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).
95 lines
4.9 KiB
Python
95 lines
4.9 KiB
Python
from dataclasses import dataclass
|
|
|
|
|
|
@dataclass
|
|
class OccupancyConfig:
|
|
lone_confirm_s: float = 30.0 # 단독 작업이 이 시간 지속돼야 발보 (동료의 잠깐 이탈 흡수)
|
|
over_confirm_s: float = 10.0 # 정원 초과 지속 요구 (경계 통과 순간의 겹침 흡수)
|
|
dwell_absence_grace_s: float = 5.0 # 이 시간 안의 미검출은 체류 연속으로 간주 (검출 드롭아웃 흡수)
|
|
|
|
|
|
class OccupancyWatchdog:
|
|
"""존 규칙(2인 1조 · 정원 · 체류 상한) 감시.
|
|
|
|
존 멤버십은 ZoneIntrusionDetector가 이미 계산한 것을 엔진이 존별 인원으로 뒤집어
|
|
넘겨준다 — 여기서는 기하 계산이 없다. 규칙은 Zone.rules(대시보드 존 편집기에서
|
|
설정, /api/config로 동기)에서 읽으므로 존 목록 교체(apply_zones)에 자동으로 따라간다.
|
|
|
|
발보는 조건이 해소될 때까지 존(또는 트랙·존)당 1회 — 재발보는 엔진 쿨다운이 아니라
|
|
조건 해소 후 재충족으로만 일어난다.
|
|
"""
|
|
|
|
def __init__(self, cfg: OccupancyConfig):
|
|
self.cfg = cfg
|
|
self._lone_since: dict[str, float] = {}
|
|
self._lone_fired: set[str] = set()
|
|
self._over_since: dict[str, float] = {}
|
|
self._over_fired: set[str] = set()
|
|
self._dwell: dict[tuple[int, str], list] = {} # (tid, zone_id) -> [since, last_seen]
|
|
self._dwell_fired: set[tuple[int, str]] = set()
|
|
|
|
def forget(self, track_id: int) -> None:
|
|
self._dwell = {k: v for k, v in self._dwell.items() if k[0] != track_id}
|
|
self._dwell_fired = {k for k in self._dwell_fired if k[0] != track_id}
|
|
|
|
def update(self, zones: list, zone_tracks: dict, now: float) -> list[dict]:
|
|
out: list[dict] = []
|
|
known: set[str] = set()
|
|
for zone in zones:
|
|
zid = zone.zone_id
|
|
known.add(zid)
|
|
rules = zone.rules or {}
|
|
tids = zone_tracks.get(zid, set())
|
|
n = len(tids)
|
|
|
|
# 2인 1조: 0 < 인원 < 최소 인원이 지속되면 발보. 빈 존은 위반이 아니다.
|
|
min_workers = rules.get("min_workers")
|
|
if min_workers and 0 < n < min_workers:
|
|
start = self._lone_since.setdefault(zid, now)
|
|
if now - start >= self.cfg.lone_confirm_s and zid not in self._lone_fired:
|
|
self._lone_fired.add(zid)
|
|
out.append({"kind": "lone_worker", "zone_id": zid,
|
|
"track_id": next(iter(tids)) if n == 1 else None,
|
|
"metrics": {"workers": n, "required": min_workers,
|
|
"alone_s": round(now - start, 1)}})
|
|
else:
|
|
self._lone_since.pop(zid, None)
|
|
self._lone_fired.discard(zid)
|
|
|
|
# 정원 초과
|
|
max_workers = rules.get("max_workers")
|
|
if max_workers and n > max_workers:
|
|
start = self._over_since.setdefault(zid, now)
|
|
if now - start >= self.cfg.over_confirm_s and zid not in self._over_fired:
|
|
self._over_fired.add(zid)
|
|
out.append({"kind": "overcapacity", "zone_id": zid, "track_id": None,
|
|
"metrics": {"workers": n, "limit": max_workers}})
|
|
else:
|
|
self._over_since.pop(zid, None)
|
|
self._over_fired.discard(zid)
|
|
|
|
# 체류 상한 (트랙별 연속 체류). 미검출 1~2틱(run_once의 stale 필터)으로 타이머가
|
|
# 리셋되면 10분 규칙이 사실상 무력화되므로, grace 안의 부재는 연속으로 간주한다.
|
|
max_dwell = rules.get("max_dwell_s")
|
|
if max_dwell:
|
|
for tid in tids:
|
|
rec = self._dwell.setdefault((tid, zid), [now, now])
|
|
rec[1] = now
|
|
if now - rec[0] >= max_dwell and (tid, zid) not in self._dwell_fired:
|
|
self._dwell_fired.add((tid, zid))
|
|
out.append({"kind": "overstay", "zone_id": zid, "track_id": tid,
|
|
"metrics": {"dwell_s": round(now - rec[0], 1),
|
|
"limit_s": max_dwell}})
|
|
|
|
# grace를 넘겨 존을 떠난(또는 트랙이 죽은) 체류 기록 정리 — 재진입은 새 체류다.
|
|
for key, rec in list(self._dwell.items()):
|
|
if now - rec[1] > self.cfg.dwell_absence_grace_s:
|
|
self._dwell.pop(key)
|
|
self._dwell_fired.discard(key)
|
|
# 삭제된 존의 잔여 상태 정리 (24/7 장치 메모리 누수 방지).
|
|
for d, fired in ((self._lone_since, self._lone_fired), (self._over_since, self._over_fired)):
|
|
for zid in list(d):
|
|
if zid not in known:
|
|
d.pop(zid, None)
|
|
fired.intersection_update(known)
|
|
return out
|