diff --git a/backend/monitoring/services.py b/backend/monitoring/services.py index 59683dd..79f992e 100644 --- a/backend/monitoring/services.py +++ b/backend/monitoring/services.py @@ -1,206 +1,210 @@ -from __future__ import annotations - -from collections import deque -import json -import math -import os -import sys -import threading -import time -from datetime import UTC, datetime -from pathlib import Path -from typing import Any, Iterator - - -PROJECT_ROOT = Path(__file__).resolve().parents[2] -WORKSPACE_ROOT = PROJECT_ROOT.parent -JPEG_FRAME_DIR = WORKSPACE_ROOT / "RobotDataShow" / "jpeg-frames" -# GPS 数据 JSON 文件路径 -GEOSTREAM_JSON_PATH = WORKSPACE_ROOT / "GeoStream" / "gps_latest.json" -GEOSTREAM_STALE_SECONDS = 15 +from __future__ import annotations + +from collections import deque +import json +import math +import os +import sys +import threading +import time +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Iterator + + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +WORKSPACE_ROOT = PROJECT_ROOT.parent +JPEG_FRAME_DIR = WORKSPACE_ROOT / "RobotDataShow" / "jpeg-frames" +# GPS 数据 JSON 文件路径 +GEOSTREAM_JSON_PATH = WORKSPACE_ROOT / "GeoStream" / "gps_latest.json" +GEOSTREAM_STALE_SECONDS = 15 OMNISOCKET_CONFIG_PATH = PROJECT_ROOT / "config" / "omnisocket_demo.yaml" VIDEO_SOURCE_MODE = os.getenv("VIDEO_SOURCE_MODE", "auto").strip().lower() OMNISOCKET_FRAME_FRESH_SECONDS = 2.0 VIDEO_TIMESTAMP_SAMPLE_SIZE = 10 +VIDEO_TIMESTAMP_TRAILER_BYTES = 8 +VIDEO_TIMESTAMP_ENDIANNESS = "little" +VIDEO_TIMESTAMP_UNIT = "ms" +VIDEO_TIMESTAMP_MULTIPLIER_NS = 1_000_000 VIDEO_TIMESTAMP_MAX_SKEW_NS = 7 * 24 * 60 * 60 * 1_000_000_000 - - -def utc_iso_now() -> str: - return datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z") - - -class OmniSocketVideoReceiver: - def __init__(self) -> None: - self._lock = threading.Lock() - self._thread: threading.Thread | None = None - self._started = False - self._session = None - self._session_cls = None - self._binary_msg_type = None - self._video_defaults: dict[str, Any] = {} - self._latest_frame: bytes | None = None - self._latest_received_at = 0.0 - self._latest_sequence: int | None = None - self._latest_latency_ms: float | None = None - self._latest_timestamp_unit: str | None = None - self._latest_timestamp_endianness: str | None = None - self._latency_samples_ms: deque[float] = deque(maxlen=VIDEO_TIMESTAMP_SAMPLE_SIZE) - self._frames_received = 0 - self._last_error = "" - self._load_backend() - - def _load_backend(self) -> None: - # 服务启动时先尝试导入一次 Python/C 扩展。 - try: - self._import_backend() - except Exception as error: # pragma: no cover - 可选的运行时依赖 - self._last_error = f"omnisocket import failed: {error}" - - def _import_backend(self) -> None: - # 优先使用已经安装到当前 Python 环境里的 omnisocket。 - # 如果导入失败，再尝试样例目录下的本地 python 路径。 - try: - from omnisocket import MSG_TYPE_BINARY, Session, VIDEO_DEFAULTS # type: ignore - except ImportError: - python_dir = WORKSPACE_ROOT / "OmniSocketGo" / "python" - if python_dir.exists(): - sys.path.insert(0, str(python_dir)) - from omnisocket import MSG_TYPE_BINARY, Session, VIDEO_DEFAULTS # type: ignore - - self._binary_msg_type = MSG_TYPE_BINARY - self._session_cls = Session - self._video_defaults = dict(VIDEO_DEFAULTS) - - def ensure_started(self) -> None: - # 当第一次请求帧或状态时，再懒启动后台接收线程。 - if self._session_cls is None or self._binary_msg_type is None: - return - - with self._lock: - if self._started: - return - self._started = True - self._thread = threading.Thread( - target=self._run, - name="omnisocket-video-receiver", - daemon=True, - ) - self._thread.start() - - def _load_config(self) -> dict[str, Any]: - # 这里保持和 SampleCData/omnisocket_video_receiver.py 一样的配置结构： - # transport + video_receiver。 - # 即使配置文件不存在，也允许回退到默认值继续运行。 - config: dict[str, Any] = {} - if OMNISOCKET_CONFIG_PATH.exists(): - try: - try: - import yaml # type: ignore - - with OMNISOCKET_CONFIG_PATH.open("r", encoding="utf-8") as file: - config = yaml.safe_load(file) or {} - except ImportError: - # 当前配置文件结构非常简单，缺少 PyYAML 时用简化解析器兜底。 - config = self._load_simple_yaml_config(OMNISOCKET_CONFIG_PATH) - except Exception as error: # pragma: no cover - 可选依赖 - self._last_error = f"config load failed: {error}" - - transport_cfg = dict(config.get("transport", {})) - video_cfg = dict(config.get("video_receiver", {})) - - transport_cfg["server_addr"] = os.getenv( - "OMNISOCKET_SERVER_ADDR", - str(transport_cfg.get("server_addr", "127.0.0.1:10909")), - ) - transport_cfg["relay_via"] = os.getenv( - "OMNISOCKET_RELAY_VIA", - str(transport_cfg.get("relay_via", "")), - ) - transport_cfg["bind_ip"] = os.getenv( - "OMNISOCKET_BIND_IP", - str(transport_cfg.get("bind_ip", "")), - ) - transport_cfg["bind_device"] = os.getenv( - "OMNISOCKET_BIND_DEVICE", - str(transport_cfg.get("bind_device", "")), - ) - video_cfg["peer_id"] = os.getenv( - "OMNISOCKET_VIDEO_PEER_ID", - str(video_cfg.get("peer_id", "peer-a-video")), - ) - video_cfg["buffer_bytes"] = int( - os.getenv( - "OMNISOCKET_BUFFER_BYTES", - str(video_cfg.get("buffer_bytes", 1024 * 1024)), - ) - ) - - return { - "transport": transport_cfg, - "video_receiver": video_cfg, - } - - def _load_simple_yaml_config(self, path: Path) -> dict[str, Any]: - parsed: dict[str, Any] = {} - current_section: str | None = None - - with path.open("r", encoding="utf-8") as file: - for raw_line in file: - line = raw_line.split("#", 1)[0].rstrip() - if not line.strip(): - continue - - if not line.startswith(" "): - if not line.endswith(":"): - raise ValueError(f"invalid top-level yaml line: {raw_line.strip()}") - current_section = line[:-1].strip() - parsed[current_section] = {} - continue - - if current_section is None: - raise ValueError(f"yaml key outside section: {raw_line.strip()}") - - stripped = line.strip() - if ":" not in stripped: - raise ValueError(f"invalid yaml key line: {raw_line.strip()}") - - key, value = stripped.split(":", 1) - parsed[current_section][key.strip()] = self._parse_simple_yaml_scalar(value.strip()) - - return parsed - - def _parse_simple_yaml_scalar(self, value: str) -> Any: - if value in {'""', "''"}: - return "" - if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}: - return value[1:-1] - if value.lower() == "true": - return True - if value.lower() == "false": - return False - if value and value.lstrip("-").isdigit(): - return int(value) - return value - - def _connect_session(self): - # 这里和样例接收器一致：创建 Session()，然后使用 transport/video 配置建立连接。 - assert self._session_cls is not None - - config = self._load_config() - transport_cfg = config.get("transport", {}) - video_cfg = config.get("video_receiver", {}) - - session = self._session_cls() - session.connect( - server_addr=str(transport_cfg.get("server_addr", "127.0.0.1:10909")), - peer_id=str(video_cfg.get("peer_id", "peer-a-video")), - relay_via=str(transport_cfg.get("relay_via", "")), - bind_ip=str(transport_cfg.get("bind_ip", "")), - bind_device=str(transport_cfg.get("bind_device", "")), - **self._video_defaults, - ) - return session, int(video_cfg.get("buffer_bytes", 1024 * 1024)) - + + +def utc_iso_now() -> str: + return datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z") + + +class OmniSocketVideoReceiver: + def __init__(self) -> None: + self._lock = threading.Lock() + self._thread: threading.Thread | None = None + self._started = False + self._session = None + self._session_cls = None + self._binary_msg_type = None + self._video_defaults: dict[str, Any] = {} + self._latest_frame: bytes | None = None + self._latest_received_at = 0.0 + self._latest_sequence: int | None = None + self._latest_latency_ms: float | None = None + self._latest_timestamp_unit: str | None = None + self._latest_timestamp_endianness: str | None = None + self._latency_samples_ms: deque[float] = deque(maxlen=VIDEO_TIMESTAMP_SAMPLE_SIZE) + self._frames_received = 0 + self._last_error = "" + self._load_backend() + + def _load_backend(self) -> None: + # 服务启动时先尝试导入一次 Python/C 扩展。 + try: + self._import_backend() + except Exception as error: # pragma: no cover - 可选的运行时依赖 + self._last_error = f"omnisocket import failed: {error}" + + def _import_backend(self) -> None: + # 优先使用已经安装到当前 Python 环境里的 omnisocket。 + # 如果导入失败，再尝试样例目录下的本地 python 路径。 + try: + from omnisocket import MSG_TYPE_BINARY, Session, VIDEO_DEFAULTS # type: ignore + except ImportError: + python_dir = WORKSPACE_ROOT / "OmniSocketGo" / "python" + if python_dir.exists(): + sys.path.insert(0, str(python_dir)) + from omnisocket import MSG_TYPE_BINARY, Session, VIDEO_DEFAULTS # type: ignore + + self._binary_msg_type = MSG_TYPE_BINARY + self._session_cls = Session + self._video_defaults = dict(VIDEO_DEFAULTS) + + def ensure_started(self) -> None: + # 当第一次请求帧或状态时，再懒启动后台接收线程。 + if self._session_cls is None or self._binary_msg_type is None: + return + + with self._lock: + if self._started: + return + self._started = True + self._thread = threading.Thread( + target=self._run, + name="omnisocket-video-receiver", + daemon=True, + ) + self._thread.start() + + def _load_config(self) -> dict[str, Any]: + # 这里保持和 SampleCData/omnisocket_video_receiver.py 一样的配置结构： + # transport + video_receiver。 + # 即使配置文件不存在，也允许回退到默认值继续运行。 + config: dict[str, Any] = {} + if OMNISOCKET_CONFIG_PATH.exists(): + try: + try: + import yaml # type: ignore + + with OMNISOCKET_CONFIG_PATH.open("r", encoding="utf-8") as file: + config = yaml.safe_load(file) or {} + except ImportError: + # 当前配置文件结构非常简单，缺少 PyYAML 时用简化解析器兜底。 + config = self._load_simple_yaml_config(OMNISOCKET_CONFIG_PATH) + except Exception as error: # pragma: no cover - 可选依赖 + self._last_error = f"config load failed: {error}" + + transport_cfg = dict(config.get("transport", {})) + video_cfg = dict(config.get("video_receiver", {})) + + transport_cfg["server_addr"] = os.getenv( + "OMNISOCKET_SERVER_ADDR", + str(transport_cfg.get("server_addr", "127.0.0.1:10909")), + ) + transport_cfg["relay_via"] = os.getenv( + "OMNISOCKET_RELAY_VIA", + str(transport_cfg.get("relay_via", "")), + ) + transport_cfg["bind_ip"] = os.getenv( + "OMNISOCKET_BIND_IP", + str(transport_cfg.get("bind_ip", "")), + ) + transport_cfg["bind_device"] = os.getenv( + "OMNISOCKET_BIND_DEVICE", + str(transport_cfg.get("bind_device", "")), + ) + video_cfg["peer_id"] = os.getenv( + "OMNISOCKET_VIDEO_PEER_ID", + str(video_cfg.get("peer_id", "peer-a-video")), + ) + video_cfg["buffer_bytes"] = int( + os.getenv( + "OMNISOCKET_BUFFER_BYTES", + str(video_cfg.get("buffer_bytes", 1024 * 1024)), + ) + ) + + return { + "transport": transport_cfg, + "video_receiver": video_cfg, + } + + def _load_simple_yaml_config(self, path: Path) -> dict[str, Any]: + parsed: dict[str, Any] = {} + current_section: str | None = None + + with path.open("r", encoding="utf-8") as file: + for raw_line in file: + line = raw_line.split("#", 1)[0].rstrip() + if not line.strip(): + continue + + if not line.startswith(" "): + if not line.endswith(":"): + raise ValueError(f"invalid top-level yaml line: {raw_line.strip()}") + current_section = line[:-1].strip() + parsed[current_section] = {} + continue + + if current_section is None: + raise ValueError(f"yaml key outside section: {raw_line.strip()}") + + stripped = line.strip() + if ":" not in stripped: + raise ValueError(f"invalid yaml key line: {raw_line.strip()}") + + key, value = stripped.split(":", 1) + parsed[current_section][key.strip()] = self._parse_simple_yaml_scalar(value.strip()) + + return parsed + + def _parse_simple_yaml_scalar(self, value: str) -> Any: + if value in {'""', "''"}: + return "" + if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}: + return value[1:-1] + if value.lower() == "true": + return True + if value.lower() == "false": + return False + if value and value.lstrip("-").isdigit(): + return int(value) + return value + + def _connect_session(self): + # 这里和样例接收器一致：创建 Session()，然后使用 transport/video 配置建立连接。 + assert self._session_cls is not None + + config = self._load_config() + transport_cfg = config.get("transport", {}) + video_cfg = config.get("video_receiver", {}) + + session = self._session_cls() + session.connect( + server_addr=str(transport_cfg.get("server_addr", "127.0.0.1:10909")), + peer_id=str(video_cfg.get("peer_id", "peer-a-video")), + relay_via=str(transport_cfg.get("relay_via", "")), + bind_ip=str(transport_cfg.get("bind_ip", "")), + bind_device=str(transport_cfg.get("bind_device", "")), + **self._video_defaults, + ) + return session, int(video_cfg.get("buffer_bytes", 1024 * 1024)) + def _extract_jpeg_payload(self, frame: bytes) -> bytes | None: # 同时兼容两种帧格式： # 1. 纯 JPEG 二进制 @@ -211,16 +215,35 @@ class OmniSocketVideoReceiver: return frame[8:] return None - def _extract_jpeg_frame(self, frame: bytes) -> bytes | None: + def _split_jpeg_frame_and_trailer(self, frame: bytes) -> tuple[bytes, bytes] | None: jpeg_payload = self._extract_jpeg_payload(frame) if jpeg_payload is None: return None + # 当前发送端协议是 JPEG 末尾固定追加 8 字节 little-endian 毫秒时间戳。 + if jpeg_payload.endswith(b"\xff\xd9"): + return jpeg_payload, b"" + + if ( + len(jpeg_payload) >= VIDEO_TIMESTAMP_TRAILER_BYTES + 2 + and jpeg_payload[-(VIDEO_TIMESTAMP_TRAILER_BYTES + 2) : -VIDEO_TIMESTAMP_TRAILER_BYTES] == b"\xff\xd9" + ): + return jpeg_payload[:-VIDEO_TIMESTAMP_TRAILER_BYTES], jpeg_payload[-VIDEO_TIMESTAMP_TRAILER_BYTES:] + + # 兜底兼容旧格式，避免直接丢帧。 eoi_index = jpeg_payload.rfind(b"\xff\xd9") if eoi_index < 0: - return jpeg_payload + return jpeg_payload, b"" - return jpeg_payload[: eoi_index + 2] + trailer_start = eoi_index + 2 + return jpeg_payload[:trailer_start], jpeg_payload[trailer_start:] + + def _extract_jpeg_frame(self, frame: bytes) -> bytes | None: + split_payload = self._split_jpeg_frame_and_trailer(frame) + if split_payload is None: + return None + jpeg_frame, _ = split_payload + return jpeg_frame def _extract_sequence(self, frame: bytes) -> int | None: if len(frame) >= 8 and not frame.startswith(b"\xff\xd8"): @@ -228,297 +251,273 @@ class OmniSocketVideoReceiver: return None def _extract_frame_tail(self, frame: bytes) -> bytes: - jpeg_payload = self._extract_jpeg_payload(frame) - if jpeg_payload is None: + split_payload = self._split_jpeg_frame_and_trailer(frame) + if split_payload is None: return b"" - - eoi_index = jpeg_payload.rfind(b"\xff\xd9") - if eoi_index < 0: - return b"" - - trailer_start = eoi_index + 2 - if trailer_start >= len(jpeg_payload): - return b"" - - return jpeg_payload[trailer_start:] + _, trailer = split_payload + return trailer def _extract_frame_timestamp(self, frame: bytes) -> tuple[int, str, str] | None: trailer = self._extract_frame_tail(frame) - if len(trailer) < 8: + if len(trailer) != VIDEO_TIMESTAMP_TRAILER_BYTES: return None - now_ns = time.time_ns() - raw_timestamp = trailer[-8:] - best_candidate: tuple[int, str, str] | None = None - best_distance_ns: int | None = None - - for endianness in ("big", "little"): - value = int.from_bytes(raw_timestamp, endianness, signed=True) - if value <= 0: - continue - - for unit, multiplier in ( - ("ns", 1), - ("us", 1_000), - ("ms", 1_000_000), - ("s", 1_000_000_000), - ): - timestamp_ns = value * multiplier - distance_ns = abs(now_ns - timestamp_ns) - if distance_ns > VIDEO_TIMESTAMP_MAX_SKEW_NS: - continue - if best_distance_ns is None or distance_ns < best_distance_ns: - best_distance_ns = distance_ns - best_candidate = (timestamp_ns, unit, endianness) - - return best_candidate - - def _run(self) -> None: - # 后台持续接收循环： - # connect -> recv_into(buffer) -> 按 body_len 截出有效内容 -> 把最新 JPEG 帧缓存在内存里 - while True: - try: - session, buffer_bytes = self._connect_session() - self._session = session - self._last_error = "" - buffer = bytearray(buffer_bytes) - - while True: - # 从 OmniSocket / C 侧收一帧原始二进制数据到缓冲区 - meta = session.recv_into(buffer, timeout_ms=1000) - if meta is None: - continue - if meta.get("msg_type") != self._binary_msg_type: - continue - - # 真正收到的有效部分切出来，形成当前这一帧 - frame = bytes(buffer[: meta["body_len"]]) - jpeg_frame = self._extract_jpeg_frame(frame) - if jpeg_frame is None: - self._last_error = "received non-JPEG binary frame" - continue - - timestamp_meta = self._extract_frame_timestamp(frame) - latency_ms = None - if timestamp_meta is not None: - timestamp_ns, unit, endianness = timestamp_meta - latency_ms = round((time.time_ns() - timestamp_ns) / 1_000_000, 3) - else: - unit = None - endianness = None - - with self._lock: - # 缓存：这里只保留最新的一张 JPEG 帧，供 Web 接口直接返回给前端。 - self._latest_frame = jpeg_frame - self._latest_received_at = time.time() - self._latest_sequence = self._extract_sequence(frame) - self._latest_latency_ms = latency_ms - self._latest_timestamp_unit = unit - self._latest_timestamp_endianness = endianness - if latency_ms is not None: - self._latency_samples_ms.append(latency_ms) - self._frames_received += 1 - except Exception as error: # pragma: no cover - 运行时集成路径 - self._last_error = str(error) - time.sleep(2) - finally: - if self._session is not None: - try: - self._session.close() - except Exception: - pass - self._session = None - - def get_latest_frame(self) -> bytes | None: - # 把内存里最新的一张真实 JPEG 帧暴露给 Django 视图层。 - # 如果这张帧已经过旧，就返回 None，让上层回退到本地模拟帧。 - self.ensure_started() - with self._lock: - if self._latest_frame is None: - return None - if time.time() - self._latest_received_at > OMNISOCKET_FRAME_FRESH_SECONDS: - return None - return self._latest_frame - - def get_status(self) -> dict[str, Any]: - self.ensure_started() - config = self._load_config() - transport_cfg = config.get("transport", {}) - video_cfg = config.get("video_receiver", {}) - with self._lock: - has_recent_frame = self._latest_frame is not None and ( - time.time() - self._latest_received_at <= OMNISOCKET_FRAME_FRESH_SECONDS - ) - if has_recent_frame and self._latest_latency_ms is not None: - timing_status = { - "available": True, - "latest_delta_ms": self._latest_latency_ms, - "delta_samples_ms": list(reversed(self._latency_samples_ms)), - "sample_count": len(self._latency_samples_ms), - "sample_window_size": VIDEO_TIMESTAMP_SAMPLE_SIZE, - "timestamp_unit": self._latest_timestamp_unit, - "timestamp_endianness": self._latest_timestamp_endianness, - } - else: - timing_status = { - "available": False, - "latest_delta_ms": None, - "delta_samples_ms": [], - "sample_count": 0, - "sample_window_size": VIDEO_TIMESTAMP_SAMPLE_SIZE, - "timestamp_unit": None, - "timestamp_endianness": None, - } - return { - "backend_ready": self._session_cls is not None, - "mode": VIDEO_SOURCE_MODE, - "connected": self._session is not None, - "has_recent_frame": has_recent_frame, - "frames_received": self._frames_received, - "latest_sequence": self._latest_sequence, - "last_error": self._last_error, - "config_path": str(OMNISOCKET_CONFIG_PATH), - "server_addr": str(transport_cfg.get("server_addr", "")), - "relay_via": str(transport_cfg.get("relay_via", "")), - "peer_id": str(video_cfg.get("peer_id", "")), - "buffer_bytes": int(video_cfg.get("buffer_bytes", 0)), - "timing": timing_status, - } - - -class VideoFrameService: - def __init__(self) -> None: - self._receiver = OmniSocketVideoReceiver() - - def get_status(self) -> dict[str, Any]: - receiver_status = self._receiver.get_status() - receiver_frame = self._receiver.get_latest_frame() - - # 如果已经收到了真实视频帧，就把当前状态标记为实时模式，给前端显示。 - if receiver_frame is not None: - return { - "available": True, - "source_mode": "omnisocket-jpeg-live", - "frame_count": receiver_status["frames_received"], - "fps": 30, - "frame_dir": str(JPEG_FRAME_DIR), - "source_detail": f"peer stream active, frames={receiver_status['frames_received']}", - "receiver": receiver_status, - "timing": receiver_status["timing"], - } - - wait_detail = receiver_status["last_error"] or ( - "未实时获取真实值，请检查 OmniSocket 服务、视频发送端和接收配置。" - ) - return { - "available": False, - "source_mode": "omnisocket-waiting", - "frame_count": receiver_status["frames_received"], - "fps": 30, - "frame_dir": str(JPEG_FRAME_DIR), - "source_detail": wait_detail, - "receiver": receiver_status, - "timing": receiver_status["timing"], - } - - def get_next_frame(self) -> bytes: - # 优先返回从 Python/C 视频接收器拿到的最新真实 JPEG 帧。 - receiver_frame = self._receiver.get_latest_frame() - if receiver_frame is not None: - return receiver_frame - - raise RuntimeError("未实时获取真实值，当前没有可用的真实 JPEG 帧。") - - def iter_mjpeg(self, fps: float = 6.0) -> Iterator[bytes]: - frame_interval = 1.0 / max(1.0, min(fps, 30.0)) - - while True: - frame = self.get_next_frame() - header = ( - b"--frame\r\n" - b"Content-Type: image/jpeg\r\n" - + f"Content-Length: {len(frame)}\r\n\r\n".encode("ascii") - ) - yield header + frame + b"\r\n" - time.sleep(frame_interval) - - -class GpsDataService: - def get_latest(self) -> dict[str, Any]: - # 优先使用由 GeoStream C 解析器生成的最新数据包。 - payload = self._read_geostream_payload() - if payload is not None: - payload["source_mode"] = "geostream-json" - payload["updated_at"] = utc_iso_now() - return payload - - # 当没有最新的 GPS 文件可用时，退回到使用一个移动的演示点位。 - return self._build_simulated_payload() - - def _read_geostream_payload(self) -> dict[str, Any] | None: - # Django 后端目前还不直接解析串口数据流。 - # 它只读取由 GeoStream/parse_gps.c 生成的 JSON 文件。 - if not GEOSTREAM_JSON_PATH.exists(): + value = int.from_bytes(trailer, VIDEO_TIMESTAMP_ENDIANNESS, signed=False) + if value <= 0: return None - # 忽略过期的文件，以便 UI 可以退回到使用模拟数据， - # 而不是把旧位置当作实时位置来显示。 - age_seconds = time.time() - GEOSTREAM_JSON_PATH.stat().st_mtime - if age_seconds > GEOSTREAM_STALE_SECONDS: + timestamp_ns = value * VIDEO_TIMESTAMP_MULTIPLIER_NS + if abs(time.time_ns() - timestamp_ns) > VIDEO_TIMESTAMP_MAX_SKEW_NS: return None - try: - with GEOSTREAM_JSON_PATH.open("r", encoding="utf-8") as file: - return json.load(file) - except (OSError, json.JSONDecodeError): - return None - - def _build_simulated_payload(self) -> dict[str, Any]: - # 为本地 UI 开发构建一个平滑的伪造轨迹。 - tick = time.time() / 12.0 - latitude = 31.2304 + math.sin(tick) * 0.0014 - longitude = 121.4737 + math.cos(tick) * 0.0018 - - return { - "has_fix": True, - "utc_time": datetime.now(UTC).strftime("%H:%M:%S"), - "latitude": round(latitude, 6), - "longitude": round(longitude, 6), - "satellites": 14 + int((math.sin(tick * 0.7) + 1.0) * 2), - "altitude_m": round(6.5 + math.cos(tick * 0.5) * 1.2, 2), - "coordinate_system": "WGS84", - "source_sentence": "SIMULATED", - "raw_coordinate_format": "decimal degrees", - "source_mode": "simulated", - "updated_at": utc_iso_now(), - } - - -class NetworkTelemetryService: - def get_latest(self) -> dict[str, Any]: - tick = time.time() - - latency_ms = 28 + math.sin(tick / 4.0) * 6 - jitter_ms = 3 + math.cos(tick / 5.0) * 1.4 - tx_kbps = 780 + math.sin(tick / 6.0) * 160 - rx_kbps = 720 + math.cos(tick / 7.0) * 140 - packet_loss_pct = max(0.0, 0.35 + math.sin(tick / 9.0) * 0.25) - signal_dbm = -53 - abs(math.sin(tick / 8.0) * 7) - - return { - "peer_status": "online", - "latency_ms": round(latency_ms, 1), - "jitter_ms": round(jitter_ms, 1), - "packet_loss_pct": round(packet_loss_pct, 2), - "tx_kbps": int(tx_kbps), - "rx_kbps": int(rx_kbps), - "signal_dbm": round(signal_dbm, 1), - "transport": "OmniSocket / simulated", - "source_mode": "simulated", - "updated_at": utc_iso_now(), - } - - -video_service = VideoFrameService() -gps_service = GpsDataService() -network_service = NetworkTelemetryService() + return timestamp_ns, VIDEO_TIMESTAMP_UNIT, VIDEO_TIMESTAMP_ENDIANNESS + + def _run(self) -> None: + # 后台持续接收循环： + # connect -> recv_into(buffer) -> 按 body_len 截出有效内容 -> 把最新 JPEG 帧缓存在内存里 + while True: + try: + session, buffer_bytes = self._connect_session() + self._session = session + self._last_error = "" + buffer = bytearray(buffer_bytes) + + while True: + # 从 OmniSocket / C 侧收一帧原始二进制数据到缓冲区 + meta = session.recv_into(buffer, timeout_ms=1000) + if meta is None: + continue + if meta.get("msg_type") != self._binary_msg_type: + continue + + # 真正收到的有效部分切出来，形成当前这一帧 + frame = bytes(buffer[: meta["body_len"]]) + jpeg_frame = self._extract_jpeg_frame(frame) + if jpeg_frame is None: + self._last_error = "received non-JPEG binary frame" + continue + + timestamp_meta = self._extract_frame_timestamp(frame) + latency_ms = None + if timestamp_meta is not None: + timestamp_ns, unit, endianness = timestamp_meta + latency_ms = round((time.time_ns() - timestamp_ns) / 1_000_000, 3) + else: + unit = None + endianness = None + + with self._lock: + # 缓存：这里只保留最新的一张 JPEG 帧，供 Web 接口直接返回给前端。 + self._latest_frame = jpeg_frame + self._latest_received_at = time.time() + self._latest_sequence = self._extract_sequence(frame) + self._latest_latency_ms = latency_ms + self._latest_timestamp_unit = unit + self._latest_timestamp_endianness = endianness + if latency_ms is not None: + self._latency_samples_ms.append(latency_ms) + self._frames_received += 1 + except Exception as error: # pragma: no cover - 运行时集成路径 + self._last_error = str(error) + time.sleep(2) + finally: + if self._session is not None: + try: + self._session.close() + except Exception: + pass + self._session = None + + def get_latest_frame(self) -> bytes | None: + # 把内存里最新的一张真实 JPEG 帧暴露给 Django 视图层。 + # 如果这张帧已经过旧，就返回 None，让上层回退到本地模拟帧。 + self.ensure_started() + with self._lock: + if self._latest_frame is None: + return None + if time.time() - self._latest_received_at > OMNISOCKET_FRAME_FRESH_SECONDS: + return None + return self._latest_frame + + def get_status(self) -> dict[str, Any]: + self.ensure_started() + config = self._load_config() + transport_cfg = config.get("transport", {}) + video_cfg = config.get("video_receiver", {}) + with self._lock: + has_recent_frame = self._latest_frame is not None and ( + time.time() - self._latest_received_at <= OMNISOCKET_FRAME_FRESH_SECONDS + ) + if has_recent_frame and self._latest_latency_ms is not None: + timing_status = { + "available": True, + "latest_delta_ms": self._latest_latency_ms, + "delta_samples_ms": list(reversed(self._latency_samples_ms)), + "sample_count": len(self._latency_samples_ms), + "sample_window_size": VIDEO_TIMESTAMP_SAMPLE_SIZE, + "timestamp_unit": self._latest_timestamp_unit, + "timestamp_endianness": self._latest_timestamp_endianness, + } + else: + timing_status = { + "available": False, + "latest_delta_ms": None, + "delta_samples_ms": [], + "sample_count": 0, + "sample_window_size": VIDEO_TIMESTAMP_SAMPLE_SIZE, + "timestamp_unit": None, + "timestamp_endianness": None, + } + return { + "backend_ready": self._session_cls is not None, + "mode": VIDEO_SOURCE_MODE, + "connected": self._session is not None, + "has_recent_frame": has_recent_frame, + "frames_received": self._frames_received, + "latest_sequence": self._latest_sequence, + "last_error": self._last_error, + "config_path": str(OMNISOCKET_CONFIG_PATH), + "server_addr": str(transport_cfg.get("server_addr", "")), + "relay_via": str(transport_cfg.get("relay_via", "")), + "peer_id": str(video_cfg.get("peer_id", "")), + "buffer_bytes": int(video_cfg.get("buffer_bytes", 0)), + "timing": timing_status, + } + + +class VideoFrameService: + def __init__(self) -> None: + self._receiver = OmniSocketVideoReceiver() + + def get_status(self) -> dict[str, Any]: + receiver_status = self._receiver.get_status() + receiver_frame = self._receiver.get_latest_frame() + + # 如果已经收到了真实视频帧，就把当前状态标记为实时模式，给前端显示。 + if receiver_frame is not None: + return { + "available": True, + "source_mode": "omnisocket-jpeg-live", + "frame_count": receiver_status["frames_received"], + "fps": 30, + "frame_dir": str(JPEG_FRAME_DIR), + "source_detail": f"peer stream active, frames={receiver_status['frames_received']}", + "receiver": receiver_status, + "timing": receiver_status["timing"], + } + + wait_detail = receiver_status["last_error"] or ( + "未实时获取真实值，请检查 OmniSocket 服务、视频发送端和接收配置。" + ) + return { + "available": False, + "source_mode": "omnisocket-waiting", + "frame_count": receiver_status["frames_received"], + "fps": 30, + "frame_dir": str(JPEG_FRAME_DIR), + "source_detail": wait_detail, + "receiver": receiver_status, + "timing": receiver_status["timing"], + } + + def get_next_frame(self) -> bytes: + # 优先返回从 Python/C 视频接收器拿到的最新真实 JPEG 帧。 + receiver_frame = self._receiver.get_latest_frame() + if receiver_frame is not None: + return receiver_frame + + raise RuntimeError("未实时获取真实值，当前没有可用的真实 JPEG 帧。") + + def iter_mjpeg(self, fps: float = 6.0) -> Iterator[bytes]: + frame_interval = 1.0 / max(1.0, min(fps, 30.0)) + + while True: + frame = self.get_next_frame() + header = ( + b"--frame\r\n" + b"Content-Type: image/jpeg\r\n" + + f"Content-Length: {len(frame)}\r\n\r\n".encode("ascii") + ) + yield header + frame + b"\r\n" + time.sleep(frame_interval) + + +class GpsDataService: + def get_latest(self) -> dict[str, Any]: + # 优先使用由 GeoStream C 解析器生成的最新数据包。 + payload = self._read_geostream_payload() + if payload is not None: + payload["source_mode"] = "geostream-json" + payload["updated_at"] = utc_iso_now() + return payload + + # 当没有最新的 GPS 文件可用时，退回到使用一个移动的演示点位。 + return self._build_simulated_payload() + + def _read_geostream_payload(self) -> dict[str, Any] | None: + # Django 后端目前还不直接解析串口数据流。 + # 它只读取由 GeoStream/parse_gps.c 生成的 JSON 文件。 + if not GEOSTREAM_JSON_PATH.exists(): + return None + + # 忽略过期的文件，以便 UI 可以退回到使用模拟数据， + # 而不是把旧位置当作实时位置来显示。 + age_seconds = time.time() - GEOSTREAM_JSON_PATH.stat().st_mtime + if age_seconds > GEOSTREAM_STALE_SECONDS: + return None + + try: + with GEOSTREAM_JSON_PATH.open("r", encoding="utf-8") as file: + return json.load(file) + except (OSError, json.JSONDecodeError): + return None + + def _build_simulated_payload(self) -> dict[str, Any]: + # 为本地 UI 开发构建一个平滑的伪造轨迹。 + tick = time.time() / 12.0 + latitude = 31.2304 + math.sin(tick) * 0.0014 + longitude = 121.4737 + math.cos(tick) * 0.0018 + + return { + "has_fix": True, + "utc_time": datetime.now(UTC).strftime("%H:%M:%S"), + "latitude": round(latitude, 6), + "longitude": round(longitude, 6), + "satellites": 14 + int((math.sin(tick * 0.7) + 1.0) * 2), + "altitude_m": round(6.5 + math.cos(tick * 0.5) * 1.2, 2), + "coordinate_system": "WGS84", + "source_sentence": "SIMULATED", + "raw_coordinate_format": "decimal degrees", + "source_mode": "simulated", + "updated_at": utc_iso_now(), + } + + +class NetworkTelemetryService: + def get_latest(self) -> dict[str, Any]: + tick = time.time() + + latency_ms = 28 + math.sin(tick / 4.0) * 6 + jitter_ms = 3 + math.cos(tick / 5.0) * 1.4 + tx_kbps = 780 + math.sin(tick / 6.0) * 160 + rx_kbps = 720 + math.cos(tick / 7.0) * 140 + packet_loss_pct = max(0.0, 0.35 + math.sin(tick / 9.0) * 0.25) + signal_dbm = -53 - abs(math.sin(tick / 8.0) * 7) + + return { + "peer_status": "online", + "latency_ms": round(latency_ms, 1), + "jitter_ms": round(jitter_ms, 1), + "packet_loss_pct": round(packet_loss_pct, 2), + "tx_kbps": int(tx_kbps), + "rx_kbps": int(rx_kbps), + "signal_dbm": round(signal_dbm, 1), + "transport": "OmniSocket / simulated", + "source_mode": "simulated", + "updated_at": utc_iso_now(), + } + + +video_service = VideoFrameService() +gps_service = GpsDataService() +network_service = NetworkTelemetryService()