186 lines
8.0 KiB
Python
186 lines
8.0 KiB
Python
"""Signal processing for the Kamil ADC reference-channel acquisition.
|
|
|
|
Each sweep arrives as two aligned complex arrays — the *main* signal and a
|
|
*reference* signal — sampled at the same step indices (see
|
|
:mod:`python_app.hardware_full.kamil_adc.protocol`). Turning that into a stable,
|
|
comparable S21 trace is a fixed three-stage pipeline:
|
|
|
|
1. **Frequency axis from the reference phase.** The reference arm has a constant
|
|
electrical delay, so its unwrapped phase is an affine function of frequency.
|
|
Two fixed calibration anchors ``(phase0, freq0)`` and ``(phase1, freq1)`` —
|
|
supplied by config, *never* derived from the live sweep — define that law::
|
|
|
|
f(phase) = freq0 + (phase - phase0) * (freq1 - freq0) / (phase1 - phase0)
|
|
|
|
Trigger jitter shifts every sample's absolute phase together, so the measured
|
|
band floats from sweep to sweep around the fixed calibration.
|
|
|
|
2. **Amplitude normalization.** ``S = main / |reference|`` divides out the
|
|
stimulus amplitude. Only the magnitude is removed; the reference phase is used
|
|
solely for the axis above.
|
|
|
|
3. **Crop + resample onto a fixed grid.** The floating per-sweep axis is resampled
|
|
(linear, on real and imaginary parts) onto a single hardcoded grid
|
|
``linspace(band_start, band_stop, band_points)``. Every sweep then shares a
|
|
byte-identical frequency axis, so traces can be averaged and subtracted. A
|
|
sweep whose floated range does not fully span the band is *rejected* rather
|
|
than edge-extrapolated.
|
|
|
|
Everything here is pure and free of I/O so it can be unit-tested in isolation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
import numpy as np
|
|
|
|
from python_app.models.run_config_schema import KamilAdcModel
|
|
|
|
# A reference sample whose magnitude falls to (effectively) zero carries no usable
|
|
# amplitude or phase; such points are dropped before normalization rather than
|
|
# producing a division blow-up. The threshold only guards genuine zeros — real
|
|
# reference frames are emitted from settled measurements and sit far above it.
|
|
_REFERENCE_AMPLITUDE_FLOOR = 1e-9
|
|
|
|
# Linear interpolation onto the band needs at least two distinct source samples;
|
|
# a sweep yielding fewer usable points is malformed and rejected.
|
|
_MIN_USABLE_POINTS = 2
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class KamilAdcProcessingParams:
|
|
"""Immutable phase→frequency calibration and output-grid definition.
|
|
|
|
Built from :class:`~python_app.models.run_config_schema.KamilAdcModel` via
|
|
:meth:`from_kamil_model`, but kept decoupled from the config types so the
|
|
processing can be exercised with plain numbers in tests.
|
|
"""
|
|
|
|
phase0_rad: float
|
|
freq0_hz: float
|
|
phase1_rad: float
|
|
freq1_hz: float
|
|
band_start_hz: float
|
|
band_stop_hz: float
|
|
band_points: int
|
|
reference_amplitude_floor: float = _REFERENCE_AMPLITUDE_FLOOR
|
|
|
|
def __post_init__(self) -> None:
|
|
if not np.isfinite([self.phase0_rad, self.phase1_rad, self.freq0_hz, self.freq1_hz]).all():
|
|
raise ValueError("Kamil ADC phase calibration anchors must be finite")
|
|
if self.phase0_rad == self.phase1_rad:
|
|
raise ValueError("Kamil ADC phase calibration anchors must use distinct phases")
|
|
if self.freq0_hz == self.freq1_hz:
|
|
raise ValueError("Kamil ADC phase calibration anchors must map to distinct frequencies")
|
|
if not (np.isfinite(self.band_start_hz) and np.isfinite(self.band_stop_hz)):
|
|
raise ValueError("Kamil ADC band edges must be finite")
|
|
if self.band_stop_hz <= self.band_start_hz:
|
|
raise ValueError("Kamil ADC band stop_hz must be greater than start_hz")
|
|
if self.band_points < 2:
|
|
raise ValueError("Kamil ADC band points must be >= 2")
|
|
if self.reference_amplitude_floor <= 0.0:
|
|
raise ValueError("Kamil ADC reference amplitude floor must be > 0")
|
|
|
|
@classmethod
|
|
def from_kamil_model(cls, kamil_adc: KamilAdcModel) -> KamilAdcProcessingParams:
|
|
"""Build parameters from the ``radar.kamil_adc`` config section."""
|
|
calibration = kamil_adc.phase_calibration
|
|
band = kamil_adc.band
|
|
return cls(
|
|
phase0_rad=float(calibration.phase0_rad),
|
|
freq0_hz=float(calibration.freq0_hz),
|
|
phase1_rad=float(calibration.phase1_rad),
|
|
freq1_hz=float(calibration.freq1_hz),
|
|
band_start_hz=float(band.start_hz),
|
|
band_stop_hz=float(band.stop_hz),
|
|
band_points=int(band.points),
|
|
)
|
|
|
|
@property
|
|
def hz_per_rad(self) -> float:
|
|
"""Frequency change per radian of reference phase (the calibration slope)."""
|
|
return (self.freq1_hz - self.freq0_hz) / (self.phase1_rad - self.phase0_rad)
|
|
|
|
|
|
class KamilAdcSweepProcessor:
|
|
"""Turns aligned (main, reference) sweeps into S21 traces on a fixed grid.
|
|
|
|
The output grid is computed once from the parameters and reused for every
|
|
sweep, so all traces this processor emits share one identical frequency axis.
|
|
"""
|
|
|
|
__slots__ = ("_params", "_grid_hz")
|
|
|
|
def __init__(self, params: KamilAdcProcessingParams) -> None:
|
|
self._params = params
|
|
self._grid_hz = np.linspace(
|
|
params.band_start_hz, params.band_stop_hz, params.band_points, dtype=np.float64
|
|
)
|
|
|
|
@property
|
|
def params(self) -> KamilAdcProcessingParams:
|
|
return self._params
|
|
|
|
@property
|
|
def grid_hz(self) -> np.ndarray:
|
|
"""The fixed output frequency axis (float32), identical for every sweep."""
|
|
return self._grid_hz.astype(np.float32)
|
|
|
|
def reference_frequency_axis(self, reference: np.ndarray) -> np.ndarray:
|
|
"""Map a reference signal's absolute unwrapped phase to frequency (Hz).
|
|
|
|
Returns frequencies in *step order* (not sorted); see the module docstring
|
|
for the calibration law.
|
|
"""
|
|
phase = np.unwrap(np.angle(np.asarray(reference)))
|
|
return self._params.freq0_hz + (phase - self._params.phase0_rad) * self._params.hz_per_rad
|
|
|
|
def process(self, main: np.ndarray, reference: np.ndarray) -> np.ndarray | None:
|
|
"""Return the S21 trace resampled onto the fixed grid, or ``None`` to reject.
|
|
|
|
``main`` and ``reference`` are equal-length complex arrays ordered by
|
|
ascending step index. ``None`` is returned when the sweep is malformed
|
|
(too few usable points) or does not fully cover the configured band.
|
|
"""
|
|
main = np.asarray(main, dtype=np.complex128)
|
|
reference = np.asarray(reference, dtype=np.complex128)
|
|
if main.size < _MIN_USABLE_POINTS or main.size != reference.size:
|
|
return None
|
|
|
|
# Frequency axis from the absolute unwrapped reference phase (step order).
|
|
freqs = self.reference_frequency_axis(reference)
|
|
|
|
# Amplitude-only normalization; drop points where the reference vanished.
|
|
reference_amplitude = np.abs(reference)
|
|
with np.errstate(divide="ignore", invalid="ignore"):
|
|
s21 = main / reference_amplitude
|
|
usable = (
|
|
(reference_amplitude > self._params.reference_amplitude_floor)
|
|
& np.isfinite(freqs)
|
|
& np.isfinite(s21.real)
|
|
& np.isfinite(s21.imag)
|
|
)
|
|
if int(np.count_nonzero(usable)) < _MIN_USABLE_POINTS:
|
|
return None
|
|
|
|
freqs = freqs[usable]
|
|
s21 = s21[usable]
|
|
|
|
# Sort onto a monotonically increasing axis (handles either sweep
|
|
# direction) so interpolation and the coverage check are well defined.
|
|
order = np.argsort(freqs, kind="stable")
|
|
freqs = freqs[order]
|
|
s21 = s21[order]
|
|
|
|
# Reject sweeps that do not span the whole band: interpolating past the
|
|
# measured edge would inject flat, non-physical points.
|
|
if freqs[0] > self._params.band_start_hz or freqs[-1] < self._params.band_stop_hz:
|
|
return None
|
|
|
|
# Linear interpolation on real/imaginary parts. Because the band lies
|
|
# within [freqs[0], freqs[-1]], np.interp never extrapolates here.
|
|
real = np.interp(self._grid_hz, freqs, s21.real)
|
|
imag = np.interp(self._grid_hz, freqs, s21.imag)
|
|
return (real + 1j * imag).astype(np.complex64)
|