Feat/sat mock scan

.github/workflows/release.yml

@@ -42,8 +42,8 @@ jobs:
       - name: Lint
         run: ruff check src
 
-      - name: Run unit tests (exclude integration and BLE)
-        run: pytest -m "not integration and not ble" -v --tb=short
+      - name: Run unit tests (exclude integration, BLE, and Docker)
+        run: pytest -m "not integration and not ble and not docker" -v --tb=short
 
       - name: Build sdist and wheel
         run: |

.github/workflows/sat-mock-test.yml

@@ -0,0 +1,49 @@
+name: Satellite Mock Tests
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+  schedule:
+    - cron: '0 0 * * 1'  # Weekly on Monday at midnight UTC
+  workflow_dispatch:
+
+jobs:
+  sat-mock-test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Set up Python 3.12
+        uses: actions/setup-python@v6
+        with:
+          python-version: "3.12"
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -e ".[dev]"
+
+      - name: Cache Docker image
+        id: docker-cache
+        uses: actions/cache@v4
+        with:
+          path: /tmp/sdr-docker-image.tar
+          key: sdr-docker-${{ hashFiles('.github/workflows/sat-mock-test.yml') }}
+
+      - name: Load cached Docker image
+        if: steps.docker-cache.outputs.cache-hit == 'true'
+        run: docker load -i /tmp/sdr-docker-image.tar
+
+      - name: Pull Docker image
+        if: steps.docker-cache.outputs.cache-hit != 'true'
+        run: docker pull ghcr.io/hubblenetwork/sdr-docker:latest
+
+      - name: Save Docker image to cache
+        if: steps.docker-cache.outputs.cache-hit != 'true'
+        run: docker save ghcr.io/hubblenetwork/sdr-docker:latest -o /tmp/sdr-docker-image.tar
+
+      - name: Run satellite mock integration tests
+        run: pytest tests/test_sat_integration.py -v --tb=short

pyproject.toml

@@ -49,6 +49,7 @@ testpaths = ["tests"]
 pythonpath = ["src"]
 markers = [
   "ble: tests that require BLE hardware/permissions",
+  "docker: tests that require a running Docker daemon",
   "integration: slow or environment-dependent tests",
 ]
 

src/hubblenetwork/cli.py

@@ -2795,63 +2795,18 @@ def sat() -> None:
     """Satellite (PlutoSDR) utilities."""
 
 
