146 lines
3.9 KiB
Python
146 lines
3.9 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
|
|
|
|
PROTOCOL_VERSION = "motion.v2"
|
|
|
|
GestureName = Literal[
|
|
"rotate_left",
|
|
"rotate_right",
|
|
"rotate_up",
|
|
"rotate_down",
|
|
"zoom_in",
|
|
"zoom_out",
|
|
"focus_prev",
|
|
"focus_next",
|
|
"layer_prev",
|
|
"layer_next",
|
|
"confirm",
|
|
]
|
|
GesturePhase = Literal["start", "active", "end", "discrete"]
|
|
InputMode = Literal["single", "dual_redundant", "single_fallback", "calibrated_3d"]
|
|
|
|
|
|
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"
|
|
protocol_version: str = PROTOCOL_VERSION
|
|
camera_id: str = "unknown"
|
|
input_mode: str = "single"
|
|
fusion: dict[str, Any] | None = None
|
|
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"
|
|
protocol_version: str = PROTOCOL_VERSION
|
|
input_mode: str = "single"
|
|
armed: bool = False
|
|
paused: bool = False
|
|
devices_open: bool = False
|
|
recognizer: str = "mediapipe-opencv"
|
|
fps: float = 0.0
|
|
recognition_fps: float = 0.0
|
|
last_gesture: str | None = None
|
|
last_fusion_reason: str | None = None
|
|
enabled_gestures: tuple[str, ...] = ()
|
|
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)
|
|
protocol_version: str = PROTOCOL_VERSION
|
|
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"
|
|
protocol_version: str = PROTOCOL_VERSION
|
|
input_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=(",", ":"))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CommandResultEvent:
|
|
command: str
|
|
request_id: str | None = None
|
|
ok: bool = True
|
|
status: dict[str, Any] = field(default_factory=dict)
|
|
error: str | None = None
|
|
timestamp_ms: int = field(default_factory=now_ms)
|
|
protocol_version: str = PROTOCOL_VERSION
|
|
source: str = "motion-agent"
|
|
type: Literal["command_result"] = "command_result"
|
|
|
|
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=(",", ":"))
|