a203c42bf3
Per-zone declarative safety rules, set in the zone editor and enforced on the edge from the tracking the pipeline already does (no new geometry, no new model): - min_workers (2인 1조): 0 < count < minimum sustained 30s -> lone_worker/HIGH. A solo worker in a buddy-rule zone is a compliance violation and a rescue- delay risk, so it outranks the other two. - max_workers: count > limit sustained 10s -> overcapacity/WARNING. - max_dwell_s: one track's continuous stay past the limit -> overstay/WARNING. A short detection dropout (<=5s grace) does NOT reset the dwell timer - without that, run_once's stale-track filter would quietly neuter every 10-minute rule at real-world detection rates. Edge: detectors/occupancy.py consumes Zone.rules; the engine flips the zone detector's per-track membership into per-zone headcounts (stale tracks already excluded), so counts always match what was actually detected this tick. Fired states clear on condition recovery - re-alerts need a fresh violation, not a cooldown expiry. Malformed rules_json keeps the zone's intrusion detection alive and logs the rules as inactive (never fail dark). Server: move_zones.rules_json (nullable JSON, migration included) validated strictly in ZoneConfigController (bounds, min<=max, unknown keys rejected - a silently dropped rule is one the operator believes is active). Delivered to the edge unchanged via GET /api/config. EventValidator accepts the 3 new types. UI: zone editor gains the three rule fields (dwell entered in minutes, wired in seconds) and shows each zone's rules in its list row; the events console and detail view now render every event type in Korean (wire codes fall back). Verified: pytest 216 pass / PHPUnit 123 pass + real-render harness in Chrome (editor save payload, list labels, console type labels).
91 lines
4.4 KiB
Python
91 lines
4.4 KiB
Python
import json
|
|
import logging
|
|
|
|
from edge.detectors.zone import Zone
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# Zones are stored NORMALIZED (0..1 of the image) by the current dashboard. Polygons drawn by the
|
|
# pre-normalization dashboard were saved in canvas PIXELS (values in the hundreds); after the edge
|
|
# normalizes the foot point, a pixel polygon can never match — silently killing that zone's
|
|
# intrusion detection with no error. Flag such rows loudly so the operator redraws them, instead
|
|
# of failing dark. A normalized coordinate may round slightly past 1.0, so allow a small margin.
|
|
_NORMALIZED_MAX = 1.5
|
|
|
|
|
|
def _looks_like_pixels(polygon: list[tuple[float, float]]) -> bool:
|
|
return any(abs(x) > _NORMALIZED_MAX or abs(y) > _NORMALIZED_MAX for x, y in polygon)
|
|
|
|
|
|
_RULE_KEYS = ("min_workers", "max_workers", "max_dwell_s")
|
|
|
|
|
|
def _rules_of(row: dict) -> dict | None:
|
|
"""Optional occupancy rules from the zone row's rules_json. A malformed rules blob must
|
|
not kill the zone's INTRUSION detection (the polygon is still valid) — raise here and the
|
|
caller keeps the zone but logs the rules loudly as inactive."""
|
|
raw = row.get("rules_json")
|
|
if raw in (None, "", "null"):
|
|
return None
|
|
data = json.loads(raw) if isinstance(raw, str) else raw
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"rules_json is not an object: {data!r}")
|
|
rules = {}
|
|
for key in _RULE_KEYS:
|
|
if data.get(key) is None:
|
|
continue
|
|
v = int(data[key])
|
|
if v <= 0:
|
|
raise ValueError(f"{key} must be positive, got {v}")
|
|
rules[key] = v
|
|
return rules or None
|
|
|
|
|
|
def zones_from_config(config: dict, camera_id: str | None = None) -> list[Zone]:
|
|
"""Parse server config zone rows into Zone objects. When camera_id is given, only
|
|
rows for that camera are kept (multi-camera routing); None keeps all rows."""
|
|
out: list[Zone] = []
|
|
rows = config.get("zones", []) or []
|
|
if not isinstance(rows, list):
|
|
return []
|
|
for row in rows:
|
|
try:
|
|
if camera_id is not None and str(row["camera_id"]) != str(camera_id):
|
|
continue
|
|
raw = row["polygon_json"]
|
|
pts = json.loads(raw) if isinstance(raw, str) else raw
|
|
polygon = [(float(p[0]), float(p[1])) for p in pts]
|
|
if len(polygon) < 3:
|
|
continue
|
|
if _looks_like_pixels(polygon):
|
|
log.warning(
|
|
"zone %s on camera %s has un-normalized (pixel-space) coordinates %r — it "
|
|
"would match nothing after foot normalization. Skipping; redraw this zone in "
|
|
"the dashboard so it is stored normalized (0..1).",
|
|
row.get("zone_id"), row.get("camera_id"), polygon[:2])
|
|
continue
|
|
try:
|
|
rules = _rules_of(row)
|
|
except (ValueError, TypeError, json.JSONDecodeError) as e:
|
|
# Same "never fail dark" principle: the polygon is fine, so intrusion detection
|
|
# stays live — but the occupancy rules the operator THINKS are active are not.
|
|
log.warning("zone %s on camera %s has malformed rules_json (%s: %s) — its "
|
|
"occupancy rules are INACTIVE until fixed in the dashboard.",
|
|
row.get("zone_id"), row.get("camera_id"), type(e).__name__, e)
|
|
rules = None
|
|
out.append(Zone(str(row["zone_id"]), str(row["level"]), polygon, rules))
|
|
except (KeyError, ValueError, TypeError, IndexError, AttributeError, json.JSONDecodeError) as e:
|
|
# Same principle as the pixel-space branch above: never fail dark. A row missing a key
|
|
# (level/zone_id), with unparseable polygon_json, or degenerate points would otherwise be
|
|
# dropped silently — intrusion detection for that zone dead with no log, no dashboard
|
|
# signal. Flag it so it can be fixed instead of discovered after a worker enters the zone.
|
|
zid = row.get("zone_id") if isinstance(row, dict) else None
|
|
log.warning("skipping malformed zone row %r (%s: %s) — its intrusion detection is "
|
|
"inactive until the row is fixed in the dashboard.", zid, type(e).__name__, e)
|
|
continue
|
|
return out
|
|
|
|
|
|
def thresholds_from_config(config: dict) -> dict:
|
|
t = config.get("thresholds")
|
|
return t if isinstance(t, dict) else {}
|