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

Refactor store dwell alert management API and dwell engine

Browse Source 
- Updated argument parsing in manage_api.py to include new threshold parameters.
- Enhanced _config_payload to include thresholds and webhook configurations.
- Modified _build_summary to track queue metrics and adjust alert reporting.
- Refactored DwellEngine to utilize queue thresholds for alerting and reporting.
- Added queue metrics calculations and status change tracking in dwell_engine.py.
- Updated notifier.py to support posting JSON events to webhooks.
- Adjusted example configuration to reflect new threshold parameters.
- Enhanced Docker entrypoint script for better process management.
- Updated tests to cover new queue metrics and thresholds.
- Improved ManagedServiceDetail and ManagedServices Vue components to display queue metrics.
This commit is contained in:
skye.yue
2026-05-09 11:35:55 +08:00
parent be5014c582
commit ea618fd674
26 changed files with 1605 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
managed/people_flow_project/src/people_flow/config.py
View File

@@ -10,8 +10,10 @@ from .models import (
AttributeConfig,
CountingConfig,
OutputConfig,
QueueConfig,
RtspConfig,
RuntimeConfig,
WebhookConfig,
YoloConfig,
)
@@ -59,6 +61,8 @@ def load_config(config_path: Path) -> AppConfig:
attributes=AttributeConfig(**data.get("attributes", {})),
output=OutputConfig(**data.get("output", {})),
rtsp=RtspConfig(**data.get("rtsp", {})),
queue=QueueConfig(**_normalize_queue_config(data.get("queue", {}))),
webhook=WebhookConfig(**data.get("webhook", {})),
runtime=RuntimeConfig(**data.get("runtime", {})),
config_path=config_path.resolve(),
)
@@ -73,6 +77,14 @@ def _normalize_counting_config(data: dict) -> dict:
return normalized
def _normalize_queue_config(data: dict) -> dict:
normalized = dict(data)
area = normalized.get("area")
if area is not None:
normalized["area"] = tuple(float(value) for value in area)
return normalized
def parse_line_override(raw_line: str) -> tuple[float, float, float, float]:
parts = [part.strip() for part in raw_line.split(",")]
if len(parts) != 4:

69
managed/people_flow_project/src/people_flow/manage_api.py
View File

