awe/radar_system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

some fixes

Browse Source
This commit is contained in:
Ayzen
2026-06-05 14:40:10 +03:00
parent 22942d9dc9
commit bbea744459
35 changed files with 1797 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python_app/models/run_config_validation.py
+129 -23
View File
@@ -8,25 +8,68 @@ from python_app.models.run_config_schema import (
ComboModel,
ControlButtonModel,
GprModel,
RadarSweepModel,
RingEndpointModel,
SwitchModel,
)
# Wire-format bounds shared with the C++ pipeline. The ring header stores the
# slot size as a uint32, and capacity * slot_size must address into a single
# shared-memory mapping, so reject values the C++ side cannot represent.
_UINT32_MAX = (1 << 32) - 1
_RING_SEGMENT_MAX_BYTES = 1 << 40 # 1 TiB upper bound on a single ring mapping.
# Defensive ceiling so a malformed combos string cannot expand into a list that
# stalls the GUI or the downstream acquisition loop.
_MAX_COMBOS = 4096
def _require_int(payload: dict[str, Any], key: str, default: int) -> int:
"""Read an integer field, rejecting JSON arrays/objects with a named ValueError.
Bare ``int()`` raises ``TypeError`` on a list/dict, which escapes the
config-error contract; surface it as a ValueError naming the field instead.
"""
value = payload.get(key, default)
if isinstance(value, bool) or not isinstance(value, (int, float, str)):
raise ValueError(f"{key} must be a JSON integer")
try:
return int(value)
except (TypeError, ValueError) as exc:
raise ValueError(f"{key} must be a JSON integer") from exc
def _require_str(payload: dict[str, Any], key: str, default: str) -> str:
"""Read a string field, rejecting JSON arrays/objects with a named ValueError."""
value = payload.get(key, default)
if isinstance(value, (dict, list)):
raise ValueError(f"{key} must be a JSON string")
return str(value)
def _require_bool(payload: dict[str, Any], key: str, default: bool) -> bool:
"""Read a boolean field, rejecting non-boolean JSON types with a named ValueError."""
value = payload.get(key, default)
if not isinstance(value, bool):
raise ValueError(f"{key} must be a JSON boolean")
return value
def load_switch_payload(
payload: dict[str, Any],
target: SwitchModel,
) -> None:
"""Populate switch model from payload preserving defaults for missing values."""
target.name = str(payload.get("name", target.name))
target.driver_mode = str(payload.get("driver_mode", target.driver_mode))
target.driver = str(payload.get("driver", target.driver))
target.radar_port = int(payload.get("radar_port", target.radar_port))
target.positions = int(payload.get("positions", target.positions))
target.default_position = int(payload.get("default_position", target.default_position))
target.gpio_chip = str(payload.get("gpio_chip", target.gpio_chip))
target.pin_a = int(payload.get("pin_a", target.pin_a))
target.pin_b = int(payload.get("pin_b", target.pin_b))
target.invert_logic = bool(payload.get("invert_logic", target.invert_logic))
# #53: scalar reads reject array/object JSON types as ValueError (not TypeError).
target.name = _require_str(payload, "name", target.name)
target.driver_mode = _require_str(payload, "driver_mode", target.driver_mode)
target.driver = _require_str(payload, "driver", target.driver)
target.radar_port = _require_int(payload, "radar_port", target.radar_port)
target.positions = _require_int(payload, "positions", target.positions)
target.default_position = _require_int(payload, "default_position", target.default_position)
target.gpio_chip = _require_str(payload, "gpio_chip", target.gpio_chip)
target.pin_a = _require_int(payload, "pin_a", target.pin_a)
target.pin_b = _require_int(payload, "pin_b", target.pin_b)
target.invert_logic = _require_bool(payload, "invert_logic", target.invert_logic)
def load_control_button_payload(
@@ -34,20 +77,56 @@ def load_control_button_payload(
target: ControlButtonModel,
) -> None:
"""Populate control-button model from payload preserving defaults."""
target.enabled = bool(payload.get("enabled", target.enabled))
target.gpio_chip = str(payload.get("gpio_chip", target.gpio_chip))
target.pin = int(payload.get("pin", target.pin))
target.active_low = bool(payload.get("active_low", target.active_low))
target.bias = str(payload.get("bias", target.bias))
target.debounce_ms = int(payload.get("debounce_ms", target.debounce_ms))
target.action = str(payload.get("action", target.action))
# #53: scalar reads reject array/object JSON types as ValueError (not TypeError).
target.enabled = _require_bool(payload, "enabled", target.enabled)
target.gpio_chip = _require_str(payload, "gpio_chip", target.gpio_chip)
target.pin = _require_int(payload, "pin", target.pin)
target.active_low = _require_bool(payload, "active_low", target.active_low)
target.bias = _require_str(payload, "bias", target.bias)
target.debounce_ms = _require_int(payload, "debounce_ms", target.debounce_ms)
target.action = _require_str(payload, "action", target.action)
def load_ring_payload(payload: dict[str, Any], target: RingEndpointModel) -> None:
"""Populate ring endpoint model from payload preserving defaults."""
target.name = str(payload.get("name", target.name))
target.capacity = int(payload.get("capacity", target.capacity))
target.slot_size_bytes = int(payload.get("slot_size_bytes", target.slot_size_bytes))
# #53: scalar reads reject array/object JSON types as ValueError (not TypeError).
target.name = _require_str(payload, "name", target.name)
target.capacity = _require_int(payload, "capacity", target.capacity)
target.slot_size_bytes = _require_int(payload, "slot_size_bytes", target.slot_size_bytes)
# #36: enforce ring sizing in Python so a bad config fails here (in GUI/save and
# at config load) instead of crashing the C++ ring allocator at boot.
validate_ring_endpoint(target)
def validate_ring_endpoint(ring: RingEndpointModel) -> None:
"""Validate ring sizing against the constraints the C++ allocator requires."""
field = ring.name or "ring"
if ring.capacity <= 0:
raise ValueError(f"rings.{field}.capacity must be > 0")
if ring.slot_size_bytes <= 0:
raise ValueError(f"rings.{field}.slot_size_bytes must be > 0")
if ring.slot_size_bytes > _UINT32_MAX:
raise ValueError(f"rings.{field}.slot_size_bytes exceeds the uint32 wire limit")
# Overflow-safe: compare against the ceiling without ever forming the full
# product, so an attacker-sized capacity cannot wrap a fixed-width index.
if ring.capacity > _RING_SEGMENT_MAX_BYTES // ring.slot_size_bytes:
raise ValueError(
f"rings.{field} capacity * slot_size_bytes exceeds the maximum ring segment size"
)
def validate_sweep_model(sweep: RadarSweepModel) -> None:
"""Validate radar sweep bounds in Python so a bad sweep fails in the GUI/save
and at config load rather than aborting the C++ acquisition process at boot.
"""
# #36: points must be a positive, integral count of frequency samples.
points = sweep.points
if isinstance(points, bool) or not isinstance(points, int):
raise ValueError("radar.sweep.points must be an integer")
if points <= 0:
raise ValueError("radar.sweep.points must be > 0")
if float(sweep.stop_hz) < float(sweep.start_hz):
raise ValueError("radar.sweep.stop_hz must be >= radar.sweep.start_hz")
def validate_gpr_model(
@@ -55,8 +134,17 @@ def validate_gpr_model(
*,
input_switch_positions: int,
output_switch_positions: int,
sweep: RadarSweepModel | None = None,
) -> None:
"""Validate stable GPR config against current switch dimensions."""
"""Validate stable GPR config against current switch dimensions.
When ``sweep`` is supplied (load and GUI/save paths share this chokepoint),
its bounds are validated here too so #36 sweep failures surface alongside the
GPR checks instead of as a C++ boot crash.
"""
if sweep is not None:
validate_sweep_model(sweep)
if float(gpr.relative_permittivity) <= 0.0:
raise ValueError("gpr.relative_permittivity must be > 0")
@@ -93,8 +181,26 @@ def parse_combos_from_text(text: str) -> list[ComboModel]:
if ":" not in pair:
raise ValueError(f"Invalid combo syntax: {pair!r}. Expected input:output")
input_text, output_text = pair.split(":", 1)
combos.append(ComboModel(input=int(input_text.strip()), output=int(output_text.strip())))
# #57: cap the combo count so a pathological string cannot expand into a
# list large enough to stall the GUI or the acquisition loop.
if len(combos) >= _MAX_COMBOS:
raise ValueError(f"Too many combos: limit is {_MAX_COMBOS}")
input_text, output_text = (side.strip() for side in pair.split(":", 1))
# #57: reject empty sides and re-raise non-integer values naming the pair/side.
if not input_text:
raise ValueError(f"Invalid combo {pair!r}: input side is empty")
if not output_text:
raise ValueError(f"Invalid combo {pair!r}: output side is empty")
try:
input_value = int(input_text)
except ValueError as exc:
raise ValueError(f"Invalid combo {pair!r}: input {input_text!r} is not an integer") from exc
try:
output_value = int(output_text)
except ValueError as exc:
raise ValueError(f"Invalid combo {pair!r}: output {output_text!r} is not an integer") from exc
combos.append(ComboModel(input=input_value, output=output_value))
if not combos:
raise ValueError("No valid combos were provided")