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).
90 lines
3.6 KiB
Python
90 lines
3.6 KiB
Python
from dataclasses import dataclass
|
|
|
|
from edge.tracking import Track
|
|
from edge.models import L_ANKLE, R_ANKLE
|
|
|
|
# Below this an ankle keypoint is treated as unusable (typically hallucinated at conf~0).
|
|
MIN_ANKLE_CONF = 0.2
|
|
|
|
|
|
def point_in_polygon(x: float, y: float, poly: list[tuple[float, float]]) -> bool:
|
|
inside = False
|
|
n = len(poly)
|
|
j = n - 1
|
|
for i in range(n):
|
|
xi, yi = poly[i]
|
|
xj, yj = poly[j]
|
|
if ((yi > y) != (yj > y)) and (x < (xj - xi) * (y - yi) / ((yj - yi) or 1e-9) + xi):
|
|
inside = not inside
|
|
j = i
|
|
return inside
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Zone:
|
|
zone_id: str
|
|
level: str # "warning" | "restricted"
|
|
polygon: list # list[tuple[float,float]]
|
|
# 선택적 존 규칙 (occupancy.py가 소비): {"min_workers": 2, "max_workers": 4,
|
|
# "max_dwell_s": 600} — 키가 없으면 그 규칙은 꺼진 것. None = 규칙 없음.
|
|
rules: dict | None = None
|
|
|
|
|
|
@dataclass
|
|
class ZoneConfig:
|
|
debounce_s: float = 0.75
|
|
|
|
|
|
class ZoneIntrusionDetector:
|
|
def __init__(self, zones: list[Zone], cfg: ZoneConfig):
|
|
self.zones = zones
|
|
self.cfg = cfg
|
|
self._inside_since: dict[tuple[int, str], float] = {}
|
|
self._fired: set[tuple[int, str]] = set()
|
|
|
|
def forget(self, track_id: int) -> None:
|
|
self._inside_since = {k: v for k, v in self._inside_since.items() if k[0] != track_id}
|
|
self._fired = {k for k in self._fired if k[0] != track_id}
|
|
|
|
def _foot(self, track: Track, frame_wh=None) -> tuple[float, float]:
|
|
p = track.latest
|
|
a, b = p.keypoints[L_ANKLE], p.keypoints[R_ANKLE]
|
|
# Average only the ankles the model is confident about. A hallucinated ankle (conf~0,
|
|
# often pinned to (0,0) or a frame corner) would otherwise drag the foot point into or
|
|
# out of a zone. With neither ankle trustworthy, fall back to the bbox bottom-centre — a
|
|
# robust standing-foot proxy (same bbox-fallback discipline as fall.py's cues).
|
|
good = [k for k in (a, b) if k.conf >= MIN_ANKLE_CONF]
|
|
if good:
|
|
fx = sum(k.x for k in good) / len(good)
|
|
fy = sum(k.y for k in good) / len(good)
|
|
else:
|
|
fx, fy = p.bbox.cx, p.bbox.y2
|
|
# Zones are stored NORMALIZED (0..1 of the image) by the dashboard, which stretches the
|
|
# camera snapshot to a fixed canvas. The foot is camera-native pixels, so normalize it
|
|
# by the frame size to compare in the same space — otherwise every zone is misplaced
|
|
# (a zone drawn at image-centre would be checked against a raw pixel like 480, matching
|
|
# nothing on a 1280-wide frame). frame_wh=None keeps raw coords for unit tests that
|
|
# supply their own matching space.
|
|
if frame_wh:
|
|
w, h = frame_wh
|
|
if w > 0 and h > 0:
|
|
fx, fy = fx / w, fy / h
|
|
return (fx, fy)
|
|
|
|
def active_zone_ids(self, track_id: int) -> set:
|
|
return {z for (tid, z) in self._inside_since if tid == track_id}
|
|
|
|
def update(self, track: Track, now: float, frame_wh=None) -> list[dict]:
|
|
fx, fy = self._foot(track, frame_wh)
|
|
out = []
|
|
for zone in self.zones:
|
|
key = (track.track_id, zone.zone_id)
|
|
if point_in_polygon(fx, fy, zone.polygon):
|
|
start = self._inside_since.setdefault(key, now)
|
|
if now - start >= self.cfg.debounce_s and key not in self._fired:
|
|
self._fired.add(key)
|
|
out.append({"zone_id": zone.zone_id, "level": zone.level})
|
|
else:
|
|
self._inside_since.pop(key, None)
|
|
self._fired.discard(key)
|
|
return out
|