ben
c36fdcf607
Some checks failed
Build Sphinx Docs Set / Build Docs (pull_request) Successful in 16s
Test with tox / Test with tox (3.10) (pull_request) Failing after 17m6s
Build Project / Build Project (3.10) (pull_request) Successful in 17m26s
Build Project / Build Project (3.11) (pull_request) Successful in 17m25s
Build Project / Build Project (3.12) (pull_request) Successful in 17m27s
Test with tox / Test with tox (3.12) (pull_request) Successful in 17m21s
Test with tox / Test with tox (3.11) (pull_request) Failing after 21m50s
445 lines
15 KiB
Python
445 lines
15 KiB
Python
"""Campaign executor: runs a capture campaign end-to-end."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import subprocess
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Callable, Optional
|
|
|
|
from ria_toolkit_oss.datatypes.recording import Recording
|
|
from ria_toolkit_oss.io.recording import to_sigmf
|
|
|
|
from .campaign import CampaignConfig, CaptureStep, TransmitterConfig
|
|
from .labeler import build_output_filename, label_recording
|
|
from .qa import QAResult, check_recording
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Device name aliases: campaign YAML names → get_sdr_device() names
|
|
_DEVICE_ALIASES = {
|
|
"usrp_b210": "usrp",
|
|
"usrp_b200": "usrp",
|
|
"usrp": "usrp",
|
|
"plutosdr": "pluto",
|
|
"pluto": "pluto",
|
|
"hackrf": "hackrf",
|
|
"hackrf_one": "hackrf",
|
|
"bladerf": "bladerf",
|
|
"rtlsdr": "rtlsdr",
|
|
"rtl_sdr": "rtlsdr",
|
|
"thinkrf": "thinkrf",
|
|
# Simulated device — no hardware required
|
|
"mock": "mock",
|
|
"sim": "mock",
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class StepResult:
|
|
"""Outcome of a single capture step."""
|
|
|
|
transmitter_id: str
|
|
step_label: str
|
|
output_path: Optional[str]
|
|
qa: QAResult
|
|
capture_timestamp: float
|
|
error: Optional[str] = None
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return self.error is None and self.qa.passed
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"transmitter_id": self.transmitter_id,
|
|
"step_label": self.step_label,
|
|
"output_path": self.output_path,
|
|
"capture_timestamp": self.capture_timestamp,
|
|
"qa": self.qa.to_dict(),
|
|
"error": self.error,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class CampaignResult:
|
|
"""Aggregate outcome of a full campaign."""
|
|
|
|
campaign_name: str
|
|
steps: list[StepResult] = field(default_factory=list)
|
|
start_time: float = field(default_factory=time.time)
|
|
end_time: Optional[float] = None
|
|
|
|
@property
|
|
def total_steps(self) -> int:
|
|
return len(self.steps)
|
|
|
|
@property
|
|
def passed(self) -> int:
|
|
return sum(1 for s in self.steps if s.ok)
|
|
|
|
@property
|
|
def flagged(self) -> int:
|
|
return sum(1 for s in self.steps if not s.error and s.qa.flagged)
|
|
|
|
@property
|
|
def failed(self) -> int:
|
|
return sum(1 for s in self.steps if s.error or not s.qa.passed)
|
|
|
|
@property
|
|
def duration_s(self) -> float:
|
|
if self.end_time:
|
|
return self.end_time - self.start_time
|
|
return time.time() - self.start_time
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"campaign_name": self.campaign_name,
|
|
"total_steps": self.total_steps,
|
|
"passed": self.passed,
|
|
"flagged": self.flagged,
|
|
"failed": self.failed,
|
|
"duration_s": round(self.duration_s, 1),
|
|
"steps": [s.to_dict() for s in self.steps],
|
|
}
|
|
|
|
def write_report(self, path: str | Path) -> None:
|
|
"""Write a JSON QA report to disk."""
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(path, "w") as f:
|
|
json.dump(self.to_dict(), f, indent=2)
|
|
logger.info(f"QA report written to {path}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# External script interface
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _run_script(script: str, *args: str, timeout: float = 15.0) -> str:
|
|
"""Run an external control script and return stdout.
|
|
|
|
The script is called as::
|
|
|
|
<script> <arg1> <arg2> ...
|
|
|
|
A non-zero return code raises RuntimeError.
|
|
|
|
Args:
|
|
script: Path to executable script. Must be an absolute path to an
|
|
existing regular file. Relative paths are rejected to prevent
|
|
accidentally executing files that are not the intended script.
|
|
*args: Positional arguments forwarded to the script.
|
|
timeout: Maximum seconds to wait.
|
|
|
|
Returns:
|
|
Script stdout as a string.
|
|
"""
|
|
if not Path(script).is_absolute():
|
|
raise RuntimeError(f"Script path must be absolute: {script}")
|
|
script_path = Path(script).resolve()
|
|
if not script_path.is_file():
|
|
raise RuntimeError(f"Script not found or is not a regular file: {script}")
|
|
|
|
cmd = [str(script_path), *args]
|
|
logger.debug(f"Running script: {' '.join(cmd)}")
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
raise RuntimeError(f"Script timed out after {timeout}s: {script}")
|
|
except FileNotFoundError:
|
|
raise RuntimeError(f"Script not found: {script}")
|
|
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"Script exited {result.returncode}: {result.stderr.strip() or result.stdout.strip()}")
|
|
return result.stdout.strip()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Campaign executor
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class CampaignExecutor:
|
|
"""Executes a :class:`CampaignConfig` end-to-end.
|
|
|
|
Initialises the SDR recorder once, then for each (transmitter, step):
|
|
1. Configures the transmitter (via external script or SDR TX)
|
|
2. Records IQ samples
|
|
3. Labels the recording with device/config metadata
|
|
4. Runs QA checks
|
|
5. Saves the recording to disk
|
|
6. Stops/resets the transmitter
|
|
|
|
Args:
|
|
config: Parsed campaign configuration.
|
|
progress_cb: Optional callback ``(step_index, total_steps, step_result)``
|
|
called after each step completes. Useful for status reporting.
|
|
verbose: Enable debug logging.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
config: CampaignConfig,
|
|
progress_cb: Optional[Callable[[int, int, StepResult], None]] = None,
|
|
verbose: bool = False,
|
|
):
|
|
self.config = config
|
|
self.progress_cb = progress_cb
|
|
self._sdr = None
|
|
|
|
if verbose:
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
else:
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Public interface
|
|
# ------------------------------------------------------------------
|
|
|
|
def run(self) -> CampaignResult:
|
|
"""Execute the full campaign and return a :class:`CampaignResult`.
|
|
|
|
Initialises the SDR, runs all steps across all transmitters,
|
|
then closes the SDR. If SDR initialisation fails the exception
|
|
propagates immediately (nothing is captured).
|
|
"""
|
|
result = CampaignResult(campaign_name=self.config.name)
|
|
|
|
logger.info(
|
|
f"Starting campaign '{self.config.name}': "
|
|
f"{self.config.total_steps()} steps, "
|
|
f"~{self.config.total_capture_time_s():.0f}s capture time"
|
|
)
|
|
|
|
self._init_sdr()
|
|
try:
|
|
total = self.config.total_steps()
|
|
step_index = 0
|
|
|
|
for transmitter in self.config.transmitters:
|
|
logger.info(f"Transmitter: {transmitter.id} ({len(transmitter.schedule)} steps)")
|
|
for step in transmitter.schedule:
|
|
step_result = self._execute_step(transmitter, step)
|
|
result.steps.append(step_result)
|
|
step_index += 1
|
|
|
|
if self.progress_cb:
|
|
self.progress_cb(step_index, total, step_result)
|
|
|
|
if step_result.error:
|
|
logger.warning(f"Step '{step.label}' error: {step_result.error}")
|
|
elif step_result.qa.flagged:
|
|
logger.warning(f"Step '{step.label}' flagged for review: " + "; ".join(step_result.qa.issues))
|
|
else:
|
|
logger.info(
|
|
f"Step '{step.label}' OK "
|
|
f"(SNR {step_result.qa.snr_db:.1f} dB, "
|
|
f"{step_result.qa.duration_s:.1f}s)"
|
|
)
|
|
finally:
|
|
self._close_sdr()
|
|
|
|
result.end_time = time.time()
|
|
logger.info(
|
|
f"Campaign complete: {result.passed}/{result.total_steps} passed, "
|
|
f"{result.flagged} flagged, {result.failed} failed"
|
|
)
|
|
return result
|
|
|
|
# ------------------------------------------------------------------
|
|
# SDR management
|
|
# ------------------------------------------------------------------
|
|
|
|
def _init_sdr(self) -> None:
|
|
"""Initialise and configure the SDR recorder."""
|
|
from ria_toolkit_oss.sdr import get_sdr_device
|
|
|
|
rec = self.config.recorder
|
|
device_name = _DEVICE_ALIASES.get(rec.device.lower(), rec.device.lower())
|
|
logger.info(f"Initialising SDR: {device_name} @ {rec.center_freq/1e6:.2f} MHz")
|
|
|
|
self._sdr = get_sdr_device(device_name)
|
|
gain = None if rec.gain == "auto" else float(rec.gain)
|
|
self._sdr.init_rx(
|
|
sample_rate=rec.sample_rate,
|
|
center_frequency=rec.center_freq,
|
|
gain=gain,
|
|
channel=0,
|
|
)
|
|
if rec.bandwidth and hasattr(self._sdr, "set_rx_bandwidth"):
|
|
self._sdr.set_rx_bandwidth(rec.bandwidth)
|
|
|
|
def _close_sdr(self) -> None:
|
|
if self._sdr is not None:
|
|
try:
|
|
self._sdr.close()
|
|
except Exception as e:
|
|
logger.warning(f"SDR close error: {e}")
|
|
self._sdr = None
|
|
|
|
def _record(self, duration_s: float) -> Recording:
|
|
"""Capture ``duration_s`` seconds of IQ samples."""
|
|
num_samples = int(duration_s * self.config.recorder.sample_rate)
|
|
return self._sdr.record(num_samples=num_samples)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Step execution
|
|
# ------------------------------------------------------------------
|
|
|
|
def _execute_step(self, transmitter: TransmitterConfig, step: CaptureStep) -> StepResult:
|
|
"""Run a single capture step.
|
|
|
|
Returns:
|
|
StepResult with QA outcome and output path (or error string).
|
|
"""
|
|
capture_timestamp = time.time()
|
|
output_path: Optional[str] = None
|
|
|
|
try:
|
|
self._start_transmitter(transmitter, step)
|
|
recording = self._record(step.duration)
|
|
self._stop_transmitter(transmitter, step)
|
|
except Exception as e:
|
|
# Best-effort stop on error
|
|
try:
|
|
self._stop_transmitter(transmitter, step)
|
|
except Exception:
|
|
pass
|
|
return StepResult(
|
|
transmitter_id=transmitter.id,
|
|
step_label=step.label,
|
|
output_path=None,
|
|
qa=QAResult(passed=False, flagged=True, snr_db=0.0, duration_s=0.0, issues=[f"Capture error: {e}"]),
|
|
capture_timestamp=capture_timestamp,
|
|
error=str(e),
|
|
)
|
|
|
|
# Label recording
|
|
recording = label_recording(
|
|
recording=recording,
|
|
device_id=transmitter.id,
|
|
step=step,
|
|
capture_timestamp=capture_timestamp,
|
|
campaign_name=self.config.name,
|
|
)
|
|
|
|
# QA
|
|
qa_result = check_recording(recording, self.config.qa)
|
|
|
|
# Save
|
|
try:
|
|
output_path = self._save(recording, transmitter.id, step)
|
|
except Exception as e:
|
|
return StepResult(
|
|
transmitter_id=transmitter.id,
|
|
step_label=step.label,
|
|
output_path=None,
|
|
qa=qa_result,
|
|
capture_timestamp=capture_timestamp,
|
|
error=f"Save failed: {e}",
|
|
)
|
|
|
|
return StepResult(
|
|
transmitter_id=transmitter.id,
|
|
step_label=step.label,
|
|
output_path=output_path,
|
|
qa=qa_result,
|
|
capture_timestamp=capture_timestamp,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Transmitter control (external script interface)
|
|
# ------------------------------------------------------------------
|
|
|
|
def _start_transmitter(self, transmitter: TransmitterConfig, step: CaptureStep) -> None:
|
|
"""Configure the transmitter for this step.
|
|
|
|
For ``external_script`` control method the script is called as::
|
|
|
|
<script> configure <step_params_json>
|
|
|
|
where ``step_params_json`` is a JSON object with channel, bandwidth,
|
|
traffic, etc. The script is responsible for applying the configuration
|
|
and returning promptly (i.e. not blocking for the capture duration).
|
|
|
|
For SDR transmitters this is a no-op placeholder (TX not yet implemented).
|
|
"""
|
|
if transmitter.control_method == "external_script":
|
|
if not transmitter.script:
|
|
logger.debug(f"No script configured for {transmitter.id}, skipping configure")
|
|
return
|
|
params = self._step_params_json(transmitter, step)
|
|
_run_script(transmitter.script, "configure", params)
|
|
|
|
elif transmitter.control_method == "sdr":
|
|
logger.debug("SDR TX not yet implemented — skipping start")
|
|
|
|
else:
|
|
logger.warning(f"Unknown control method '{transmitter.control_method}' — skipping")
|
|
|
|
def _stop_transmitter(self, transmitter: TransmitterConfig, step: CaptureStep) -> None:
|
|
"""Signal the transmitter to stop.
|
|
|
|
Calls ``<script> stop`` for external_script transmitters.
|
|
"""
|
|
if transmitter.control_method == "external_script":
|
|
if not transmitter.script:
|
|
return
|
|
try:
|
|
_run_script(transmitter.script, "stop")
|
|
except Exception as e:
|
|
logger.warning(f"Script stop failed for {transmitter.id}: {e}")
|
|
|
|
@staticmethod
|
|
def _step_params_json(transmitter: TransmitterConfig, step: CaptureStep) -> str:
|
|
"""Serialise step parameters to a JSON string for the control script."""
|
|
params: dict = {"device": transmitter.device or ""}
|
|
if step.channel is not None:
|
|
params["channel"] = step.channel
|
|
if step.bandwidth_mhz is not None:
|
|
params["bandwidth_mhz"] = step.bandwidth_mhz
|
|
if step.traffic is not None:
|
|
params["traffic"] = step.traffic
|
|
if step.power_dbm is not None:
|
|
params["power_dbm"] = step.power_dbm
|
|
return json.dumps(params)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Output
|
|
# ------------------------------------------------------------------
|
|
|
|
def _save(self, recording: Recording, device_id: str, step: CaptureStep) -> str:
|
|
"""Save a recording to disk and return the data file path."""
|
|
out = self.config.output
|
|
rel_filename = build_output_filename(device_id, step)
|
|
out_dir = Path(out.path).resolve()
|
|
|
|
# build_output_filename returns "<device_id>/<label>"
|
|
# to_sigmf needs filename (base) and path (dir) separately
|
|
parts = Path(rel_filename)
|
|
subdir = (out_dir / parts.parent).resolve()
|
|
|
|
# Prevent path traversal: the resolved subdir must stay within the configured output directory.
|
|
try:
|
|
subdir.relative_to(out_dir)
|
|
except ValueError:
|
|
raise RuntimeError(
|
|
f"Output path escape detected: '{subdir}' is outside configured output directory '{out_dir}'"
|
|
)
|
|
|
|
subdir.mkdir(parents=True, exist_ok=True)
|
|
base = parts.name
|
|
|
|
to_sigmf(recording, filename=base, path=str(subdir), overwrite=True)
|
|
return str(subdir / f"{base}.sigmf-data")
|