Spaces:
Running
Running
File size: 2,501 Bytes
f492127 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 | """
HaramGuard — Shared Data Models
================================
All dataclasses used across agents. Import from here to avoid circular imports.
"""
import numpy as np
from dataclasses import dataclass, field
@dataclass
class FrameResult:
"""Output of PerceptionAgent for a single frame."""
frame_id: int
timestamp: float
person_count: int
density_score: float
avg_spacing: float
boxes: list # [{'x1','y1','x2','y2','conf'}]
annotated: np.ndarray # frame with drawn boxes (BGR)
guardrail_flags: list = field(default_factory=list)
track_ids: list = field(default_factory=list) # YOLO tracking IDs for this frame
occupation_pct: float = 0.0 # Σ(box areas) / frame area × 100
# Condition-based features
compression_ratio: float = 0.0 # 0.0-1.0: how compressed the crowd is
flow_velocity: float = 0.0 # pixels/frame: average movement speed
distribution_score: float = 0.0 # 0.0-1.0: spatial distribution (clustered vs spread)
# Spatial grid — UQU Haram research (local hotspot detection)
grid_counts: list = field(default_factory=list) # 3×3 nested list of per-cell counts
grid_max: int = 0 # max persons in any single cell
hotspot_zone: str = '' # e.g. 'center', 'top-left', 'bottom-right'
@dataclass
class RiskResult:
"""Output of RiskAgent for current sliding window."""
frame_id: int
timestamp: float
risk_score: float # 0.0 - 1.0
risk_level: str # HIGH / MEDIUM / LOW
trend: str # rising / stable / falling
level_changed: bool # True -> trigger OperationsAgent
window_avg: float
window_max: int
density_ema: float = 0.0 # EMA of peak person_count (smoothed)
density_pct: float = 0.0 # density_ema / 50 * 100 (percentage, capped at 150)
@dataclass
class Decision:
"""Single operational decision from OperationsAgent."""
frame_id: int
timestamp: str
context: str
priority: str # P0 / P1 / P2
actions: list # Arabic action strings (filled by CoordinatorAgent)
risk_score: float
risk_level: str
justification: str = "" # Arabic reasoning behind chosen actions (filled by CoordinatorAgent)
selected_gates: list = field(default_factory=list) # gate names chosen by LLM |