events.py

# -*- coding: utf-8 -*-
"""Event dataclasses consumed/produced by the Task Manager.

對應 interfaces.md §2 的 channel schema。
本檔只放純 dataclass + 解析函式，**不** import collaborator / flask。
"""
from __future__ import annotations

import json
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional


BARCODE_FACE_VALUES = {"up", "down", "side", "unknown"}
ZONE_VALUES = {"left_conveyor", "workspace", "right_conveyor", "unknown"}
SKILL_PHASE_VALUES = {"begin", "progress", "end"}
SKILL_STATUS_VALUES = {"ok", "fail", "timeout", "cancelled"}


@dataclass
class VisionObservation:
    """單一包裹的視覺觀測（從 vision_events.objects[] 取出的「我們關心的欄位」）。"""

    package_id: str
    zone: str = "unknown"
    barcode_face: str = "unknown"
    score: float = 0.0
    barcode_score: float = 0.0
    bbox: Optional[List[int]] = None
    pose_3d: Optional[Dict[str, Any]] = None
    size: Optional[Dict[str, float]] = None
    ts: float = field(default_factory=time.time)
    frame_id: str = ""

    @classmethod
    def from_object_dict(
        cls, obj: Dict[str, Any], ts: float, frame_id: str
    ) -> "VisionObservation":
        zone = obj.get("zone", "unknown")
        if zone not in ZONE_VALUES:
            zone = "unknown"
        face = obj.get("barcode_face", "unknown")
        if face not in BARCODE_FACE_VALUES:
            face = "unknown"
        return cls(
            package_id=obj.get("id", "unknown"),
            zone=zone,
            barcode_face=face,
            score=float(obj.get("score", 0.0)),
            barcode_score=float(obj.get("barcode_score", 0.0) or 0.0),
            bbox=obj.get("bbox"),
            pose_3d=obj.get("pose_3d"),
            size=obj.get("size"),
            ts=ts,
            frame_id=frame_id,
        )


@dataclass
class VisionFrame:
    """整幀視覺事件。"""

    ts: float
    frame_id: str
    camera: str
    image_path: Optional[str]
    observations: List[VisionObservation]

    @classmethod
    def from_json(cls, raw: str) -> "VisionFrame":
        data = json.loads(raw)
        ts = float(data.get("ts", time.time()))
        frame_id = data.get("frame_id", "")
        observations = [
            VisionObservation.from_object_dict(o, ts, frame_id)
            for o in data.get("objects", [])
        ]
        return cls(
            ts=ts,
            frame_id=frame_id,
            camera=data.get("camera", "head"),
            image_path=data.get("image_path"),
            observations=observations,
        )


@dataclass
class SkillEvent:
    """來自 Skill Manager 的技能事件（interfaces.md §2.2）。"""

    skill_name: str
    phase: str = "end"
    status: str = "ok"
    task_id: str = ""
    subtask_id: str = ""
    robot_name: str = ""
    fault_code: Optional[str] = None
    ts: float = field(default_factory=time.time)
    details: Dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_json(cls, raw: str) -> "SkillEvent":
        data = json.loads(raw)
        phase = data.get("phase", "end")
        if phase not in SKILL_PHASE_VALUES:
            phase = "end"
        status = data.get("status", "ok")
        if status not in SKILL_STATUS_VALUES:
            status = "fail"
        return cls(
            skill_name=data.get("skill_name", ""),
            phase=phase,
            status=status,
            task_id=data.get("task_id", ""),
            subtask_id=data.get("subtask_id", ""),
            robot_name=data.get("robot_name", ""),
            fault_code=data.get("fault_code"),
            ts=float(data.get("ts", time.time())),
            details=data.get("details", {}) or {},
        )

    @property
    def is_terminal(self) -> bool:
        return self.phase == "end"


@dataclass
class FaultRecord:
    """任務級故障記錄（寫到 task:{task_id}:faults，也廣播到 task_faults channel）。"""

    code: str
    name: str
    severity: str  # info / warn / error / critical
    task_id: str
    subtask_id: Optional[str] = None
    package_id: Optional[str] = None
    ts: float = field(default_factory=time.time)
    details: Dict[str, Any] = field(default_factory=dict)
    action_taken: str = "none"

    def to_dict(self) -> Dict[str, Any]:
        return {
            "event": "task_fault",
            "ts": self.ts,
            "code": self.code,
            "name": self.name,
            "severity": self.severity,
            "task_id": self.task_id,
            "subtask_id": self.subtask_id,
            "package_id": self.package_id,
            "details": self.details,
            "action_taken": self.action_taken,
        }


@dataclass
class ProgressEvent:
    """推送到 task_progress channel / WebSocket 的事件（interfaces.md §2.3）。"""

    task_id: str
    fsm_state: str
    fsm_prev_state: Optional[str] = None
    subtask_id: Optional[str] = None
    package_id: Optional[str] = None
    progress: float = 0.0
    barcode_face: Optional[str] = None
    zone: Optional[str] = None
    queue_active: int = 0
    queue_pending: int = 0
    ts: float = field(default_factory=time.time)

    def to_dict(self) -> Dict[str, Any]:
        return {
            "event": "task_progress",
            "ts": self.ts,
            "task_id": self.task_id,
            "subtask_id": self.subtask_id,
            "package_id": self.package_id,
            "fsm_state": self.fsm_state,
            "fsm_prev_state": self.fsm_prev_state,
            "progress": self.progress,
            "barcode_face": self.barcode_face,
            "zone": self.zone,
            "queue": {"active": self.queue_active, "pending": self.queue_pending},
        }


__all__ = [
    "VisionObservation",
    "VisionFrame",
    "SkillEvent",
    "FaultRecord",
    "ProgressEvent",
    "BARCODE_FACE_VALUES",
    "ZONE_VALUES",
]