move_detect/edge/zones.py
유창욱 fe6a144754 fix(edge): stop hiding a failed camera; fix ghost dedup leak + silent zone drop
- Heartbeat reports the CONFIGURED camera count, not just the started ones: a
  camera that failed to construct at boot (wrong index / unseated ribbon) now
  shows as total/erroring instead of vanishing, so the server can see it is dark.
- GHOST emits without an engine cooldown key (ghost.update already fires once per
  never-reused track id) — removes a permanent _last_emit key that forget() could
  never reclaim (a slow 24/7 leak).
- Malformed zone rows are logged loudly instead of silently swallowed, matching
  the pixel-space branch's "don't fail dark" — a dead zone is no longer discovered
  only after a worker enters it.
2026-07-25 23:18:19 +09:00

58 lines
2.9 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)
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
out.append(Zone(str(row["zone_id"]), str(row["level"]), polygon))
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 {}