From 2617dace7b9fc9a24fe02d95b7ffb2b1f75465a3 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 2 Dec 2025 23:21:10 +0000 Subject: [PATCH] Implement Phase 3: Collector Component This commit adds the complete Collector component for storing MeshCore events: MQTT Subscriber (collector/subscriber.py): - Subscribes to all event topics from MQTT broker - Routes events to appropriate handlers - Manages database persistence Event Handlers: - advertisement.py: Handles node advertisements, upserts nodes - message.py: Handles contact and channel messages - trace.py: Handles network trace path data - telemetry.py: Handles sensor telemetry responses - contacts.py: Handles contacts sync events - event_log.py: Generic handler for informational events CLI (collector/cli.py): - Click command for running the collector - Environment variable support for all options - Integrated with main CLI Tests: - Subscriber tests - Handler tests for advertisement, message, telemetry --- src/meshcore_hub/__main__.py | 49 +--- src/meshcore_hub/collector/cli.py | 94 +++++++ .../collector/handlers/__init__.py | 36 +++ .../collector/handlers/advertisement.py | 100 ++++++++ .../collector/handlers/contacts.py | 75 ++++++ .../collector/handlers/event_log.py | 61 +++++ .../collector/handlers/message.py | 130 ++++++++++ .../collector/handlers/telemetry.py | 103 ++++++++ src/meshcore_hub/collector/handlers/trace.py | 77 ++++++ src/meshcore_hub/collector/subscriber.py | 230 ++++++++++++++++++ tests/test_collector/conftest.py | 25 ++ .../test_handlers/test_advertisement.py | 87 +++++++ .../test_handlers/test_message.py | 89 +++++++ .../test_handlers/test_telemetry.py | 61 +++++ tests/test_collector/test_subscriber.py | 78 ++++++ 15 files changed, 1249 insertions(+), 46 deletions(-) create mode 100644 src/meshcore_hub/collector/cli.py create mode 100644 src/meshcore_hub/collector/handlers/advertisement.py create mode 100644 src/meshcore_hub/collector/handlers/contacts.py create mode 100644 src/meshcore_hub/collector/handlers/event_log.py create mode 100644 src/meshcore_hub/collector/handlers/message.py create mode 100644 src/meshcore_hub/collector/handlers/telemetry.py create mode 100644 src/meshcore_hub/collector/handlers/trace.py create mode 100644 src/meshcore_hub/collector/subscriber.py create mode 100644 tests/test_collector/conftest.py create mode 100644 tests/test_collector/test_handlers/test_advertisement.py create mode 100644 tests/test_collector/test_handlers/test_message.py create mode 100644 tests/test_collector/test_handlers/test_telemetry.py create mode 100644 tests/test_collector/test_subscriber.py diff --git a/src/meshcore_hub/__main__.py b/src/meshcore_hub/__main__.py index e29d2ca..81d5b45 100644 --- a/src/meshcore_hub/__main__.py +++ b/src/meshcore_hub/__main__.py @@ -29,55 +29,12 @@ def cli(ctx: click.Context, log_level: str) -> None: configure_logging(level=ctx.obj["log_level"]) -# Import and register interface CLI +# Import and register component CLIs from meshcore_hub.interface.cli import interface +from meshcore_hub.collector.cli import collector cli.add_command(interface) - - -@cli.command() -@click.option( - "--mqtt-host", - type=str, - default="localhost", - envvar="MQTT_HOST", - help="MQTT broker host", -) -@click.option( - "--mqtt-port", - type=int, - default=1883, - envvar="MQTT_PORT", - help="MQTT broker port", -) -@click.option( - "--prefix", - type=str, - default="meshcore", - envvar="MQTT_PREFIX", - help="MQTT topic prefix", -) -@click.option( - "--database-url", - type=str, - default="sqlite:///./meshcore.db", - envvar="DATABASE_URL", - help="Database connection URL", -) -def collector( - mqtt_host: str, - mqtt_port: int, - prefix: str, - database_url: str, -) -> None: - """Run the collector component. - - Subscribes to MQTT broker and stores events in database. - """ - click.echo("Starting collector...") - click.echo(f"MQTT: {mqtt_host}:{mqtt_port} (prefix: {prefix})") - click.echo(f"Database: {database_url}") - click.echo("Collector component not yet implemented.") +cli.add_command(collector) @cli.command() diff --git a/src/meshcore_hub/collector/cli.py b/src/meshcore_hub/collector/cli.py new file mode 100644 index 0000000..7f41fc2 --- /dev/null +++ b/src/meshcore_hub/collector/cli.py @@ -0,0 +1,94 @@ +"""CLI for the Collector component.""" + +import click + +from meshcore_hub.common.logging import configure_logging + + +@click.command("collector") +@click.option( + "--mqtt-host", + type=str, + default="localhost", + envvar="MQTT_HOST", + help="MQTT broker host", +) +@click.option( + "--mqtt-port", + type=int, + default=1883, + envvar="MQTT_PORT", + help="MQTT broker port", +) +@click.option( + "--mqtt-username", + type=str, + default=None, + envvar="MQTT_USERNAME", + help="MQTT username", +) +@click.option( + "--mqtt-password", + type=str, + default=None, + envvar="MQTT_PASSWORD", + help="MQTT password", +) +@click.option( + "--prefix", + type=str, + default="meshcore", + envvar="MQTT_PREFIX", + help="MQTT topic prefix", +) +@click.option( + "--database-url", + type=str, + default="sqlite:///./meshcore.db", + envvar="DATABASE_URL", + help="Database connection URL", +) +@click.option( + "--log-level", + type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]), + default="INFO", + envvar="LOG_LEVEL", + help="Log level", +) +def collector( + mqtt_host: str, + mqtt_port: int, + mqtt_username: str | None, + mqtt_password: str | None, + prefix: str, + database_url: str, + log_level: str, +) -> None: + """Run the collector component. + + The collector subscribes to MQTT broker and stores + MeshCore events in the database for later retrieval. + + Events stored include: + - Node advertisements + - Contact and channel messages + - Trace path data + - Telemetry responses + - Informational events (battery, status, etc.) + """ + configure_logging(level=log_level) + + click.echo("Starting MeshCore Collector") + click.echo(f"MQTT: {mqtt_host}:{mqtt_port} (prefix: {prefix})") + click.echo(f"Database: {database_url}") + + from meshcore_hub.collector.subscriber import run_collector + + run_collector( + mqtt_host=mqtt_host, + mqtt_port=mqtt_port, + mqtt_username=mqtt_username, + mqtt_password=mqtt_password, + mqtt_prefix=prefix, + database_url=database_url, + ) diff --git a/src/meshcore_hub/collector/handlers/__init__.py b/src/meshcore_hub/collector/handlers/__init__.py index 8586818..b857aff 100644 --- a/src/meshcore_hub/collector/handlers/__init__.py +++ b/src/meshcore_hub/collector/handlers/__init__.py @@ -1 +1,37 @@ """Event handlers for processing MQTT messages.""" + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from meshcore_hub.collector.subscriber import Subscriber + + +def register_all_handlers(subscriber: "Subscriber") -> None: + """Register all event handlers with the subscriber. + + Args: + subscriber: Subscriber instance + """ + from meshcore_hub.collector.handlers.advertisement import handle_advertisement + from meshcore_hub.collector.handlers.message import ( + handle_contact_message, + handle_channel_message, + ) + from meshcore_hub.collector.handlers.trace import handle_trace_data + from meshcore_hub.collector.handlers.telemetry import handle_telemetry + from meshcore_hub.collector.handlers.contacts import handle_contacts + from meshcore_hub.collector.handlers.event_log import handle_event_log + + # Persisted events with specific handlers + subscriber.register_handler("advertisement", handle_advertisement) + subscriber.register_handler("contact_msg_recv", handle_contact_message) + subscriber.register_handler("channel_msg_recv", handle_channel_message) + subscriber.register_handler("trace_data", handle_trace_data) + subscriber.register_handler("telemetry_response", handle_telemetry) + subscriber.register_handler("contacts", handle_contacts) + + # Informational events (logged only) + subscriber.register_handler("send_confirmed", handle_event_log) + subscriber.register_handler("status_response", handle_event_log) + subscriber.register_handler("battery", handle_event_log) + subscriber.register_handler("path_updated", handle_event_log) diff --git a/src/meshcore_hub/collector/handlers/advertisement.py b/src/meshcore_hub/collector/handlers/advertisement.py new file mode 100644 index 0000000..1588a61 --- /dev/null +++ b/src/meshcore_hub/collector/handlers/advertisement.py @@ -0,0 +1,100 @@ +"""Handler for advertisement events.""" + +import logging +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy import select + +from meshcore_hub.common.database import DatabaseManager +from meshcore_hub.common.models import Advertisement, Node + +logger = logging.getLogger(__name__) + + +def handle_advertisement( + public_key: str, + event_type: str, + payload: dict[str, Any], + db: DatabaseManager, +) -> None: + """Handle an advertisement event. + + 1. Upserts the node in the nodes table + 2. Creates an advertisement record + 3. Updates node last_seen timestamp + + Args: + public_key: Receiver node's public key (from MQTT topic) + event_type: Event type name + payload: Advertisement payload + db: Database manager + """ + adv_public_key = payload.get("public_key") + if not adv_public_key: + logger.warning("Advertisement missing public_key") + return + + name = payload.get("name") + adv_type = payload.get("adv_type") + flags = payload.get("flags") + now = datetime.now(timezone.utc) + + with db.session_scope() as session: + # Find or create receiver node + receiver_node = None + if public_key: + receiver_query = select(Node).where(Node.public_key == public_key) + receiver_node = session.execute(receiver_query).scalar_one_or_none() + + if not receiver_node: + receiver_node = Node( + public_key=public_key, + first_seen=now, + last_seen=now, + ) + session.add(receiver_node) + session.flush() + + # Find or create advertised node + node_query = select(Node).where(Node.public_key == adv_public_key) + node = session.execute(node_query).scalar_one_or_none() + + if node: + # Update existing node + if name: + node.name = name + if adv_type: + node.adv_type = adv_type + if flags is not None: + node.flags = flags + node.last_seen = now + else: + # Create new node + node = Node( + public_key=adv_public_key, + name=name, + adv_type=adv_type, + flags=flags, + first_seen=now, + last_seen=now, + ) + session.add(node) + session.flush() + + # Create advertisement record + advertisement = Advertisement( + receiver_node_id=receiver_node.id if receiver_node else None, + node_id=node.id, + public_key=adv_public_key, + name=name, + adv_type=adv_type, + flags=flags, + received_at=now, + ) + session.add(advertisement) + + logger.info( + f"Stored advertisement from {name or adv_public_key[:12]!r} " + f"(type={adv_type})" + ) diff --git a/src/meshcore_hub/collector/handlers/contacts.py b/src/meshcore_hub/collector/handlers/contacts.py new file mode 100644 index 0000000..096fb3e --- /dev/null +++ b/src/meshcore_hub/collector/handlers/contacts.py @@ -0,0 +1,75 @@ +"""Handler for contacts sync events.""" + +import logging +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy import select + +from meshcore_hub.common.database import DatabaseManager +from meshcore_hub.common.models import Node + +logger = logging.getLogger(__name__) + + +def handle_contacts( + public_key: str, + event_type: str, + payload: dict[str, Any], + db: DatabaseManager, +) -> None: + """Handle a contacts sync event. + + Upserts all contacts in the contacts list. + + Args: + public_key: Receiver node's public key (from MQTT topic) + event_type: Event type name + payload: Contacts payload + db: Database manager + """ + contacts = payload.get("contacts", []) + if not contacts: + logger.debug("Empty contacts list received") + return + + now = datetime.now(timezone.utc) + created_count = 0 + updated_count = 0 + + with db.session_scope() as session: + for contact in contacts: + contact_key = contact.get("public_key") + if not contact_key: + continue + + name = contact.get("name") + node_type = contact.get("node_type") + + # Find or create node + node_query = select(Node).where(Node.public_key == contact_key) + node = session.execute(node_query).scalar_one_or_none() + + if node: + # Update existing node + if name and not node.name: + node.name = name + if node_type and not node.adv_type: + node.adv_type = node_type + node.last_seen = now + updated_count += 1 + else: + # Create new node + node = Node( + public_key=contact_key, + name=name, + adv_type=node_type, + first_seen=now, + last_seen=now, + ) + session.add(node) + created_count += 1 + + logger.info( + f"Processed contacts sync: {created_count} new, {updated_count} updated" + ) diff --git a/src/meshcore_hub/collector/handlers/event_log.py b/src/meshcore_hub/collector/handlers/event_log.py new file mode 100644 index 0000000..8553780 --- /dev/null +++ b/src/meshcore_hub/collector/handlers/event_log.py @@ -0,0 +1,61 @@ +"""Generic event log handler for informational events.""" + +import logging +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy import select + +from meshcore_hub.common.database import DatabaseManager +from meshcore_hub.common.models import EventLog, Node + +logger = logging.getLogger(__name__) + + +def handle_event_log( + public_key: str, + event_type: str, + payload: dict[str, Any], + db: DatabaseManager, +) -> None: + """Handle an event by logging it to the events_log table. + + This is used for informational events that don't need + specific processing but should be recorded. + + Args: + public_key: Receiver node's public key (from MQTT topic) + event_type: Event type name + payload: Event payload + db: Database manager + """ + now = datetime.now(timezone.utc) + + with db.session_scope() as session: + # Find receiver node + receiver_node = None + if public_key: + receiver_query = select(Node).where(Node.public_key == public_key) + receiver_node = session.execute(receiver_query).scalar_one_or_none() + + if not receiver_node: + receiver_node = Node( + public_key=public_key, + first_seen=now, + last_seen=now, + ) + session.add(receiver_node) + session.flush() + else: + receiver_node.last_seen = now + + # Create event log record + event_log = EventLog( + receiver_node_id=receiver_node.id if receiver_node else None, + event_type=event_type, + payload=payload, + received_at=now, + ) + session.add(event_log) + + logger.debug(f"Logged event: {event_type}") diff --git a/src/meshcore_hub/collector/handlers/message.py b/src/meshcore_hub/collector/handlers/message.py new file mode 100644 index 0000000..28614f9 --- /dev/null +++ b/src/meshcore_hub/collector/handlers/message.py @@ -0,0 +1,130 @@ +"""Handler for message events.""" + +import logging +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy import select + +from meshcore_hub.common.database import DatabaseManager +from meshcore_hub.common.models import Message, Node + +logger = logging.getLogger(__name__) + + +def handle_contact_message( + public_key: str, + event_type: str, + payload: dict[str, Any], + db: DatabaseManager, +) -> None: + """Handle a contact message event. + + Args: + public_key: Receiver node's public key (from MQTT topic) + event_type: Event type name + payload: Message payload + db: Database manager + """ + _handle_message(public_key, "contact", payload, db) + + +def handle_channel_message( + public_key: str, + event_type: str, + payload: dict[str, Any], + db: DatabaseManager, +) -> None: + """Handle a channel message event. + + Args: + public_key: Receiver node's public key (from MQTT topic) + event_type: Event type name + payload: Message payload + db: Database manager + """ + _handle_message(public_key, "channel", payload, db) + + +def _handle_message( + public_key: str, + message_type: str, + payload: dict[str, Any], + db: DatabaseManager, +) -> None: + """Handle a message event (contact or channel). + + Args: + public_key: Receiver node's public key + message_type: Message type ('contact' or 'channel') + payload: Message payload + db: Database manager + """ + text = payload.get("text") + if not text: + logger.warning(f"Message missing text content") + return + + now = datetime.now(timezone.utc) + + # Extract fields based on message type + pubkey_prefix = payload.get("pubkey_prefix") if message_type == "contact" else None + channel_idx = payload.get("channel_idx") if message_type == "channel" else None + path_len = payload.get("path_len") + txt_type = payload.get("txt_type") + signature = payload.get("signature") + snr = payload.get("SNR") or payload.get("snr") + + # Parse sender timestamp + sender_ts = payload.get("sender_timestamp") + sender_timestamp = None + if sender_ts: + try: + sender_timestamp = datetime.fromtimestamp(sender_ts, tz=timezone.utc) + except (ValueError, OSError): + pass + + with db.session_scope() as session: + # Find receiver node + receiver_node = None + if public_key: + receiver_query = select(Node).where(Node.public_key == public_key) + receiver_node = session.execute(receiver_query).scalar_one_or_none() + + if not receiver_node: + receiver_node = Node( + public_key=public_key, + first_seen=now, + last_seen=now, + ) + session.add(receiver_node) + session.flush() + else: + receiver_node.last_seen = now + + # Create message record + message = Message( + receiver_node_id=receiver_node.id if receiver_node else None, + message_type=message_type, + pubkey_prefix=pubkey_prefix, + channel_idx=channel_idx, + text=text, + path_len=path_len, + txt_type=txt_type, + signature=signature, + snr=snr, + sender_timestamp=sender_timestamp, + received_at=now, + ) + session.add(message) + + if message_type == "contact": + logger.info( + f"Stored contact message from {pubkey_prefix!r}: " + f"{text[:30]}{'...' if len(text) > 30 else ''}" + ) + else: + logger.info( + f"Stored channel {channel_idx} message: " + f"{text[:30]}{'...' if len(text) > 30 else ''}" + ) diff --git a/src/meshcore_hub/collector/handlers/telemetry.py b/src/meshcore_hub/collector/handlers/telemetry.py new file mode 100644 index 0000000..40c14e1 --- /dev/null +++ b/src/meshcore_hub/collector/handlers/telemetry.py @@ -0,0 +1,103 @@ +"""Handler for telemetry events.""" + +import logging +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy import select + +from meshcore_hub.common.database import DatabaseManager +from meshcore_hub.common.models import Node, Telemetry + +logger = logging.getLogger(__name__) + + +def handle_telemetry( + public_key: str, + event_type: str, + payload: dict[str, Any], + db: DatabaseManager, +) -> None: + """Handle a telemetry response event. + + Args: + public_key: Receiver node's public key (from MQTT topic) + event_type: Event type name + payload: Telemetry payload + db: Database manager + """ + node_public_key = payload.get("node_public_key") + if not node_public_key: + logger.warning("Telemetry missing node_public_key") + return + + now = datetime.now(timezone.utc) + + lpp_data = payload.get("lpp_data") + parsed_data = payload.get("parsed_data") + + # Convert lpp_data to bytes if it's a string or list + lpp_bytes = None + if lpp_data: + if isinstance(lpp_data, bytes): + lpp_bytes = lpp_data + elif isinstance(lpp_data, list): + lpp_bytes = bytes(lpp_data) + elif isinstance(lpp_data, str): + try: + lpp_bytes = bytes.fromhex(lpp_data) + except ValueError: + lpp_bytes = lpp_data.encode() + + with db.session_scope() as session: + # Find receiver node + receiver_node = None + if public_key: + receiver_query = select(Node).where(Node.public_key == public_key) + receiver_node = session.execute(receiver_query).scalar_one_or_none() + + if not receiver_node: + receiver_node = Node( + public_key=public_key, + first_seen=now, + last_seen=now, + ) + session.add(receiver_node) + session.flush() + else: + receiver_node.last_seen = now + + # Find or create reporting node + reporting_node = None + if node_public_key: + node_query = select(Node).where(Node.public_key == node_public_key) + reporting_node = session.execute(node_query).scalar_one_or_none() + + if not reporting_node: + reporting_node = Node( + public_key=node_public_key, + first_seen=now, + last_seen=now, + ) + session.add(reporting_node) + session.flush() + else: + reporting_node.last_seen = now + + # Create telemetry record + telemetry = Telemetry( + receiver_node_id=receiver_node.id if receiver_node else None, + node_id=reporting_node.id if reporting_node else None, + node_public_key=node_public_key, + lpp_data=lpp_bytes, + parsed_data=parsed_data, + received_at=now, + ) + session.add(telemetry) + + # Log telemetry values + if parsed_data: + values = ", ".join(f"{k}={v}" for k, v in parsed_data.items()) + logger.info(f"Stored telemetry from {node_public_key[:12]!r}: {values}") + else: + logger.info(f"Stored telemetry from {node_public_key[:12]!r}") diff --git a/src/meshcore_hub/collector/handlers/trace.py b/src/meshcore_hub/collector/handlers/trace.py new file mode 100644 index 0000000..13e42b4 --- /dev/null +++ b/src/meshcore_hub/collector/handlers/trace.py @@ -0,0 +1,77 @@ +"""Handler for trace data events.""" + +import logging +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy import select + +from meshcore_hub.common.database import DatabaseManager +from meshcore_hub.common.models import Node, TracePath + +logger = logging.getLogger(__name__) + + +def handle_trace_data( + public_key: str, + event_type: str, + payload: dict[str, Any], + db: DatabaseManager, +) -> None: + """Handle a trace data event. + + Args: + public_key: Receiver node's public key (from MQTT topic) + event_type: Event type name + payload: Trace data payload + db: Database manager + """ + initiator_tag = payload.get("initiator_tag") + if initiator_tag is None: + logger.warning("Trace data missing initiator_tag") + return + + now = datetime.now(timezone.utc) + + path_len = payload.get("path_len") + flags = payload.get("flags") + auth = payload.get("auth") + path_hashes = payload.get("path_hashes") + snr_values = payload.get("snr_values") + hop_count = payload.get("hop_count") + + with db.session_scope() as session: + # Find receiver node + receiver_node = None + if public_key: + receiver_query = select(Node).where(Node.public_key == public_key) + receiver_node = session.execute(receiver_query).scalar_one_or_none() + + if not receiver_node: + receiver_node = Node( + public_key=public_key, + first_seen=now, + last_seen=now, + ) + session.add(receiver_node) + session.flush() + else: + receiver_node.last_seen = now + + # Create trace path record + trace_path = TracePath( + receiver_node_id=receiver_node.id if receiver_node else None, + initiator_tag=initiator_tag, + path_len=path_len, + flags=flags, + auth=auth, + path_hashes=path_hashes, + snr_values=snr_values, + hop_count=hop_count, + received_at=now, + ) + session.add(trace_path) + + logger.info( + f"Stored trace data: tag={initiator_tag}, hops={hop_count}" + ) diff --git a/src/meshcore_hub/collector/subscriber.py b/src/meshcore_hub/collector/subscriber.py new file mode 100644 index 0000000..456c9cc --- /dev/null +++ b/src/meshcore_hub/collector/subscriber.py @@ -0,0 +1,230 @@ +"""MQTT Subscriber for collecting MeshCore events. + +The subscriber: +1. Connects to MQTT broker +2. Subscribes to all event topics +3. Routes events to appropriate handlers +4. Persists data to database +""" + +import logging +import signal +import threading +import time +from typing import Any, Callable, Optional + +from meshcore_hub.common.database import DatabaseManager +from meshcore_hub.common.mqtt import MQTTClient, MQTTConfig + +logger = logging.getLogger(__name__) + + +# Handler type: receives (public_key, event_type, payload, db_manager) +EventHandler = Callable[[str, str, dict[str, Any], DatabaseManager], None] + + +class Subscriber: + """MQTT Subscriber for collecting and storing MeshCore events.""" + + def __init__( + self, + mqtt_client: MQTTClient, + db_manager: DatabaseManager, + ): + """Initialize subscriber. + + Args: + mqtt_client: MQTT client instance + db_manager: Database manager instance + """ + self.mqtt = mqtt_client + self.db = db_manager + self._running = False + self._shutdown_event = threading.Event() + self._handlers: dict[str, EventHandler] = {} + + def register_handler(self, event_type: str, handler: EventHandler) -> None: + """Register a handler for an event type. + + Args: + event_type: Event type name (e.g., 'advertisement') + handler: Handler function + """ + self._handlers[event_type] = handler + logger.debug(f"Registered handler for {event_type}") + + def _handle_mqtt_message( + self, + topic: str, + pattern: str, + payload: dict[str, Any], + ) -> None: + """Handle incoming MQTT event message. + + Args: + topic: MQTT topic + pattern: Subscription pattern + payload: Message payload + """ + # Parse event from topic + parsed = self.mqtt.topic_builder.parse_event_topic(topic) + if not parsed: + logger.warning(f"Could not parse event topic: {topic}") + return + + public_key, event_type = parsed + logger.debug(f"Received event: {event_type} from {public_key[:12]}...") + + # Find and call handler + handler = self._handlers.get(event_type) + if handler: + try: + handler(public_key, event_type, payload, self.db) + except Exception as e: + logger.error(f"Error handling {event_type}: {e}") + else: + # Use generic event log handler if no specific handler + from meshcore_hub.collector.handlers.event_log import handle_event_log + + try: + handle_event_log(public_key, event_type, payload, self.db) + except Exception as e: + logger.error(f"Error logging event {event_type}: {e}") + + def start(self) -> None: + """Start the subscriber.""" + logger.info("Starting collector subscriber") + + # Create database tables if needed + self.db.create_tables() + + # Connect to MQTT broker + try: + self.mqtt.connect() + self.mqtt.start_background() + logger.info("Connected to MQTT broker") + except Exception as e: + logger.error(f"Failed to connect to MQTT broker: {e}") + raise + + # Subscribe to all event topics + event_topic = self.mqtt.topic_builder.all_events_topic() + self.mqtt.subscribe(event_topic, self._handle_mqtt_message) + logger.info(f"Subscribed to event topic: {event_topic}") + + self._running = True + + def run(self) -> None: + """Run the subscriber event loop (blocking).""" + if not self._running: + self.start() + + logger.info("Collector running. Press Ctrl+C to stop.") + + try: + while self._running and not self._shutdown_event.is_set(): + time.sleep(0.1) + except KeyboardInterrupt: + logger.info("Keyboard interrupt received") + finally: + self.stop() + + def stop(self) -> None: + """Stop the subscriber.""" + if not self._running: + return + + logger.info("Stopping collector subscriber") + self._running = False + self._shutdown_event.set() + + # Stop MQTT + self.mqtt.stop() + self.mqtt.disconnect() + + logger.info("Collector subscriber stopped") + + +def create_subscriber( + mqtt_host: str = "localhost", + mqtt_port: int = 1883, + mqtt_username: Optional[str] = None, + mqtt_password: Optional[str] = None, + mqtt_prefix: str = "meshcore", + database_url: str = "sqlite:///./meshcore.db", +) -> Subscriber: + """Create a configured subscriber instance. + + Args: + mqtt_host: MQTT broker host + mqtt_port: MQTT broker port + mqtt_username: MQTT username + mqtt_password: MQTT password + mqtt_prefix: MQTT topic prefix + database_url: Database connection URL + + Returns: + Configured Subscriber instance + """ + # Create MQTT client + mqtt_config = MQTTConfig( + host=mqtt_host, + port=mqtt_port, + username=mqtt_username, + password=mqtt_password, + prefix=mqtt_prefix, + client_id="meshcore-collector", + ) + mqtt_client = MQTTClient(mqtt_config) + + # Create database manager + db_manager = DatabaseManager(database_url) + + # Create subscriber + subscriber = Subscriber(mqtt_client, db_manager) + + # Register handlers + from meshcore_hub.collector.handlers import register_all_handlers + + register_all_handlers(subscriber) + + return subscriber + + +def run_collector( + mqtt_host: str = "localhost", + mqtt_port: int = 1883, + mqtt_username: Optional[str] = None, + mqtt_password: Optional[str] = None, + mqtt_prefix: str = "meshcore", + database_url: str = "sqlite:///./meshcore.db", +) -> None: + """Run the collector (blocking). + + Args: + mqtt_host: MQTT broker host + mqtt_port: MQTT broker port + mqtt_username: MQTT username + mqtt_password: MQTT password + mqtt_prefix: MQTT topic prefix + database_url: Database connection URL + """ + subscriber = create_subscriber( + mqtt_host=mqtt_host, + mqtt_port=mqtt_port, + mqtt_username=mqtt_username, + mqtt_password=mqtt_password, + mqtt_prefix=mqtt_prefix, + database_url=database_url, + ) + + # Set up signal handlers + def signal_handler(signum: int, frame: Any) -> None: + logger.info(f"Received signal {signum}") + subscriber.stop() + + signal.signal(signal.SIGINT, signal_handler) + signal.signal(signal.SIGTERM, signal_handler) + + # Run + subscriber.run() diff --git a/tests/test_collector/conftest.py b/tests/test_collector/conftest.py new file mode 100644 index 0000000..c8c7592 --- /dev/null +++ b/tests/test_collector/conftest.py @@ -0,0 +1,25 @@ +"""Fixtures for collector component tests.""" + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from meshcore_hub.common.database import DatabaseManager +from meshcore_hub.common.models import Base + + +@pytest.fixture +def db_manager(): + """Create an in-memory database manager for testing.""" + manager = DatabaseManager("sqlite:///:memory:") + manager.create_tables() + yield manager + manager.dispose() + + +@pytest.fixture +def db_session(db_manager): + """Create a database session for testing.""" + session = db_manager.get_session() + yield session + session.close() diff --git a/tests/test_collector/test_handlers/test_advertisement.py b/tests/test_collector/test_handlers/test_advertisement.py new file mode 100644 index 0000000..c4a7d50 --- /dev/null +++ b/tests/test_collector/test_handlers/test_advertisement.py @@ -0,0 +1,87 @@ +"""Tests for advertisement handler.""" + +import pytest +from sqlalchemy import select + +from meshcore_hub.common.models import Advertisement, Node +from meshcore_hub.collector.handlers.advertisement import handle_advertisement + + +class TestHandleAdvertisement: + """Tests for handle_advertisement.""" + + def test_creates_new_node(self, db_manager, db_session): + """Test that new nodes are created.""" + payload = { + "public_key": "a" * 64, + "name": "TestNode", + "adv_type": "chat", + "flags": 218, + } + + handle_advertisement("b" * 64, "advertisement", payload, db_manager) + + # Check node was created + node = db_session.execute( + select(Node).where(Node.public_key == "a" * 64) + ).scalar_one_or_none() + + assert node is not None + assert node.name == "TestNode" + assert node.adv_type == "chat" + assert node.flags == 218 + + def test_updates_existing_node(self, db_manager, db_session): + """Test that existing nodes are updated.""" + # Create initial node + node = Node(public_key="a" * 64, name="OldName", adv_type="repeater") + db_session.add(node) + db_session.commit() + + # Handle advertisement with new data + payload = { + "public_key": "a" * 64, + "name": "NewName", + "adv_type": "chat", + "flags": 100, + } + + handle_advertisement("b" * 64, "advertisement", payload, db_manager) + + # Refresh node + db_session.refresh(node) + + assert node.name == "NewName" + assert node.adv_type == "chat" + assert node.flags == 100 + + def test_creates_advertisement_record(self, db_manager, db_session): + """Test that advertisement records are created.""" + payload = { + "public_key": "a" * 64, + "name": "TestNode", + "adv_type": "chat", + } + + handle_advertisement("b" * 64, "advertisement", payload, db_manager) + + # Check advertisement was created + ad = db_session.execute(select(Advertisement)).scalar_one_or_none() + + assert ad is not None + assert ad.public_key == "a" * 64 + assert ad.name == "TestNode" + + def test_handles_missing_public_key(self, db_manager, db_session): + """Test that missing public_key is handled gracefully.""" + payload = { + "name": "TestNode", + "adv_type": "chat", + } + + # Should not raise + handle_advertisement("b" * 64, "advertisement", payload, db_manager) + + # No advertisement should be created + ads = db_session.execute(select(Advertisement)).scalars().all() + assert len(ads) == 0 diff --git a/tests/test_collector/test_handlers/test_message.py b/tests/test_collector/test_handlers/test_message.py new file mode 100644 index 0000000..fdf32d7 --- /dev/null +++ b/tests/test_collector/test_handlers/test_message.py @@ -0,0 +1,89 @@ +"""Tests for message handlers.""" + +import pytest +from sqlalchemy import select + +from meshcore_hub.common.models import Message, Node +from meshcore_hub.collector.handlers.message import ( + handle_contact_message, + handle_channel_message, +) + + +class TestHandleContactMessage: + """Tests for handle_contact_message.""" + + def test_creates_contact_message(self, db_manager, db_session): + """Test that contact messages are stored.""" + payload = { + "pubkey_prefix": "01ab2186c4d5", + "text": "Hello World!", + "path_len": 3, + "SNR": 15.5, + } + + handle_contact_message("a" * 64, "contact_msg_recv", payload, db_manager) + + # Check message was created + msg = db_session.execute(select(Message)).scalar_one_or_none() + + assert msg is not None + assert msg.message_type == "contact" + assert msg.pubkey_prefix == "01ab2186c4d5" + assert msg.text == "Hello World!" + assert msg.path_len == 3 + assert msg.snr == 15.5 + + def test_handles_missing_text(self, db_manager, db_session): + """Test that missing text is handled gracefully.""" + payload = { + "pubkey_prefix": "01ab2186c4d5", + "path_len": 3, + } + + handle_contact_message("a" * 64, "contact_msg_recv", payload, db_manager) + + # No message should be created + msgs = db_session.execute(select(Message)).scalars().all() + assert len(msgs) == 0 + + +class TestHandleChannelMessage: + """Tests for handle_channel_message.""" + + def test_creates_channel_message(self, db_manager, db_session): + """Test that channel messages are stored.""" + payload = { + "channel_idx": 4, + "text": "Channel broadcast", + "path_len": 10, + "SNR": 8.5, + } + + handle_channel_message("a" * 64, "channel_msg_recv", payload, db_manager) + + # Check message was created + msg = db_session.execute(select(Message)).scalar_one_or_none() + + assert msg is not None + assert msg.message_type == "channel" + assert msg.channel_idx == 4 + assert msg.text == "Channel broadcast" + assert msg.path_len == 10 + assert msg.snr == 8.5 + + def test_creates_receiver_node_if_needed(self, db_manager, db_session): + """Test that receiver node is created if it doesn't exist.""" + payload = { + "channel_idx": 4, + "text": "Test message", + } + + handle_channel_message("a" * 64, "channel_msg_recv", payload, db_manager) + + # Check receiver node was created + node = db_session.execute( + select(Node).where(Node.public_key == "a" * 64) + ).scalar_one_or_none() + + assert node is not None diff --git a/tests/test_collector/test_handlers/test_telemetry.py b/tests/test_collector/test_handlers/test_telemetry.py new file mode 100644 index 0000000..17f89bd --- /dev/null +++ b/tests/test_collector/test_handlers/test_telemetry.py @@ -0,0 +1,61 @@ +"""Tests for telemetry handler.""" + +import pytest +from sqlalchemy import select + +from meshcore_hub.common.models import Node, Telemetry +from meshcore_hub.collector.handlers.telemetry import handle_telemetry + + +class TestHandleTelemetry: + """Tests for handle_telemetry.""" + + def test_creates_telemetry_record(self, db_manager, db_session): + """Test that telemetry records are stored.""" + payload = { + "node_public_key": "b" * 64, + "parsed_data": { + "temperature": 22.5, + "humidity": 65, + "battery": 3.8, + }, + } + + handle_telemetry("a" * 64, "telemetry_response", payload, db_manager) + + # Check telemetry was created + telemetry = db_session.execute(select(Telemetry)).scalar_one_or_none() + + assert telemetry is not None + assert telemetry.node_public_key == "b" * 64 + assert telemetry.parsed_data["temperature"] == 22.5 + assert telemetry.parsed_data["humidity"] == 65 + assert telemetry.parsed_data["battery"] == 3.8 + + def test_creates_reporting_node(self, db_manager, db_session): + """Test that reporting node is created if needed.""" + payload = { + "node_public_key": "b" * 64, + "parsed_data": {"temperature": 20.0}, + } + + handle_telemetry("a" * 64, "telemetry_response", payload, db_manager) + + # Check node was created + node = db_session.execute( + select(Node).where(Node.public_key == "b" * 64) + ).scalar_one_or_none() + + assert node is not None + + def test_handles_missing_node_public_key(self, db_manager, db_session): + """Test that missing node_public_key is handled gracefully.""" + payload = { + "parsed_data": {"temperature": 20.0}, + } + + handle_telemetry("a" * 64, "telemetry_response", payload, db_manager) + + # No telemetry should be created + records = db_session.execute(select(Telemetry)).scalars().all() + assert len(records) == 0 diff --git a/tests/test_collector/test_subscriber.py b/tests/test_collector/test_subscriber.py new file mode 100644 index 0000000..722161e --- /dev/null +++ b/tests/test_collector/test_subscriber.py @@ -0,0 +1,78 @@ +"""Tests for the collector subscriber.""" + +import pytest +from unittest.mock import MagicMock, patch + +from meshcore_hub.collector.subscriber import Subscriber, create_subscriber + + +class TestSubscriber: + """Tests for Subscriber class.""" + + @pytest.fixture + def mock_mqtt_client(self): + """Create a mock MQTT client.""" + client = MagicMock() + client.topic_builder = MagicMock() + client.topic_builder.all_events_topic.return_value = "meshcore/+/event/#" + client.topic_builder.parse_event_topic.return_value = ("a" * 64, "advertisement") + return client + + @pytest.fixture + def subscriber(self, mock_mqtt_client, db_manager): + """Create a subscriber instance.""" + return Subscriber(mock_mqtt_client, db_manager) + + def test_register_handler(self, subscriber): + """Test handler registration.""" + handler = MagicMock() + + subscriber.register_handler("advertisement", handler) + + assert "advertisement" in subscriber._handlers + + def test_start_connects_mqtt(self, subscriber, mock_mqtt_client): + """Test that start connects to MQTT.""" + subscriber.start() + + mock_mqtt_client.connect.assert_called_once() + mock_mqtt_client.start_background.assert_called_once() + mock_mqtt_client.subscribe.assert_called_once() + + def test_stop_disconnects_mqtt(self, subscriber, mock_mqtt_client): + """Test that stop disconnects MQTT.""" + subscriber.start() + subscriber.stop() + + mock_mqtt_client.stop.assert_called_once() + mock_mqtt_client.disconnect.assert_called_once() + + def test_handle_mqtt_message_calls_handler(self, subscriber, mock_mqtt_client, db_manager): + """Test that MQTT messages are routed to handlers.""" + handler = MagicMock() + subscriber.register_handler("advertisement", handler) + subscriber.start() + + subscriber._handle_mqtt_message( + topic="meshcore/abc/event/advertisement", + pattern="meshcore/+/event/#", + payload={"public_key": "b" * 64, "name": "Test"}, + ) + + handler.assert_called_once() + + +class TestCreateSubscriber: + """Tests for create_subscriber factory function.""" + + def test_creates_subscriber(self): + """Test creating a subscriber.""" + with patch("meshcore_hub.collector.subscriber.MQTTClient") as MockMQTT: + subscriber = create_subscriber( + mqtt_host="localhost", + mqtt_port=1883, + database_url="sqlite:///:memory:", + ) + + assert subscriber is not None + MockMQTT.assert_called_once()