diff --git a/edge/__main__.py b/edge/__main__.py new file mode 100644 index 0000000..8744750 --- /dev/null +++ b/edge/__main__.py @@ -0,0 +1,4 @@ +from edge.daemon import main + +if __name__ == "__main__": + main() diff --git a/edge/daemon.py b/edge/daemon.py new file mode 100644 index 0000000..408892b --- /dev/null +++ b/edge/daemon.py @@ -0,0 +1,54 @@ +import os + + +def run_loop(runtime, client, queue, *, clock, sleep, iterations=None, + sync_every=150, heartbeat_every=900, flush_every=300, + frame_interval=0.0): + counters = {"frames": 0, "syncs": 0, "beats": 0, "flushes": 0, "errors": 0} + i = 0 + while iterations is None or i < iterations: + i += 1 + try: + runtime.step(clock()) + counters["frames"] += 1 + except Exception: + counters["errors"] += 1 + if sync_every and i % sync_every == 0: + cfg = client.fetch_config() + if cfg is not None: + from edge.zones import zones_from_config + runtime.apply_zones(zones_from_config(cfg)) + counters["syncs"] += 1 + if heartbeat_every and i % heartbeat_every == 0: + if client.send_heartbeat({"fps": 0.0, "frames": counters["frames"]}): + counters["beats"] += 1 + if flush_every and i % flush_every == 0: + queue.flush() + counters["flushes"] += 1 + if frame_interval: + sleep(frame_interval) + return counters + + +def main() -> None: # pragma: no cover (integration entrypoint, exercised on device) + import time + from edge.config import load_config + from edge.cloudclient import CloudClient + from edge.alert import PostClient + from edge.queue import OfflineQueue + from edge.runtime import PipelineRuntime + from edge.adapters.uvc_source import UvcSource + from edge.adapters.movenet_pose import MoveNetPoseAdapter + + cfg = load_config(dict(os.environ)) + nonce = lambda: os.urandom(8).hex() + now = time.time + base = cfg.cafe24_url.rstrip("/") + client = CloudClient(base, cfg.device_id, cfg.secret, nonce, now) + post = PostClient(base + "/api/events", cfg.device_id, cfg.secret, nonce, now) + queue = OfflineQueue(os.environ.get("QUEUE_DB", "queue.db"), post) + source = UvcSource(int(os.environ.get("UVC_DEVICE", "0"))) + pose = MoveNetPoseAdapter(os.environ["MOVENET_ONNX"]) + runtime = PipelineRuntime(cfg, source, pose, queue) + run_loop(runtime, client, queue, clock=now, sleep=time.sleep, iterations=None, + frame_interval=1.0 / float(os.environ.get("TARGET_FPS", "12"))) diff --git a/pyproject.toml b/pyproject.toml index a75cd4d..510cccc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,6 +7,9 @@ dependencies = ["numpy", "opencv-python", "requests"] [project.optional-dependencies] dev = ["pytest"] +[project.scripts] +move-detect-edge = "edge.daemon:main" + [tool.pytest.ini_options] testpaths = ["tests"] addopts = "-q" diff --git a/tests/test_daemon.py b/tests/test_daemon.py new file mode 100644 index 0000000..1a675be --- /dev/null +++ b/tests/test_daemon.py @@ -0,0 +1,43 @@ +from edge.daemon import run_loop + + +class _Runtime: + def __init__(self): self.steps = 0; self.applied = None + def step(self, now): self.steps += 1; return [] + def apply_zones(self, zones): self.applied = zones + + +class _Client: + def __init__(self): self.configs = 0; self.beats = 0 + def fetch_config(self): self.configs += 1; return {"zones": [], "thresholds": {}} + def send_heartbeat(self, m): self.beats += 1; return True + + +class _Queue: + def __init__(self): self.flushes = 0 + def flush(self): self.flushes += 1; return 0 + + +def test_loop_steps_and_periodics(): + rt, cl, q = _Runtime(), _Client(), _Queue() + ticks = iter(range(1000)) + counters = run_loop(rt, cl, q, clock=lambda: float(next(ticks)), + sleep=lambda s: None, iterations=10, + sync_every=5, heartbeat_every=5, flush_every=5) + assert rt.steps == 10 + assert cl.configs == 2 and cl.beats == 2 and q.flushes == 2 # at frames 5 and 10 + assert rt.applied == [] # zones applied from fetched config + + +def test_loop_survives_step_exception(): + class Boom(_Runtime): + def step(self, now): + self.steps += 1 + if self.steps == 2: raise RuntimeError("frame glitch") + return [] + rt = Boom() + counters = run_loop(rt, _Client(), _Queue(), clock=lambda: 0.0, + sleep=lambda s: None, iterations=3, + sync_every=99, heartbeat_every=99, flush_every=99) + assert rt.steps == 3 # kept going past the exception + assert counters["errors"] == 1