"""Compose input, decode, anonymous vision, rules and local event output.""" from __future__ import annotations from dataclasses import dataclass from pathlib import Path from typing import Any, Mapping from yovision_brain.config import BrainInputConfig, parse_input_config from yovision_brain.decode import decode_packets from yovision_brain.events import EventSink, candidate_from_decision from yovision_brain.input import CancellationToken, build_input_source from yovision_brain.rules import ( AreaDefinition, DirectionalLineDefinition, NormalizedPoint, RuleEngine, RuleSet, ) from yovision_brain.vision import LumaBlobDetector, SingleStreamTracker, TorchLumaBlobDetector @dataclass(frozen=True, slots=True) class RunSummary: status: str frames: int detections: int events: int def _rules(config: BrainInputConfig, version: str) -> RuleSet: return RuleSet( version=version, profile_id=config.profile.profile_id, width=config.profile.width, height=config.profile.height, areas=tuple( AreaDefinition(area.rule_id, tuple(NormalizedPoint(point.x, point.y) for point in area.points)) for area in config.areas ), directional_lines=tuple( DirectionalLineDefinition( line.rule_id, NormalizedPoint(line.start.x, line.start.y), NormalizedPoint(line.end.x, line.end.y), line.trigger_direction, ) for line in config.directional_lines ), ) def run_pipeline( raw: Mapping[str, Any], sink: EventSink, *, base_dir: Path | None = None, cancellation: CancellationToken | None = None, ) -> RunSummary: input_raw = raw.get("input") if not isinstance(input_raw, Mapping): raise ValueError("input must be an object") config = parse_input_config(input_raw, base_dir=base_dir) version = raw.get("rules_version") if not isinstance(version, str) or not version: raise ValueError("rules_version must be a non-empty string") detector_raw = raw.get("detector", {}) if not isinstance(detector_raw, Mapping): raise ValueError("detector must be an object") backend = detector_raw.get("backend", "python") threshold = detector_raw.get("threshold", 200) minimum_area = detector_raw.get("minimum_area", 1) if not isinstance(threshold, int) or not isinstance(minimum_area, int): raise ValueError("detector threshold and minimum_area must be integers") if backend == "python": detector = LumaBlobDetector(threshold=threshold, minimum_area=minimum_area) elif backend == "torch_cpu": detector = TorchLumaBlobDetector(threshold=threshold, minimum_area=minimum_area, device="cpu") else: raise ValueError("detector.backend must be python or torch_cpu") source = build_input_source(config) tracker = SingleStreamTracker() engine = RuleEngine(_rules(config, version)) frame_count = detection_count = event_count = 0 for frame in decode_packets(source.packets(cancellation), cancellation): frame_count += 1 detections = detector.detect(frame) detection_count += len(detections) tracks = tracker.update(detections, frame_sequence=frame.sequence, timestamp_ns=frame.timestamp_ns) decisions = engine.evaluate( tracks, profile_id=frame.profile_id, width=frame.width, height=frame.height, ) tracks_by_id = {track.track_id: track for track in tracks} for decision in decisions: if not decision.triggered: continue sink.write(candidate_from_decision( decision, tracks_by_id[decision.track_id], logical_input_id=config.logical_device_id, detector=detector.metadata, )) event_count += 1 tracker.finish() status = "cancelled" if cancellation is not None and cancellation.cancelled else "completed" return RunSummary(status, frame_count, detection_count, event_count)