207 lines
7.3 KiB
Python
207 lines
7.3 KiB
Python
"""Validated project-internal input configuration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Mapping, Sequence
|
|
|
|
INTERNAL_SCHEMA = "brain.internal.input/v1"
|
|
_SECRET_KEYS = {"credential", "password", "secret", "token", "username"}
|
|
|
|
|
|
class ConfigError(ValueError):
|
|
"""Raised when Brain's project-internal test/runtime config is invalid."""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Point:
|
|
x: float
|
|
y: float
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class AreaRule:
|
|
rule_id: str
|
|
points: tuple[Point, ...]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DirectionalLineRule:
|
|
rule_id: str
|
|
start: Point
|
|
end: Point
|
|
trigger_direction: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class VideoProfile:
|
|
profile_id: str
|
|
width: int
|
|
height: int
|
|
fps: float
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SourceConfig:
|
|
kind: str
|
|
seed: int | None = None
|
|
frame_count: int | None = None
|
|
path: Path | None = None
|
|
chunk_size: int = 64 * 1024
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class BrainInputConfig:
|
|
schema: str
|
|
logical_device_id: str
|
|
profile: VideoProfile
|
|
source: SourceConfig
|
|
areas: tuple[AreaRule, ...] = ()
|
|
directional_lines: tuple[DirectionalLineRule, ...] = ()
|
|
|
|
|
|
def _mapping(value: object, field: str) -> Mapping[str, Any]:
|
|
if not isinstance(value, Mapping):
|
|
raise ConfigError(f"{field} must be an object")
|
|
return value
|
|
|
|
|
|
def _text(value: object, field: str) -> str:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ConfigError(f"{field} must be a non-empty string")
|
|
return value.strip()
|
|
|
|
|
|
def _integer(value: object, field: str, *, minimum: int = 1) -> int:
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
|
|
raise ConfigError(f"{field} must be an integer >= {minimum}")
|
|
return value
|
|
|
|
|
|
def _number(value: object, field: str, *, minimum: float = 0.0) -> float:
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise ConfigError(f"{field} must be a number")
|
|
result = float(value)
|
|
if result <= minimum:
|
|
raise ConfigError(f"{field} must be greater than {minimum}")
|
|
return result
|
|
|
|
|
|
def _reject_secrets(value: object, field: str = "config") -> None:
|
|
if isinstance(value, Mapping):
|
|
for key, child in value.items():
|
|
normalized = str(key).strip().lower()
|
|
if normalized in _SECRET_KEYS or any(
|
|
marker in normalized for marker in ("password", "secret", "token")
|
|
):
|
|
raise ConfigError(f"{field} must not contain credential field {key!r}")
|
|
_reject_secrets(child, f"{field}.{key}")
|
|
elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
|
|
for index, child in enumerate(value):
|
|
_reject_secrets(child, f"{field}[{index}]")
|
|
|
|
|
|
def _point(value: object, field: str) -> Point:
|
|
if (
|
|
not isinstance(value, Sequence)
|
|
or isinstance(value, (str, bytes, bytearray))
|
|
or len(value) != 2
|
|
):
|
|
raise ConfigError(f"{field} must be [x, y]")
|
|
x, y = value
|
|
if isinstance(x, bool) or isinstance(y, bool) or not isinstance(x, (int, float)) or not isinstance(y, (int, float)):
|
|
raise ConfigError(f"{field} coordinates must be numbers")
|
|
point = Point(float(x), float(y))
|
|
if not 0.0 <= point.x <= 1.0 or not 0.0 <= point.y <= 1.0:
|
|
raise ConfigError(f"{field} coordinates must be normalized to 0..1")
|
|
return point
|
|
|
|
|
|
def parse_input_config(raw: Mapping[str, Any], *, base_dir: Path | None = None) -> BrainInputConfig:
|
|
"""Parse the explicitly versioned Brain-internal input configuration."""
|
|
_reject_secrets(raw)
|
|
schema = _text(raw.get("schema"), "schema")
|
|
if schema != INTERNAL_SCHEMA:
|
|
raise ConfigError(f"schema must be {INTERNAL_SCHEMA!r}")
|
|
|
|
profile_raw = _mapping(raw.get("profile"), "profile")
|
|
profile = VideoProfile(
|
|
profile_id=_text(profile_raw.get("id"), "profile.id"),
|
|
width=_integer(profile_raw.get("width"), "profile.width"),
|
|
height=_integer(profile_raw.get("height"), "profile.height"),
|
|
fps=_number(profile_raw.get("fps"), "profile.fps"),
|
|
)
|
|
|
|
source_raw = _mapping(raw.get("source"), "source")
|
|
kind = _text(source_raw.get("kind"), "source.kind")
|
|
if kind == "synthetic":
|
|
seed = source_raw.get("seed", 0)
|
|
if isinstance(seed, bool) or not isinstance(seed, int):
|
|
raise ConfigError("source.seed must be an integer")
|
|
source = SourceConfig(
|
|
kind=kind,
|
|
seed=seed,
|
|
frame_count=_integer(source_raw.get("frame_count"), "source.frame_count"),
|
|
)
|
|
elif kind == "local_file":
|
|
configured_path = Path(_text(source_raw.get("path"), "source.path"))
|
|
if not configured_path.is_absolute() and base_dir is not None:
|
|
configured_path = base_dir / configured_path
|
|
source = SourceConfig(
|
|
kind=kind,
|
|
path=configured_path,
|
|
chunk_size=_integer(source_raw.get("chunk_size", 64 * 1024), "source.chunk_size"),
|
|
)
|
|
else:
|
|
raise ConfigError("source.kind must be 'synthetic' or 'local_file'")
|
|
|
|
areas_raw = raw.get("areas", [])
|
|
if not isinstance(areas_raw, list):
|
|
raise ConfigError("areas must be an array")
|
|
areas: list[AreaRule] = []
|
|
for index, item in enumerate(areas_raw):
|
|
area = _mapping(item, f"areas[{index}]")
|
|
points_raw = area.get("points")
|
|
if not isinstance(points_raw, list) or len(points_raw) < 3:
|
|
raise ConfigError(f"areas[{index}].points must contain at least three points")
|
|
areas.append(
|
|
AreaRule(
|
|
rule_id=_text(area.get("id"), f"areas[{index}].id"),
|
|
points=tuple(_point(point, f"areas[{index}].points[{point_index}]") for point_index, point in enumerate(points_raw)),
|
|
)
|
|
)
|
|
|
|
lines_raw = raw.get("directional_lines", [])
|
|
if not isinstance(lines_raw, list):
|
|
raise ConfigError("directional_lines must be an array")
|
|
lines: list[DirectionalLineRule] = []
|
|
for index, item in enumerate(lines_raw):
|
|
line = _mapping(item, f"directional_lines[{index}]")
|
|
direction = _text(line.get("trigger_direction"), f"directional_lines[{index}].trigger_direction")
|
|
if direction not in {"left_to_right", "right_to_left"}:
|
|
raise ConfigError(
|
|
f"directional_lines[{index}].trigger_direction must be left_to_right or right_to_left"
|
|
)
|
|
lines.append(
|
|
DirectionalLineRule(
|
|
rule_id=_text(line.get("id"), f"directional_lines[{index}].id"),
|
|
start=_point(line.get("start"), f"directional_lines[{index}].start"),
|
|
end=_point(line.get("end"), f"directional_lines[{index}].end"),
|
|
trigger_direction=direction,
|
|
)
|
|
)
|
|
|
|
identifiers = [area.rule_id for area in areas] + [line.rule_id for line in lines]
|
|
if len(set(identifiers)) != len(identifiers):
|
|
raise ConfigError("rule ids must be unique")
|
|
|
|
return BrainInputConfig(
|
|
schema=schema,
|
|
logical_device_id=_text(raw.get("logical_device_id"), "logical_device_id"),
|
|
profile=profile,
|
|
source=source,
|
|
areas=tuple(areas),
|
|
directional_lines=tuple(lines),
|
|
)
|