26 lines
973 B
Python
26 lines
973 B
Python
"""One acquisition producer selection shared by supervisor and launcher."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
from python_app.models.run_config_schema import RadarModel
|
|
from python_app.models.json_fields import json_object, read_str
|
|
|
|
|
|
def acquisition_command(config_path: Path) -> list[str]:
|
|
"""Select the configured model's acquisition producer."""
|
|
payload = json_object(json.loads(config_path.read_text(encoding="utf-8")), "run config")
|
|
radar = json_object(payload.get("radar"), "radar")
|
|
model = read_str(radar, "model", RadarModel().model)
|
|
producers = {
|
|
"librevna": "matrix_raw_producer",
|
|
"librevna_multi": "matrix_raw_producer",
|
|
"rph_radar": "rph_radar_raw_producer",
|
|
}
|
|
if model not in producers:
|
|
raise ValueError(f"Unsupported radar model: {model}")
|
|
return [sys.executable, "-m", f"python_app.scripts.{producers[model]}", "--config", str(config_path)]
|