feat: multi-format capture ingestion (rtl_433, CSV, ZIP) + dedup
Adds src/api/ingest.py and refactors the live upload handler to auto-detect and route each submission by format instead of assuming .sub: - Flipper .sub -> existing parser + signature matcher (unchanged) - rtl_433 .json/.ndjson -> decoded; model is the device (conf 1.0) - Wigle-style .csv -> decoded; one observation per row (conf 0.9) - .zip batch -> recursed; any mix of the above Also adds stable dedup (SHA256 for whole files, composite key for decoded records) so re-submissions are skipped rather than stored twice, optional GPS privacy rounding via manifest privacy_gps_decimals, a session-level GPS fallback, and helpful errors for unsupported formats. Documented in docs/SUBMISSION_FORMAT.md with rtl_433/CSV sample fixtures. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
leetcrypt
@@ -0,0 +1,266 @@
|
||||
"""
|
||||
Submission ingestion helpers for GigLez.
|
||||
|
||||
Turns a variety of popular Sub-GHz capture formats into the platform's common
|
||||
capture record shape:
|
||||
|
||||
- Flipper Zero .sub (handled in main_simple via SubFileParser)
|
||||
- rtl_433 .json / .ndjson (decoded observations, one JSON object per line)
|
||||
- tabular .csv (Wigle-style pre-decoded rows)
|
||||
- batch .zip (any mix of the above + a manifest)
|
||||
|
||||
rtl_433 and CSV rows arrive already decoded, so no signature matching is needed:
|
||||
the reported model IS the device. This module maps those into capture records,
|
||||
assigns a device category, applies optional GPS privacy rounding, and computes a
|
||||
stable dedup key so the same observation is never stored twice.
|
||||
"""
|
||||
|
||||
import csv
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from typing import Iterable, Optional
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Device-category mapping for decoded (rtl_433 / CSV) submissions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# rtl_433 catalog category -> GigLez display category
|
||||
_RTL433_CAT = {
|
||||
"weather": "Weather Sensor",
|
||||
"tpms": "TPMS",
|
||||
"automotive": "Vehicle",
|
||||
"security": "Security",
|
||||
"home_automation": "Home Automation",
|
||||
"energy": "Energy Monitor",
|
||||
"lighting": "Lighting",
|
||||
"sensors": "Sensor",
|
||||
"other": "Other",
|
||||
}
|
||||
|
||||
# Keyword heuristics applied to a decoded model string (checked in order).
|
||||
_KEYWORD_CATEGORY = [
|
||||
(("tpms", "schrader", "toyota", "ford", "citroen", "renault", "hyundai",
|
||||
"jansite", "abarth", "pmv", "truck-tpms"), "TPMS"),
|
||||
(("acurite", "oregon", "lacrosse", "nexus", "ambient", "fineoffset",
|
||||
"fine offset", "ecowitt", "bresser", "tfa", "auriol", "prologue",
|
||||
"rubicson", "inkbird", "thermopro", "wt450", "ws2032", "rain", "temperature",
|
||||
"humidity", "weather"), "Weather Sensor"),
|
||||
# Garage/gate and doorbell come before Security so "doorbell" isn't swallowed
|
||||
# by the Security rule's "door" keyword (contact sensors).
|
||||
(("garage", "gate", "chamberlain", "liftmaster", "genie", "linear"),
|
||||
"Garage/Gate"),
|
||||
(("doorbell", "chime", "bell"), "Doorbell"),
|
||||
(("honeywell", "dsc", "visonic", "interlogix", "secplus", "contact",
|
||||
"door", "pir", "motion", "smoke", "alarm", "security"), "Security"),
|
||||
(("efergy", "owl", "power", "energy", "watt"), "Energy Monitor"),
|
||||
(("led", "light", "lamp"), "Lighting"),
|
||||
(("switch", "socket", "plug", "blind", "shade", "relay", "remote"),
|
||||
"Home Automation"),
|
||||
]
|
||||
|
||||
|
||||
def map_model_to_category(model: str, explicit: Optional[str] = None) -> str:
|
||||
"""Best-effort device category for a decoded model name."""
|
||||
if explicit:
|
||||
# normalise a raw rtl_433 category if that's what was passed
|
||||
return _RTL433_CAT.get(explicit.lower(), explicit)
|
||||
m = (model or "").lower()
|
||||
for keywords, category in _KEYWORD_CATEGORY:
|
||||
if any(k in m for k in keywords):
|
||||
return category
|
||||
return "Sensor"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Format detection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def detect_format(filename: str, content: bytes) -> str:
|
||||
"""Return one of: sub, rtl433_json, csv, zip, unknown."""
|
||||
name = (filename or "").lower()
|
||||
if name.endswith(".zip") or content[:2] == b"PK":
|
||||
return "zip"
|
||||
if name.endswith((".ndjson", ".jsonl")):
|
||||
return "rtl433_json"
|
||||
if name.endswith(".csv"):
|
||||
return "csv"
|
||||
if name.endswith(".sub"):
|
||||
return "sub"
|
||||
if name.endswith(".json"):
|
||||
return "rtl433_json"
|
||||
|
||||
# Fall back to sniffing the first non-empty line.
|
||||
head = content[:4096].lstrip()
|
||||
if head.startswith(b"Filetype:"):
|
||||
return "sub"
|
||||
if head[:1] in (b"{", b"["):
|
||||
return "rtl433_json"
|
||||
if b"," in head.split(b"\n", 1)[0]:
|
||||
return "csv"
|
||||
return "unknown"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GPS helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def round_gps(lat, lon, decimals: Optional[int]):
|
||||
"""Round coordinates for privacy. `decimals=None` leaves them untouched."""
|
||||
if lat is None or lon is None or decimals is None:
|
||||
return lat, lon
|
||||
try:
|
||||
return round(float(lat), decimals), round(float(lon), decimals)
|
||||
except (TypeError, ValueError):
|
||||
return lat, lon
|
||||
|
||||
|
||||
def _record_gps(record: dict, fallback_entry: dict, session_gps: dict):
|
||||
"""Resolve GPS for a decoded record: record fields > manifest entry > session."""
|
||||
for lat_k, lon_k in (("latitude", "longitude"), ("lat", "lon"), ("lat", "lng")):
|
||||
if record.get(lat_k) is not None and record.get(lon_k) is not None:
|
||||
return record[lat_k], record[lon_k], "record"
|
||||
if fallback_entry.get("latitude") is not None:
|
||||
return fallback_entry.get("latitude"), fallback_entry.get("longitude"), "manifest"
|
||||
if session_gps.get("latitude") is not None:
|
||||
return session_gps.get("latitude"), session_gps.get("longitude"), "session"
|
||||
return None, None, "none"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Parsers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def parse_rtl433(content: bytes) -> list:
|
||||
"""Parse rtl_433 output: newline-delimited JSON objects OR a JSON array."""
|
||||
text = content.decode("utf-8", errors="replace").strip()
|
||||
if not text:
|
||||
return []
|
||||
# JSON array form
|
||||
if text[0] == "[":
|
||||
try:
|
||||
data = json.loads(text)
|
||||
return [r for r in data if isinstance(r, dict)]
|
||||
except json.JSONDecodeError:
|
||||
return []
|
||||
# NDJSON form (one object per line)
|
||||
records = []
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
obj = json.loads(line)
|
||||
if isinstance(obj, dict):
|
||||
records.append(obj)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
return records
|
||||
|
||||
|
||||
def parse_csv(content: bytes) -> list:
|
||||
"""Parse a Wigle-style CSV into a list of row dicts."""
|
||||
text = content.decode("utf-8", errors="replace")
|
||||
reader = csv.DictReader(io.StringIO(text))
|
||||
return [dict(row) for row in reader]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Decoded record -> capture record
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _to_float(v):
|
||||
try:
|
||||
return float(v)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def build_decoded_capture(record: dict, *, source_format: str, filename: str,
|
||||
manifest_entry: dict, session_gps: dict,
|
||||
data_source: str, session_id: str,
|
||||
gps_decimals: Optional[int]) -> dict:
|
||||
"""Map one decoded rtl_433/CSV record into a GigLez capture record."""
|
||||
model = (record.get("model") or record.get("device_name")
|
||||
or record.get("protocol") or "Unknown")
|
||||
manufacturer = record.get("manufacturer") or record.get("brand") or ""
|
||||
|
||||
# frequency: rtl_433 emits MHz in `freq`/`freq1` only with -M level; CSV may
|
||||
# provide Hz or MHz. Normalise to Hz.
|
||||
freq_hz = 0
|
||||
for key in ("frequency", "freq", "freq1"):
|
||||
val = _to_float(record.get(key))
|
||||
if val:
|
||||
freq_hz = int(val if val > 1e6 else val * 1e6)
|
||||
break
|
||||
|
||||
lat, lon, gps_source = _record_gps(record, manifest_entry, session_gps)
|
||||
lat, lon = round_gps(lat, lon, gps_decimals)
|
||||
|
||||
method = "rtl_433_decoded" if source_format == "rtl433_json" else "csv_import"
|
||||
conf = _to_float(record.get("confidence"))
|
||||
if conf is None:
|
||||
conf = 1.0 if source_format == "rtl433_json" else 0.9
|
||||
|
||||
return {
|
||||
"filename": filename,
|
||||
"frequency": freq_hz,
|
||||
"protocol": model,
|
||||
"preset": record.get("mod") or record.get("modulation") or source_format,
|
||||
"latitude": lat,
|
||||
"longitude": lon,
|
||||
"gps_source": gps_source,
|
||||
"timestamp": record.get("time") or record.get("timestamp")
|
||||
or manifest_entry.get("timestamp", ""),
|
||||
"data_source": data_source,
|
||||
"session_id": session_id,
|
||||
"device_name": model,
|
||||
"device_category": map_model_to_category(model, record.get("category")),
|
||||
"match_confidence": conf,
|
||||
"match_method": method,
|
||||
"device_description": (f"{manufacturer} {model}".strip()),
|
||||
"matched_devices": [],
|
||||
# keep a couple of useful decoded fields for training/inspection
|
||||
"decoded": {k: record.get(k) for k in ("id", "channel", "battery_ok",
|
||||
"temperature_C", "humidity", "pressure_kPa") if k in record},
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Deduplication
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def dedup_key_for_file(content: bytes) -> str:
|
||||
"""SHA256 of raw file bytes (for whole-file formats like .sub)."""
|
||||
return "sha256:" + hashlib.sha256(content).hexdigest()
|
||||
|
||||
|
||||
def dedup_key_for_capture(cap: dict) -> str:
|
||||
"""Stable identity for a decoded observation (model + id + place + time)."""
|
||||
ident = "|".join(str(cap.get(k, "")) for k in (
|
||||
"protocol", "device_name", "frequency"))
|
||||
decoded = cap.get("decoded") or {}
|
||||
ident += "|" + str(decoded.get("id", "")) + "|" + str(decoded.get("channel", ""))
|
||||
lat, lon = cap.get("latitude"), cap.get("longitude")
|
||||
if lat is not None and lon is not None:
|
||||
ident += f"|{round(float(lat), 4)}|{round(float(lon), 4)}"
|
||||
ident += "|" + str(cap.get("timestamp", ""))[:19] # to the second
|
||||
return "rec:" + hashlib.sha256(ident.encode()).hexdigest()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ZIP batch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def iter_zip_members(content: bytes) -> Iterable:
|
||||
"""Yield (member_filename, member_bytes) for non-manifest files in a zip."""
|
||||
with zipfile.ZipFile(io.BytesIO(content)) as zf:
|
||||
for info in zf.infolist():
|
||||
if info.is_dir():
|
||||
continue
|
||||
name = Path(info.filename).name
|
||||
if not name or name.startswith(".") or name.lower() == "manifest.json":
|
||||
continue
|
||||
yield name, zf.read(info)
|
||||
+183
-108
@@ -31,6 +31,16 @@ from src.matcher.strategies import (
|
||||
RTL433DecoderStrategy,
|
||||
PatternBasedStrategy
|
||||
)
|
||||
from src.api.ingest import (
|
||||
detect_format,
|
||||
parse_rtl433,
|
||||
parse_csv,
|
||||
build_decoded_capture,
|
||||
dedup_key_for_file,
|
||||
dedup_key_for_capture,
|
||||
round_gps,
|
||||
iter_zip_members,
|
||||
)
|
||||
|
||||
# =============================================================================
|
||||
# APPLICATION INSTANCE
|
||||
@@ -56,6 +66,9 @@ STORAGE_FILE.parent.mkdir(exist_ok=True)
|
||||
# In-memory storage for uploaded captures (simplified mode)
|
||||
captures_storage = []
|
||||
upload_counter = 0
|
||||
# Dedup keys for everything already stored (rebuilt on load). A submission whose
|
||||
# key is already present is skipped rather than stored twice.
|
||||
seen_dedup_keys = set()
|
||||
|
||||
# Initialize unified matcher engine with all strategies
|
||||
_matcher_engine = None
|
||||
@@ -93,7 +106,7 @@ def get_matcher_engine():
|
||||
|
||||
def load_captures():
|
||||
"""Load captures from JSON file"""
|
||||
global captures_storage, upload_counter
|
||||
global captures_storage, upload_counter, seen_dedup_keys
|
||||
|
||||
if STORAGE_FILE.exists():
|
||||
try:
|
||||
@@ -109,6 +122,12 @@ def load_captures():
|
||||
else:
|
||||
print(f"ℹ️ No existing captures file at {STORAGE_FILE}")
|
||||
|
||||
# Rebuild the dedup index so re-submitting the same file/observation is a no-op.
|
||||
seen_dedup_keys = set()
|
||||
for cap in captures_storage:
|
||||
key = cap.get("dedup_key") or dedup_key_for_capture(cap)
|
||||
seen_dedup_keys.add(key)
|
||||
|
||||
|
||||
def save_captures():
|
||||
"""Save captures to JSON file"""
|
||||
@@ -465,140 +484,194 @@ async def rtl433_protocols():
|
||||
}
|
||||
|
||||
|
||||
def _store_capture(cap: dict, dedup_key: str):
|
||||
"""Assign an id and persist a capture, unless its dedup key already exists.
|
||||
|
||||
Returns the stored capture dict, or None if it was a duplicate.
|
||||
"""
|
||||
global upload_counter
|
||||
if dedup_key in seen_dedup_keys:
|
||||
return None
|
||||
upload_counter += 1
|
||||
cap["id"] = upload_counter
|
||||
cap["dedup_key"] = dedup_key
|
||||
seen_dedup_keys.add(dedup_key)
|
||||
captures_storage.append(cap)
|
||||
return cap
|
||||
|
||||
|
||||
def _build_sub_capture(content: bytes, filename: str, entry: dict,
|
||||
manifest_data: dict, gps_decimals):
|
||||
"""Parse a Flipper .sub file, run the matcher, and build a capture record."""
|
||||
import tempfile
|
||||
import os
|
||||
|
||||
sub_parser = SubFileParser()
|
||||
gps_extractor = GPSFilenameExtractor()
|
||||
|
||||
temp_path = None
|
||||
try:
|
||||
with tempfile.NamedTemporaryFile(mode='wb', delete=False, suffix='.sub') as temp_file:
|
||||
temp_path = temp_file.name
|
||||
temp_file.write(content)
|
||||
metadata = sub_parser.parse(temp_path)
|
||||
finally:
|
||||
if temp_path and os.path.exists(temp_path):
|
||||
os.unlink(temp_path)
|
||||
|
||||
gps_coords = gps_extractor.extract(filename)
|
||||
|
||||
frequency = metadata.frequency if hasattr(metadata, 'frequency') else 0
|
||||
protocol = metadata.protocol if hasattr(metadata, 'protocol') else "RAW"
|
||||
preset = metadata.preset if hasattr(metadata, 'preset') else "Unknown"
|
||||
|
||||
# Unified device matching (Exact, Frequency, BitPattern, Timing, RTL_433, Pattern Decoder)
|
||||
matcher_engine = get_matcher_engine()
|
||||
match_results = matcher_engine.match(metadata, max_results=10)
|
||||
best_match = match_results[0] if match_results else None
|
||||
best_category = (
|
||||
best_match.match_details.get("predicted_category") if best_match else None
|
||||
)
|
||||
|
||||
lat = gps_coords.latitude if gps_coords else entry.get("latitude")
|
||||
lon = gps_coords.longitude if gps_coords else entry.get("longitude")
|
||||
lat, lon = round_gps(lat, lon, gps_decimals)
|
||||
|
||||
return {
|
||||
"filename": filename,
|
||||
"frequency": frequency,
|
||||
"protocol": protocol,
|
||||
"preset": preset,
|
||||
"latitude": lat,
|
||||
"longitude": lon,
|
||||
"gps_source": gps_coords.source if gps_coords else "manual",
|
||||
"timestamp": entry.get("timestamp", ""),
|
||||
"data_source": manifest_data.get("data_source", "production"),
|
||||
"session_id": manifest_data.get("session_uuid", ""),
|
||||
"device_name": best_match.device_name if best_match else None,
|
||||
"device_category": best_category,
|
||||
"match_confidence": best_match.confidence if best_match else None,
|
||||
"match_method": best_match.match_method if best_match else None,
|
||||
"device_description": f"{best_match.manufacturer} {best_match.device_name}" if best_match else None,
|
||||
"matched_devices": [
|
||||
{
|
||||
"device_name": m.device_name,
|
||||
"manufacturer": m.manufacturer,
|
||||
"category": m.match_details.get("predicted_category"),
|
||||
"confidence": m.confidence,
|
||||
"method": m.match_method,
|
||||
"details": m.match_details,
|
||||
}
|
||||
for m in match_results[:10]
|
||||
] if match_results else [],
|
||||
}
|
||||
|
||||
|
||||
def _ingest_one(filename: str, content: bytes, manifest_map: dict,
|
||||
manifest_data: dict, session_gps: dict, gps_decimals,
|
||||
successful: list, skipped: list, failed: list):
|
||||
"""Detect a submission's format and route it to the right parser.
|
||||
|
||||
Appends resulting capture records to successful / skipped / failed. ZIP
|
||||
batches recurse into their members.
|
||||
"""
|
||||
fmt = detect_format(filename, content)
|
||||
entry = manifest_map.get(filename)
|
||||
if entry is None:
|
||||
entry = (manifest_data.get("captures") or [{}])[0]
|
||||
|
||||
if fmt == "zip":
|
||||
for member_name, member_bytes in iter_zip_members(content):
|
||||
_ingest_one(member_name, member_bytes, manifest_map, manifest_data,
|
||||
session_gps, gps_decimals, successful, skipped, failed)
|
||||
return
|
||||
|
||||
if fmt == "sub":
|
||||
cap = _build_sub_capture(content, filename, entry, manifest_data, gps_decimals)
|
||||
stored = _store_capture(cap, dedup_key_for_file(content))
|
||||
(successful if stored else skipped).append(cap)
|
||||
return
|
||||
|
||||
if fmt in ("rtl433_json", "csv"):
|
||||
records = parse_rtl433(content) if fmt == "rtl433_json" else parse_csv(content)
|
||||
if not records:
|
||||
failed.append({
|
||||
"filename": filename,
|
||||
"error": f"No usable records parsed from {fmt} file",
|
||||
})
|
||||
return
|
||||
for rec in records:
|
||||
cap = build_decoded_capture(
|
||||
rec,
|
||||
source_format=fmt,
|
||||
filename=filename,
|
||||
manifest_entry=entry,
|
||||
session_gps=session_gps,
|
||||
data_source=manifest_data.get("data_source", "production"),
|
||||
session_id=manifest_data.get("session_uuid", ""),
|
||||
gps_decimals=gps_decimals,
|
||||
)
|
||||
stored = _store_capture(cap, dedup_key_for_capture(cap))
|
||||
(successful if stored else skipped).append(cap)
|
||||
return
|
||||
|
||||
failed.append({
|
||||
"filename": filename,
|
||||
"error": (f"Unsupported format for '{filename}'. Supported: Flipper .sub, "
|
||||
f"rtl_433 .json/.ndjson, Wigle-style .csv, or a .zip batch."),
|
||||
})
|
||||
|
||||
|
||||
@app.post("/api/v1/captures/upload")
|
||||
async def upload_captures(
|
||||
files: List[UploadFile] = File(...),
|
||||
manifest: str = Form(...)
|
||||
):
|
||||
"""
|
||||
Mock upload endpoint - parses files and returns success
|
||||
Does not save to database (simplified version)
|
||||
"""
|
||||
import tempfile
|
||||
import os
|
||||
"""Ingest one or more captures.
|
||||
|
||||
Accepts Flipper `.sub`, rtl_433 `.json`/`.ndjson`, Wigle-style `.csv`, and
|
||||
`.zip` batches (any mix of the above). Format is auto-detected per file.
|
||||
Duplicate submissions are skipped via a stable dedup key. Optional privacy
|
||||
GPS rounding is controlled by the manifest's `privacy_gps_decimals`.
|
||||
"""
|
||||
try:
|
||||
# Parse manifest
|
||||
manifest_data = json.loads(manifest)
|
||||
|
||||
# Initialize parsers
|
||||
sub_parser = SubFileParser()
|
||||
gps_extractor = GPSFilenameExtractor()
|
||||
|
||||
# Map manifest entries by filename so each file gets its own GPS/timestamp
|
||||
# (previously every file was assigned captures[0], breaking multi-file uploads)
|
||||
# Per-file manifest entries carry each file's GPS/timestamp.
|
||||
manifest_map = {
|
||||
c.get("filename"): c
|
||||
for c in manifest_data.get("captures", [])
|
||||
}
|
||||
# Session-level GPS fallback (used by decoded records without their own coords).
|
||||
session_gps = {
|
||||
"latitude": manifest_data.get("latitude"),
|
||||
"longitude": manifest_data.get("longitude"),
|
||||
}
|
||||
gps_decimals = manifest_data.get("privacy_gps_decimals")
|
||||
|
||||
successful = []
|
||||
failed = []
|
||||
successful, skipped, failed = [], [], []
|
||||
|
||||
# Process each file
|
||||
for file in files:
|
||||
temp_path = None
|
||||
try:
|
||||
# Read file content
|
||||
content = await file.read()
|
||||
|
||||
# Save to temporary file for parsing
|
||||
with tempfile.NamedTemporaryFile(mode='wb', delete=False, suffix='.sub') as temp_file:
|
||||
temp_path = temp_file.name
|
||||
temp_file.write(content)
|
||||
|
||||
# Parse .sub file
|
||||
metadata = sub_parser.parse(temp_path)
|
||||
|
||||
# Try to extract GPS from filename
|
||||
gps_coords = gps_extractor.extract(file.filename)
|
||||
|
||||
# Extract signal parameters
|
||||
frequency = metadata.frequency if hasattr(metadata, 'frequency') else 0
|
||||
protocol = metadata.protocol if hasattr(metadata, 'protocol') else "RAW"
|
||||
preset = metadata.preset if hasattr(metadata, 'preset') else "Unknown"
|
||||
|
||||
# Perform unified device matching with all strategies
|
||||
# This includes: Exact, Frequency, BitPattern, Timing, RTL_433, and Pattern Decoder!
|
||||
matcher_engine = get_matcher_engine()
|
||||
match_results = matcher_engine.match(metadata, max_results=10)
|
||||
|
||||
# Get best match from unified results
|
||||
best_match = match_results[0] if match_results else None
|
||||
|
||||
# Per-file manifest entry (falls back to first entry, then empty)
|
||||
entry = manifest_map.get(file.filename)
|
||||
if entry is None:
|
||||
entry = manifest_data.get("captures", [{}])[0]
|
||||
|
||||
# The real device category comes from the category router (e.g.
|
||||
# "Weather Sensor", "Remote Control"), carried in match_details.
|
||||
best_category = (
|
||||
best_match.match_details.get("predicted_category")
|
||||
if best_match else None
|
||||
)
|
||||
|
||||
# Use GPS from manifest or filename
|
||||
global upload_counter
|
||||
upload_counter += 1
|
||||
|
||||
capture_info = {
|
||||
"id": upload_counter,
|
||||
"filename": file.filename,
|
||||
"frequency": frequency,
|
||||
"protocol": protocol,
|
||||
"preset": preset,
|
||||
"latitude": gps_coords.latitude if gps_coords else entry.get("latitude"),
|
||||
"longitude": gps_coords.longitude if gps_coords else entry.get("longitude"),
|
||||
"gps_source": gps_coords.source if gps_coords else "manual",
|
||||
"timestamp": entry.get("timestamp", ""),
|
||||
"data_source": manifest_data.get("data_source", "production"), # production, test, or mock
|
||||
"session_id": manifest_data.get("session_uuid", ""),
|
||||
# Device identification fields (from unified matcher)
|
||||
"device_name": best_match.device_name if best_match else None,
|
||||
"device_category": best_category,
|
||||
"match_confidence": best_match.confidence if best_match else None,
|
||||
"match_method": best_match.match_method if best_match else None,
|
||||
"device_description": f"{best_match.manufacturer} {best_match.device_name}" if best_match else None,
|
||||
# All matched devices from unified matcher (includes RTL_433, Pattern Decoder, etc.)
|
||||
"matched_devices": [
|
||||
{
|
||||
"device_name": m.device_name,
|
||||
"manufacturer": m.manufacturer,
|
||||
"category": m.match_details.get("predicted_category"),
|
||||
"confidence": m.confidence,
|
||||
"method": m.match_method,
|
||||
"details": m.match_details
|
||||
}
|
||||
for m in match_results[:10] # Top 10 matches
|
||||
] if match_results else []
|
||||
}
|
||||
|
||||
# Store in memory
|
||||
captures_storage.append(capture_info)
|
||||
successful.append(capture_info)
|
||||
|
||||
_ingest_one(file.filename, content, manifest_map, manifest_data,
|
||||
session_gps, gps_decimals, successful, skipped, failed)
|
||||
except Exception as e:
|
||||
failed.append({
|
||||
"filename": file.filename,
|
||||
"error": str(e)
|
||||
})
|
||||
finally:
|
||||
# Clean up temp file
|
||||
if temp_path and os.path.exists(temp_path):
|
||||
os.unlink(temp_path)
|
||||
failed.append({"filename": file.filename, "error": str(e)})
|
||||
|
||||
# Save to persistent storage
|
||||
save_captures()
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"message": f"Processed {len(successful)} files successfully",
|
||||
"message": (f"Processed {len(successful)} records "
|
||||
f"({len(skipped)} duplicates skipped, {len(failed)} failed)"),
|
||||
"successful": successful,
|
||||
"skipped": skipped,
|
||||
"failed": failed,
|
||||
"total_uploaded": len(files),
|
||||
"total_successful": len(successful),
|
||||
"total_failed": len(failed)
|
||||
"total_skipped": len(skipped),
|
||||
"total_failed": len(failed),
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
@@ -606,10 +679,12 @@ async def upload_captures(
|
||||
"success": False,
|
||||
"message": f"Upload failed: {str(e)}",
|
||||
"successful": [],
|
||||
"skipped": [],
|
||||
"failed": [],
|
||||
"total_uploaded": 0,
|
||||
"total_successful": 0,
|
||||
"total_failed": 0
|
||||
"total_skipped": 0,
|
||||
"total_failed": 0,
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user