mypy and tests

Author: LC-Linkous

.github/workflows/tests.yml

@@ -72,7 +72,7 @@ jobs:
         run: uv run ruff check .
 
       - name: Mypy
-        # Non-blocking for now: flip `continue-on-error` to false once the
-        # type surface is clean and you want mypy to gate merges.
-        continue-on-error: true
-        run: uv run mypy src/hackrfpy
+        # Blocking. The package ships a py.typed marker, which promises
+        # downstream type-checkers that these annotations are real; this gate is
+        # what keeps that promise true.
+        run: uv run mypy

CHANGELOG.md

@@ -11,6 +11,15 @@ Work on the current development branch. Entries move to a versioned section on
 release.
 
 ### Added
+- Type annotations across the entire shipped package, and `mypy` promoted to a
+  blocking CI gate (`disallow_untyped_defs`). The package has always shipped a
+  `py.typed` marker, which tells downstream type-checkers the inline annotations
+  are real; previously only 1 of ~114 definitions was annotated, so that marker
+  was a false promise. It is now enforced.
+- `HostOps` protocol (`_host.py`) making the contract between `HackRF` and the
+  command mixins explicit and type-checkable. The mixins call ~27 methods on
+  `self` that live on the host class; that dependency was previously implicit in
+  a comment. Runtime composition and MRO are unchanged.
 - Continuous integration: GitHub Actions workflow running the test suite across
   Windows, Linux, and macOS on Python 3.11-3.13, plus a ruff + mypy lint job.
 - CLI test suite covering argument parsing, the mode state file, preset
@@ -42,6 +51,17 @@ release.
   the codebase made lint-clean so the CI lint job is meaningful.
 
 ### Fixed
+- The `atexit` backstop never reaped an orphaned `PersistentReceiver`.
+  `PersistentReceiver` registers itself in the live-handle registry, whose
+  shutdown hook calls `if h.is_alive(): h.stop()` -- but `is_alive()` did not
+  exist on the class, so the resulting `AttributeError` was swallowed by the
+  hook's bare `except Exception` and the receiver was silently left running.
+  Found by the typing pass; `is_alive()` added and pinned with a regression test.
+- Reading from a closed `PersistentReceiver` raised a bare
+  `TypeError: 'NoneType' object is not iterable` instead of a typed error; it now
+  raises `HackRFDeviceError` with an actionable message.
+- `sweep_stream(..., print_cmd=True)` would hand `StreamCtx` a `None` and crash;
+  it now raises `HackRFValueError` pointing at `sweep(..., print_cmd=True)`.
 - Unbounded memory growth on long-lived RX/TX handles: `_Process` drained child
   stdout/stderr into lists that were never trimmed, so an open-ended capture or
   repeat transmit retained every per-second stats line for the life of the

hackrfpy/pyproject.toml

@@ -54,6 +54,20 @@ markers = [
 ]
 testpaths = ["tests"]
 
+[tool.mypy]
+python_version = "3.11"
+files = ["src/hackrfpy"]
+# The point of this config: the package ships a py.typed marker, which promises
+# downstream type-checkers that the inline annotations are real. These settings
+# are what make that promise true -- every def in the shipped package must be
+# annotated, or CI fails.
+disallow_untyped_defs = true
+disallow_incomplete_defs = true
+check_untyped_defs = true
+no_implicit_optional = true
+warn_redundant_casts = true
+warn_unused_ignores = true
+
 [tool.ruff]
 # Match the codebase's existing style rather than reformatting it.
 line-length = 100

hackrfpy/src/hackrfpy/_commands/capture.py

@@ -18,18 +18,30 @@
 #   Author(s): <you>
 ##--------------------------------------------------------------------\
 
+from __future__ import annotations
+
 import os
+from typing import TYPE_CHECKING, Any, Callable, Generator
+
+import numpy as np
 
 from .. import constants as C
+from .._host import HostOps
 from ..sigmf import write_sigmf_meta
 
+if TYPE_CHECKING:
+    from .._receiver import PersistentReceiver
+    from .._stream_ctx import StreamCtx
+
 
