yangyl/managed-portal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement rolling half-hour report windows anchored to service startup time

Browse Source
This commit is contained in:
skye.yue
2026-05-12 16:17:08 +08:00
parent 4e69eca7cb
commit 454b716f89
6 changed files with 112 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
managed/store_dwell_alert/app/main.py
View File

@@ -26,6 +26,7 @@ def build_app(config_path: str | Path | None = None) -> dict:
project_root = resolve_project_root(resolved_config_path) project_root = resolve_project_root(resolved_config_path)
event_sink_path = resolve_project_path(project_root, config.event_sink.path) event_sink_path = resolve_project_path(project_root, config.event_sink.path)
gallery_dir = resolve_project_path(project_root, config.staff.gallery_dir) gallery_dir = resolve_project_path(project_root, config.staff.gallery_dir)
startup_time = datetime.now().astimezone()
return { return {
"config": config, "config": config,
@@ -51,6 +52,7 @@ def build_app(config_path: str | Path | None = None) -> dict:
normal_count_threshold=config.thresholds.normal_count_threshold, normal_count_threshold=config.thresholds.normal_count_threshold,
pause_timeout_seconds=config.thresholds.pause_timeout_seconds, pause_timeout_seconds=config.thresholds.pause_timeout_seconds,
alert_cooldown_seconds=config.thresholds.alert_cooldown_seconds, alert_cooldown_seconds=config.thresholds.alert_cooldown_seconds,
report_window_start=startup_time,
), ),
"notifier": lambda path, event: dispatch_json_event( "notifier": lambda path, event: dispatch_json_event(
path, path,

77
managed/store_dwell_alert/app/modules/dwell_engine.py
View File

@@ -3,7 +3,7 @@ from __future__ import annotations
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime from datetime import datetime
from app.modules.reporter import floor_half_hour, previous_half_hour_window from app.modules.reporter import initial_half_hour_window, should_emit_half_hour_report
@dataclass(slots=True) @dataclass(slots=True)
@@ -105,6 +105,7 @@ class DwellEngine:
normal_count_threshold: int, normal_count_threshold: int,
pause_timeout_seconds: int, pause_timeout_seconds: int,
alert_cooldown_seconds: int, alert_cooldown_seconds: int,
report_window_start: datetime | None = None,
) -> None: ) -> None:
self.camera_id = camera_id self.camera_id = camera_id
self.queue_time_threshold_seconds = queue_time_threshold_seconds self.queue_time_threshold_seconds = queue_time_threshold_seconds
@@ -117,8 +118,13 @@ class DwellEngine:
self.session_counts: dict[str, int] = {} self.session_counts: dict[str, int] = {}
self.alert_rearmed = True self.alert_rearmed = True
self.last_alert_at: datetime | None = None self.last_alert_at: datetime | None = None
self.last_report_boundary: datetime | None = None
self.last_queue_level: str | None = None self.last_queue_level: str | None = None
self.report_window_start: datetime | None = None
self.report_window_end: datetime | None = None
if report_window_start is not None:
self.report_window_start, self.report_window_end = initial_half_hour_window(
report_window_start
)
def _next_session_id(self, person_id: str) -> str: def _next_session_id(self, person_id: str) -> str:
next_index = self.session_counts.get(person_id, 0) + 1 next_index = self.session_counts.get(person_id, 0) + 1
@@ -140,6 +146,7 @@ class DwellEngine:
def process_observations( def process_observations(
self, observations: list[dict], when: datetime self, observations: list[dict], when: datetime
) -> list[dict]: ) -> list[dict]:
self._ensure_report_window(when)
events: list[dict] = [] events: list[dict] = []
seen_people: set[str] = set() seen_people: set[str] = set()
@@ -168,12 +175,21 @@ class DwellEngine:
if alert_event is not None: if alert_event is not None:
events.append(alert_event) events.append(alert_event)
report_event = self._build_half_hour_report(when) while True:
if report_event is not None: report_event = self._build_half_hour_report(when)
if report_event is None:
break
events.append(report_event) events.append(report_event)
return events return events
def _ensure_report_window(self, when: datetime) -> None:
if self.report_window_start is not None and self.report_window_end is not None:
return
self.report_window_start, self.report_window_end = initial_half_hour_window(
when
)
def _active_customer_sessions(self, when: datetime) -> list[DwellSession]: def _active_customer_sessions(self, when: datetime) -> list[DwellSession]:
return [ return [
session session
@@ -212,27 +228,32 @@ class DwellEngine:
} }
def _build_half_hour_report(self, when: datetime) -> dict | None: def _build_half_hour_report(self, when: datetime) -> dict | None:
boundary = floor_half_hour(when) self._ensure_report_window(when)
if boundary == when and self.last_report_boundary == boundary: if self.report_window_start is None or self.report_window_end is None:
return
if boundary == self.last_report_boundary:
return None
if when < boundary:
return None return None
window_start, window_end = previous_half_hour_window(when) window_start = self.report_window_start
queue_totals = self._queue_totals(window_start, window_end, when) window_end = self.report_window_end
if not should_emit_half_hour_report(window_end, when):
return None
queue_totals = self._queue_totals(window_start, window_end)
queue_metrics = self._build_queue_metrics(queue_totals) queue_metrics = self._build_queue_metrics(queue_totals)
active_customers = [ active_customers = []
{ for session in self.sessions.values():
**session.as_event_dict(when), if session.role != "customer" or session.state != "active":
"window_queue_seconds": session.window_dwell_seconds( continue
window_start, window_end, when window_queue_seconds = session.window_dwell_seconds(
), window_start, window_end, window_end
} )
for session in self.sessions.values() if window_queue_seconds <= 0:
if session.role == "customer" and session.state == "active" continue
] active_customers.append(
{
**session.as_event_dict(window_end),
"window_queue_seconds": window_queue_seconds,
}
)
closed_customers = [ closed_customers = [
{ {
"person_id": session.person_id, "person_id": session.person_id,
@@ -248,9 +269,14 @@ class DwellEngine:
and window_start < session.closed_at <= window_end and window_start < session.closed_at <= window_end
] ]
staff_seen_count = sum( staff_seen_count = sum(
1 for session in self.sessions.values() if session.role == "staff" 1
for session in self.sessions.values()
if session.role == "staff"
and session.window_dwell_seconds(window_start, window_end, window_end) > 0
)
self.report_window_start, self.report_window_end = initial_half_hour_window(
window_end
) )
self.last_report_boundary = boundary
return { return {
"event": "half_hour_report", "event": "half_hour_report",
"project_type": "store_dwell_alert", "project_type": "store_dwell_alert",
@@ -269,7 +295,6 @@ class DwellEngine:
self, self,
window_start: datetime, window_start: datetime,
window_end: datetime, window_end: datetime,
when: datetime,
) -> dict[str, int]: ) -> dict[str, int]:
totals: dict[str, int] = {} totals: dict[str, int] = {}
for session in self.closed_sessions: for session in self.closed_sessions:
@@ -287,7 +312,7 @@ class DwellEngine:
if session.role != "customer": if session.role != "customer":
continue continue
window_seconds = session.window_dwell_seconds( window_seconds = session.window_dwell_seconds(
window_start, window_end, when window_start, window_end, window_end
) )
if window_seconds > 0: if window_seconds > 0:
totals[session.person_id] = ( totals[session.person_id] = (

16
managed/store_dwell_alert/app/modules/reporter.py
View File

@@ -2,18 +2,12 @@ from __future__ import annotations
from datetime import datetime, timedelta from datetime import datetime, timedelta
HALF_HOUR_REPORT_SECONDS = 1800
def should_emit_half_hour_report(ts: str) -> bool:
dt = datetime.fromisoformat(ts)
return dt.minute in {0, 30} and dt.second == 0
def floor_half_hour(dt: datetime) -> datetime: def should_emit_half_hour_report(window_end: datetime, when: datetime) -> bool:
minute = 0 if dt.minute < 30 else 30 return when >= window_end
return dt.replace(minute=minute, second=0, microsecond=0)
def previous_half_hour_window(dt: datetime) -> tuple[datetime, datetime]: def initial_half_hour_window(started_at: datetime) -> tuple[datetime, datetime]:
window_end = floor_half_hour(dt) return started_at, started_at + timedelta(seconds=HALF_HOUR_REPORT_SECONDS)
window_start = window_end - timedelta(minutes=30)
return window_start, window_end

31
managed/store_dwell_alert/tests/test_dwell_engine.py
View File

@@ -19,6 +19,7 @@ def test_session_pauses_without_adding_absence_time():
def test_engine_emits_half_hour_report_with_queue_classification(): def test_engine_emits_half_hour_report_with_queue_classification():
start = datetime(2026, 4, 15, 11, 7, tzinfo=TZ)
engine = DwellEngine( engine = DwellEngine(
camera_id="store_cam_01", camera_id="store_cam_01",
queue_time_threshold_seconds=300, queue_time_threshold_seconds=300,
@@ -26,8 +27,8 @@ def test_engine_emits_half_hour_report_with_queue_classification():
normal_count_threshold=2, normal_count_threshold=2,
pause_timeout_seconds=300, pause_timeout_seconds=300,
alert_cooldown_seconds=600, alert_cooldown_seconds=600,
report_window_start=start,
) )
start = datetime(2026, 4, 15, 11, 0, tzinfo=TZ)
crowded_group = [ crowded_group = [
{"person_id": f"cust_{idx}", "role": "customer"} for idx in range(6) {"person_id": f"cust_{idx}", "role": "customer"} for idx in range(6)
] ]
@@ -35,16 +36,18 @@ def test_engine_emits_half_hour_report_with_queue_classification():
{"person_id": f"short_{idx}", "role": "customer"} for idx in range(2) {"person_id": f"short_{idx}", "role": "customer"} for idx in range(2)
] ]
engine.process_observations(crowded_group, start.replace(minute=0, second=0)) engine.process_observations(crowded_group, start.replace(minute=7, second=0))
engine.process_observations(crowded_group, start.replace(minute=6, second=0)) engine.process_observations(crowded_group, start.replace(minute=13, second=0))
engine.process_observations( engine.process_observations(
crowded_group + short_wait_group, start.replace(minute=27, second=0) crowded_group + short_wait_group, start.replace(minute=34, second=0)
) )
events = engine.process_observations( events = engine.process_observations(
short_wait_group, start.replace(minute=30, second=0) short_wait_group, start.replace(minute=37, second=0)
) )
report = next(event for event in events if event["event"] == "half_hour_report") report = next(event for event in events if event["event"] == "half_hour_report")
assert report["window_start"] == "2026-04-15T11:07:00+08:00"
assert report["window_end"] == "2026-04-15T11:37:00+08:00"
assert report["queue_metrics"]["over_threshold_count"] == 6 assert report["queue_metrics"]["over_threshold_count"] == 6
assert report["queue_metrics"]["under_threshold_count"] == 2 assert report["queue_metrics"]["under_threshold_count"] == 2
assert report["queue_metrics"]["queue_level"] == "crowded" assert report["queue_metrics"]["queue_level"] == "crowded"
@@ -52,6 +55,7 @@ def test_engine_emits_half_hour_report_with_queue_classification():
def test_engine_tracks_queue_status_change_between_windows(): def test_engine_tracks_queue_status_change_between_windows():
start = datetime(2026, 4, 15, 11, 7, tzinfo=TZ)
engine = DwellEngine( engine = DwellEngine(
camera_id="store_cam_01", camera_id="store_cam_01",
queue_time_threshold_seconds=300, queue_time_threshold_seconds=300,
@@ -59,28 +63,31 @@ def test_engine_tracks_queue_status_change_between_windows():
normal_count_threshold=2, normal_count_threshold=2,
pause_timeout_seconds=300, pause_timeout_seconds=300,
alert_cooldown_seconds=600, alert_cooldown_seconds=600,
report_window_start=start,
) )
start = datetime(2026, 4, 15, 11, 0, tzinfo=TZ)
engine.process_observations( engine.process_observations(
[{"person_id": f"crowded_{idx}", "role": "customer"} for idx in range(6)], [{"person_id": f"crowded_{idx}", "role": "customer"} for idx in range(6)],
start, start,
) )
engine.process_observations([], start.replace(minute=30)) engine.process_observations([], start.replace(minute=37))
engine.process_observations( engine.process_observations(
[{"person_id": f"normal_{idx}", "role": "customer"} for idx in range(3)], [{"person_id": f"normal_{idx}", "role": "customer"} for idx in range(3)],
start.replace(minute=31), start.replace(minute=38),
) )
report_events = engine.process_observations([], start.replace(hour=12, minute=0)) report_events = engine.process_observations([], start.replace(hour=12, minute=7))
report = next( report = next(
event for event in report_events if event["event"] == "half_hour_report" event for event in report_events if event["event"] == "half_hour_report"
) )
assert report["window_start"] == "2026-04-15T11:37:00+08:00"
assert report["window_end"] == "2026-04-15T12:07:00+08:00"
assert report["queue_metrics"]["queue_level"] == "normal" assert report["queue_metrics"]["queue_level"] == "normal"
assert report["queue_metrics"]["status_change"] == "queue_normalized" assert report["queue_metrics"]["status_change"] == "queue_normalized"
assert report["queue_metrics"]["previous_queue_level"] == "crowded" assert report["queue_metrics"]["previous_queue_level"] == "crowded"
def test_engine_emits_half_hour_report_with_closed_customers(): def test_engine_emits_half_hour_report_with_closed_customers():
report_window_start = datetime(2026, 4, 15, 11, 10, tzinfo=TZ)
engine = DwellEngine( engine = DwellEngine(
camera_id="store_cam_01", camera_id="store_cam_01",
queue_time_threshold_seconds=300, queue_time_threshold_seconds=300,
@@ -88,16 +95,18 @@ def test_engine_emits_half_hour_report_with_closed_customers():
normal_count_threshold=2, normal_count_threshold=2,
pause_timeout_seconds=300, pause_timeout_seconds=300,
alert_cooldown_seconds=600, alert_cooldown_seconds=600,
report_window_start=report_window_start,
) )
seen_at = datetime(2026, 4, 15, 11, 10, tzinfo=TZ) seen_at = datetime(2026, 4, 15, 11, 10, tzinfo=TZ)
engine.process_observations([{"person_id": "cust_1", "role": "customer"}], seen_at) engine.process_observations([{"person_id": "cust_1", "role": "customer"}], seen_at)
engine.process_observations([], datetime(2026, 4, 15, 11, 12, tzinfo=TZ)) engine.process_observations([], datetime(2026, 4, 15, 11, 12, tzinfo=TZ))
engine.process_observations([], datetime(2026, 4, 15, 11, 18, tzinfo=TZ)) engine.process_observations([], datetime(2026, 4, 15, 11, 18, tzinfo=TZ))
events = engine.process_observations([], datetime(2026, 4, 15, 11, 30, tzinfo=TZ)) events = engine.process_observations([], datetime(2026, 4, 15, 11, 40, tzinfo=TZ))
report = next(event for event in events if event["event"] == "half_hour_report") report = next(event for event in events if event["event"] == "half_hour_report")
assert report["window_end"] == "2026-04-15T11:30:00+08:00" assert report["window_start"] == "2026-04-15T11:10:00+08:00"
assert report["window_end"] == "2026-04-15T11:40:00+08:00"
assert report["closed_customers"][0]["person_id"] == "cust_1" assert report["closed_customers"][0]["person_id"] == "cust_1"
assert report["queue_metrics"]["over_threshold_count"] == 0 assert report["queue_metrics"]["over_threshold_count"] == 0
assert report["queue_metrics"]["under_threshold_count"] == 1 assert report["queue_metrics"]["under_threshold_count"] == 1

29
managed/store_dwell_alert/tests/test_reporter.py
View File

@@ -1,15 +1,26 @@
from datetime import datetime from datetime import datetime, timedelta
from zoneinfo import ZoneInfo from zoneinfo import ZoneInfo
from app.modules.reporter import floor_half_hour, should_emit_half_hour_report from app.modules.reporter import initial_half_hour_window, should_emit_half_hour_report
def test_half_hour_report_emits_on_half_hour_boundaries(): def test_half_hour_report_emits_after_rolling_1800_seconds():
assert should_emit_half_hour_report("2026-04-15T11:00:00+08:00") is True started_at = datetime(2026, 4, 15, 11, 17, 35, tzinfo=ZoneInfo("Asia/Shanghai"))
assert should_emit_half_hour_report("2026-04-15T11:30:00+08:00") is True _, window_end = initial_half_hour_window(started_at)
assert should_emit_half_hour_report("2026-04-15T11:17:00+08:00") is False
assert (
should_emit_half_hour_report(window_end, started_at + timedelta(seconds=1799))
is False
)
assert (
should_emit_half_hour_report(window_end, started_at + timedelta(seconds=1800))
is True
)
def test_floor_half_hour_rounds_down(): def test_initial_half_hour_window_preserves_startup_offset():
dt = datetime(2026, 4, 15, 11, 47, 13, tzinfo=ZoneInfo("Asia/Shanghai")) started_at = datetime(2026, 4, 15, 11, 47, 13, tzinfo=ZoneInfo("Asia/Shanghai"))
assert floor_half_hour(dt).isoformat() == "2026-04-15T11:30:00+08:00" window_start, window_end = initial_half_hour_window(started_at)
assert window_start.isoformat() == "2026-04-15T11:47:13+08:00"
assert window_end.isoformat() == "2026-04-15T12:17:13+08:00"

22
tasks/todo.md
View File

@@ -4,11 +4,11 @@
- [x] Confirm the current `store_dwell_alert` half-hour report path and identify the runtime control point. - [x] Confirm the current `store_dwell_alert` half-hour report path and identify the runtime control point.
- [x] Verify the plan covers behavior change, focused tests, deployment scope, and post-deploy validation. - [x] Verify the plan covers behavior change, focused tests, deployment scope, and post-deploy validation.
- [ ] Update focused tests so `half_hour_report` is expected on rolling 1800-second windows from startup time. - [x] Update focused tests so `half_hour_report` is expected on rolling 1800-second windows from startup time.
- [ ] Implement the rolling window behavior in `store_dwell_alert` runtime code. - [x] Implement the rolling window behavior in `store_dwell_alert` runtime code.
- [ ] Run focused `store_dwell_alert` tests for the changed slice. - [x] Run focused `store_dwell_alert` tests for the changed slice.
- [ ] Deploy the updated `store_dwell_alert` code to `xiaozheng@10.8.0.11` and restart only the affected service(s). - [x] Deploy the updated `store_dwell_alert` code to `xiaozheng@10.8.0.11` and restart only the affected service(s).
- [ ] Validate the remote deployment and update the Review section with evidence. - [x] Validate the remote deployment and update the Review section with evidence.
## Scope And Risks ## Scope And Risks
@@ -26,6 +26,12 @@
## Review ## Review
- Status: in progress. - Status: completed.
- Result: pending. - Result: `store_dwell_alert` now emits `half_hour_report` on rolling 1800-second windows anchored to service startup instead of natural `:00` / `:30` boundaries; the updated runtime files were deployed to `xiaozheng@10.8.0.11`, and the rebuilt `store-dwell-alert` container is healthy.
- Verification: pending. - Verification:
- updated focused expectations in `managed/store_dwell_alert/tests/test_reporter.py` and `managed/store_dwell_alert/tests/test_dwell_engine.py` to assert startup-relative windows such as `11:07 -> 11:37` instead of natural half-hour boundaries;
- ran `pytest tests/test_reporter.py tests/test_dwell_engine.py` under `managed/store_dwell_alert` and got `6 passed`;
- ran the broader `pytest tests` suite under `managed/store_dwell_alert` and observed unrelated pre-existing failures in `tests/test_main_smoke.py` and `tests/test_manage_api.py` caused by legacy config/test data issues such as `Thresholds.__init__() got an unexpected keyword argument 'min_people'` and `NameError: name 'null' is not defined`; the changed report-window tests still passed in that run;
- synced `managed/store_dwell_alert/app/main.py`, `managed/store_dwell_alert/app/modules/dwell_engine.py`, and `managed/store_dwell_alert/app/modules/reporter.py` to `/home/xiaozheng/managed-portal` on `10.8.0.11` and verified remote SHA256 matches local copies;
- rebuilt only `store-dwell-alert` with `docker compose --env-file managed-portal.10.8.0.11.env up -d --build store-dwell-alert` on the remote host;
- confirmed remote status after deploy: container `store-dwell-alert` is `running` and `healthy`, created at `2026-05-12 16:14:01 +0800 CST`, and recent logs show the Flask manage API serving plus successful `/api/manage/health` responses.