159 lines
5.1 KiB
Python
159 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from sqlmodel import Session, col, select
|
|
|
|
from app.models import TerminalDeviceState, get_datetime_utc
|
|
|
|
|
|
def _to_int(value: Any) -> int | None:
|
|
if isinstance(value, bool):
|
|
return None
|
|
if isinstance(value, int):
|
|
return value
|
|
if isinstance(value, float):
|
|
return int(value)
|
|
if isinstance(value, str):
|
|
stripped = value.strip()
|
|
if stripped.isdigit():
|
|
return int(stripped)
|
|
return None
|
|
|
|
|
|
def _normalize_satellite_summary(summary: Any) -> dict[str, Any]:
|
|
if not isinstance(summary, dict):
|
|
return {
|
|
"satellites_in_view": None,
|
|
"snr_values": [],
|
|
"cn0_values": [],
|
|
"constellations": {},
|
|
"used_satellite_ids": [],
|
|
"satellites": [],
|
|
}
|
|
|
|
raw_satellites = summary.get("satellites", [])
|
|
satellites: list[dict[str, Any]] = []
|
|
if isinstance(raw_satellites, list):
|
|
for satellite in raw_satellites:
|
|
if not isinstance(satellite, dict):
|
|
continue
|
|
satellite_id = str(satellite.get("satellite_id", "")).strip()
|
|
if not satellite_id:
|
|
continue
|
|
satellites.append(
|
|
{
|
|
"satellite_id": satellite_id,
|
|
"constellation": str(satellite.get("constellation", "Unknown")),
|
|
"azimuth_deg": _to_int(satellite.get("azimuth_deg")),
|
|
"elevation_deg": _to_int(satellite.get("elevation_deg")),
|
|
"snr_dbhz": _to_int(satellite.get("snr_dbhz")),
|
|
"used_in_navigation": bool(
|
|
satellite.get("used_in_navigation", False)
|
|
),
|
|
}
|
|
)
|
|
|
|
snr_values: list[int] = []
|
|
raw_snr_values = summary.get("snr_values", [])
|
|
if isinstance(raw_snr_values, list):
|
|
for value in raw_snr_values:
|
|
converted = _to_int(value)
|
|
if converted is not None:
|
|
snr_values.append(converted)
|
|
|
|
cn0_values: list[int] = []
|
|
raw_cn0_values = summary.get("cn0_values", [])
|
|
if isinstance(raw_cn0_values, list):
|
|
for value in raw_cn0_values:
|
|
converted = _to_int(value)
|
|
if converted is not None:
|
|
cn0_values.append(converted)
|
|
if not cn0_values:
|
|
cn0_values = list(snr_values)
|
|
|
|
satellites_in_view = _to_int(summary.get("satellites_in_view"))
|
|
if satellites_in_view is None:
|
|
satellites_in_view = len(satellites)
|
|
|
|
constellations = summary.get("constellations", {})
|
|
if not isinstance(constellations, dict):
|
|
constellations = {}
|
|
|
|
used_satellite_ids = summary.get("used_satellite_ids", [])
|
|
if not isinstance(used_satellite_ids, list):
|
|
used_satellite_ids = []
|
|
|
|
return {
|
|
"satellites_in_view": satellites_in_view,
|
|
"snr_values": snr_values,
|
|
"cn0_values": cn0_values,
|
|
"constellations": constellations,
|
|
"used_satellite_ids": used_satellite_ids,
|
|
"satellites": satellites,
|
|
}
|
|
|
|
|
|
def _build_signal_health(snr_values: list[int]) -> dict[str, Any]:
|
|
strong_signal_count = sum(1 for value in snr_values if value >= 40)
|
|
average_snr = (sum(snr_values) / len(snr_values)) if snr_values else None
|
|
best_snr = max(snr_values) if snr_values else None
|
|
|
|
if strong_signal_count >= 20:
|
|
level = "good"
|
|
message = "At least 20 satellites have CN0 >= 40"
|
|
elif strong_signal_count >= 10:
|
|
level = "fair"
|
|
message = "At least 10 satellites have CN0 >= 40"
|
|
else:
|
|
level = "poor"
|
|
message = "Less than 10 satellites have CN0 >= 40"
|
|
|
|
score = min(strong_signal_count * 5, 100)
|
|
|
|
return {
|
|
"level": level,
|
|
"score": score,
|
|
"strong_signal_count": strong_signal_count,
|
|
"average_snr_dbhz": round(average_snr, 1) if average_snr is not None else None,
|
|
"best_snr_dbhz": best_snr,
|
|
"message": message,
|
|
}
|
|
|
|
|
|
def _serialize_device_state(state: TerminalDeviceState) -> dict[str, Any]:
|
|
signal_summary = _normalize_satellite_summary(state.satellite_signal_summary)
|
|
signal_health = _build_signal_health(signal_summary["snr_values"])
|
|
|
|
return {
|
|
"device_id": state.device_id,
|
|
"last_event_time": state.last_event_time,
|
|
"updated_at": state.updated_at,
|
|
"fix_status": state.fix_status,
|
|
"lat": state.lat,
|
|
"lon": state.lon,
|
|
"alt_m": state.alt_m,
|
|
"satellites_used": state.satellites_used,
|
|
"satellite_signal_summary": signal_summary,
|
|
"signal_health": signal_health,
|
|
}
|
|
|
|
|
|
def build_telemetry_snapshot(
|
|
*,
|
|
session: Session,
|
|
device_id: str | None = None,
|
|
limit: int = 50,
|
|
) -> dict[str, Any]:
|
|
statement = select(TerminalDeviceState).order_by(
|
|
col(TerminalDeviceState.updated_at).desc()
|
|
)
|
|
if device_id:
|
|
statement = statement.where(TerminalDeviceState.device_id == device_id)
|
|
statement = statement.limit(limit)
|
|
|
|
states = session.exec(statement).all()
|
|
devices = [_serialize_device_state(state) for state in states]
|
|
|
|
return {"generated_at": get_datetime_utc(), "devices": devices}
|