97 lines
2.6 KiB
Python
97 lines
2.6 KiB
Python
"""Stable event models emitted by the local motion capture agent."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from dataclasses import asdict, dataclass, field
|
|
from typing import Any, Literal
|
|
|
|
GestureName = Literal["rotate_left", "rotate_right", "zoom_in", "zoom_out", "confirm"]
|
|
GesturePhase = Literal["start", "active", "end", "discrete"]
|
|
|
|
|
|
def now_ms() -> int:
|
|
return int(time.time() * 1000)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GestureEvent:
|
|
gesture: GestureName
|
|
phase: GesturePhase = "discrete"
|
|
confidence: float = 1.0
|
|
intensity: float = 1.0
|
|
timestamp_ms: int = field(default_factory=now_ms)
|
|
seq: int = 0
|
|
source: str = "motion-agent"
|
|
mode: str = "single"
|
|
payload: dict[str, Any] = field(default_factory=dict)
|
|
type: Literal["gesture"] = "gesture"
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.to_dict(), ensure_ascii=False, separators=(",", ":"))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class StatusEvent:
|
|
connected: bool
|
|
camera_count: int
|
|
active_camera_ids: tuple[str, ...] = ()
|
|
mode: str = "single"
|
|
recognizer: str = "mediapipe-opencv"
|
|
fps: float = 0.0
|
|
last_gesture: str | None = None
|
|
error: str | None = None
|
|
timestamp_ms: int = field(default_factory=now_ms)
|
|
source: str = "motion-agent"
|
|
type: Literal["status"] = "status"
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.to_dict(), ensure_ascii=False, separators=(",", ":"))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HeartbeatEvent:
|
|
timestamp_ms: int = field(default_factory=now_ms)
|
|
source: str = "motion-agent"
|
|
type: Literal["heartbeat"] = "heartbeat"
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.to_dict(), ensure_ascii=False, separators=(",", ":"))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SkeletonJoint:
|
|
id: str
|
|
x: float
|
|
y: float
|
|
confidence: float = 1.0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SkeletonEvent:
|
|
joints: list[SkeletonJoint]
|
|
bones: list[tuple[str, str]]
|
|
matched_gesture: GestureName | None = None
|
|
confidence: float = 0.0
|
|
camera_id: str = "unknown"
|
|
timestamp_ms: int = field(default_factory=now_ms)
|
|
source: str = "motion-agent"
|
|
mode: str = "single"
|
|
type: Literal["skeleton"] = "skeleton"
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.to_dict(), ensure_ascii=False, separators=(",", ":"))
|