added generator mode
This commit is contained in:
Ayzen
@@ -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}")
|
||||
@@ -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),
|
||||
)
|
||||
@@ -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"
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user