65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
"""Smart alert profile matching engine."""
|
|
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
from vigilar.config import AlertProfileConfig, AlertProfileRule
|
|
from vigilar.constants import HouseholdState
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def is_in_time_window(window: str, current_time: str) -> bool:
|
|
if not window:
|
|
return True
|
|
|
|
parts = window.split("-")
|
|
if len(parts) != 2:
|
|
return True
|
|
|
|
start, end = parts[0].strip(), parts[1].strip()
|
|
t = current_time
|
|
|
|
if start <= end:
|
|
return start <= t < end
|
|
else:
|
|
return t >= start or t < end
|
|
|
|
|
|
def match_profile(
|
|
profiles: list[AlertProfileConfig],
|
|
household_state: HouseholdState,
|
|
current_time: str,
|
|
) -> AlertProfileConfig | None:
|
|
for profile in profiles:
|
|
if not profile.enabled:
|
|
continue
|
|
if household_state.value not in profile.presence_states:
|
|
continue
|
|
if not is_in_time_window(profile.time_window, current_time):
|
|
continue
|
|
return profile
|
|
return None
|
|
|
|
|
|
def get_action_for_event(
|
|
profile: AlertProfileConfig,
|
|
detection_type: str,
|
|
camera_id: str,
|
|
camera_location: str | None = None,
|
|
) -> tuple[str, str]:
|
|
for rule in profile.rules:
|
|
if rule.detection_type != detection_type:
|
|
continue
|
|
# Check camera_location match
|
|
if rule.camera_location == "any":
|
|
return rule.action, rule.recipients
|
|
if rule.camera_location == camera_id:
|
|
return rule.action, rule.recipients
|
|
if camera_location and rule.camera_location == camera_location:
|
|
return rule.action, rule.recipients
|
|
if not camera_location and rule.camera_location == "any":
|
|
return rule.action, rule.recipients
|
|
continue
|
|
return "quiet_log", "none"
|