"""FIREV2: independent Flame/Smoke media-PTS evidence with OR Event semantics. Call only after Frame admission. Missing/deadline-dropped Frames are never negative evidence. An active episode clears only after accepted Frames observe both branches negative continuously for clear_seconds. """ from collections import Counter, deque from dataclasses import asdict, dataclass import math import time import uuid from app.fire_config import FIRE_MODEL, FIRE_NAMES 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 flame_confidence: float = .5 flame_window_seconds: float = 2. flame_positive_ratio: float = .6 smoke_confidence: float = .5 smoke_window_seconds: float = 2. smoke_positive_ratio: float = .6 roi: tuple = FULL_FRAME roi_id: str = 'primary' roi_name: str = 'Fire ROI' debug_roi: bool = True min_observation_seconds: float = 1. clear_seconds: float = 5. 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=min(self.flame_confidence, self.smoke_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 ('flame_confidence', 'smoke_confidence', 'flame_positive_ratio', 'smoke_positive_ratio'): value = getattr(self, name) if type(value) not in (int, float) or not math.isfinite(value) or not 0 <= value <= 1: raise ValueError('Invalid FIRE ratio.') if not 0 < self.flame_positive_ratio <= 1 or not 0 < self.smoke_positive_ratio <= 1: raise ValueError('Invalid FIRE positive ratio.') for name in ('flame_window_seconds', 'smoke_window_seconds', 'min_observation_seconds', 'clear_seconds', 'max_observation_gap_seconds'): nanoseconds(getattr(self, name)) if (self.min_observation_seconds > min(self.flame_window_seconds, self.smoke_window_seconds) or self.max_observation_gap_seconds > min(self.flame_window_seconds, self.smoke_window_seconds)): raise ValueError('FIRE observation/gap must fit both windows.') 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: raise ValueError('Invalid FIRE scenario settings.') # Saved pre-PR13 internal settings are migrated locally. Public Web # compatibility for the removed top-level confidence is intentionally absent. value = dict(value) legacy = { 'confidence': 'flame_confidence', 'window_seconds': 'flame_window_seconds', 'positive_ratio': 'flame_positive_ratio', } for old, new in legacy.items(): if old in value: if new in value and value[new] != value[old]: raise ValueError('Conflicting legacy FIRE setting.') value.setdefault(new, value.pop(old)) if 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): """Preserve the existing compact scenarioConfig wire shape.""" 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=min(self.flame_confidence, self.smoke_confidence)) @property def scenario_name(self): return 'FIRE' def confidence_for(self, class_id): if class_id == 0: return self.smoke_confidence if class_id == 1: return self.flame_confidence raise ValueError('Invalid FIRE class.') def window_for(self, class_id): return self.smoke_window_seconds if class_id == 0 else self.flame_window_seconds def ratio_for(self, class_id): return self.smoke_positive_ratio if class_id == 0 else self.flame_positive_ratio @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): category_key: str = 'flame_visible' class_id: int = 1 trigger_reason: str = 'FLAME' def payload(self): # trigger_reason stays AI-local diagnostics; public Event keys remain unchanged. return dict(super().payload(), scenario='FIRE', categoryKey=self.category_key, classId=self.class_id) @dataclass(frozen=True) class FireEvaluation: indices: tuple = () decisions: tuple = () events: tuple = () accepted: bool = True trigger_reason: str | None = None class Fire: def __init__(self, config, *, clock=time.monotonic, wall_ns=time.time_ns, event_id_factory=lambda: str(uuid.uuid4()), contains=None): if not isinstance(config, FireScenarioConfig): raise ValueError('FIRE requires validated scenario configuration.') self.config, self.clock, self.wall_ns = config, clock, wall_ns self.contains = contains self.event_id_factory = event_id_factory self.metrics = Counter() self.clear() def clear(self): self.samples = {0: deque(), 1: deque()} self.identity = None self.last_seq = -1 self.last_pts = None self.clear_from = None self.event_id = None self.last_trigger_reason = None self._break = True def expire(self): pass def _trim_and_append(self, class_id, pts, positive): cfg = self.config samples = self.samples[class_id] cutoff = pts - nanoseconds(cfg.window_for(class_id)) while samples and samples[0][0] < cutoff: samples.popleft() if len(samples) >= cfg.max_samples: samples.clear() self.metrics[f'class_{class_id}_sample_capacity_reset'] += 1 samples.append((pts, positive)) self.metrics[f'class_{class_id}_samples'] += 1 self.metrics[f'class_{class_id}_positive_samples'] += int(positive) def _ready(self, class_id, current_positive): if not current_positive: return False samples = self.samples[class_id] if not samples: return False span = samples[-1][0] - samples[0][0] ratio = sum(p for _, p in samples) / len(samples) return (span >= nanoseconds(self.config.min_observation_seconds) and ratio >= self.config.ratio_for(class_id)) def evaluate(self, frame, observations, *, gap_before=False): 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 qualifying = {0: [], 1: []} points = {} for i, obs in enumerate(observations): x1, y1, x2, y2 = obs.bbox_xyxy point = ((x1+x2)/2, (y1+y2)/2) points[i] = point if (obs.score >= cfg.confidence_for(obs.class_id) and (self.contains(point) if self.contains else inside(point, cfg.roi))): qualifying[obs.class_id].append(i) 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 = () trigger_reason = None if not time_ok: self.metrics['invalid_time'] += 1 for samples in self.samples.values(): 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 or gap_before: for samples in self.samples.values(): samples.clear() self.clear_from = None self.metrics['evidence_reset'] += 1 current = {cid: bool(qualifying[cid]) for cid in (0, 1)} for cid in (0, 1): self._trim_and_append(cid, pts, current[cid]) # Only an accepted Frame with both branches valid-negative advances clear. both_negative = not current[0] and not current[1] if both_negative: if self.clear_from is None: self.clear_from = pts else: self.clear_from = None if (self.event_id is not None and both_negative and pts-self.clear_from >= nanoseconds(cfg.clear_seconds)): self.event_id = None self.last_trigger_reason = None for samples in self.samples.values(): samples.clear() samples.append((pts, False)) self.metrics['episodes_cleared'] += 1 ready = {cid: self._ready(cid, current[cid]) for cid in (0, 1)} triggered_classes = [cid for cid in (0, 1) if ready[cid]] from app.event_cooldown import allow_event if self.event_id is None and triggered_classes and allow_event(self, 'FIRE'): candidates = [(i, cid) for cid in triggered_classes for i in qualifying[cid]] representative, representative_class = max( candidates, key=lambda row: (observations[row[0]].score, -row[0])) obs = observations[representative] event_id = self.event_id_factory() if type(event_id) is not str or not event_id or len(event_id) > 100: raise ValueError('Invalid FIRE event ID.') trigger_reason = ('FLAME_AND_SMOKE' if set(triggered_classes) == {0, 1} else 'FLAME' if triggered_classes == [1] else 'SMOKE') self.event_id = event_id self.last_trigger_reason = trigger_reason events = (FireEvent( event_id, frame, None, cfg.roi_id, str(representative), obs.score, obs.bbox_xyxy, FIRE_MODEL, self.wall_ns() // 1_000_000, FIRE_NAMES[representative_class], representative_class, trigger_reason),) self.metrics['events_created'] += 1 self.metrics['trigger_' + trigger_reason.lower()] += 1 self.last_pts, self._break = pts, False state = 'FIRE' if self.event_id or triggered_classes else 'CONFIRMING' indices = tuple(i for cid in (0, 1) for i in qualifying[cid]) decisions = tuple( FireDecision(points[i], True, state, cfg.confidence_for(observations[i].class_id), self.event_id if time_ok else None) for i in indices) return FireEvaluation(indices, decisions, events, True, trigger_reason or self.last_trigger_reason)