generated from Grigo/AndroidTemplate
67 lines
1.8 KiB
Python
67 lines
1.8 KiB
Python
"""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)
|
|
label = body.get("device_label") or body.get("label")
|
|
device_label = str(label).strip() if label else None
|
|
if device_label == "":
|
|
device_label = None
|
|
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")),
|
|
device_label=device_label,
|
|
)
|