diff --git a/libs/config/settings.py b/libs/config/settings.py index 56e98ad..110e2fb 100644 --- a/libs/config/settings.py +++ b/libs/config/settings.py @@ -30,6 +30,14 @@ class Settings(BaseSettings): tracker_max_cosine_distance: float = 0.4 camera_id: str = "cam_01" + # Action classifier settings + lingering_threshold_sec: float = 5.0 + movement_threshold_px: float = 15.0 + near_keypad_dist_px: float = 75.0 + keypad_center_x: float = 500.0 + keypad_center_y: float = 500.0 + + # Reasoning trigger settings reasoning_dwell_threshold_seconds: float = 5.0 reasoning_cooldown_seconds: float = 5.0 @@ -45,12 +53,10 @@ class Settings(BaseSettings): kafka_bootstrap_servers: str = "localhost:9092" kafka_topic: str = "track-events" - @property - def REDIS_URL(self) -> str: - """Backward-compatible uppercase alias for redis_url.""" - return self.redis_url - - model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8") + model_config = SettingsConfigDict( + env_file=".env", + env_file_encoding="utf-8", + ) settings = Settings() diff --git a/services/memory/trigger.py b/services/memory/trigger.py index f9da301..6644980 100644 --- a/services/memory/trigger.py +++ b/services/memory/trigger.py @@ -2,93 +2,67 @@ from time import monotonic -from libs.schemas.tracking import TrackLifecycleEvent from libs.config.settings import settings - +from libs.schemas.memory import ( + TrackSequence, + ActionHint, +) _reasoning_cooldowns: dict[int, float] = {} SUSPICIOUS_ACTIONS = { - "LINGERING", - "NEAR_KEYPAD", - "REPEATED_APPROACH", + ActionHint.LINGERING, + ActionHint.NEAR_KEYPAD, + ActionHint.REPEATED_APPROACH, } -""" -Clear cooldown state for a track after reasoning completes. -""" - def reset_cooldown(track_id: int) -> None: - """Clear cooldown state after reasoning completes.""" + """ + Clear cooldown state after reasoning completes. + """ _reasoning_cooldowns.pop(track_id, None) -""" -Determine whether VLM/LLM reasoning should be triggered -for a suspicious track sequence. - -Conditions: -- track must be inside a restricted zone -- dwell time must exceed configured threshold -- at least one suspicious action must exist -- track must not be inside cooldown window -""" def should_trigger_reasoning( - event: TrackLifecycleEvent, - suspicious_actions: set[str], + seq: TrackSequence, ) -> bool: """ - Determine whether VLM/LLM reasoning should be triggered - for a suspicious track sequence. + Determine whether VLM/LLM reasoning should be triggered. Conditions: - - track must be inside a restricted zone - - dwell time must exceed configured threshold - - at least one suspicious action must exist - - track must not be inside cooldown window + - Track is inside a restricted zone + - Dwell time exceeds configured threshold + - At least one suspicious action exists + - Track is not inside cooldown window """ - if settings.reasoning_dwell_threshold_seconds < 0: - raise ValueError( - "reasoning_dwell_threshold_seconds must be >= 0" - ) - - if settings.reasoning_cooldown_seconds < 0: - raise ValueError( - "reasoning_cooldown_seconds must be >= 0" - ) + if not seq.events: + return False - # Must be inside at least one restricted zone - if not event.zones_present: + # Zone check + if not seq.zones_visited: return False - # Dwell time must exceed configured threshold - if ( - event.dwell_time_seconds - < settings.reasoning_dwell_threshold_seconds - ): + # Dwell threshold + if seq.total_dwell < settings.reasoning_dwell_threshold_seconds: return False - # At least one suspicious action must exist - if not (SUSPICIOUS_ACTIONS & suspicious_actions): + # Suspicious action check + has_suspicious_action = any(event.action_hint in SUSPICIOUS_ACTIONS for event in seq.events) + + if not has_suspicious_action: return False now = monotonic() - # Cooldown protection - last_trigger = _reasoning_cooldowns.get(event.track_id) + # Cooldown check + last_trigger = _reasoning_cooldowns.get(seq.track_id) - if ( - last_trigger is not None - and ( - now - last_trigger - < settings.reasoning_cooldown_seconds - ) - ): + if last_trigger is not None and (now - last_trigger < settings.reasoning_cooldown_seconds): return False - # Store latest trigger timestamp - _reasoning_cooldowns[event.track_id] = now + # Start cooldown + _reasoning_cooldowns[seq.track_id] = now - return True \ No newline at end of file + return True diff --git a/tests/test_trigger.py b/tests/test_trigger.py index b18b7df..8f788f2 100644 --- a/tests/test_trigger.py +++ b/tests/test_trigger.py @@ -1,15 +1,17 @@ import pytest + from libs.config.settings import settings +from libs.schemas.memory import ( + TrackSequence, + TrackEvent, + ActionHint, +) from services.memory.trigger import ( should_trigger_reasoning, reset_cooldown, ) -from libs.schemas.tracking import ( - TrackLifecycleEvent, - TrackState, -) @pytest.fixture(autouse=True) def fixed_reasoning_gate_config(): @@ -25,107 +27,86 @@ def fixed_reasoning_gate_config(): settings.reasoning_cooldown_seconds = old_cooldown -def make_event( +def make_sequence( dwell: float = 10.0, zones: list[str] | None = None, track_id: int = 1, + action: ActionHint = ActionHint.LINGERING, ): - return TrackLifecycleEvent( - event=TrackState.LOST, + event = TrackEvent( track_id=track_id, frame_id=1, - zones_present=zones if zones is not None else ["restricted_zone"], + timestamp_ms=1000, + action_hint=action, dwell_time_seconds=dwell, ) + return TrackSequence( + track_id=track_id, + events=[event], + total_dwell=dwell, + zones_visited=(zones if zones is not None else ["restricted_zone"]), + ) + def test_returns_false_without_zone(): - event = make_event(zones=[]) + seq = make_sequence(zones=[]) - result = should_trigger_reasoning( - event, - {"LINGERING"}, - ) + result = should_trigger_reasoning(seq) assert result is False def test_returns_false_below_dwell_threshold(): - event = make_event(dwell=1.0) + seq = make_sequence(dwell=1.0) - result = should_trigger_reasoning( - event, - {"LINGERING"}, - ) + result = should_trigger_reasoning(seq) assert result is False def test_returns_false_without_suspicious_actions(): - event = make_event() + seq = make_sequence(action=ActionHint.WALKING) - result = should_trigger_reasoning( - event, - {"NORMAL_WALKING"}, - ) + result = should_trigger_reasoning(seq) assert result is False def test_returns_true_for_valid_suspicious_sequence(): - event = make_event(track_id=100) + seq = make_sequence(track_id=100) reset_cooldown(100) - result = should_trigger_reasoning( - event, - {"LINGERING"}, - ) + result = should_trigger_reasoning(seq) assert result is True def test_returns_false_during_cooldown(): - event = make_event(track_id=200) + seq = make_sequence(track_id=200) reset_cooldown(200) - first = should_trigger_reasoning( - event, - {"LINGERING"}, - ) - - second = should_trigger_reasoning( - event, - {"LINGERING"}, - ) + first = should_trigger_reasoning(seq) + second = should_trigger_reasoning(seq) assert first is True assert second is False def test_reset_cooldown_allows_retrigger(): - event = make_event(track_id=300) + seq = make_sequence(track_id=300) reset_cooldown(300) - first = should_trigger_reasoning( - event, - {"LINGERING"}, - ) - - second = should_trigger_reasoning( - event, - {"LINGERING"}, - ) + first = should_trigger_reasoning(seq) + second = should_trigger_reasoning(seq) reset_cooldown(300) - third = should_trigger_reasoning( - event, - {"LINGERING"}, - ) + third = should_trigger_reasoning(seq) assert first is True assert second is False - assert third is True \ No newline at end of file + assert third is True