-@sat.command("scan")
-@click.option(
-    "--timeout",
-    "-t",
-    type=int,
-    show_default=False,
-    help="Timeout in seconds (default: no timeout)",
-)
-@click.option(
-    "--count",
-    "-n",
-    type=int,
-    default=None,
-    show_default=False,
-    help="Stop after receiving N packets",
-)
-@click.option(
-    "--format",
-    "-o",
-    "output_format",
-    type=click.Choice(["tabular", "json"], case_sensitive=False),
-    default="tabular",
-    show_default=True,
-    help="Output format for packets",
-)
-@click.option(
-    "--poll-interval",
-    type=float,
-    default=2.0,
-    show_default=True,
-    help="Seconds between API polls",
-)
-@click.option(
-    "--payload-format",
-    "payload_format",
-    type=click.Choice(["base64", "hex", "string"], case_sensitive=False),
-    default="base64",
-    show_default=True,
-    help="Encoding format for packet payload",
-)
-def sat_scan(
-    timeout: Optional[int] = None,
-    count: Optional[int] = None,
-    output_format: str = "tabular",
-    poll_interval: float = 2.0,
-    payload_format: str = "base64",
+def _run_sat_scan(
+    *,
+    mock: bool,
+    timeout: Optional[int],
+    count: Optional[int],
+    output_format: str,
+    poll_interval: float,
+    payload_format: str,
 ) -> None:
-    """
-    Start the satellite receiver and stream decoded packets.
+    """Shared implementation for ``sat scan`` and ``sat mock-scan``."""
+    mode_label = "mock satellite receiver" if mock else "satellite receiver"
 
-    Requires Docker and a PlutoSDR device connected via USB.
-
-    Example:
-      hubblenetwork sat scan --timeout 30
-      hubblenetwork sat scan -o json --timeout 10
-      hubblenetwork sat scan -n 5
-    """
     printer_class = _SAT_STREAMING_PRINTERS.get(
         output_format.lower(), _SatStreamingTablePrinter
     )
@@ -2870,15 +2825,15 @@ def sat_scan(
 
     if not printer.suppress_info_messages:
         click.secho(
-            "[INFO] Starting satellite receiver... (Press Ctrl+C to stop)"
+            f"[INFO] Starting {mode_label}... (Press Ctrl+C to stop)"
         )
 
     _stop_msg_shown = [False]
 
     def _on_interrupt(sig, frame):
         if not _stop_msg_shown[0] and not printer.suppress_info_messages:
             click.secho(
-                "\n[INFO] Stopping satellite receiver...", fg="yellow", err=True
+                f"\n[INFO] Stopping {mode_label}...", fg="yellow", err=True
             )
             _stop_msg_shown[0] = True
         raise KeyboardInterrupt()
@@ -2887,7 +2842,7 @@ def _on_interrupt(sig, frame):
     error_occurred = False
     try:
         for pkt in sat_mod.scan(
-            timeout=timeout, poll_interval=poll_interval
+            timeout=timeout, poll_interval=poll_interval, mock=mock
         ):
             printer.print_row(pkt)
             if count is not None and printer.packet_count >= count:
@@ -2921,6 +2876,87 @@ def _on_interrupt(sig, frame):
             )
 
 
+def _sat_scan_options(fn):
+    """Apply the common sat scan/mock-scan Click options."""
+    for decorator in reversed([
+        click.option("--timeout", "-t", type=int, show_default=False,
+                     help="Timeout in seconds (default: no timeout)"),
+        click.option("--count", "-n", type=int, default=None,
+                     show_default=False, help="Stop after receiving N packets"),
+        click.option("--format", "-o", "output_format",
+                     type=click.Choice(["tabular", "json"], case_sensitive=False),
+                     default="tabular", show_default=True,
+                     help="Output format for packets"),
+        click.option("--poll-interval", type=float, default=2.0,
+                     show_default=True, help="Seconds between API polls"),
+        click.option("--payload-format", "payload_format",
+                     type=click.Choice(["base64", "hex", "string"],
+                                       case_sensitive=False),
+                     default="base64", show_default=True,
+                     help="Encoding format for packet payload"),
+    ]):
+        fn = decorator(fn)
+    return fn
+
+
+@sat.command("scan")
+@_sat_scan_options
+def sat_scan(
+    timeout: Optional[int] = None,
+    count: Optional[int] = None,
+    output_format: str = "tabular",
+    poll_interval: float = 2.0,
+    payload_format: str = "base64",
+) -> None:
+    """
+    Start the satellite receiver and stream decoded packets.
+
+    Requires Docker and a PlutoSDR device connected via USB.
+
+    Example:
+      hubblenetwork sat scan --timeout 30
+      hubblenetwork sat scan -o json --timeout 10
+      hubblenetwork sat scan -n 5
+    """
+    _run_sat_scan(
+        mock=False,
+        timeout=timeout,
+        count=count,
+        output_format=output_format,
+        poll_interval=poll_interval,
+        payload_format=payload_format,
+    )
+
+
+@sat.command("mock-scan")
+@_sat_scan_options
+def sat_mock_scan(
+    timeout: Optional[int] = None,
+    count: Optional[int] = None,
+    output_format: str = "tabular",
+    poll_interval: float = 2.0,
+    payload_format: str = "base64",
+) -> None:
+    """
+    Start the satellite receiver in mock mode and stream synthetic packets.
+
+    Uses simulated data -- no PlutoSDR hardware required. Useful for testing
+    the satellite scanning interface.
+
+    Example:
+      hubblenetwork sat mock-scan --timeout 30
+      hubblenetwork sat mock-scan -o json -n 5
+    """
+    _run_sat_scan(
+        mock=True,
+        timeout=timeout,
+        count=count,
+        output_format=output_format,
+        poll_interval=poll_interval,
+        payload_format=payload_format,
+    )
+
+
 def main(argv: Optional[list[str]] = None) -> int:
     """
     Entry point used by console_scripts.

src/hubblenetwork/sat.py

@@ -14,7 +14,7 @@
 import logging
 import time
 from pathlib import Path
-from typing import Generator, List, Optional, Set, Tuple
+from typing import Dict, Generator, List, Optional, Set, Tuple
 
 import httpx
 
@@ -25,6 +25,7 @@
 
 DOCKER_IMAGE = "ghcr.io/hubblenetwork/sdr-docker:latest"
 CONTAINER_NAME = "hubble-pluto-sdr"
+MOCK_CONTAINER_NAME = "hubble-pluto-sdr-mock"
 API_PORT = 8050
 _CONTAINER_INTERNAL_PORT = 8050  # fixed by the Docker image
 
@@ -106,6 +107,10 @@ def pull_image(image: str = DOCKER_IMAGE) -> None:
 def start_container(
     image: str = DOCKER_IMAGE,
     port: int = API_PORT,
+    *,
+    environment: Optional[Dict[str, str]] = None,
+    privileged: bool = True,
+    name: str = CONTAINER_NAME,
 ) -> str:
     """Start the PlutoSDR container and return the container ID.
 
@@ -119,8 +124,9 @@ def start_container(
             detach=True,
             auto_remove=True,
             ports={f"{_CONTAINER_INTERNAL_PORT}/tcp": port},
-            name=CONTAINER_NAME,
-            privileged=True,
+            name=name,
+            privileged=privileged,
+            environment=environment or {},
         )
         logger.debug("Started container %s", container.short_id)
         return container.id
@@ -214,17 +220,32 @@ def scan(
     poll_interval: float = 2.0,
     port: int = API_PORT,
     image: str = DOCKER_IMAGE,
+    *,
+    mock: bool = False,
 ) -> Generator[SatellitePacket, None, None]:
     """Scan for satellite packets, managing the Docker container lifecycle.
 
     Yields new ``SatellitePacket`` objects as they are discovered.  The
     container is guaranteed to be stopped when the generator is closed or
     an exception occurs.
+
+    When *mock* is ``True`` the container is started in mock mode
+    (``SDR_TYPE=mock``) which emits synthetic packets without requiring
+    PlutoSDR hardware.
     """
     ensure_docker_available()
     pull_image(image)
 
-    container_id = start_container(image=image, port=port)
+    container_name = MOCK_CONTAINER_NAME if mock else CONTAINER_NAME
+    environment: Optional[Dict[str, str]] = {"SDR_TYPE": "mock"} if mock else None
+
+    container_id = start_container(
+        image=image,
+        port=port,
+        environment=environment,
+        privileged=not mock,
+        name=container_name,
+    )
     try:
         _wait_for_api(port=port)