-class CaptureMixin:
-    def capture(self, freq, sample_rate, *, out="capture.iq",
-                num_samples=None, duration=None,
-                lna=16, vga=20, amp=False, bias_tee=False,
-                baseband_bw=None, to_stdout=False, sigmf=True,
-                segment_secs=None, print_cmd=False):
+class CaptureMixin(HostOps):
+    def capture(self, freq: float, sample_rate: float, *,
+                out: str = "capture.iq", num_samples: int | None = None,
+                duration: float | None = None, lna: int = 16, vga: int = 20,
+                amp: bool = False, bias_tee: bool = False,
+                baseband_bw: float | None = None, to_stdout: bool = False,
+                sigmf: bool = True, segment_secs: float | None = None,
+                print_cmd: bool = False) -> Any:
         self.require_mode(C.MODE_RX)
         freq, sample_rate, lna, vga = self.validate_rx(freq, sample_rate, lna, vga)
         bw = self._auto_baseband(sample_rate, baseband_bw)
@@ -57,7 +69,7 @@ def capture(self, freq, sample_rate, *, out="capture.iq",
             # The inner stream generator is closed EXPLICITLY on the way out;
             # leaving it to GC can orphan a receiving hackrf_transfer after
             # the caller breaks out of the loop.
-            def _gen():
+            def _gen() -> Generator[np.ndarray, None, None]:   # decoded, not raw
                 tail = b""
                 inner = self._run(argv, mode="stream")
                 try:
@@ -97,17 +109,21 @@ def _gen():
         return res
 
     # ---- aliases ----
-    def rx(self, *a, **k):
+    def rx(self, *a: Any, **k: Any) -> Any:
         return self.capture(*a, **k)
 
-    def capture_samples(self, freq, sample_rate, num_samples, **k):
+    def capture_samples(self, freq: float, sample_rate: float,
+                        num_samples: int, **k: Any) -> Any:
         return self.capture(freq, sample_rate, num_samples=num_samples, **k)
 
-    def capture_seconds(self, freq, sample_rate, duration, **k):
+    def capture_seconds(self, freq: float, sample_rate: float,
+                        duration: float, **k: Any) -> Any:
         return self.capture(freq, sample_rate, duration=duration, **k)
 
-    def scan_frequencies(self, freqs, sample_rate, num_samples, *,
-                         on_capture=None, **k):
+    def scan_frequencies(self, freqs: list[float], sample_rate: float,
+                         num_samples: int, *,
+                         on_capture: Callable[..., Any] | None = None,
+                         **k: Any) -> dict[float, np.ndarray] | None:
         # Sequentially capture a fixed sample count at each frequency in
         # `freqs`, retuning between them. This does NOT close the gapless-
         # retune gap (each retune is a fresh hackrf_transfer with a short
@@ -130,8 +146,8 @@ def scan_frequencies(self, freqs, sample_rate, num_samples, *,
         return results if on_capture is None else None
 
     # ---- in-memory + context-managed entry points ----
-    def capture_array(self, freq, sample_rate, num_samples, *,
-                      return_params=False, **k):
+    def capture_array(self, freq: float, sample_rate: float, num_samples: int,
+                      *, return_params: bool = False, **k: Any) -> Any:
         # Scripting entry point: return EXACTLY num_samples complex64 samples
         # in RAM, no file. Built on the stdout-stream path so it shares the
         # odd-byte carry + clean-reap logic. The stream is closed as soon as
@@ -164,8 +180,10 @@ def capture_array(self, freq, sample_rate, num_samples, *,
             return iq, self.last_params
         return iq
 
-    def open_receiver(self, freq, sample_rate, *, lna=16, vga=20, amp=False,
-                      baseband_bw=None, read_samples=131072):
+    def open_receiver(self, freq: float, sample_rate: float, *, lna: int = 16,
+                      vga: int = 20, amp: bool = False,
+                      baseband_bw: float | None = None,
+                      read_samples: int = 131072) -> PersistentReceiver:
         # Open a PERSISTENT fixed-frequency receiver: one long-lived
         # hackrf_transfer you drain in segments over time, so you don't pay the
         # ~1-2 s process spin-up per capture. Use as a context manager:
@@ -184,7 +202,8 @@ def open_receiver(self, freq, sample_rate, *, lna=16, vga=20, amp=False,
                                   amp=amp, baseband_bw=baseband_bw,
                                   read_samples=read_samples)
 
-    def capture_stream(self, freq, sample_rate, **k):
+    def capture_stream(self, freq: float, sample_rate: float,
+                       **k: Any) -> StreamCtx:
         # Context manager wrapping the live stdout stream so the receiving
         # hackrf_transfer is ALWAYS reaped on exit, even on exception:
         #     with h.capture_stream(433.92e6, 8e6) as blocks:
@@ -194,8 +213,10 @@ def capture_stream(self, freq, sample_rate, **k):
         gen = self.capture(freq, sample_rate, to_stdout=True, **k)
         return StreamCtx(gen)
 
-    def capture_callback(self, freq, sample_rate, on_block, *,
-                         max_samples=None, max_blocks=None, **k):
+    def capture_callback(self, freq: float, sample_rate: float,
+                         on_block: Callable[..., Any], *,
+                         max_samples: int | None = None,
+                         max_blocks: int | None = None, **k: Any) -> int:
         # Binder-style ergonomics over the subprocess stream: instead of the
         # caller writing the receive loop, register a callback that fires with
         # each decoded complex64 block as it arrives. This is the usability
@@ -228,8 +249,9 @@ def capture_callback(self, freq, sample_rate, on_block, *,
         return total
 
     # ---- internals ----
-    def _rx_argv(self, freq, sample_rate, target, lna, vga, amp, bias_tee,
-                 bw, num_samples):
+    def _rx_argv(self, freq: float, sample_rate: float, target: str, lna: int,
+                 vga: int, amp: bool, bias_tee: bool, bw: float,
+                 num_samples: int | None) -> list[Any]:
         argv = ["transfer", "-r", target,
                 "-f", int(freq), "-s", int(sample_rate),
                 "-l", lna, "-g", vga,
@@ -241,8 +263,10 @@ def _rx_argv(self, freq, sample_rate, target, lna, vga, amp, bias_tee,
             argv += ["-n", int(num_samples)]
         return argv
 
-    def _capture_segmented(self, freq, sample_rate, out, segment_secs, lna,
-                           vga, amp, bias_tee, bw, sigmf, print_cmd):
+    def _capture_segmented(self, freq: float, sample_rate: float, out: str,
+                           segment_secs: float, lna: int, vga: int, amp: bool,
+                           bias_tee: bool, bw: float, sigmf: bool,
+                           print_cmd: bool) -> list[str]:
         # Rolling files out_000.iq, out_001.iq, ... each `segment_secs` long.
         # Each segment is a bounded blocking capture, so files are whole.
         base, ext = os.path.splitext(out)

hackrfpy/src/hackrfpy/_commands/device.py

@@ -12,29 +12,34 @@
 #   Author(s): <you>
 ##--------------------------------------------------------------------\
 
+from __future__ import annotations
+
 import re
 import shutil
+from typing import Any
 
 from .. import constants as C
+from .._host import HostOps
 from ..exceptions import HackRFDeviceError, HackRFValueError
 
 
-class DeviceMixin:
+class DeviceMixin(HostOps):
     # ---- clock ----
-    def clock(self, *args, print_cmd=False):
+    def clock(self, *args: Any, print_cmd: bool = False) -> Any:
         argv = ["clock"] + list(args)
         return self._run(argv, mode="blocking", text=True, print_cmd=print_cmd)
 
     # ---- operacake antenna switch ----
-    def operacake(self, *args, print_cmd=False):
+    def operacake(self, *args: Any, print_cmd: bool = False) -> Any:
         argv = ["operacake"] + list(args)
         return self._run(argv, mode="blocking", text=True, print_cmd=print_cmd)
 
-    def operacake_list(self):
+    def operacake_list(self) -> Any:
         return self.operacake("-l")
 
     # ---- cpld jtag (CPLD firmware) ----
-    def cpldjtag(self, firmware, *, confirm=False, print_cmd=False):
+    def cpldjtag(self, firmware: str, *, confirm: bool = False,
+                 print_cmd: bool = False) -> Any:
         if not confirm and not print_cmd:
             raise HackRFValueError(
                 "cpldjtag flashes the CPLD and can brick the board. "
@@ -43,7 +48,8 @@ def cpldjtag(self, firmware, *, confirm=False, print_cmd=False):
                          text=True, print_cmd=print_cmd)
 
     # ---- spiflash (firmware) ----
-    def spiflash_write(self, firmware, *, confirm=False, print_cmd=False):
+    def spiflash_write(self, firmware: str, *, confirm: bool = False,
+                       print_cmd: bool = False) -> Any:
         # Writing firmware. The single most dangerous operation in the library.
         if not confirm and not print_cmd:
             raise HackRFValueError(
@@ -53,30 +59,31 @@ def spiflash_write(self, firmware, *, confirm=False, print_cmd=False):
         return self._run(["spiflash", "-w", firmware], mode="blocking",
                          text=True, print_cmd=print_cmd)
 
-    def spiflash_read(self, out, length=None, print_cmd=False):
-        argv = ["spiflash", "-r", out]
+    def spiflash_read(self, out: str, length: int | None = None,
+                      print_cmd: bool = False) -> Any:
+        argv: list[Any] = ["spiflash", "-r", out]
         if length is not None:
             argv += ["-l", int(length)]
         return self._run(argv, mode="blocking", text=True, print_cmd=print_cmd)
 
-    def spiflash_reset(self, print_cmd=False):
+    def spiflash_reset(self, print_cmd: bool = False) -> Any:
         return self._run(["spiflash", "-R"], mode="blocking", text=True,
                          print_cmd=print_cmd)
 
     # ---- debug register access ----
-    def debug(self, *args, print_cmd=False):
+    def debug(self, *args: Any, print_cmd: bool = False) -> Any:
         argv = ["debug"] + list(args)
         return self._run(argv, mode="blocking", text=True, print_cmd=print_cmd)
 
     # =================================================================
     # preflight: environment / readiness check
     # (was 'doctor' -- kept as an alias below for the familiar CLI verb)
     # =================================================================
-    def preflight(self, capture_path="."):
+    def preflight(self, capture_path: str = ".") -> dict[str, Any]:
         # Checks tooling, board presence, free disk, and reports the active
         # mode. Returns a structured report; raises only on hard environment
         # failures the user must fix.
-        report = {"tools": {}, "boards": [], "mode": self.mode,
+        report: dict[str, Any] = {"tools": {}, "boards": [], "mode": self.mode,
                   "tool_version": None, "disk_free_bytes": None,
                   "features": {}, "problems": []}
 
@@ -133,10 +140,11 @@ def preflight(self, capture_path="."):
 
     # doctor: familiar alias for preflight (brew/flutter-style verb). The CLI
     # still exposes `hrf doctor`; the honest method name is preflight().
-    def doctor(self, capture_path="."):
+    def doctor(self, capture_path: str = ".") -> dict[str, Any]:
         return self.preflight(capture_path=capture_path)
 
-    def features(self, tool_version=None, firmware_version=None):
+    def features(self, tool_version: str | None = None,
+                 firmware_version: str | None = None) -> dict[str, Any]:
         # Map version strings -> capability flags. The reliable year signal is
         # the FIRMWARE version (e.g. "2024.02.1"); the tools version is often a
         # git tag (e.g. "git-b1dbb47") with no parseable year. If nothing is
@@ -157,7 +165,7 @@ def features(self, tool_version=None, firmware_version=None):
         # Prefer the firmware version's year; fall back to the tools version.
         # A "git-" build is a from-source build, which is current by
         # definition -- treat it as modern rather than merely "unknown".
-        def _year(v):
+        def _year(v: str | None) -> int | None:
             m = re.match(r"(\d{4})", v or "")
             return int(m.group(1)) if m else None
 
@@ -177,7 +185,7 @@ def _year(v):
             "bias_tee": is_git or year is None or year >= 2018,  # -p
         }
 
-    def _print_preflight(self, r):
+    def _print_preflight(self, r: dict[str, Any]) -> None:
         print("hackrfpy preflight")
         print("-" * 40)
         for name, path in r["tools"].items():