@@ -16,7 +16,6 @@ from .config import (
save_config_document,
)
PROJECT_TYPE = "people_flow_project"
DEFAULT_MANAGE_PORT = 18082
MAX_PREVIEW_LINES = 2000
@@ -135,7 +134,12 @@ def parse_args() -> ArgumentParser:
parser = ArgumentParser(description="People flow management API")
parser.add_argument("--config", required=True, help="Path to YAML config file")
parser.add_argument("--host", default="0.0.0.0", help="Host for the management API")
parser.add_argument("--port", type=int, default=DEFAULT_MANAGE_PORT, help="Port for the management API")
parser.add_argument(
"--port",
type=int,
default=DEFAULT_MANAGE_PORT,
help="Port for the management API",
)
return parser
@@ -160,6 +164,19 @@ def _config_payload(ctx: ManageContext) -> dict:
"output_subdir": config.rtsp.output_subdir,
"window_seconds": config.rtsp.window_seconds,
},
"queue": {
"source_id": config.queue.source_id,
"queue_time_threshold_seconds": config.queue.queue_time_threshold_seconds,
"crowded_count_threshold": config.queue.crowded_count_threshold,
"normal_count_threshold": config.queue.normal_count_threshold,
},
"webhook": {
"url": config.webhook.url,
"event_log_path": str(
resolve_project_path(ctx.project_root, config.webhook.event_log_path)
),
"timeout_seconds": config.webhook.timeout_seconds,
},
}
@@ -191,15 +208,33 @@ def _build_summary(ctx: ManageContext) -> dict:
total_people = _int_value(payload.get("total_people"))
window_end = _string_value(payload.get("window_end"))
queue_metrics = (
payload.get("queue_metrics")
if isinstance(payload.get("queue_metrics"), dict)
else {}
)
return {
"result_type": PROJECT_TYPE,
"headline": f"Latest window counted {total_people} people",
"headline": (
"Latest report shows "
f"{_string_value(queue_metrics.get('queue_level')) or 'few'} queue, "
f"{_int_value(queue_metrics.get('over_threshold_count'))} over 5 min and "
f"{_int_value(queue_metrics.get('under_threshold_count'))} under 5 min"
),
"last_result_time": window_end,
"metrics": {
"summary_path": str(summary_path) if summary_path else "",
"window_start": _string_value(payload.get("window_start")),
"window_end": window_end,
"total_people": total_people,
"queue_level": _string_value(queue_metrics.get("queue_level")),
"over_threshold_count": _int_value(
queue_metrics.get("over_threshold_count")
),
"under_threshold_count": _int_value(
queue_metrics.get("under_threshold_count")
),
"status_change": _string_value(queue_metrics.get("status_change")),
"direction_counts": direction_counts,
"age_counts": _map_string_int(payload.get("age_counts")),
"gender_counts": _map_string_int(payload.get("gender_counts")),
@@ -246,6 +281,14 @@ def _load_window_stats(ctx: ManageContext) -> list[dict]:
"window_start": _string_value(payload.get("window_start")),
"window_end": _string_value(payload.get("window_end")),
"total_people": _int_value(payload.get("total_people")),
"queue_level": _queue_metric_value(payload, "queue_level"),
"over_threshold_count": _queue_metric_int(
payload, "over_threshold_count"
),
"under_threshold_count": _queue_metric_int(
payload, "under_threshold_count"
),
"status_change": _queue_metric_value(payload, "status_change"),
"age_counts": _map_string_int(payload.get("age_counts")),
"gender_counts": _map_string_int(payload.get("gender_counts")),
"unknown_attributes": _int_value(payload.get("unknown_attributes")),
@@ -259,6 +302,7 @@ def _list_result_files(ctx: ManageContext) -> list[dict]:
files: list[dict] = []
for path, label in (
(_latest_json_path(ctx), "Latest Summary"),
(_webhook_log_path(ctx), "Webhook Event Log"),
(_runtime_log_path(ctx), "Runtime Log"),
):
if path.exists() and path.is_file():
@@ -305,6 +349,11 @@ def _runtime_log_path(ctx: ManageContext) -> Path:
return _output_root(ctx) / "rtsp_run.log"
def _webhook_log_path(ctx: ManageContext) -> Path:
config = load_config(ctx.config_path)
return resolve_project_path(ctx.project_root, config.webhook.event_log_path)
def _window_files(ctx: ManageContext) -> list[Path]:
windows_dir = _windows_dir(ctx)
if not windows_dir.exists():
@@ -385,5 +434,19 @@ def _map_string_int(value) -> dict[str, int]:
return {str(key): _int_value(raw) for key, raw in value.items()}
def _queue_metric_value(payload: dict, field: str) -> str:
queue_metrics = payload.get("queue_metrics")
if not isinstance(queue_metrics, dict):
return ""
return _string_value(queue_metrics.get(field))
def _queue_metric_int(payload: dict, field: str) -> int:
queue_metrics = payload.get("queue_metrics")
if not isinstance(queue_metrics, dict):
return 0
return _int_value(queue_metrics.get(field))
if __name__ == "__main__":
raise SystemExit(main())

33
managed/people_flow_project/src/people_flow/models.py
View File

@@ -20,7 +20,9 @@ class CountingConfig:
line_mode: str = "normalized"
crossing_tolerance: float = 12.0
def to_pixel_line(self, width: int, height: int) -> tuple[float, float, float, float]:
def to_pixel_line(
self, width: int, height: int
) -> tuple[float, float, float, float]:
x1, y1, x2, y2 = self.line
if self.line_mode == "pixel":
return x1, y1, x2, y2
@@ -58,6 +60,33 @@ class RtspConfig:
output_subdir: str = "rtsp_stream"
@dataclass
class QueueConfig:
enabled: bool = True
area: tuple[float, float, float, float] = (0.0, 0.0, 1.0, 1.0)
area_mode: str = "normalized"
queue_time_threshold_seconds: int = 300
crowded_count_threshold: int = 5
normal_count_threshold: int = 2
pause_timeout_seconds: int = 5
source_id: str = "people_flow_queue"
def to_pixel_area(
self, width: int, height: int
) -> tuple[float, float, float, float]:
x1, y1, x2, y2 = self.area
if self.area_mode == "pixel":
return x1, y1, x2, y2
return x1 * width, y1 * height, x2 * width, y2 * height
@dataclass
class WebhookConfig:
url: str = ""
timeout_seconds: float = 5.0
event_log_path: str = "outputs/rtsp_stream/webhook_events.jsonl"
@dataclass
class RuntimeConfig:
rtsp_url: str = "rtsp://user:password@camera-ip:554/h264/ch1/main/av_stream"
@@ -71,6 +100,8 @@ class AppConfig:
attributes: AttributeConfig = field(default_factory=AttributeConfig)
output: OutputConfig = field(default_factory=OutputConfig)
rtsp: RtspConfig = field(default_factory=RtspConfig)
queue: QueueConfig = field(default_factory=QueueConfig)
webhook: WebhookConfig = field(default_factory=WebhookConfig)
runtime: RuntimeConfig = field(default_factory=RuntimeConfig)
config_path: Path | None = None

99
managed/people_flow_project/src/people_flow/pipeline.py
View File

@@ -20,8 +20,9 @@ from .io_utils import (
write_window_json,
)
from .models import AppConfig
from .queue_analytics import QueueWindowTracker
from .tracking import extract_person_tracks
from .webhook import dispatch_json_event
SUPPORTED_EXTENSIONS = {".mp4", ".mov", ".mkv", ".avi"}
@@ -104,7 +105,9 @@ class PeopleFlowPipeline:
writer = None
if self.config.output.save_video:
writer = make_video_writer(video_output_path, width=width, height=height, fps=fps)
writer = make_video_writer(
video_output_path, width=width, height=height, fps=fps
)
counter = LineCrossCounter(pixel_line, self.config.counting)
attributes = AttributeAggregator(self.config.attributes)
@@ -118,7 +121,9 @@ class PeopleFlowPipeline:
observations = self._track_frame(frame)
for observation in observations:
attributes.maybe_collect(frame=frame, frame_index=frame_index, track=observation)
attributes.maybe_collect(
frame=frame, frame_index=frame_index, track=observation
)
counter.update(observations)
@@ -154,7 +159,9 @@ class PeopleFlowPipeline:
sample_interval = max(float(self.config.rtsp.sample_interval_seconds), 0.01)
window_seconds = max(int(self.config.rtsp.window_seconds), 1)
reconnect_delay = max(float(self.config.rtsp.reconnect_delay_seconds), 0.1)
open_timeout_seconds = max(float(self.config.rtsp.stream_open_timeout_seconds), 1.0)
open_timeout_seconds = max(
float(self.config.rtsp.stream_open_timeout_seconds), 1.0
)
idle_sleep = max(float(self.config.rtsp.idle_sleep_seconds), 0.0)
window_index = 0
@@ -168,7 +175,18 @@ class PeopleFlowPipeline:
capture = None
pixel_line = None
counter = None
queue_tracker = None
attributes = AttributeAggregator(self.config.attributes)
project_root = (
self.config.config_path.parent.parent
if self.config.config_path is not None
else self.output_root
)
webhook_event_log_path = (
project_root / self.config.webhook.event_log_path
if not Path(self.config.webhook.event_log_path).is_absolute()
else Path(self.config.webhook.event_log_path)
)
try:
while True:
@@ -181,6 +199,7 @@ class PeopleFlowPipeline:
window_end=window_end,
counter=counter,
attributes=attributes,
queue_tracker=queue_tracker,
)
json_path = write_window_json(
rtsp_paths["windows"],
@@ -188,6 +207,12 @@ class PeopleFlowPipeline:
payload,
window_end,
)
dispatch_json_event(
webhook_event_log_path,
payload,
webhook_url=self.config.webhook.url,
timeout_seconds=self.config.webhook.timeout_seconds,
)
print(f"window_json={json_path}", flush=True)
print(f"window_total_people={payload['total_people']}", flush=True)
window_index += 1
@@ -195,6 +220,8 @@ class PeopleFlowPipeline:
window_end = window_start + timedelta(seconds=window_seconds)
if counter is not None:
counter.reset()
if queue_tracker is not None:
queue_tracker.reset()
attributes.reset()
now = datetime.now().astimezone()
@@ -213,8 +240,14 @@ class PeopleFlowPipeline:
if pixel_line is None:
height, width = frame.shape[:2]
pixel_line = self.config.counting.to_pixel_line(width=width, height=height)
pixel_line = self.config.counting.to_pixel_line(
width=width, height=height
)
counter = LineCrossCounter(pixel_line, self.config.counting)
queue_tracker = QueueWindowTracker(
self.config.queue,
self.config.queue.to_pixel_area(width=width, height=height),
)
current_time = time.monotonic()
if current_time - last_processed_at < sample_interval:
@@ -225,9 +258,13 @@ class PeopleFlowPipeline:
last_processed_at = current_time
observations = self._track_frame(frame)
for observation in observations:
attributes.maybe_collect(frame=frame, frame_index=frame_index, track=observation)
attributes.maybe_collect(
frame=frame, frame_index=frame_index, track=observation
)
if counter is not None:
counter.update(observations)
if queue_tracker is not None and self.config.queue.enabled:
queue_tracker.observe(observations, now)
if current_time >= next_heartbeat_at:
self._print_rtsp_heartbeat(
process_started_at=process_started_at,
@@ -283,7 +320,9 @@ class PeopleFlowPipeline:
capture.release()
return None
def _build_live_stats(self, counter: LineCrossCounter, attributes: AttributeAggregator) -> dict:
def _build_live_stats(
self, counter: LineCrossCounter, attributes: AttributeAggregator
) -> dict:
age_counts = {"minor": 0, "adult": 0, "senior": 0}
gender_counts = {"male": 0, "female": 0}
unknown_attributes = 0
@@ -313,7 +352,9 @@ class PeopleFlowPipeline:
last_processed_wall_time: datetime | None,
) -> None:
stats = self._build_live_stats(counter, attributes)
runtime_seconds = int((datetime.now().astimezone() - process_started_at).total_seconds())
runtime_seconds = int(
(datetime.now().astimezone() - process_started_at).total_seconds()
)
last_processed = (
last_processed_wall_time.isoformat(timespec="seconds")
if last_processed_wall_time is not None
@@ -387,20 +428,41 @@ class PeopleFlowPipeline:
window_end: datetime,
counter: LineCrossCounter | None,
attributes: AttributeAggregator,
queue_tracker: QueueWindowTracker | None,
) -> dict:
age_counts, gender_counts, unknown_attributes, track_summaries = self._collect_track_summaries(
counter,
attributes,
age_counts, gender_counts, unknown_attributes, track_summaries = (
self._collect_track_summaries(
counter,
attributes,
)
)
total_people = 0 if counter is None else counter.total_people
queue_metrics = (
queue_tracker.build_queue_metrics(window_start, window_end)
if queue_tracker is not None and self.config.queue.enabled
else {
"queue_time_threshold_seconds": self.config.queue.queue_time_threshold_seconds,
"over_threshold_count": 0,
"under_threshold_count": 0,
"queue_level": "few",
"previous_queue_level": None,
"status_change": "initial",
"people": [],
}
)
return {
"event": "half_hour_report",
"project_type": "people_flow_project",
"source_type": "rtsp",
"source": source,
"source_id": self.config.queue.source_id,
"window_index": window_index,
"window_start": window_start.isoformat(),
"window_end": window_end.isoformat(),
"window_duration_seconds": int((window_end - window_start).total_seconds()),
"config_path": str(self.config.config_path) if self.config.config_path else None,
"config_path": (
str(self.config.config_path) if self.config.config_path else None
),
"line": {
"coordinates": list(self.config.counting.line),
"mode": self.config.counting.line_mode,
@@ -410,6 +472,7 @@ class PeopleFlowPipeline:
"gender_counts": gender_counts,
"unknown_attributes": unknown_attributes,
"tracks": track_summaries,
"queue_metrics": queue_metrics,
}
def _finalize_summary(
@@ -419,15 +482,19 @@ class PeopleFlowPipeline:
attributes: AttributeAggregator,
json_path: Path,
) -> dict:
age_counts, gender_counts, unknown_attributes, track_summaries = self._collect_track_summaries(
counter,
attributes,
age_counts, gender_counts, unknown_attributes, track_summaries = (
self._collect_track_summaries(
counter,
attributes,
)
)
payload = {
"video_name": video_path.name,
"video_path": str(video_path),
"config_path": str(self.config.config_path) if self.config.config_path else None,
"config_path": (
str(self.config.config_path) if self.config.config_path else None
),
"line": {
"coordinates": list(self.config.counting.line),
"mode": self.config.counting.line_mode,

201
managed/people_flow_project/src/people_flow/queue_analytics.py Normal file
View File

@@ -0,0 +1,201 @@
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from .models import QueueConfig, TrackObservation
@dataclass
class QueueTrackState:
track_id: int
entered_at: datetime
accumulated_queue_seconds: int = 0
active_started_at: datetime | None = None
last_seen_at: datetime | None = None
pause_started_at: datetime | None = None
completed_periods: list[tuple[datetime, datetime]] = field(default_factory=list)
def __post_init__(self) -> None:
if self.active_started_at is None:
self.active_started_at = self.entered_at
if self.last_seen_at is None:
self.last_seen_at = self.entered_at
def mark_seen(self, when: datetime) -> None:
if self.active_started_at is None:
self.active_started_at = when
self.last_seen_at = when
self.pause_started_at = None
def pause(self, when: datetime) -> None:
if self.active_started_at is None:
return
self.completed_periods.append((self.active_started_at, when))
self.accumulated_queue_seconds += max(
0,
int((when - self.active_started_at).total_seconds()),
)
self.active_started_at = None
self.pause_started_at = when
self.last_seen_at = when
def expire(self, when: datetime, pause_timeout_seconds: int) -> bool:
if self.pause_started_at is None:
return False
return (
int((when - self.pause_started_at).total_seconds()) > pause_timeout_seconds
)
def window_queue_seconds(self, window_start: datetime, window_end: datetime) -> int:
total = 0
for period_start, period_end in self.completed_periods:
total += _overlap_seconds(
period_start, period_end, window_start, window_end
)
if self.active_started_at is not None:
current_end = self.last_seen_at or window_end
total += _overlap_seconds(
self.active_started_at, current_end, window_start, window_end
)
return total
class QueueWindowTracker:
def __init__(
self, config: QueueConfig, pixel_area: tuple[float, float, float, float]
) -> None:
self.config = config
self.pixel_area = pixel_area
self.states: dict[int, QueueTrackState] = {}
self.closed_states: list[QueueTrackState] = []
self.last_queue_level: str | None = None
def observe(self, observations: list[TrackObservation], when: datetime) -> None:
seen_ids: set[int] = set()
for observation in observations:
if not _point_in_area(observation.center, self.pixel_area):
continue
seen_ids.add(observation.track_id)
state = self.states.get(observation.track_id)
if state is None:
state = QueueTrackState(track_id=observation.track_id, entered_at=when)
self.states[observation.track_id] = state
state.mark_seen(when)
for track_id, state in list(self.states.items()):
if track_id in seen_ids:
continue
if state.active_started_at is not None:
state.pause(when)
if state.expire(when, self.config.pause_timeout_seconds):
self.closed_states.append(state)
del self.states[track_id]
def build_queue_metrics(self, window_start: datetime, window_end: datetime) -> dict:
totals: dict[int, int] = {}
for state in self.closed_states:
queue_seconds = state.window_queue_seconds(window_start, window_end)
if queue_seconds > 0:
totals[state.track_id] = totals.get(state.track_id, 0) + queue_seconds
for track_id, state in self.states.items():
queue_seconds = state.window_queue_seconds(window_start, window_end)
if queue_seconds > 0:
totals[track_id] = queue_seconds
over_threshold_count = sum(
1
for queue_seconds in totals.values()
if queue_seconds >= self.config.queue_time_threshold_seconds
)
under_threshold_count = sum(
1
for queue_seconds in totals.values()
if 0 < queue_seconds < self.config.queue_time_threshold_seconds
)
queue_level = _queue_level(
over_threshold_count,
crowded_count_threshold=self.config.crowded_count_threshold,
normal_count_threshold=self.config.normal_count_threshold,
)
previous_queue_level = self.last_queue_level
status_change = _queue_status_change(previous_queue_level, queue_level)
self.last_queue_level = queue_level
return {
"queue_time_threshold_seconds": self.config.queue_time_threshold_seconds,
"over_threshold_count": over_threshold_count,
"under_threshold_count": under_threshold_count,
"queue_level": queue_level,
"previous_queue_level": previous_queue_level,
"status_change": status_change,
"people": [
{
"person_id": f"track_{track_id}",
"queue_seconds": queue_seconds,
"bucket": (
"over_threshold"
if queue_seconds >= self.config.queue_time_threshold_seconds
else "under_threshold"
),
}
for track_id, queue_seconds in sorted(
totals.items(), key=lambda item: item[1], reverse=True
)
],
}
def reset(self) -> None:
self.states.clear()
self.closed_states.clear()
def _point_in_area(
point: tuple[float, float],
area: tuple[float, float, float, float],
) -> bool:
px, py = point
x1, y1, x2, y2 = area
left = min(x1, x2)
right = max(x1, x2)
top = min(y1, y2)
bottom = max(y1, y2)
return left <= px <= right and top <= py <= bottom
def _overlap_seconds(
period_start: datetime,
period_end: datetime,
window_start: datetime,
window_end: datetime,
) -> int:
overlap_start = max(period_start, window_start)
overlap_end = min(period_end, window_end)
if overlap_end <= overlap_start:
return 0
return int((overlap_end - overlap_start).total_seconds())
def _queue_level(
over_threshold_count: int,
crowded_count_threshold: int,
normal_count_threshold: int,
) -> str:
if over_threshold_count > crowded_count_threshold:
return "crowded"
if over_threshold_count >= normal_count_threshold:
return "normal"
return "few"
def _queue_status_change(previous_level: str | None, current_level: str) -> str:
if previous_level is None:
return "initial"
if previous_level == current_level:
return "unchanged"
if current_level == "crowded" and previous_level in {"normal", "few"}:
return "queue_increased"
if current_level == "few" and previous_level in {"normal", "crowded"}:
return "queue_decreased"
if current_level == "normal" and previous_level in {"crowded", "few"}:
return "queue_normalized"
return "changed"

29
managed/people_flow_project/src/people_flow/webhook.py Normal file
View File

@@ -0,0 +1,29 @@
from __future__ import annotations
import json
from pathlib import Path
from urllib import request
def dispatch_json_event(
path: str | Path,
payload: dict,
webhook_url: str = "",
timeout_seconds: float = 5.0,
) -> None:
output_path = Path(path)
output_path.parent.mkdir(parents=True, exist_ok=True)
with output_path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(payload, ensure_ascii=False) + "\n")
if not webhook_url.strip():
return
req = request.Request(
url=webhook_url,
data=json.dumps(payload).encode("utf-8"),
method="POST",
)
req.add_header("Content-Type", "application/json")
with request.urlopen(req, timeout=timeout_seconds):
return