"""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() logger.debug("Generator sweep marker GPIO lines opened") def close(self) -> None: """Release GPIO lines.""" self._is_open = False self._lines.close() logger.debug("Generator sweep marker GPIO lines closed") 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() logger.debug("Completed generator sweep pass over %d frequencies", len(self._config.frequencies_hz)) 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