"""محرك VDE — المضيف: يسجّل الموديولات، يوزّع أحداث MQTT + البثّ، يعزل الأخطاء. - الوضع الحدثي: MQTT من Frigate → dispatch_event → on_event. - الوضع البثّي: عامل لكل (موديول، كاميرا) يقرأ البث بمعدّل fps → on_frame. - P4: كل استدعاء معزول (try/except)؛ موديول يفشل تكراراً يُعطَّل تلقائياً بلا إسقاط الباقي. """ import json import threading import time from .api import Context class Services: """خدمات مشتركة يوفّرها المحرك للموديولات عبر Context (لا شبكة مباشرة للموديول).""" def __init__(self, frame_fn, vlm_fn, log_fn, onnx_cls=None): self._frame = frame_fn # (camera) -> jpeg bytes | None self._vlm = vlm_fn # (feature_key, frames) -> dict self._log = log_fn self._onnx_cls = onnx_cls # OnnxDetector class (يُحقن لتسهيل الاختبار) self._onnx_cache = {} def frame(self, camera): return self._frame(camera) def burst(self, camera, n, gap): out = [] for i in range(max(1, n)): f = self._frame(camera) if f: out.append(f) if i < n - 1: time.sleep(gap) return out def onnx(self, path): if path not in self._onnx_cache: if self._onnx_cls is None: from analyzer import OnnxDetector self._onnx_cls = OnnxDetector self._onnx_cache[path] = self._onnx_cls(path) return self._onnx_cache[path] def vlm(self, feature_key, frames, camera=None): return self._vlm(feature_key, frames, camera) def log(self, msg): self._log(msg) class Slot: __slots__ = ("key", "camera", "detector", "ctx", "triggers", "cooldown_s", "version", "fails", "enabled", "_last") def __init__(self, key, camera, detector, ctx, triggers, cooldown_s, version): self.key = key self.camera = camera self.detector = detector self.ctx = ctx self.triggers = triggers self.cooldown_s = cooldown_s self.version = version self.fails = 0 self.enabled = True self._last = 0.0 class Engine: def __init__(self, services: Services, output, max_fails=5): self.services = services self.output = output # (key, camera, Detection) -> None self.max_fails = max_fails self.slots = {} # (key, camera) -> Slot self._compound = [] # slots التنبيهات المركّبة (تُغذّى بإطلاقات غيرها) self._threads = [] self._stop = threading.Event() # ---- تسجيل / إلغاء الموديولات (يُستدعى من التحميل الساخن) ---- def register(self, key, detector, camera, config, model_path, version, triggers=None): """key + triggers من التخصيص/الmanifest (سلطة)، وإلا من صنف الموديول.""" trig = triggers or detector.triggers detector.key = key # الموديول يعرف مفتاح ميزته ctx = Context(config, model_path, self.services, camera) try: detector.setup(ctx) except Exception as exc: self.services.log(f"[vde] setup failed {key}/{camera}: {exc}") return None slot = Slot(key, camera, detector, ctx, trig, float((config or {}).get("cooldown_s", 30)), version) self.slots[(key, camera)] = slot if getattr(detector, "conditions", None): # تنبيه مركّب: يستمع لإطلاقات غيره self._compound.append(slot) if "stream" in trig and not self._stop.is_set(): self._spawn_stream(slot) self.services.log(f"[vde] registered {key}/{camera} v{version} triggers={trig}") return slot def unregister(self, key, camera): slot = self.slots.pop((key, camera), None) if slot: slot.enabled = False # يُسكت عامل البثّ إن وُجد (ponytail: يبقى خاملاً) if slot in self._compound: self._compound.remove(slot) # ---- التوزيع (الوضع الحدثي) ---- def dispatch_event(self, camera, trigger, ev=None): """يوقظ الموديولات التي triggers لها يطابق trigger على هذه الكاميرا.""" ev = dict(ev or {}); ev.update(camera=camera, trigger=trigger) for slot in list(self.slots.values()): if slot.camera != camera or not slot.enabled: continue if trigger in slot.triggers: self._run(slot, slot.detector.on_event, ev) def tick_schedule(self): """نبضة الموديولات المجدولة (تُستدعى دورياً).""" now = time.time() for slot in list(self.slots.values()): det = slot.detector if "schedule" in slot.triggers and slot.enabled \ and now - slot._last >= max(1, det.interval_s): self._run(slot, det.on_event, {"camera": slot.camera, "trigger": "schedule"}) # ---- العزل والتنفيذ (P4) ---- def _run(self, slot, method, arg): now = time.time() # الكواشف العادية (ONNX/VLM): تُهدّأ بتخطّي الاستدعاء (توفير حوسبة). # الكواشف ذات الحالة (خمول/مركّب): تُستدعى دائماً، وتُهدّأ مخرجاتها فقط. if not slot.detector.always_run and now - slot._last < slot.cooldown_s: return try: det = method(arg, slot.ctx) slot.fails = 0 except Exception as exc: # عزل: خطأ موديول لا يُسقط المحرك slot.fails += 1 self.services.log(f"[vde] error {slot.key}/{slot.camera}: {exc}") if slot.fails >= self.max_fails: slot.enabled = False # تعطيل تلقائي self.services.log(f"[vde] AUTO-DISABLED {slot.key}/{slot.camera} " f"بعد {slot.fails} أخطاء") return if det and det.flagged and now - slot._last >= slot.cooldown_s: slot._last = now self._emit(slot, det) def _emit(self, slot, det): if det.snapshot is None: det.snapshot = self.services.frame(slot.camera) try: self.output(slot.key, slot.camera, det) except Exception as exc: self.services.log(f"[vde] output failed {slot.key}: {exc}") # تغذية التنبيهات المركّبة (لا نغذّي مركّباً من إطلاق مركّب — تجنّب الحلقة) if not getattr(slot.detector, "conditions", None): for cs in list(self._compound): if cs.camera == slot.camera and cs.enabled \ and slot.key in cs.detector.conditions: self._run(cs, cs.detector.on_event, {"trigger": "condition", "fired_key": slot.key, "camera": slot.camera}) # ---- الوضع البثّي ---- def _spawn_stream(self, slot): def worker(): period = 1.0 / max(0.1, slot.detector.fps) while not self._stop.is_set(): if slot.enabled: frame = self.services.frame(slot.camera) if frame: self._run(slot, slot.detector.on_frame, frame) time.sleep(period) t = threading.Thread(target=worker, daemon=True, name=f"vde-stream-{slot.key}-{slot.camera}") t.start() self._threads.append(t) # ---- تشغيل MQTT (Frigate) ---- def run_mqtt(self, host="localhost", port=1883): """يشترك في أحداث Frigate ويوزّعها. يحتاج mosquitto + تفعيل MQTT في Frigate.""" import paho.mqtt.client as mqtt def on_connect(client, *_): client.subscribe("frigate/events") client.subscribe("frigate/+/motion") client.subscribe("frigate/+/audio/+") # أحداث الصوت (صراخ/زجاج/طلق) def on_message(client, _u, msg): try: self._on_mqtt(msg.topic, msg.payload) except Exception as exc: self.services.log(f"[vde] mqtt parse: {exc}") try: # paho 2.x يتطلب إصدار API c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1) except (AttributeError, TypeError): c = mqtt.Client() # paho 1.x c.on_connect = on_connect c.on_message = on_message c.connect(host, port, 60) c.loop_forever() def start_schedule(self): """نبضة الموديولات المجدولة في خيط جانبي (تُبدأ مرة).""" threading.Thread(target=self._schedule_loop, daemon=True, name="vde-schedule").start() def _on_mqtt(self, topic, payload): if topic == "frigate/events": data = json.loads(payload) after = data.get("after") or data.get("before") or {} cam, label = after.get("camera"), after.get("label") if cam and label: self.dispatch_event(cam, f"object:{label}", {"label": label, "zones": after.get("current_zones", [])}) for z in after.get("current_zones", []): self.dispatch_event(cam, f"zone:{z}", {"zones": [z]}) elif topic.endswith("/motion"): if payload.decode(errors="ignore").strip().upper() == "ON": cam = topic.split("/")[1] self.dispatch_event(cam, "motion") else: # frigate//audio/