diff --git a/.env.production.example b/.env.production.example index 18999b4..c653307 100644 --- a/.env.production.example +++ b/.env.production.example @@ -42,6 +42,18 @@ POSTGRES_PASSWORD=password_CYmsGt SENTRY_DSN= +# MQTT ingestion pipeline +MQTT_HOST=101.35.119.226 +MQTT_PORT=1883 +MQTT_USERNAME= +MQTT_PASSWORD= +MQTT_TOPIC_PATTERN=terminal/+/raw +MQTT_CLIENT_ID=spatialhub-mqtt-ingestor +RAW_HOT_RETENTION_DAYS=90 +RAW_ARCHIVE_BASE_DIR=/data/archive/gnss +PARSER_BATCH_SIZE=200 +PARSER_POLL_INTERVAL_MS=300 + # Docker images (本地构建，不需要远程 registry) DOCKER_IMAGE_BACKEND=backend DOCKER_IMAGE_FRONTEND=frontend diff --git a/backend/README.md b/backend/README.md index 521ee3f..dcc7794 100644 --- a/backend/README.md +++ b/backend/README.md @@ -170,3 +170,40 @@ The email templates are in `./backend/app/email-templates/`. Here, there are two Before continuing, ensure you have the [MJML extension](https://github.com/mjmlio/vscode-mjml) installed in your VS Code. Once you have the MJML extension installed, you can create a new email template in the `src` directory. After creating the new email template and with the `.mjml` file open in your editor, open the command palette with `Ctrl+Shift+P` and search for `MJML: Export to HTML`. This will convert the `.mjml` file to a `.html` file and now you can save it in the build directory. + +## MQTT Ingestion Pipeline + +This project includes a two-stage MQTT pipeline: + +- `mqtt-ingestor` subscribes to `terminal/{device_id}/raw` and stores payload bytes losslessly in `terminal_message_raw`. +- `mqtt-parser-worker` consumes pending rows and writes parsed events/state. + +### Run locally with Docker Compose + +```console +$ docker compose up -d db backend mqtt-ingestor mqtt-parser-worker +``` + +### Unit tests for MQTT modules + +The existing `tests/conftest.py` expects a PostgreSQL connection. For parser/repository unit tests that run with in-memory SQLite fixtures, use: + +```console +$ cd backend +$ PYTHONPATH=. ../.venv/bin/pytest tests/mqtt --confcutdir=tests/mqtt -q +``` + +### Verify end-to-end ingestion quickly + +1. Start services (`db`, `backend`, `mqtt-ingestor`, `mqtt-parser-worker`). +2. Publish a sample message to `terminal/device-001/raw`. +3. Query PostgreSQL tables: + - `terminal_message_raw` + - `terminal_observation_event` + - `terminal_device_state` + +### Archive run (manual trigger) + +```console +$ docker compose run --rm mqtt-parser-worker python -m app.mqtt.archive_job +``` diff --git a/backend/app/models.py b/backend/app/models.py index 5945c5c..bedfd0e 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -121,7 +121,7 @@ class TerminalMessageRaw(SQLModel, table=True): default_factory=get_datetime_utc, sa_type=DateTime(timezone=True), # type: ignore ) - payload_bytes: bytes = Field(sa_type=LargeBinary) # type: ignore + payload_bytes: bytes = Field(sa_type=LargeBinary) payload_size: int = 0 payload_sha256: str = Field(default="", max_length=64, index=True) qos: int = 0 @@ -154,7 +154,7 @@ class TerminalObservationEvent(SQLModel, table=True): index=True, ) event_type: str = Field(max_length=64, index=True) - payload_json: dict[str, Any] = Field(default_factory=dict, sa_type=JSON) # type: ignore + payload_json: dict[str, Any] = Field(default_factory=dict, sa_type=JSON) created_at: datetime = Field( default_factory=get_datetime_utc, sa_type=DateTime(timezone=True), # type: ignore @@ -167,14 +167,14 @@ class TerminalDeviceState(SQLModel, table=True): device_id: str = Field(max_length=255, primary_key=True) last_event_time: datetime | None = Field( default=None, sa_type=DateTime(timezone=True) - ) # type: ignore + ) # type: ignore[call-overload] fix_status: str | None = Field(default=None, max_length=64) lat: float | None = None lon: float | None = None alt_m: float | None = None satellites_used: int | None = None - satellite_signal_summary: dict[str, Any] | None = Field(default=None, sa_type=JSON) # type: ignore - rtcm_type_summary: dict[str, Any] | None = Field(default=None, sa_type=JSON) # type: ignore + satellite_signal_summary: dict[str, Any] | None = Field(default=None, sa_type=JSON) + rtcm_type_summary: dict[str, Any] | None = Field(default=None, sa_type=JSON) updated_at: datetime = Field( default_factory=get_datetime_utc, sa_type=DateTime(timezone=True), # type: ignore diff --git a/backend/app/mqtt/archive_job.py b/backend/app/mqtt/archive_job.py new file mode 100644 index 0000000..a8ff69d --- /dev/null +++ b/backend/app/mqtt/archive_job.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import hashlib +import logging +from datetime import datetime, timedelta +from pathlib import Path + +from sqlmodel import Session, col, select + +from app.core.config import settings +from app.core.db import engine +from app.models import TerminalMessageRaw, TerminalRawArchiveManifest, get_datetime_utc + +logger = logging.getLogger(__name__) + + +def _format_timestamp(value: datetime) -> str: + return value.strftime("%Y-%m-%dT%H-%M-%S") + + +def _build_archive_path( + *, + archive_base_dir: Path, + device_id: str, + range_start: datetime, + range_end: datetime, +) -> Path: + day = range_start.strftime("%Y-%m-%d") + output_dir = archive_base_dir / device_id / day + output_dir.mkdir(parents=True, exist_ok=True) + filename = ( + f"device_{device_id}_{_format_timestamp(range_start)}" + f"_{_format_timestamp(range_end)}.txt" + ) + return output_dir / filename + + +def run_archive_once( + *, + session: Session, + archive_base_dir: Path | str, + retention_days: int, + now: datetime | None = None, +) -> list[TerminalRawArchiveManifest]: + archive_root = Path(archive_base_dir) + ts_now = now or get_datetime_utc() + cutoff = ts_now - timedelta(days=retention_days) + + statement = ( + select(TerminalMessageRaw) + .where(TerminalMessageRaw.received_at < cutoff) + .order_by( + col(TerminalMessageRaw.device_id), col(TerminalMessageRaw.received_at) + ) + ) + rows = session.exec(statement).all() + if not rows: + return [] + + grouped: dict[tuple[str, str], list[TerminalMessageRaw]] = {} + for row in rows: + key = (row.device_id, row.received_at.strftime("%Y-%m-%d")) + grouped.setdefault(key, []).append(row) + + manifests: list[TerminalRawArchiveManifest] = [] + for (_, _), group_rows in grouped.items(): + range_start = group_rows[0].received_at + range_end = group_rows[-1].received_at + output_path = _build_archive_path( + archive_base_dir=archive_root, + device_id=group_rows[0].device_id, + range_start=range_start, + range_end=range_end, + ) + + digest = hashlib.sha256() + record_count = 0 + with output_path.open("wb") as out: + for row in group_rows: + out.write(row.payload_bytes) + digest.update(row.payload_bytes) + record_count += 1 + + manifest = TerminalRawArchiveManifest( + device_id=group_rows[0].device_id, + range_start=range_start, + range_end=range_end, + file_path=str(output_path), + file_size=output_path.stat().st_size, + sha256=digest.hexdigest(), + record_count=record_count, + ) + session.add(manifest) + manifests.append(manifest) + + for row in group_rows: + session.delete(row) + + session.commit() + + for manifest in manifests: + session.refresh(manifest) + + return manifests + + +def run() -> None: + logging.basicConfig(level=logging.INFO) + with Session(engine) as session: + manifests = run_archive_once( + session=session, + archive_base_dir=settings.RAW_ARCHIVE_BASE_DIR, + retention_days=settings.RAW_HOT_RETENTION_DAYS, + ) + logger.info("Archived %s file(s)", len(manifests)) + + +if __name__ == "__main__": + run() diff --git a/backend/app/mqtt/ingestor.py b/backend/app/mqtt/ingestor.py new file mode 100644 index 0000000..a50fa02 --- /dev/null +++ b/backend/app/mqtt/ingestor.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import logging + +import paho.mqtt.client as mqtt # type: ignore[import-not-found,import-untyped] +from sqlmodel import Session + +from app.core.config import settings +from app.core.db import engine +from app.mqtt.repository import insert_raw_message +from app.mqtt.topic import extract_device_id + +logger = logging.getLogger(__name__) + + +def _on_connect( + client: mqtt.Client, + _userdata: object, + _flags: dict[str, int], + reason_code: int, + _properties: mqtt.Properties | None = None, +) -> None: + if reason_code != 0: + logger.error("MQTT connect failed: rc=%s", reason_code) + return + + logger.info( + "Connected to MQTT broker %s:%s", settings.MQTT_HOST, settings.MQTT_PORT + ) + client.subscribe(settings.MQTT_TOPIC_PATTERN, qos=1) + + +def _on_message( + _client: mqtt.Client, + _userdata: object, + message: mqtt.MQTTMessage, +) -> None: + try: + device_id = extract_device_id(message.topic) + except ValueError: + logger.warning("Skipping unexpected topic: %s", message.topic) + return + + with Session(engine) as session: + insert_raw_message( + session=session, + device_id=device_id, + topic=message.topic, + payload=bytes(message.payload), + qos=message.qos, + retain=message.retain, + ) + + +def create_client() -> mqtt.Client: + client = mqtt.Client( + callback_api_version=mqtt.CallbackAPIVersion.VERSION2, + client_id=settings.MQTT_CLIENT_ID, + clean_session=False, + ) + if settings.MQTT_USERNAME: + client.username_pw_set(settings.MQTT_USERNAME, settings.MQTT_PASSWORD) + + client.on_connect = _on_connect + client.on_message = _on_message + client.reconnect_delay_set(min_delay=1, max_delay=30) + return client + + +def run() -> None: + logging.basicConfig(level=logging.INFO) + client = create_client() + client.connect(settings.MQTT_HOST, settings.MQTT_PORT, keepalive=60) + client.loop_forever() + + +if __name__ == "__main__": + run() diff --git a/backend/app/mqtt/parser_worker.py b/backend/app/mqtt/parser_worker.py new file mode 100644 index 0000000..644352f --- /dev/null +++ b/backend/app/mqtt/parser_worker.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import logging +import time +from datetime import timedelta + +from sqlmodel import Session + +from app.core.config import settings +from app.core.db import engine +from app.models import TerminalMessageRaw, get_datetime_utc +from app.mqtt.parsers.mixed import parse_mixed_payload +from app.mqtt.repository import ( + fetch_pending_rows_for_update, + mark_failed, + mark_processing, + mark_succeeded, + write_events_and_update_state, +) + +logger = logging.getLogger(__name__) + + +def _next_retry_delay_seconds(attempts: int) -> int: + # Bounded exponential backoff: 30s, 60s, 120s, ... capped to 15m. + return int(min(30 * (2 ** max(attempts - 1, 0)), 900)) + + +def _process_one_row(*, session: Session, row: TerminalMessageRaw) -> None: + mark_processing(session=session, row=row) + session.commit() + + try: + events = parse_mixed_payload(row.payload_bytes) + write_events_and_update_state(session=session, row=row, events=events) + mark_succeeded(session=session, row=row) + except Exception as exc: # pragma: no cover - safety net path + delay = _next_retry_delay_seconds(row.parse_attempts) + mark_failed( + session=session, + row=row, + parse_error=str(exc), + next_retry_at=get_datetime_utc() + timedelta(seconds=delay), + ) + + session.commit() + + +def process_pending_batch(*, session: Session, batch_size: int) -> int: + rows = fetch_pending_rows_for_update(session=session, batch_size=batch_size) + for row in rows: + _process_one_row(session=session, row=row) + return len(rows) + + +def run() -> None: + logging.basicConfig(level=logging.INFO) + sleep_seconds = max(settings.PARSER_POLL_INTERVAL_MS, 50) / 1000.0 + + while True: + with Session(engine) as session: + processed = process_pending_batch( + session=session, batch_size=settings.PARSER_BATCH_SIZE + ) + if processed == 0: + time.sleep(sleep_seconds) + else: + logger.info("Processed %s pending MQTT raw messages", processed) + + +if __name__ == "__main__": + run() diff --git a/backend/app/mqtt/parsers/__init__.py b/backend/app/mqtt/parsers/__init__.py new file mode 100644 index 0000000..04bc14e --- /dev/null +++ b/backend/app/mqtt/parsers/__init__.py @@ -0,0 +1,5 @@ +from app.mqtt.parsers.mixed import parse_mixed_payload +from app.mqtt.parsers.nmea import parse_nmea_sentences +from app.mqtt.parsers.rtcm import summarize_rtcm_types + +__all__ = ["parse_mixed_payload", "parse_nmea_sentences", "summarize_rtcm_types"] diff --git a/backend/app/mqtt/parsers/mixed.py b/backend/app/mqtt/parsers/mixed.py new file mode 100644 index 0000000..df0f864 --- /dev/null +++ b/backend/app/mqtt/parsers/mixed.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +from typing import Any + +from app.mqtt.parsers.nmea import parse_nmea_sentences +from app.mqtt.parsers.rtcm import summarize_rtcm_types + + +def parse_mixed_payload(payload: bytes) -> list[dict[str, Any]]: + events = parse_nmea_sentences(payload) + + rtcm_types = summarize_rtcm_types(payload) + if rtcm_types: + events.append( + { + "event_type": "rtcm_summary", + "payload": { + "rtcm_message_count": sum(rtcm_types.values()), + "rtcm_types": rtcm_types, + }, + } + ) + + return events diff --git a/backend/app/mqtt/parsers/nmea.py b/backend/app/mqtt/parsers/nmea.py new file mode 100644 index 0000000..218f70d --- /dev/null +++ b/backend/app/mqtt/parsers/nmea.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +from typing import Any + +SUPPORTED_NMEA_TYPES = {"GGA", "RMC", "GSV", "GSA", "GST"} + + +def _to_float(value: str) -> float | None: + if not value: + return None + try: + return float(value) + except ValueError: + return None + + +def _to_int(value: str) -> int | None: + if not value: + return None + try: + return int(value) + except ValueError: + return None + + +def _parse_lat_lon(value: str, hemisphere: str) -> float | None: + if not value or not hemisphere: + return None + + try: + degree_len = 2 if hemisphere in {"N", "S"} else 3 + degrees = float(value[:degree_len]) + minutes = float(value[degree_len:]) + except ValueError: + return None + + decimal = degrees + (minutes / 60.0) + if hemisphere in {"S", "W"}: + decimal *= -1 + return decimal + + +def _extract_sentence_type(sentence: str) -> str | None: + if not sentence.startswith("$"): + return None + if len(sentence) < 6: + return None + nmea_type = sentence[3:6].upper() + if nmea_type not in SUPPORTED_NMEA_TYPES: + return None + return nmea_type + + +def _parse_nmea_fields(nmea_type: str, fields: list[str]) -> dict[str, Any]: + payload: dict[str, Any] = {"sentence_type": nmea_type, "raw_fields": fields} + + if nmea_type == "GGA": + payload.update( + { + "utc_time": fields[1] if len(fields) > 1 else None, + "lat": _parse_lat_lon( + fields[2] if len(fields) > 2 else "", + fields[3] if len(fields) > 3 else "", + ), + "lon": _parse_lat_lon( + fields[4] if len(fields) > 4 else "", + fields[5] if len(fields) > 5 else "", + ), + "fix_quality": _to_int(fields[6] if len(fields) > 6 else ""), + "satellites_used": _to_int(fields[7] if len(fields) > 7 else ""), + "hdop": _to_float(fields[8] if len(fields) > 8 else ""), + "altitude_m": _to_float(fields[9] if len(fields) > 9 else ""), + } + ) + elif nmea_type == "RMC": + payload.update( + { + "utc_time": fields[1] if len(fields) > 1 else None, + "status": fields[2] if len(fields) > 2 else None, + "lat": _parse_lat_lon( + fields[3] if len(fields) > 3 else "", + fields[4] if len(fields) > 4 else "", + ), + "lon": _parse_lat_lon( + fields[5] if len(fields) > 5 else "", + fields[6] if len(fields) > 6 else "", + ), + "speed_knots": _to_float(fields[7] if len(fields) > 7 else ""), + "course_deg": _to_float(fields[8] if len(fields) > 8 else ""), + "date": fields[9] if len(fields) > 9 else None, + } + ) + elif nmea_type == "GSV": + payload.update( + { + "total_messages": _to_int(fields[1] if len(fields) > 1 else ""), + "message_index": _to_int(fields[2] if len(fields) > 2 else ""), + "satellites_in_view": _to_int(fields[3] if len(fields) > 3 else ""), + "snr_values": [ + _to_int(fields[i]) + for i in range(7, len(fields), 4) + if i < len(fields) + ], + } + ) + elif nmea_type == "GSA": + payload.update( + { + "mode": fields[1] if len(fields) > 1 else None, + "fix_type": _to_int(fields[2] if len(fields) > 2 else ""), + "pdop": _to_float(fields[15] if len(fields) > 15 else ""), + "hdop": _to_float(fields[16] if len(fields) > 16 else ""), + "vdop": _to_float(fields[17] if len(fields) > 17 else ""), + } + ) + elif nmea_type == "GST": + payload.update( + { + "utc_time": fields[1] if len(fields) > 1 else None, + "rms": _to_float(fields[2] if len(fields) > 2 else ""), + "std_major": _to_float(fields[3] if len(fields) > 3 else ""), + "std_minor": _to_float(fields[4] if len(fields) > 4 else ""), + } + ) + + return payload + + +def parse_nmea_sentences(payload: bytes) -> list[dict[str, Any]]: + events: list[dict[str, Any]] = [] + + for raw_line in payload.splitlines(): + if not raw_line.startswith(b"$"): + continue + + sentence = raw_line.decode("ascii", errors="ignore").strip() + nmea_type = _extract_sentence_type(sentence) + if nmea_type is None: + continue + + main_part = sentence.split("*", 1)[0] + fields = main_part.split(",") + parsed_payload = _parse_nmea_fields(nmea_type, fields) + events.append( + {"event_type": f"nmea_{nmea_type.lower()}", "payload": parsed_payload} + ) + + return events diff --git a/backend/app/mqtt/parsers/rtcm.py b/backend/app/mqtt/parsers/rtcm.py new file mode 100644 index 0000000..839b259 --- /dev/null +++ b/backend/app/mqtt/parsers/rtcm.py @@ -0,0 +1,22 @@ +from __future__ import annotations + + +def summarize_rtcm_types(payload: bytes) -> dict[int, int]: + counts: dict[int, int] = {} + i = 0 + + while i + 6 <= len(payload): + if payload[i] != 0xD3: + i += 1 + continue + + length = ((payload[i + 1] & 0x03) << 8) | payload[i + 2] + frame_end = i + 3 + length + 3 + if frame_end > len(payload): + break + + message_type = ((payload[i + 3] << 4) | (payload[i + 4] >> 4)) & 0x0FFF + counts[message_type] = counts.get(message_type, 0) + 1 + i = frame_end + + return counts diff --git a/backend/app/mqtt/repository.py b/backend/app/mqtt/repository.py new file mode 100644 index 0000000..14e2d51 --- /dev/null +++ b/backend/app/mqtt/repository.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +import hashlib +from datetime import datetime +from typing import Any + +from sqlmodel import Session, col, select + +from app.models import ( + TerminalDeviceState, + TerminalMessageRaw, + TerminalObservationEvent, + get_datetime_utc, +) + + +def insert_raw_message( + *, + session: Session, + device_id: str, + topic: str, + payload: bytes, + qos: int, + retain: bool, + received_at: datetime | None = None, +) -> TerminalMessageRaw: + ts = received_at or get_datetime_utc() + row = TerminalMessageRaw( + device_id=device_id, + topic=topic, + received_at=ts, + payload_bytes=payload, + payload_size=len(payload), + payload_sha256=hashlib.sha256(payload).hexdigest(), + qos=qos, + retain=retain, + parse_status="pending", + created_at=ts, + ) + session.add(row) + session.commit() + session.refresh(row) + return row + + +def fetch_pending_rows_for_update( + *, session: Session, batch_size: int +) -> list[TerminalMessageRaw]: + statement = ( + select(TerminalMessageRaw) + .where(TerminalMessageRaw.parse_status == "pending") + .order_by(col(TerminalMessageRaw.received_at)) + .limit(batch_size) + ) + + bind = session.get_bind() + if bind is not None and bind.dialect.name != "sqlite": + statement = statement.with_for_update(skip_locked=True) + + rows = session.exec(statement).all() + return list(rows) + + +def mark_processing(*, session: Session, row: TerminalMessageRaw) -> None: + row.parse_status = "processing" + row.parse_attempts += 1 + row.parse_error = None + session.add(row) + + +def mark_succeeded(*, session: Session, row: TerminalMessageRaw) -> None: + row.parse_status = "succeeded" + row.parse_error = None + row.next_retry_at = None + session.add(row) + + +def mark_failed( + *, + session: Session, + row: TerminalMessageRaw, + parse_error: str, + next_retry_at: datetime | None, +) -> None: + row.parse_status = "failed" + row.parse_error = parse_error[:1024] + row.next_retry_at = next_retry_at + session.add(row) + + +def write_events_and_update_state( + *, + session: Session, + row: TerminalMessageRaw, + events: list[dict[str, Any]], +) -> None: + if not events: + state = session.get(TerminalDeviceState, row.device_id) + if state is None: + state = TerminalDeviceState(device_id=row.device_id) + state.last_event_time = row.received_at + state.updated_at = get_datetime_utc() + session.add(state) + return + + state = session.get(TerminalDeviceState, row.device_id) + if state is None: + state = TerminalDeviceState(device_id=row.device_id) + + for event in events: + payload = event.get("payload", {}) + event_type = str(event.get("event_type", "unknown")) + db_event = TerminalObservationEvent( + raw_id=row.id, + device_id=row.device_id, + event_time=row.received_at, + event_type=event_type, + payload_json=payload if isinstance(payload, dict) else {"value": payload}, + ) + session.add(db_event) + + if event_type == "nmea_gga": + state.lat = payload.get("lat") + state.lon = payload.get("lon") + state.alt_m = payload.get("altitude_m") + state.satellites_used = payload.get("satellites_used") + fix_quality = payload.get("fix_quality") + if fix_quality is not None: + state.fix_status = str(fix_quality) + elif event_type == "nmea_rmc": + if payload.get("lat") is not None: + state.lat = payload.get("lat") + if payload.get("lon") is not None: + state.lon = payload.get("lon") + status = payload.get("status") + if status is not None: + state.fix_status = str(status) + elif event_type == "nmea_gsv": + state.satellite_signal_summary = { + "snr_values": payload.get("snr_values", []), + "satellites_in_view": payload.get("satellites_in_view"), + } + elif event_type == "rtcm_summary": + state.rtcm_type_summary = payload.get("rtcm_types", {}) + + state.last_event_time = row.received_at + state.updated_at = get_datetime_utc() + session.add(state) diff --git a/backend/tests/mqtt/test_archive_job.py b/backend/tests/mqtt/test_archive_job.py new file mode 100644 index 0000000..e68059c --- /dev/null +++ b/backend/tests/mqtt/test_archive_job.py @@ -0,0 +1,54 @@ +from datetime import timedelta +from pathlib import Path + +from sqlmodel import Session, SQLModel, create_engine, select + +from app.models import TerminalMessageRaw, get_datetime_utc +from app.mqtt.archive_job import run_archive_once +from app.mqtt.repository import insert_raw_message + + +def _session() -> Session: + engine = create_engine("sqlite://") + SQLModel.metadata.create_all(engine) + return Session(engine) + + +def test_archive_writes_raw_bytes_and_manifest(tmp_path: Path) -> None: + with _session() as session: + current = get_datetime_utc() + old_ts = current - timedelta(days=91) + + insert_raw_message( + session=session, + device_id="device-001", + topic="terminal/device-001/raw", + payload=b"abc", + qos=1, + retain=False, + received_at=old_ts, + ) + insert_raw_message( + session=session, + device_id="device-001", + topic="terminal/device-001/raw", + payload=b"\x01\x02", + qos=1, + retain=False, + received_at=old_ts, + ) + + manifests = run_archive_once( + session=session, + archive_base_dir=tmp_path, + retention_days=90, + now=current, + ) + + assert len(manifests) == 1 + archived_file = Path(manifests[0].file_path) + assert archived_file.exists() + assert archived_file.read_bytes() == b"abc\x01\x02" + + remaining = session.exec(select(TerminalMessageRaw)).all() + assert remaining == [] diff --git a/backend/tests/mqtt/test_mixed_parser.py b/backend/tests/mqtt/test_mixed_parser.py new file mode 100644 index 0000000..34567cf --- /dev/null +++ b/backend/tests/mqtt/test_mixed_parser.py @@ -0,0 +1,27 @@ +from app.mqtt.parsers.mixed import parse_mixed_payload +from app.mqtt.parsers.nmea import parse_nmea_sentences + + +def test_parse_nmea_sentences_extracts_gga_and_rmc() -> None: + payload = ( + b"$GPGGA,123519,3723.2475,N,12158.3416,W,1,12,0.8,10.0,M,-25.0,M,,*65\r\n" + b"$GPRMC,123520,A,3723.2475,N,12158.3416,W,0.13,309.62,120598,,,A*74\r\n" + ) + + events = parse_nmea_sentences(payload) + + assert {e["event_type"] for e in events} >= {"nmea_gga", "nmea_rmc"} + + +def test_parse_mixed_payload_returns_rtcm_summary() -> None: + payload = ( + b"$GPGSV,1,1,04,02,17,213,40,05,29,121,37,12,66,314,44,25,14,048,39*71\r\n" + + b"\xd3\x00\x04" + + b"\x43\x50\x00\x00" + + b"\x00\x00\x00" + ) + + events = parse_mixed_payload(payload) + + event_types = {event["event_type"] for event in events} + assert "rtcm_summary" in event_types diff --git a/backend/tests/mqtt/test_parser_worker.py b/backend/tests/mqtt/test_parser_worker.py new file mode 100644 index 0000000..836304a --- /dev/null +++ b/backend/tests/mqtt/test_parser_worker.py @@ -0,0 +1,53 @@ +from sqlmodel import Session, SQLModel, create_engine, select + +from app.models import TerminalDeviceState, TerminalMessageRaw, TerminalObservationEvent +from app.mqtt.parser_worker import process_pending_batch +from app.mqtt.repository import insert_raw_message + + +def _session() -> Session: + engine = create_engine("sqlite://") + SQLModel.metadata.create_all(engine) + return Session(engine) + + +def test_insert_raw_message_sets_pending_status() -> None: + with _session() as session: + row = insert_raw_message( + session=session, + device_id="device-001", + topic="terminal/device-001/raw", + payload=b"$GPRMC,123520,A,3723.2475,N,12158.3416,W,0.13,309.62,120598,,,A*74\r\n", + qos=1, + retain=False, + ) + + assert row.parse_status == "pending" + assert row.payload_size > 0 + assert len(row.payload_sha256) == 64 + + +def test_worker_marks_success_and_writes_events() -> None: + with _session() as session: + raw = insert_raw_message( + session=session, + device_id="device-001", + topic="terminal/device-001/raw", + payload=b"$GPGGA,123519,3723.2475,N,12158.3416,W,1,12,0.8,10.0,M,-25.0,M,,*65\r\n", + qos=1, + retain=False, + ) + + processed = process_pending_batch(session=session, batch_size=10) + + assert processed == 1 + + refreshed = session.get(TerminalMessageRaw, raw.id) + assert refreshed is not None + assert refreshed.parse_status == "succeeded" + + events = session.exec(select(TerminalObservationEvent)).all() + assert len(events) >= 1 + + state = session.get(TerminalDeviceState, "device-001") + assert state is not None diff --git a/compose.override.yml b/compose.override.yml index 779cc82..2fc7718 100644 --- a/compose.override.yml +++ b/compose.override.yml @@ -87,6 +87,20 @@ services: SMTP_TLS: "false" EMAILS_FROM_EMAIL: "noreply@example.com" + mqtt-ingestor: + restart: "no" + command: + - python + - -m + - app.mqtt.ingestor + + mqtt-parser-worker: + restart: "no" + command: + - python + - -m + - app.mqtt.parser_worker + mailcatcher: image: schickling/mailcatcher ports: diff --git a/compose.yml b/compose.yml index 2488fc0..05a17be 100644 --- a/compose.yml +++ b/compose.yml @@ -136,6 +136,83 @@ services: # Enable redirection for HTTP and HTTPS - traefik.http.routers.${STACK_NAME?Variable not set}-backend-http.middlewares=https-redirect + mqtt-ingestor: + image: '${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG-latest}' + restart: always + networks: + - default + depends_on: + db: + condition: service_healthy + restart: true + prestart: + condition: service_completed_successfully + command: + - python + - -m + - app.mqtt.ingestor + env_file: + - .env + environment: + - ENVIRONMENT=${ENVIRONMENT} + - SECRET_KEY=${SECRET_KEY?Variable not set} + - FIRST_SUPERUSER=${FIRST_SUPERUSER?Variable not set} + - FIRST_SUPERUSER_PASSWORD=${FIRST_SUPERUSER_PASSWORD?Variable not set} + - POSTGRES_SERVER=db + - POSTGRES_PORT=${POSTGRES_PORT} + - POSTGRES_DB=${POSTGRES_DB} + - POSTGRES_USER=${POSTGRES_USER?Variable not set} + - POSTGRES_PASSWORD=${POSTGRES_PASSWORD?Variable not set} + - MQTT_HOST=${MQTT_HOST-101.35.119.226} + - MQTT_PORT=${MQTT_PORT-1883} + - MQTT_USERNAME=${MQTT_USERNAME} + - MQTT_PASSWORD=${MQTT_PASSWORD} + - MQTT_TOPIC_PATTERN=${MQTT_TOPIC_PATTERN-terminal/+/raw} + - MQTT_CLIENT_ID=${MQTT_CLIENT_ID-spatialhub-mqtt-ingestor} + - RAW_HOT_RETENTION_DAYS=${RAW_HOT_RETENTION_DAYS-90} + - RAW_ARCHIVE_BASE_DIR=${RAW_ARCHIVE_BASE_DIR-/data/archive/gnss} + - PARSER_BATCH_SIZE=${PARSER_BATCH_SIZE-200} + - PARSER_POLL_INTERVAL_MS=${PARSER_POLL_INTERVAL_MS-300} + build: + context: . + dockerfile: backend/Dockerfile + volumes: + - mqtt-archive-data:/data/archive/gnss + + mqtt-parser-worker: + image: '${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG-latest}' + restart: always + networks: + - default + depends_on: + db: + condition: service_healthy + restart: true + prestart: + condition: service_completed_successfully + command: + - python + - -m + - app.mqtt.parser_worker + env_file: + - .env + environment: + - ENVIRONMENT=${ENVIRONMENT} + - SECRET_KEY=${SECRET_KEY?Variable not set} + - FIRST_SUPERUSER=${FIRST_SUPERUSER?Variable not set} + - FIRST_SUPERUSER_PASSWORD=${FIRST_SUPERUSER_PASSWORD?Variable not set} + - POSTGRES_SERVER=db + - POSTGRES_PORT=${POSTGRES_PORT} + - POSTGRES_DB=${POSTGRES_DB} + - POSTGRES_USER=${POSTGRES_USER?Variable not set} + - POSTGRES_PASSWORD=${POSTGRES_PASSWORD?Variable not set} + - RAW_HOT_RETENTION_DAYS=${RAW_HOT_RETENTION_DAYS-90} + - PARSER_BATCH_SIZE=${PARSER_BATCH_SIZE-200} + - PARSER_POLL_INTERVAL_MS=${PARSER_POLL_INTERVAL_MS-300} + build: + context: . + dockerfile: backend/Dockerfile + frontend: image: '${DOCKER_IMAGE_FRONTEND?Variable not set}:${TAG-latest}' restart: always @@ -167,6 +244,7 @@ services: - traefik.http.routers.${STACK_NAME?Variable not set}-frontend-http.middlewares=https-redirect volumes: app-db-data: + mqtt-archive-data: networks: traefik-public: