From 4abc95c3721cd91ec553dc0e8028ef182364898c Mon Sep 17 00:00:00 2001 From: Ayzen Date: Wed, 1 Apr 2026 13:00:01 +0300 Subject: [PATCH] added generator mode --- .codex | 0 python_app/generator_sweep/__init__.py | 31 +++ python_app/generator_sweep/config.py | 202 +++++++++++++++ python_app/generator_sweep/pwm.py | 92 +++++++ python_app/generator_sweep/runner.py | 208 ++++++++++++++++ .../hardware_full/librevna_driver/__init__.py | 2 + .../librevna_driver/api/__init__.py | 2 + .../librevna_driver/api/generator.py | 65 +++++ .../hardware_full/librevna_driver/device.py | 2 + .../hardware_full/librevna_driver/models.py | 17 ++ .../librevna_driver/protocol/__init__.py | 2 + .../librevna_driver/protocol/codec.py | 15 ++ .../scripts/convert_legacy_preprocess_sets.py | 231 ++++++++++++++++++ .../scripts/librevna_generator_sweep.py | 35 +++ .../librevna_generator_sweep_config.py | 31 +++ requirements.txt | 1 + 16 files changed, 936 insertions(+) create mode 100644 .codex create mode 100644 python_app/generator_sweep/__init__.py create mode 100644 python_app/generator_sweep/config.py create mode 100644 python_app/generator_sweep/pwm.py create mode 100644 python_app/generator_sweep/runner.py create mode 100644 python_app/hardware_full/librevna_driver/api/generator.py create mode 100644 python_app/scripts/convert_legacy_preprocess_sets.py create mode 100644 python_app/scripts/librevna_generator_sweep.py create mode 100644 python_app/scripts/librevna_generator_sweep_config.py diff --git a/.codex b/.codex new file mode 100644 index 0000000..e69de29 diff --git a/python_app/generator_sweep/__init__.py b/python_app/generator_sweep/__init__.py new file mode 100644 index 0000000..aaeb340 --- /dev/null +++ b/python_app/generator_sweep/__init__.py @@ -0,0 +1,31 @@ +"""Standalone LibreVNA signal-generator sweep support.""" + +__all__ = [ + "GeneratorSweepConfig", + "GeneratorSweepRunner", + "HardwarePwmGate", + "ResolvedGeneratorSweepConfig", + "resolve_generator_sweep_config", +] + + +def __getattr__(name: str) -> object: + """Load generator sweep helpers lazily to avoid heavy hardware imports.""" + if name in {"GeneratorSweepConfig", "ResolvedGeneratorSweepConfig", "resolve_generator_sweep_config"}: + from .config import GeneratorSweepConfig, ResolvedGeneratorSweepConfig, resolve_generator_sweep_config + + exports = { + "GeneratorSweepConfig": GeneratorSweepConfig, + "ResolvedGeneratorSweepConfig": ResolvedGeneratorSweepConfig, + "resolve_generator_sweep_config": resolve_generator_sweep_config, + } + return exports[name] + if name == "HardwarePwmGate": + from .pwm import HardwarePwmGate + + return HardwarePwmGate + if name == "GeneratorSweepRunner": + from .runner import GeneratorSweepRunner + + return GeneratorSweepRunner + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/python_app/generator_sweep/config.py b/python_app/generator_sweep/config.py new file mode 100644 index 0000000..3e43687 --- /dev/null +++ b/python_app/generator_sweep/config.py @@ -0,0 +1,202 @@ +"""Configuration resolution for the standalone generator sweep.""" + +from __future__ import annotations + +from dataclasses import dataclass + +@dataclass(slots=True, frozen=True) +class GeneratorSweepConfig: + """User-edited generator sweep configuration stored next to the script.""" + + log_level: str = "INFO" + serial: str | None = None + strict_protocol_version: int = 14 + start_hz: float = 1_000_000.0 + stop_hz: float = 6_000_000.0 + step_hz: float = 0.0 + points: int = 0 + hold_time_ms: float = 50.0 + loop: bool = True + port: int = 1 + power_dbm: float = -10.0 + amplitude_correction: bool = False + settle_timeout_ms: int = 1_000 + status_poll_ms: float = 10.0 + post_lock_delay_us: int = 0 + gpio_chip: str = "/dev/gpiochip0" + sweep_start_pin: int = 5 + curr_step_pin: int = 6 + curr_step_initial_level: int = 0 + pwm_pin: int = 12 + pwm_frequency_hz: int = 2_000_000 + pwm_duty_cycle: float = 0.5 + + +@dataclass(slots=True, frozen=True) +class ResolvedGeneratorSweepConfig: + """Fully validated generator sweep configuration.""" + + log_level: int + serial: str | None + strict_protocol_version: int + frequencies_hz: tuple[int, ...] + hold_time_s: float + loop: bool + port: int + power_dbm: float + amplitude_correction: bool + settle_timeout_s: float + status_poll_s: float + post_lock_delay_s: float + gpio_chip: str + sweep_start_pin: int + curr_step_pin: int + curr_step_initial_level: int + pwm_pin: int + pwm_frequency_hz: int + pwm_duty_cycle: float + + +def _require(condition: bool, message: str) -> None: + """Raise ``ValueError`` when a configuration invariant is violated.""" + if not condition: + raise ValueError(message) + + +def _validate_positive_int(value: int, label: str) -> None: + """Validate positive integer setting.""" + _require(value > 0, f"{label} must be > 0") + + +def _validate_non_negative_int(value: int, label: str) -> None: + """Validate non-negative integer setting.""" + _require(value >= 0, f"{label} must be >= 0") + + +def _validate_non_negative_float(value: float, label: str) -> None: + """Validate non-negative float setting.""" + _require(value >= 0.0, f"{label} must be >= 0") + + +def _validate_positive_float(value: float, label: str) -> None: + """Validate positive float setting.""" + _require(value > 0.0, f"{label} must be > 0") + + +def _resolve_log_level(value: str) -> int: + """Normalize logging level name to ``logging`` module integer constant.""" + import logging + + normalized = value.strip().upper() + if not normalized: + raise ValueError("generator_sweep.log_level must not be empty") + + level = logging.getLevelName(normalized) + if isinstance(level, int): + return level + raise ValueError("generator_sweep.log_level must be one of DEBUG, INFO, WARNING, ERROR, CRITICAL") + + +def _build_frequency_grid(model: GeneratorSweepConfig) -> tuple[int, ...]: + """Build the ordered frequency grid with inclusive stop frequency.""" + start_hz = int(round(model.start_hz)) + stop_hz = int(round(model.stop_hz)) + step_hz = int(round(model.step_hz)) + points = int(model.points) + + _validate_positive_int(start_hz, "generator_sweep.start_hz") + _validate_positive_int(stop_hz, "generator_sweep.stop_hz") + _require(stop_hz >= start_hz, "generator_sweep.stop_hz must be >= generator_sweep.start_hz") + + has_step = step_hz > 0 + has_points = points > 0 + _require(has_step != has_points, "generator_sweep must define exactly one of step_hz or points") + + if start_hz == stop_hz: + if has_points: + _require(points == 1, "generator_sweep.points must be 1 when start_hz == stop_hz") + return (start_hz,) + + if has_step: + _validate_positive_int(step_hz, "generator_sweep.step_hz") + frequencies: list[int] = [] + current_hz = start_hz + while current_hz < stop_hz: + frequencies.append(current_hz) + current_hz += step_hz + if not frequencies or frequencies[-1] != stop_hz: + frequencies.append(stop_hz) + return tuple(frequencies) + + _validate_positive_int(points, "generator_sweep.points") + _require(points >= 2, "generator_sweep.points must be >= 2 when start_hz != stop_hz") + span_hz = stop_hz - start_hz + frequencies = tuple( + int(round(start_hz + (span_hz * index) / float(points - 1))) + for index in range(points) + ) + _require(frequencies[0] == start_hz, "generator_sweep.points grid does not start at start_hz") + _require(frequencies[-1] == stop_hz, "generator_sweep.points grid does not end at stop_hz") + _require( + all(left < right for left, right in zip(frequencies, frequencies[1:])), + "generator sweep grid contains duplicate or non-monotonic frequencies", + ) + return frequencies + + +def resolve_generator_sweep_config(model: GeneratorSweepConfig) -> ResolvedGeneratorSweepConfig: + """Validate and resolve runtime configuration for generator sweep mode.""" + log_level = _resolve_log_level(model.log_level) + frequencies_hz = _build_frequency_grid(model) + hold_time_ms = float(model.hold_time_ms) + settle_timeout_ms = int(model.settle_timeout_ms) + status_poll_ms = float(model.status_poll_ms) + post_lock_delay_us = int(model.post_lock_delay_us) + pwm_frequency_hz = int(model.pwm_frequency_hz) + sweep_start_pin = int(model.sweep_start_pin) + curr_step_pin = int(model.curr_step_pin) + pwm_pin = int(model.pwm_pin) + curr_step_initial_level = int(model.curr_step_initial_level) + + _validate_non_negative_float(hold_time_ms, "generator_sweep.hold_time_ms") + _validate_positive_int(settle_timeout_ms, "generator_sweep.settle_timeout_ms") + _validate_positive_float(status_poll_ms, "generator_sweep.status_poll_ms") + _validate_non_negative_int(post_lock_delay_us, "generator_sweep.post_lock_delay_us") + _validate_positive_int(pwm_frequency_hz, "generator_sweep.pwm_frequency_hz") + _validate_positive_int(int(model.strict_protocol_version), "generator_sweep.strict_protocol_version") + _require(model.port in {1, 2}, "generator_sweep.port must be 1 or 2") + _require(curr_step_initial_level in {0, 1}, "generator_sweep.curr_step_initial_level must be 0 or 1") + _require(bool(model.gpio_chip.strip()), "generator_sweep.gpio_chip must not be empty") + _validate_non_negative_int(sweep_start_pin, "generator_sweep.sweep_start_pin") + _validate_non_negative_int(curr_step_pin, "generator_sweep.curr_step_pin") + _validate_non_negative_int(pwm_pin, "generator_sweep.pwm_pin") + _require( + len({sweep_start_pin, curr_step_pin, pwm_pin}) == 3, + "generator_sweep sweep_start_pin, curr_step_pin, and pwm_pin must be distinct", + ) + _require( + 0.0 < float(model.pwm_duty_cycle) <= 1.0, + "generator_sweep.pwm_duty_cycle must be within (0, 1]", + ) + + return ResolvedGeneratorSweepConfig( + log_level=log_level, + serial=(model.serial.strip() or None) if isinstance(model.serial, str) else model.serial, + strict_protocol_version=int(model.strict_protocol_version), + frequencies_hz=frequencies_hz, + hold_time_s=hold_time_ms / 1000.0, + loop=bool(model.loop), + port=int(model.port), + power_dbm=float(model.power_dbm), + amplitude_correction=bool(model.amplitude_correction), + settle_timeout_s=settle_timeout_ms / 1000.0, + status_poll_s=status_poll_ms / 1000.0, + post_lock_delay_s=post_lock_delay_us / 1_000_000.0, + gpio_chip=model.gpio_chip.strip(), + sweep_start_pin=sweep_start_pin, + curr_step_pin=curr_step_pin, + curr_step_initial_level=curr_step_initial_level, + pwm_pin=pwm_pin, + pwm_frequency_hz=pwm_frequency_hz, + pwm_duty_cycle=float(model.pwm_duty_cycle), + ) diff --git a/python_app/generator_sweep/pwm.py b/python_app/generator_sweep/pwm.py new file mode 100644 index 0000000..67bb52d --- /dev/null +++ b/python_app/generator_sweep/pwm.py @@ -0,0 +1,92 @@ +"""Hardware-PWM backend for Raspberry Pi 5 generator sweep gating.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path + + +_PWM_CHANNEL_BY_PIN = { + 12: 0, + 13: 1, + 18: 2, + 19: 3, +} + + +@dataclass(slots=True) +class HardwarePwmGate: + """Thin wrapper around Raspberry Pi hardware PWM output.""" + + pin: int + frequency_hz: int + duty_cycle: float + _pwm: object | None = field(init=False, default=None, repr=False) + _running: bool = field(init=False, default=False, repr=False) + + def __post_init__(self) -> None: + """Validate static PWM configuration.""" + if self.pin not in _PWM_CHANNEL_BY_PIN: + raise ValueError( + "generator_sweep.pwm_pin must be one of GPIO12, GPIO13, GPIO18, or GPIO19 on Raspberry Pi 5" + ) + if self.frequency_hz <= 0: + raise ValueError("generator_sweep.pwm_frequency_hz must be > 0") + if not (0.0 < self.duty_cycle <= 1.0): + raise ValueError("generator_sweep.pwm_duty_cycle must be within (0, 1]") + + def open(self) -> None: + """Initialize hardware PWM handle.""" + if self._pwm is not None: + return + + try: + from rpi_hardware_pwm import HardwarePWM + except ImportError as exc: + raise RuntimeError( + "rpi_hardware_pwm is required for generator_sweep PWM output. " + "Install project dependencies on the Raspberry Pi before running this script." + ) from exc + + channel = _PWM_CHANNEL_BY_PIN[self.pin] + try: + self._pwm = HardwarePWM(pwm_channel=channel, hz=self.frequency_hz, chip=0) + except Exception as exc: + config_hint = _boot_config_hint() + raise RuntimeError( + "Failed to initialize Raspberry Pi hardware PWM. " + f"Enable the PWM overlay in {config_hint} by adding " + "'dtoverlay=pwm-2chan', then reboot the Raspberry Pi." + ) from exc + self._running = False + + def enable(self) -> None: + """Start PWM output when not already running.""" + if self._pwm is None: + raise RuntimeError("Hardware PWM gate is not open") + if self._running: + return + + duty_percent = self.duty_cycle * 100.0 + self._pwm.start(duty_percent) + self._running = True + + def disable(self) -> None: + """Stop PWM output when active.""" + if self._pwm is None or not self._running: + return + self._pwm.stop() + self._running = False + + def close(self) -> None: + """Stop PWM and release runtime state.""" + self.disable() + self._pwm = None + + +def _boot_config_hint() -> str: + """Return the most likely active Raspberry Pi boot config path.""" + firmware_config = Path("/boot/firmware/config.txt") + if firmware_config.exists(): + return str(firmware_config) + return "/boot/config.txt" diff --git a/python_app/generator_sweep/runner.py b/python_app/generator_sweep/runner.py new file mode 100644 index 0000000..529ad3b --- /dev/null +++ b/python_app/generator_sweep/runner.py @@ -0,0 +1,208 @@ +"""Orchestration for standalone LibreVNA signal-generator sweeps.""" + +from __future__ import annotations + +from dataclasses import replace +import logging +import time + +from python_app.generator_sweep.config import ( + GeneratorSweepConfig, + ResolvedGeneratorSweepConfig, + resolve_generator_sweep_config, +) +from python_app.generator_sweep.pwm import HardwarePwmGate +from python_app.hardware_full.librevna_driver import GeneratorSettings, LibreVNADevice +from python_app.hardware_full.switch_drivers.gpio_uapi import GpioOutputLines + +logger = logging.getLogger(__name__) + + +class _MarkerOutputs: + """Own two GPIO marker signals used during generator sweeps.""" + + def __init__( + self, + *, + gpio_chip: str, + sweep_start_pin: int, + curr_step_pin: int, + curr_step_initial_level: int, + ) -> None: + """Store GPIO metadata and initialize marker state.""" + self._lines = GpioOutputLines( + chip=gpio_chip, + offsets=[sweep_start_pin, curr_step_pin], + consumer="generator_sweep", + ) + self._curr_step_initial_level = curr_step_initial_level + self._sweep_start_level = 0 + self._curr_step_level = curr_step_initial_level + self._is_open = False + + def open(self) -> None: + """Request GPIO lines and apply idle levels.""" + self._lines.open() + self._is_open = True + self.reset_to_idle() + + def close(self) -> None: + """Release GPIO lines.""" + self._is_open = False + self._lines.close() + + def begin_sweep(self) -> None: + """Raise sweep marker while preserving current step level.""" + time.sleep(0.001) + self._sweep_start_level = 1 + self._apply() + + def end_sweep(self) -> None: + """Drop sweep marker after a completed pass.""" + self._sweep_start_level = 0 + self._apply() + + def toggle_step(self) -> None: + """Invert current-step marker.""" + self._curr_step_level = 1 - self._curr_step_level + self._apply() + + def reset_to_idle(self) -> None: + """Return both markers to their configured idle state.""" + self._sweep_start_level = 0 + self._curr_step_level = self._curr_step_initial_level + self._apply() + + def _apply(self) -> None: + """Write current marker levels to GPIO.""" + if not self._is_open: + return + self._lines.set_values([self._sweep_start_level, self._curr_step_level]) + + +class GeneratorSweepRunner: + """Execute a signal-generator frequency sweep defined in run config.""" + + def __init__(self, config: GeneratorSweepConfig) -> None: + """Resolve configuration and prepare runtime helpers.""" + self._config: ResolvedGeneratorSweepConfig = resolve_generator_sweep_config(config) + self._device: LibreVNADevice | None = None + self._markers = _MarkerOutputs( + gpio_chip=self._config.gpio_chip, + sweep_start_pin=self._config.sweep_start_pin, + curr_step_pin=self._config.curr_step_pin, + curr_step_initial_level=self._config.curr_step_initial_level, + ) + self._pwm = HardwarePwmGate( + pin=self._config.pwm_pin, + frequency_hz=self._config.pwm_frequency_hz, + duty_cycle=self._config.pwm_duty_cycle, + ) + self._base_generator_settings = GeneratorSettings( + frequency_hz=float(self._config.frequencies_hz[0]), + power_dbm=self._config.power_dbm, + active_port=self._config.port, + apply_amplitude_correction=self._config.amplitude_correction, + ) + + def run(self) -> None: + """Open resources and execute one-shot or looping sweeps.""" + try: + self._open() + while True: + self._run_single_sweep() + if not self._config.loop: + break + except KeyboardInterrupt: + logger.info("Generator sweep interrupted by user") + finally: + self._close() + + def _open(self) -> None: + """Open GPIO, PWM, and LibreVNA device connection.""" + self._markers.open() + self._pwm.open() + self._device = LibreVNADevice() + self._device.connect( + serial=self._config.serial, + strict_protocol_version=self._config.strict_protocol_version, + timeout_s=2.0, + ) + self._device.generator.set_idle(timeout_s=1.0) + + logger.info( + "Generator sweep ready: points=%d start=%dHz stop=%dHz loop=%s", + len(self._config.frequencies_hz), + self._config.frequencies_hz[0], + self._config.frequencies_hz[-1], + self._config.loop, + ) + + def _close(self) -> None: + """Best-effort cleanup of PWM, GPIO, and device state.""" + self._pwm.disable() + + device = self._device + if device is not None: + try: + if device.is_connected: + device.generator.set_idle(timeout_s=1.0) + except Exception as exc: # noqa: BLE001 + logger.warning("Failed to set LibreVNA idle mode during shutdown: %s", exc) + finally: + device.disconnect() + self._device = None + + try: + self._markers.reset_to_idle() + finally: + self._markers.close() + self._pwm.close() + + def _run_single_sweep(self) -> None: + """Execute one pass from the first configured frequency to the last.""" + self._markers.begin_sweep() + try: + for index, frequency_hz in enumerate(self._config.frequencies_hz): + self._pwm.disable() + self._set_generator_frequency(frequency_hz) + if index > 0: + self._markers.toggle_step() + self._wait_for_generator_ready() + self._emit_pwm_window() + finally: + self._pwm.disable() + self._markers.end_sweep() + + def _set_generator_frequency(self, frequency_hz: int) -> None: + """Transmit next generator frequency to LibreVNA.""" + device = self._require_device() + settings = replace(self._base_generator_settings, frequency_hz=float(frequency_hz)) + device.generator.configure(settings, timeout_s=max(self._config.settle_timeout_s, 1.0)) + logger.debug("Generator frequency set to %d Hz", frequency_hz) + + def _wait_for_generator_ready(self) -> None: + """Wait until generator lock telemetry reports ready.""" + device = self._require_device() + device.generator.wait_until_ready( + timeout_s=self._config.settle_timeout_s, + poll_interval_s=self._config.status_poll_s, + ) + if self._config.post_lock_delay_s > 0.0: + time.sleep(self._config.post_lock_delay_s) + + def _emit_pwm_window(self) -> None: + """Enable PWM only during the configured hold window.""" + if self._config.hold_time_s <= 0.0: + return + self._pwm.enable() + try: + time.sleep(self._config.hold_time_s) + finally: + self._pwm.disable() + + def _require_device(self) -> LibreVNADevice: + """Return connected device or fail fast if runner is not open.""" + if self._device is None: + raise RuntimeError("Generator sweep runner is not open") + return self._device diff --git a/python_app/hardware_full/librevna_driver/__init__.py b/python_app/hardware_full/librevna_driver/__init__.py index 118490e..6263112 100644 --- a/python_app/hardware_full/librevna_driver/__init__.py +++ b/python_app/hardware_full/librevna_driver/__init__.py @@ -28,6 +28,7 @@ from .models import ( DeviceInfo, DeviceLimits, DeviceStatus, + GeneratorSettings, Packet, StreamHandle, SweepResult, @@ -45,6 +46,7 @@ __all__ = [ "DeviceInfo", "DeviceLimits", "DeviceStatus", + "GeneratorSettings", "HardwareFamily", "IncompleteSweepError", "LibreVNADevice", diff --git a/python_app/hardware_full/librevna_driver/api/__init__.py b/python_app/hardware_full/librevna_driver/api/__init__.py index 56ea479..550f82a 100644 --- a/python_app/hardware_full/librevna_driver/api/__init__.py +++ b/python_app/hardware_full/librevna_driver/api/__init__.py @@ -1,9 +1,11 @@ """Public controller classes.""" from .config import ConfigController +from .generator import GeneratorController from .vna import VNAController __all__ = [ "ConfigController", + "GeneratorController", "VNAController", ] diff --git a/python_app/hardware_full/librevna_driver/api/generator.py b/python_app/hardware_full/librevna_driver/api/generator.py new file mode 100644 index 0000000..9672838 --- /dev/null +++ b/python_app/hardware_full/librevna_driver/api/generator.py @@ -0,0 +1,65 @@ +"""Signal-generator controller for direct protocol packets.""" + +from __future__ import annotations + +import logging +import time + +from ..enums import PacketType +from ..exceptions import TimeoutError +from ..models import DeviceStatus, GeneratorSettings, Packet +from ..session import LibreVNASession + +logger = logging.getLogger(__name__) + + +class GeneratorController: + """Signal-generator operations built on direct packet protocol.""" + + def __init__(self, session: LibreVNASession) -> None: + """Bind generator controller to active session.""" + self._session = session + self._settings: GeneratorSettings | None = None + + def configure(self, settings: GeneratorSettings, *, timeout_s: float = 1.0) -> None: + """Configure generator mode and active CW output.""" + self._settings = settings + self._session.send(Packet(PacketType.GENERATOR, settings), require_ack=True, timeout_s=timeout_s) + logger.info( + "Generator configured: freq=%.3fHz power=%.2fdBm port=%d correction=%s", + settings.frequency_hz, + settings.power_dbm, + settings.active_port, + settings.apply_amplitude_correction, + ) + + def set_idle(self, *, timeout_s: float = 1.0) -> None: + """Return device to idle mode and stop generator output.""" + logger.info("Setting LibreVNA idle mode") + self._session.send(Packet(PacketType.SET_IDLE), require_ack=True, timeout_s=timeout_s) + + def wait_until_ready( + self, + *, + timeout_s: float, + poll_interval_s: float, + ) -> DeviceStatus: + """Wait until available lock flags report the generator is ready.""" + + deadline = time.monotonic() + timeout_s + + while True: + status = self._session.get_device_status(timeout_s=min(timeout_s, 1.0)) + lock_values = [value for value in (status.source_locked, status.lo_locked) if value is not None] + if lock_values: + if all(lock_values): + return status + else: + raise RuntimeError( + f"Generator lock telemetry is unavailable for hardware family {status.family.name}" + ) + + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("Timed out waiting for LibreVNA generator lock") + time.sleep(min(poll_interval_s, remaining)) diff --git a/python_app/hardware_full/librevna_driver/device.py b/python_app/hardware_full/librevna_driver/device.py index bc8426c..0c9b2b2 100644 --- a/python_app/hardware_full/librevna_driver/device.py +++ b/python_app/hardware_full/librevna_driver/device.py @@ -6,6 +6,7 @@ import logging from types import TracebackType from .api.config import ConfigController +from .api.generator import GeneratorController from .api.vna import VNAController from .enums import PacketType from .models import DeviceInfo, DeviceStatus, Packet, USBDeviceDescriptor @@ -22,6 +23,7 @@ class LibreVNADevice: self._session = LibreVNASession() self.vna = VNAController(self._session) + self.generator = GeneratorController(self._session) self.config = ConfigController(self._session) def __enter__(self) -> LibreVNADevice: diff --git a/python_app/hardware_full/librevna_driver/models.py b/python_app/hardware_full/librevna_driver/models.py index d06cd69..8a0d56b 100644 --- a/python_app/hardware_full/librevna_driver/models.py +++ b/python_app/hardware_full/librevna_driver/models.py @@ -120,6 +120,23 @@ class VNASweepSettings: raise ValueError("power sweep requires f_start_hz == f_stop_hz") +@dataclass(slots=True) +class GeneratorSettings: + """Configuration for LibreVNA signal-generator mode.""" + + frequency_hz: float = 1_000_000.0 + power_dbm: float = -10.0 + active_port: int = 1 + apply_amplitude_correction: bool = False + + def __post_init__(self) -> None: + """Validate generator settings before transmission.""" + if self.frequency_hz <= 0: + raise ValueError("frequency_hz must be > 0") + if self.active_port not in {1, 2}: + raise ValueError("active_port must be 1 or 2") + + @dataclass(slots=True) class DeviceConfigVariant: """Family-specific device configuration fields.""" diff --git a/python_app/hardware_full/librevna_driver/protocol/__init__.py b/python_app/hardware_full/librevna_driver/protocol/__init__.py index f295c84..b36a73d 100644 --- a/python_app/hardware_full/librevna_driver/protocol/__init__.py +++ b/python_app/hardware_full/librevna_driver/protocol/__init__.py @@ -4,6 +4,7 @@ from .codec import ( decode_packet_payload, decode_vna_datapoint_payload, encode_device_config_payload, + encode_generator_settings_payload, encode_packet_payload, encode_sweep_settings_payload, ensure_no_payload_types, @@ -22,6 +23,7 @@ __all__ = [ "decode_vna_datapoint_payload", "encode_device_config_payload", "encode_frame", + "encode_generator_settings_payload", "encode_packet_payload", "encode_sweep_settings_payload", "ensure_no_payload_types", diff --git a/python_app/hardware_full/librevna_driver/protocol/codec.py b/python_app/hardware_full/librevna_driver/protocol/codec.py index 5a31c8f..25eb5d5 100644 --- a/python_app/hardware_full/librevna_driver/protocol/codec.py +++ b/python_app/hardware_full/librevna_driver/protocol/codec.py @@ -15,6 +15,7 @@ from ..models import ( DeviceInfo, DeviceLimits, DeviceStatus, + GeneratorSettings, Packet, VNADatapointPacket, VNASweepSettings, @@ -26,6 +27,7 @@ logger = logging.getLogger(__name__) _DEVICE_INFO_STRUCT = struct.Struct(" bytes: ) +def encode_generator_settings_payload(settings: GeneratorSettings) -> bytes: + """Encode ``GeneratorSettings`` payload.""" + return _GENERATOR_SETTINGS_STRUCT.pack( + int(round(settings.frequency_hz)), + int(round(settings.power_dbm * 100.0)), + int(settings.active_port), + int(bool(settings.apply_amplitude_correction)), + ) + + def parse_device_config(payload: bytes, family: HardwareFamily) -> DeviceConfigVariant: """Decode family-specific `DeviceConfiguration` payload.""" if len(payload) != 15: @@ -378,6 +390,8 @@ def encode_packet_payload(packet_type: PacketType, payload: object) -> bytes: if packet_type == PacketType.SWEEP_SETTINGS and isinstance(payload, VNASweepSettings): return encode_sweep_settings_payload(payload) + if packet_type == PacketType.GENERATOR and isinstance(payload, GeneratorSettings): + return encode_generator_settings_payload(payload) if packet_type == PacketType.DEVICE_CONFIGURATION and isinstance(payload, DeviceConfigVariant): return encode_device_config_payload(payload) @@ -396,6 +410,7 @@ NO_PAYLOAD_PACKET_TYPES = { PacketType.REQUEST_DEVICE_CONFIGURATION, PacketType.REQUEST_DEVICE_STATUS, PacketType.INITIATE_SWEEP, + PacketType.SET_IDLE, PacketType.RESET_DEVICE_CONFIGURATION, } diff --git a/python_app/scripts/convert_legacy_preprocess_sets.py b/python_app/scripts/convert_legacy_preprocess_sets.py new file mode 100644 index 0000000..71102bf --- /dev/null +++ b/python_app/scripts/convert_legacy_preprocess_sets.py @@ -0,0 +1,231 @@ +#!/usr/bin/env python3 +"""Convert legacy preprocess set storage into the current two-channel format.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +import sys +from typing import Any + +import numpy as np + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData +from python_app.storage.npz_store import NpzStore + + +LEGACY_KIND_MAP: dict[str, str] = { + "calibration": "s21_calibration", + "reference": "s21_reference", + "s21_calibration": "s21_calibration", + "s21_reference": "s21_reference", + "s11_open": "s11_open", + "s11_short": "s11_short", + "s11_load": "s11_load", + "s11_reference": "s11_reference", +} + +S21_ONLY_TARGET_KINDS = {"s21_calibration", "s21_reference"} + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=( + "Convert old preprocess-set storage from a legacy python_app/data tree into the " + "current format required by the new GUI/runtime." + ) + ) + parser.add_argument( + "source_data_dir", + type=Path, + help="Path to legacy python_app/data directory from the old project copy.", + ) + parser.add_argument( + "output_data_dir", + type=Path, + help="Destination directory where converted sets will be written in the new format.", + ) + parser.add_argument( + "--overwrite", + action="store_true", + help="Allow overwriting already converted destination sets.", + ) + return parser + + +def _load_json(path: Path) -> dict[str, Any]: + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError(f"JSON root must be object: {path}") + return payload + + +def _read_combo_position(combo_payload: dict[str, Any], *, primary_key: str, alias_key: str) -> int: + if primary_key in combo_payload: + return int(combo_payload[primary_key]) + if alias_key in combo_payload: + return int(combo_payload[alias_key]) + raise KeyError(f"Missing combo position field: {primary_key}/{alias_key}") + + +def _combo_suffix(input_pos: int, output_pos: int) -> str: + return f"i{input_pos}_o{output_pos}" + + +def _load_array(arrays: Any, key: str, *, dtype: np.dtype[Any], label: str) -> np.ndarray: + if key not in arrays: + raise KeyError(f"Missing {label} array '{key}' in NPZ archive") + return np.asarray(arrays[key], dtype=dtype).reshape(-1) + + +def _load_legacy_collection(meta_path: Path, npz_path: Path, *, target_kind: str) -> SweepCollection: + meta = _load_json(meta_path) + combos_payload = meta.get("combos") + if not isinstance(combos_payload, list): + raise ValueError(f"Expected 'combos' list in {meta_path}") + + with np.load(npz_path) as arrays: + traces: list[TraceData] = [] + for combo_payload in combos_payload: + if not isinstance(combo_payload, dict): + raise ValueError(f"Expected combo object in {meta_path}") + + input_pos = _read_combo_position(combo_payload, primary_key="input", alias_key="input_pos") + output_pos = _read_combo_position(combo_payload, primary_key="output", alias_key="output_pos") + suffix = _combo_suffix(input_pos, output_pos) + + freq_key = str(combo_payload.get("freq_key") or f"freq_{suffix}") + s21_key = str(combo_payload.get("s21_key") or f"s21_{suffix}") + s11_key = str(combo_payload.get("s11_key") or f"s11_{suffix}") + + frequency_hz = _load_array(arrays, freq_key, dtype=np.float32, label="frequency") + s21 = _load_array(arrays, s21_key, dtype=np.complex64, label="S21") + + if frequency_hz.shape != s21.shape: + raise ValueError(f"Frequency/S21 shape mismatch in {npz_path}: {frequency_hz.shape} vs {s21.shape}") + + if s11_key in arrays: + s11 = _load_array(arrays, s11_key, dtype=np.complex64, label="S11") + elif target_kind in S21_ONLY_TARGET_KINDS: + s11 = np.zeros_like(s21, dtype=np.complex64) + else: + raise KeyError( + f"Missing S11 array '{s11_key}' in {npz_path}; " + f"cannot convert target kind '{target_kind}' without real S11 data" + ) + + if frequency_hz.shape != s11.shape: + raise ValueError(f"Frequency/S11 shape mismatch in {npz_path}: {frequency_hz.shape} vs {s11.shape}") + + traces.append( + TraceData( + combo=ComboKey(input_pos=input_pos, output_pos=output_pos), + frequency_hz=frequency_hz, + s11=s11, + s21=s21, + ) + ) + + return SweepCollection( + collection_id=int(meta.get("collection_id", 0)), + monotonic_ns=int(meta.get("monotonic_ns", 0)), + traces=traces, + ) + + +def _convert_one_set( + store: NpzStore, + *, + output_root: Path, + source_kind: str, + target_kind: str, + radar_key: str, + meta_path: Path, + overwrite: bool, +) -> None: + set_name = meta_path.stem + npz_path = meta_path.with_suffix(".npz") + if not npz_path.exists(): + raise FileNotFoundError(f"Missing NPZ archive for set '{set_name}': {npz_path}") + + target_dir = output_root / target_kind / radar_key + target_json = target_dir / f"{set_name}.json" + target_npz = target_dir / f"{set_name}.npz" + if not overwrite and (target_json.exists() or target_npz.exists()): + raise FileExistsError(f"Destination set already exists: {target_dir / set_name}") + + collection = _load_legacy_collection(meta_path, npz_path, target_kind=target_kind) + store.save_set(target_kind, radar_key, set_name, collection) + print( + f"[converted] {source_kind}/{radar_key}/{set_name} -> " + f"{target_kind}/{radar_key}/{set_name} (traces={len(collection.traces)})" + ) + + +def main() -> int: + parser = _build_parser() + args = parser.parse_args() + + source_root = args.source_data_dir.expanduser().resolve() + output_root = args.output_data_dir.expanduser().resolve() + + if not source_root.exists(): + raise FileNotFoundError(f"Source data directory does not exist: {source_root}") + if source_root == output_root: + raise ValueError("Source and output directories must be different") + + store = NpzStore(output_root) + converted_count = 0 + skipped_kind_count = 0 + error_messages: list[str] = [] + + for source_kind_dir in sorted(path for path in source_root.iterdir() if path.is_dir()): + source_kind = source_kind_dir.name + target_kind = LEGACY_KIND_MAP.get(source_kind) + if target_kind is None: + skipped_kind_count += 1 + print(f"[skip-kind] {source_kind_dir}") + continue + + for radar_key_dir in sorted(path for path in source_kind_dir.iterdir() if path.is_dir()): + radar_key = radar_key_dir.name + for meta_path in sorted(radar_key_dir.glob("*.json")): + try: + _convert_one_set( + store, + output_root=output_root, + source_kind=source_kind, + target_kind=target_kind, + radar_key=radar_key, + meta_path=meta_path, + overwrite=bool(args.overwrite), + ) + converted_count += 1 + except Exception as exc: # noqa: BLE001 + error_messages.append(f"{meta_path}: {exc}") + print(f"[error] {meta_path}: {exc}") + + print( + "\nConversion summary:\n" + f" source root: {source_root}\n" + f" output root: {output_root}\n" + f" converted sets: {converted_count}\n" + f" skipped kinds: {skipped_kind_count}\n" + f" errors: {len(error_messages)}" + ) + + if error_messages: + return 1 + if converted_count == 0: + print("No convertible preprocess sets were found.") + return 2 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python_app/scripts/librevna_generator_sweep.py b/python_app/scripts/librevna_generator_sweep.py new file mode 100644 index 0000000..3db35e8 --- /dev/null +++ b/python_app/scripts/librevna_generator_sweep.py @@ -0,0 +1,35 @@ +#!/usr/bin/env python3 +"""Standalone LibreVNA generator sweep driven by a local Python config.""" + +from __future__ import annotations + +import logging +import sys + +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from python_app.generator_sweep.config import resolve_generator_sweep_config +from python_app.scripts.librevna_generator_sweep_config import CONFIG + + +def main() -> int: + """Load local generator config and execute the sweep.""" + resolved = resolve_generator_sweep_config(CONFIG) + logging.basicConfig( + level=resolved.log_level, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + ) + + from python_app.generator_sweep.runner import GeneratorSweepRunner + + runner = GeneratorSweepRunner(CONFIG) + runner.run() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python_app/scripts/librevna_generator_sweep_config.py b/python_app/scripts/librevna_generator_sweep_config.py new file mode 100644 index 0000000..dbfb4a8 --- /dev/null +++ b/python_app/scripts/librevna_generator_sweep_config.py @@ -0,0 +1,31 @@ +"""Local editable config for ``librevna_generator_sweep.py``.""" + +from __future__ import annotations + +from python_app.generator_sweep.config import GeneratorSweepConfig + + +CONFIG = GeneratorSweepConfig( + log_level="INFO", # DEBUG, INFO, WARNING, ERROR, or CRITICAL. + serial=None, # None or exact serial string, for example "206930A15532". + strict_protocol_version=14, # Exact value for the connected device: 14. + start_hz=100_000_000.0, # Script constraint: > 0. Device reports frequency range 0.0 .. 6_000_000_000.0 Hz. + stop_hz=6_000_000_000.0, # Script constraint: >= start_hz. Device reports frequency range 0.0 .. 6_000_000_000.0 Hz. + step_hz=0, # > 0 to use step mode, or 0 to disable and use points mode. + points=501, # 0 to disable, or >= 2 to use points mode; if start_hz == stop_hz then only 1 is allowed. + hold_time_ms=50.0, # >= 0 ms. Float values like 0.5 are allowed. + loop=True, # True or False. + port=1, # Only 1 or 2. + power_dbm=-10.0, # Device reports source power range -40.0 .. 0.0 dBm. + amplitude_correction=False, # True = apply source amplitude calibration, False = use raw generator level. + settle_timeout_ms=1_000, # > 0 ms. + status_poll_ms=10.0, # > 0 ms. Float values like 0.2 are allowed. + post_lock_delay_us=0, # >= 0 us. + gpio_chip="/dev/gpiochip0", # Non-empty Linux GPIO chip path. + sweep_start_pin=5, # BCM GPIO, >= 0, must differ from curr_step_pin and pwm_pin. + curr_step_pin=6, # BCM GPIO, >= 0, must differ from sweep_start_pin and pwm_pin. + curr_step_initial_level=0, # Only 0 or 1. + pwm_pin=12, # Only 12, 13, 18, or 19 on Raspberry Pi 5. + pwm_frequency_hz=2_000_000, # > 0 Hz. + pwm_duty_cycle=0.5, # Range: 0.0 < value <= 1.0. +) diff --git a/requirements.txt b/requirements.txt index 5299828..89e90ee 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,4 @@ numpy>=1.26,<3 libusb1>=3.1 PyQt6>=6.6 pyqtgraph>=0.13.7 +rpi-hardware-pwm>=0.2.2,<1