"""FIREV1: worker-owned flame-only PTS evidence; no tracking or external I/O. Call only after frame admission. Missing observations never prove absence. An active episode survives silence until observed negative time or run clear. """ from collections import Counter, deque from dataclasses import asdict, dataclass import math import time import uuid from app.fire_config import FIRE_MODEL from app.intrusion import IntrusionConfig, IntrusionEvent, inside FULL_FRAME = ((0., 0.), (1., 0.), (1., 1.), (0., 1.)) def nanoseconds(value): if type(value) not in (int, float) or not math.isfinite(value) or not 1e-9 <= value <= 3600: raise ValueError('Invalid FIRE duration.') return round(value * 1_000_000_000) @dataclass(frozen=True) class FireScenarioConfig: enabled: bool = False confidence: float = .5 roi: tuple = FULL_FRAME roi_id: str = 'primary' roi_name: str = 'Fire ROI' debug_roi: bool = True window_seconds: float = 2. min_observation_seconds: float = 1. positive_ratio: float = .6 clear_seconds: float = 5. # Evidence guard, not a channel watchdog. This is an initial tuning value. max_observation_gap_seconds: float = .5 max_samples: int = 512 def __post_init__(self): roi = FULL_FRAME if self.roi is None or self.roi == [] or self.roi == () else self.roi geometry = IntrusionConfig(enabled=self.enabled, confidence=self.confidence, roi=roi, roi_id=self.roi_id, roi_name=self.roi_name, debug_roi=self.debug_roi, point_x=.5, point_y=.5) object.__setattr__(self, 'roi', geometry.roi) for name in ('window_seconds', 'min_observation_seconds', 'clear_seconds', 'max_observation_gap_seconds'): nanoseconds(getattr(self, name)) if self.min_observation_seconds > self.window_seconds or self.max_observation_gap_seconds > self.window_seconds: raise ValueError('FIRE observation/gap must fit window.') if type(self.positive_ratio) not in (int, float) or not math.isfinite(self.positive_ratio) or not 0 < self.positive_ratio <= 1: raise ValueError('Invalid FIRE ratio.') if type(self.max_samples) is not int or not 2 <= self.max_samples <= 4096: raise ValueError('Invalid FIRE sample capacity.') @classmethod def parse(cls, value): if type(value) is not dict or set(value) - set(cls.__dataclass_fields__): raise ValueError('Invalid FIRE scenario settings.') return cls(**value) def settings_payload(self): value = asdict(self) value['roi'] = [list(p) for p in self.roi] return value def payload(self): """Reuse shared ROI/display fields; do not mislabel window as dwell.""" return dict(enabled=self.enabled, roi_id=self.roi_id, roi_name=self.roi_name, debug_roi=self.debug_roi, roi=[list(p) for p in self.roi], point_x=.5, point_y=.5, confidence=self.confidence) @property def scenario_name(self): return 'FIRE' @dataclass(frozen=True) class FireDecision: point: tuple inside: bool state: str confidence: float event_id: str | None = None def payload(self): return dict(scenario='FIRE', point=list(self.point), inside=self.inside, state=self.state, confidence=self.confidence, eventId=self.event_id) @dataclass(frozen=True) class FireEvent(IntrusionEvent): """Same wire keys, no fabricated track ID or public temporal fields.""" def payload(self): return dict(super().payload(), scenario='FIRE', categoryKey='flame_visible', classId=1) @dataclass(frozen=True) class FireEvaluation: indices: tuple = () decisions: tuple = () events: tuple = () accepted: bool = True class Fire: def __init__(self, config, *, clock=time.monotonic, wall_ns=time.time_ns, event_id_factory=lambda: str(uuid.uuid4())): if not isinstance(config, FireScenarioConfig): raise ValueError('FIRE requires validated scenario configuration.') self.config, self.clock, self.wall_ns = config, clock, wall_ns self.event_id_factory = event_id_factory self.metrics = Counter() self.clear() def clear(self): self.samples = deque() self.identity = None self.last_seq = -1 self.last_pts = None self.clear_from = None self.event_id = None self._break = True def expire(self): # Already bounded. Wall-clock idle cannot change media-time evidence. # Run owner calls clear after worker join. No autonomous rearm on idle. pass def evaluate(self, frame, observations): identity = (frame.server_id, frame.source_id, frame.stream_id, frame.run_id, frame.run_generation) if self.identity is not None and identity != self.identity: raise ValueError('Foreign FIRE run.') if (type(frame.frame_seq) is not int or frame.frame_seq < 0 or len(observations) > 300 or any(o.frame != frame or o.model_key != FIRE_MODEL or type(o.class_id) is not int or o.class_id not in (0, 1) for o in observations)): raise ValueError('FIRE observation association differs.') if frame.frame_seq <= self.last_seq: self.metrics['stale_frame'] += 1 return FireEvaluation(accepted=False) self.identity, self.last_seq = identity, frame.frame_seq if not self.config.enabled: return FireEvaluation() cfg = self.config indices, points = [], [] for i, obs in enumerate(observations): x1, y1, x2, y2 = obs.bbox_xyxy point = ((x1+x2)/2, (y1+y2)/2) if obs.class_id == 1 and obs.score >= cfg.confidence and inside(point, cfg.roi): indices.append(i) points.append(point) pts = frame.pts_ns time_ok = type(pts) is int and 0 <= pts < 2**63 and (self.last_pts is None or pts > self.last_pts) events = () if not time_ok: self.metrics['invalid_time'] += 1 self.samples.clear() self.clear_from = None self._break = True state = 'TIME_UNAVAILABLE' else: gap = self.last_pts is not None and pts-self.last_pts > nanoseconds(cfg.max_observation_gap_seconds) if gap or self._break: self.samples.clear() self.clear_from = None self.metrics['evidence_reset'] += 1 cutoff = pts-nanoseconds(cfg.window_seconds) while self.samples and self.samples[0][0] < cutoff: self.samples.popleft() if len(self.samples) >= cfg.max_samples: # Conservative evidence reset, not eviction that inflates ratio. self.samples.clear() self.clear_from = None self.metrics['sample_capacity_reset'] += 1 positive = bool(indices) self.samples.append((pts, positive)) self.metrics['samples'] += 1 self.metrics['positive_samples'] += int(positive) if positive: self.clear_from = None elif self.clear_from is None: # Five seconds start at the FIRST observed negative, not last flame. self.clear_from = pts if self.event_id is not None and not positive and pts-self.clear_from >= nanoseconds(cfg.clear_seconds): self.event_id = None self.samples.clear() self.samples.append((pts, False)) self.metrics['episodes_cleared'] += 1 span = self.samples[-1][0] - self.samples[0][0] ratio = sum(p for _, p in self.samples) / len(self.samples) if (self.event_id is None and positive and span >= nanoseconds(cfg.min_observation_seconds) and ratio >= cfg.positive_ratio): representative = max(indices, key=lambda i: (observations[i].score, -i)) obs = observations[representative] event_id = self.event_id_factory() if not isinstance(event_id, str) or not event_id or len(event_id) > 100: raise ValueError('Invalid FIRE event ID.') self.event_id = event_id event = FireEvent(event_id, frame, None, cfg.roi_id, str(representative), obs.score, obs.bbox_xyxy, FIRE_MODEL, self.wall_ns() // 1_000_000) events = (event,) self.metrics['events_created'] += 1 self.last_pts, self._break = pts, False state = 'FIRE' if self.event_id else 'CONFIRMING' decisions = tuple(FireDecision(p, True, state, cfg.confidence, self.event_id if time_ok else None) for p in points) return FireEvaluation(tuple(indices), decisions, events)