Initial commit: LoraTester Android + server

server/core/telemetry_body.py

@@ -0,0 +1,61 @@
+"""Normalize Android telemetry POST body into TelemetryIn."""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Optional
+
+from .models import TelemetryIn
+
+
+def _float_or_none(value: Any) -> Optional[float]:
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def merge_meta(body: dict[str, Any]) -> tuple[Optional[str], Optional[str]]:
+    """Build meta JSON string and role from meta / fields keys."""
+    meta = body.get("meta")
+    fields = body.get("fields")
+    role = body.get("role")
+
+    if isinstance(meta, dict):
+        mobj = meta
+        meta = None
+    elif isinstance(meta, str) and meta.strip():
+        try:
+            mobj = json.loads(meta)
+        except json.JSONDecodeError:
+            mobj = {}
+    else:
+        mobj = {}
+
+    if isinstance(fields, dict):
+        mobj["fields"] = fields
+
+    if mobj:
+        if role is None and mobj.get("role"):
+            role = str(mobj["role"])
+        return json.dumps(mobj, ensure_ascii=False), role
+    if isinstance(meta, str):
+        return meta, role
+    return None, role
+
+
+def telemetry_from_body(body: dict[str, Any]) -> TelemetryIn:
+    meta, role = merge_meta(body)
+    return TelemetryIn(
+        device_id=str(body["device_id"]),
+        lat=_float_or_none(body.get("lat")),
+        lon=_float_or_none(body.get("lon")),
+        rssi=_float_or_none(body.get("rssi")),
+        range_m=_float_or_none(body.get("range_m")),
+        raw_frame=body.get("raw_frame"),
+        meta=meta,
+        role=role,
+        ts=_float_or_none(body.get("ts")),
+    )