from copy import deepcopy from dataclasses import replace from types import SimpleNamespace from unittest.mock import patch import numpy as np import pytest from PIL import Image from app.artifact_media import draw_frame from app.crowd_config import CrowdConfig, EVENT_PARAMETERS, validate_event_parameters from app.crowd_event import CrowdRegion from app.crowd_result import CrowdResultProcessor from app.event_cooldown import EventCooldown from app.event_artifact_contract import build_body, relative_directory, media_path from app.multi_provider_results import public_positions from app.provider_plan import plan from app.scenario_parameters import apply, legacy_projection_view from crowd_fixtures import engine, frame, observations PARAMS = dict(reference_count=50, safe_ratio=.25, caution_ratio=.5, warning_ratio=.75, danger_ratio=.9) @pytest.mark.parametrize('key', EVENT_PARAMETERS) @pytest.mark.parametrize('bad', [None, '', '0.5', True, float('nan'), float('inf'), -1]) def test_invalid_required_group_never_uses_saved_defaults(key, bad): raw = dict(PARAMS, **{key: bad}) spec = dict(crowd=CrowdConfig(enabled=True, event_enabled=True).settings_payload()) request = dict(scenarioCode='CROWD', enabled=True, parameters=raw) assert not apply(spec, request) assert not spec['crowd']['enabled'] and not spec['crowd']['event_enabled'] assert not legacy_projection_view(spec, [request])[0]['enabled'] assert plan(spec).keys == ('medium',) @pytest.mark.parametrize('key', EVENT_PARAMETERS) def test_missing_field_is_not_filled_even_safe_is_required(key): raw = dict(PARAMS) raw.pop(key) spec = {'fire': {'enabled': True}} apply(spec, dict(scenarioCode='CROWD', enabled=True, parameters=raw)) assert plan(spec).keys == ('fire',) assert spec['crowd_start_rejection'] == 'MISSING_CROWD_EVENT_PARAMETERS' def test_valid_values_zero_safe_recovery_and_cooldown_reset(monkeypatch): monkeypatch.setenv('CONTEXTRA_EVENT_COOLDOWN_SECONDS', '10') spec = {} bad = dict(scenarioCode='CROWD', enabled=True, parameters={}) apply(spec, bad) request = dict(bad, parameters=dict(PARAMS, safe_ratio=0, cooldown_seconds=7)) apply(spec, request) assert 'crowd_start_rejection' not in spec assert spec['crowd']['event_enabled'] assert legacy_projection_view(spec, [request])[0]['enabled'] assert spec['event_cooldowns']['CROWD'] == 7 apply(spec, dict(request, parameters=PARAMS)) assert spec['event_cooldowns']['CROWD'] == 10 for raw in (dict(PARAMS, reference_count=0), dict(PARAMS, danger_ratio=1.1), dict(PARAMS, warning_ratio=.5), dict(PARAMS, reference_count=1.5)): with pytest.raises(ValueError): validate_event_parameters(raw) @pytest.mark.parametrize('code', ['INTRUSION','LOITERING','FALL','FIGHT','FIRE','CROWD']) def test_cooldown_all_scenarios_default_and_independence(code, monkeypatch): monkeypatch.setenv('CONTEXTRA_EVENT_COOLDOWN_SECONDS','10') spec = {} request = dict(scenarioCode=code, enabled=False, parameters={'cooldown_seconds': 0}) apply(spec, request) assert spec['event_cooldowns'][code] == 0 for raw in (None, {}, {'cooldown_seconds':None}, {'cooldown_seconds':301}, {'cooldown_seconds':True}): apply(spec, dict(request, parameters=raw)) assert spec['event_cooldowns'][code] == 10 now = [0] gate = EventCooldown(10, overrides={code:0}, clock=lambda:now[0]) assert gate.allow(code) and gate.allow(code) other = 'FIRE' if code != 'FIRE' else 'FALL' assert gate.allow(other) and not gate.allow(other) assert EventCooldown(10).allow(other) def processor(**kwargs): e, clock = engine(event_enabled=True, reference_count=10, **kwargs) p = CrowdResultProcessor(e, camera_id=1, wall_ns=lambda:1_000_000_000) return p, clock def step(p, clock, t, heads=9, **kwargs): clock.value = t f = frame(t) return p.process(f, observations(f, p=2, h=heads), **kwargs) def test_status_boundaries_and_unclamped_ratio(): cfg = CrowdConfig() for count, status in ((0,'SAFE'),(24,'SAFE'),(25,'CAUTION'),(37,'CAUTION'), (38,'WARNING'),(45,'DANGER'),(60,'DANGER')): r = CrowdRegion.from_estimate(cfg, SimpleNamespace(head_count=count, person_count=200)) assert r.status == status and r.payload()['count'] == count assert r.density_ratio == 1.2 def test_danger_episode_cooldown_rearm_and_missing(): p, clock = processor() for t in range(4): r = step(p, clock, t) assert len(r.events) == 1 event = r.events[0] assert event.frame == r.frame and event.object_ref is None assert event.payload()['trackId'] is None assert public_positions(r.commands, r.events) == list(range(len(r.commands))) assert r.snapshot()['crowd']['regions'][0]['status'] == 'DANGER' for t in range(4,7): assert not step(p, clock, t).events for t in range(7,10): r = step(p, clock, t, heads=0) assert r.crowd_regions[0].status == 'SAFE' for t in range(10,13): r = step(p, clock, t) assert not r.events # t12: only nine monotonic seconds after t3. for t in range(13,16): r = step(p, clock, t) assert len(r.events) == 1 assert event.region.status == 'DANGER' # Original Event remains frozen. clock.value = 30 p.expire_idle() assert step(p, clock, 30).crowd_regions == () for t in range(31,34): r = step(p, clock, t, gap_before=True) assert not r.events # Missing data never rearms a DANGER episode. assert r.crowd_regions[0].status == 'DANGER' assert p.process(frame(34), (), usable=False) is None p.close() with pytest.raises(RuntimeError): step(p, clock, 35) def test_repeated_danger_off_and_small_gap(): p, clock = processor(suppress_repeated_danger=False) events = [] for t in range(16): events.extend(step(p, clock, t, gap_before=t==5).events) assert len(events) == 2 and [e.frame.pts_ns for e in events] == [3_000_000_000,15_000_000_000] assert p.engine.metrics['upstream_gap'] == 1 def test_roi_event_wire_paths_and_sidecar(tmp_path): from control_sidecar.main import validate_event from tests.test_event_artifact import event e = event() e.update(scenario='CROWD', trackId=None, crowd=CrowdRegion('Crowd',45,1,50,.9,'DANGER').payload()) path = relative_directory(e) body = build_body(e, {'clip':True,'thumbnail':True}, None, 'https://example.invalid/artifacts', path) validate_event(body, e['serverId']) row = body['events'][0] assert row['severityCode'] == 'DANGER' and row['detail']['trackId'] is None assert row['detail']['crowd'] == e['crowd'] target = tmp_path/path/'thumbnail.jpg' target.parent.mkdir(parents=True) target.write_bytes(b'test') assert media_path(tmp_path, path+'/thumbnail.jpg') == target def test_crowd_cache_activation_and_no_capture(monkeypatch): from app.event_capture import EventCapture monkeypatch.setenv('CONTEXTRA_TEST_NO_CAPTURE','false') with patch('app.event_capture.ArtifactConfig.from_environment', return_value=SimpleNamespace(enabled=True)), \ patch('app.training_capture.TrainingCapture', return_value=SimpleNamespace(store=None)), \ patch.object(EventCapture, '__init__', return_value=None) as init: spec = dict(source_id='1', crowd={'enabled':True}) assert EventCapture.from_environment(spec) is None spec['crowd']['event_enabled'] = True assert isinstance(EventCapture.from_environment(spec), EventCapture) assert init.call_count == 1 monkeypatch.setenv('CONTEXTRA_TEST_NO_CAPTURE','true') assert EventCapture.from_environment(spec).__class__.__name__ == 'NoCaptureTrial' assert init.call_count == 1 def test_overlay_masks_non_heads_and_preserves_outside_roi(): p, clock = processor(roi=((0,0),(.7,0),(.7,.7),(0,.7)), debug_roi=False) for t in range(4): r = step(p, clock, t) wire = r.snapshot() image = Image.new('RGB',(1280,720),(80,80,80)) out = np.asarray(draw_frame(image,wire,scenario='CROWD')) assert np.all(out[:,1000:] == 80) and np.all(out[600:] == 80) altered = deepcopy(wire) for obj in altered['objects']: if obj['classId'] == 0: obj['bbox'] = [.8,.8,.95,.95] assert np.array_equal(out, np.asarray(draw_frame(image,altered,scenario='CROWD'))) assert not np.array_equal(out,np.asarray(image)) assert draw_frame(image,wire,(640,360),scenario='CROWD').size == (640,360) def test_real_start_projection_skip_recovery_and_no_crowd_process(tmp_path, monkeypatch): import json from tests.test_web_start_preparation import service from tests.test_secure_bridge import start_body from app.secure_control import Start manager, control = service(tmp_path, monkeypatch) body = start_body(9876) for s in body['scenarios']: s['enabled'] = s['scenarioCode'] == 'CROWD' crowd = next(s for s in body['scenarios'] if s['scenarioCode'] == 'CROWD') group = control.start_camera(Start(json.dumps(body))) assert plan(group.spec).keys == ('medium',) assert group.spec['secure_control_projection']['skippedScenarios'] == ['CROWD'] crowd['parameters'] = dict(PARAMS, cooldown_seconds=3) group = control.start_camera(Start(json.dumps(body))) assert plan(group.spec).keys == ('crowd',) assert group.spec['crowd']['event_enabled'] and group.spec['event_cooldowns']['CROWD'] == 3 assert not group.spec['secure_control_projection']['skippedScenarios'] crowd['parameters'].pop('reference_count') group = control.start_camera(Start(json.dumps(body))) assert plan(group.spec).keys == ('medium',) assert not group.spec['crowd']['event_enabled'] def test_linear_crowd_preserves_roi_event_and_stop_handoff(monkeypatch): import threading from app.provider_plan import ProviderPlan, ProviderSpec from app.provider_linear_results import LinearProviderResults from app.native_metadata import NativeFrame from tests.test_pr13_multi_provider import _Ingress, _Collector from app.crowd_config import CROWD_MODEL p, clock = processor() for t in range(4): result = step(p, clock, t) monkeypatch.setenv('CONTEXTRA_PROVIDER_FRESHNESS_ENABLED','false') provider = ProviderSpec('crowd','det',CROWD_MODEL,('bbox',),('CROWD',)) spec = dict(camera_id=1, crowd=dict(enabled=True,event_enabled=True), tracking={'enabled':False}, event_cooldowns={'CROWD':7}) c = LinearProviderResults(spec,_Ingress(),ProviderPlan((provider,),None,('CROWD',)), _Collector(),lambda:True,lambda:None,clock=lambda:clock.value) assert c.crowd.engine.event_admission('CROWD') assert not c.crowd.engine.event_admission('CROWD') c.ingress.result_gap = lambda f: True copied = NativeFrame(result.frame.frame_seq,result.frame.pts_ns,1280,720,(),1280,720) assert c.submit('crowd',result.frame,copied,timing=(clock.value,)) entered, release = threading.Event(), threading.Event() handed = [] def process(*args, **kwargs): assert kwargs['gap_before'] entered.set() assert release.wait(2) return result c.crowd.process = process c.artifacts = SimpleNamespace(submit=lambda r:handed.append(r),close=lambda timeout:True) c._thread = threading.Thread(target=c._run) c._thread.start() assert entered.wait(2) stopped = threading.Thread(target=lambda:c.close(2)) stopped.start() release.set() stopped.join(3) assert not stopped.is_alive() and handed[0].events == result.events assert not c.submit('crowd',result.frame,copied,timing=(clock.value,))