feat(edge): parse cloud config zones/thresholds into Zone objects
This commit is contained in:
parent
316d80e696
commit
3cf702e440
2 changed files with 44 additions and 0 deletions
23
edge/zones.py
Normal file
23
edge/zones.py
Normal file
|
|
@ -0,0 +1,23 @@
|
||||||
|
import json
|
||||||
|
|
||||||
|
from edge.detectors.zone import Zone
|
||||||
|
|
||||||
|
|
||||||
|
def zones_from_config(config: dict) -> list[Zone]:
|
||||||
|
out: list[Zone] = []
|
||||||
|
for row in config.get("zones", []) or []:
|
||||||
|
try:
|
||||||
|
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
|
||||||
|
out.append(Zone(str(row["zone_id"]), str(row["level"]), polygon))
|
||||||
|
except (KeyError, ValueError, TypeError, json.JSONDecodeError):
|
||||||
|
continue
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def thresholds_from_config(config: dict) -> dict:
|
||||||
|
t = config.get("thresholds")
|
||||||
|
return t if isinstance(t, dict) else {}
|
||||||
21
tests/test_zones.py
Normal file
21
tests/test_zones.py
Normal file
|
|
@ -0,0 +1,21 @@
|
||||||
|
from edge.zones import zones_from_config, thresholds_from_config
|
||||||
|
|
||||||
|
def test_parses_zone_rows_json_string_polygon():
|
||||||
|
cfg = {"zones": [
|
||||||
|
{"zone_id": "z1", "camera_id": "cam-1", "level": "restricted",
|
||||||
|
"polygon_json": "[[0,0],[9,0],[9,9]]"},
|
||||||
|
], "thresholds": {"fall_prone_s": 3.0}}
|
||||||
|
zones = zones_from_config(cfg)
|
||||||
|
assert len(zones) == 1
|
||||||
|
assert zones[0].zone_id == "z1"
|
||||||
|
assert zones[0].level == "restricted"
|
||||||
|
assert zones[0].polygon == [(0.0, 0.0), (9.0, 0.0), (9.0, 9.0)]
|
||||||
|
assert thresholds_from_config(cfg)["fall_prone_s"] == 3.0
|
||||||
|
|
||||||
|
def test_skips_malformed_row():
|
||||||
|
cfg = {"zones": [{"zone_id": "bad"}], "thresholds": {}}
|
||||||
|
assert zones_from_config(cfg) == []
|
||||||
|
|
||||||
|
def test_empty_config():
|
||||||
|
assert zones_from_config({}) == []
|
||||||
|
assert thresholds_from_config({}) == {}
|
||||||
Loading…
Reference in a new issue