gx/vf-cuda-grid
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

controller: FrigateBridge cell borders с idle/motion state machine

Browse Source 
Permanent 1-2px рамка вокруг каждой cell для visual разделения.
При motion ON → рамка светится красным (config'и BorderTheme).

Logic:
  _ensure_borders()  Lazy init 4 borders (id="cell_N_border") при первом event
                     с idle стилем (#808080 width=2 opacity=0.4)
  _set_border_state(motion=bool) Upsert тот же overlay с motion (#FF0000 width=4 opacity=1.0)
                                  или idle styling
  _cell_states       set'у активных cams per cell ("inst:cell" → set(cam_names))
                     — border ON если хоть один cam имеет motion, OFF только когда
                     все cams cleared

BorderTheme:
  idle_color/width/opacity     subtle разделитель
  motion_color/width/opacity   alarm подсветка
  configurable через cfg.frigate.border_theme

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
gx
2026-05-20 20:43:34 +01:00
parent 96e6048b64
commit 26e9f30990
1 changed files with 73 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/cuda_grid_controller/frigate_bridge.py
+73 -20
View File
@@ -42,10 +42,24 @@ class FrigateCameraMapping(BaseModel):
bbox_overlay: bool = Field(default=True, description="Рисовать bbox для object detection events")
class BorderTheme(BaseModel):
"""Стиль cell border в трёх состояниях. Color = HEX RGB."""
idle_color: str = Field(default="#808080", description="Нейтральная разделительная рамка")
idle_width: int = Field(default=2, ge=1, le=8)
idle_opacity: float = Field(default=0.4, ge=0.0, le=1.0)
motion_color: str = Field(default="#FF0000", description="Цвет при frigate motion ON")
motion_width: int = Field(default=4, ge=1, le=16)
motion_opacity: float = Field(default=1.0, ge=0.0, le=1.0)
class FrigateBridgeCfg(BaseModel):
enabled: bool = False
base_topic: str = "frigate"
mappings: list[FrigateCameraMapping] = []
border_theme: BorderTheme = Field(default_factory=BorderTheme,
description="Стиль cell borders — idle/motion цвета")
class FrigateBridge:
@@ -60,8 +74,10 @@ class FrigateBridge:
self.cfg = cfg
self.dispatcher = dispatcher
self._by_camera = {m.frigate_camera: m for m in cfg.mappings if m.enabled}
self._motion_overlays: dict[str, str] = {}
self._motion_overlays: dict[str, str] = {} # legacy — unused if border theme active
self._event_overlays: dict[str, tuple[str, str]] = {}
self._borders_initialized: dict[str, bool] = {} # target_instance → bool
self._cell_states: dict[str, set[str]] = {} # "<inst>:<cell>" → set of active cameras с motion
def topics_to_subscribe(self) -> list[str]:
if not self.cfg.enabled:
@@ -69,10 +85,55 @@ class FrigateBridge:
base = self.cfg.base_topic.rstrip("/")
return [f"{base}/+/motion", f"{base}/events"]
def _cell_border_id(self, cell: int) -> str:
return f"cell_{cell}_border"
async def _ensure_borders(self, instance: str) -> None:
"""Lazy init: 4 idle borders для каждого cell в instance. Idempotent."""
if self._borders_initialized.get(instance) or not self.dispatcher:
return
cells = sorted({m.cell for m in self.cfg.mappings if m.target_instance == instance})
theme = self.cfg.border_theme
for cell in cells:
ov = RectOverlay(
id=self._cell_border_id(cell),
cell=cell,
x=0.0, y=0.0, w=1.0, h=1.0,
color=theme.idle_color,
border_only=True,
border_width=theme.idle_width,
opacity=theme.idle_opacity,
z_order=5,
)
await self.dispatcher.handle(instance, "overlay.add", ov.model_dump_json())
self._borders_initialized[instance] = True
log.info("frigate_bridge.borders_initialized", instance=instance, cells=cells)
async def _set_border_state(self, instance: str, cell: int, motion: bool) -> None:
"""Upsert cell border overlay в idle или motion стиль."""
if not self.dispatcher:
return
theme = self.cfg.border_theme
ov = RectOverlay(
id=self._cell_border_id(cell),
cell=cell,
x=0.0, y=0.0, w=1.0, h=1.0,
color=theme.motion_color if motion else theme.idle_color,
border_only=True,
border_width=theme.motion_width if motion else theme.idle_width,
opacity=theme.motion_opacity if motion else theme.idle_opacity,
z_order=5,
)
await self.dispatcher.handle(instance, "overlay.add", ov.model_dump_json())
async def handle_message(self, topic: str, payload: str) -> None:
if not self.cfg.enabled:
return
# Lazy init borders для всех target instances при первом event
for inst in {m.target_instance for m in self.cfg.mappings}:
await self._ensure_borders(inst)
base = self.cfg.base_topic.rstrip("/")
# frigate/<cam>/motion
@@ -100,27 +161,19 @@ class FrigateBridge:
if not self.dispatcher:
return
ov_id = f"motion_{cam}"
# Cell может содержать multiple cameras (теоретически) — border ON если
# хотя бы один cam имеет motion. Tracking через set'у per cell.
cell_key = f"{mapping.target_instance}:{mapping.cell}"
active = self._cell_states.setdefault(cell_key, set())
was_active = bool(active)
if state == "ON":
if cam in self._motion_overlays:
return # уже активен
ov = RectOverlay(
id=ov_id,
cell=mapping.cell,
x=0.0, y=0.0, w=1.0, h=1.0,
color="#FF8800",
border_only=True,
border_width=6,
z_order=10,
opacity=0.9,
)
self._motion_overlays[cam] = ov_id
await self.dispatcher.handle(mapping.target_instance, "overlay.add", ov.model_dump_json())
active.add(cam)
elif state == "OFF":
stored = self._motion_overlays.pop(cam, None)
if stored:
await self.dispatcher.handle(mapping.target_instance, "overlay.remove", stored)
active.discard(cam)
is_active = bool(active)
if is_active != was_active:
await self._set_border_state(mapping.target_instance, mapping.cell, motion=is_active)
async def _handle_event(self, event: dict) -> None:
event_type = event.get("type", "?")