mock_vision.py

# -*- coding: utf-8 -*-
"""Mock vision publisher：在 GroundingDINO 未就緒前，模擬 vision_events。

用法：
    python -m tools.mock_vision --host 127.0.0.1 --port 6379 --rate 1.0 --packages P_mock_001,P_mock_002

對應 interfaces.md §9.1。
"""
from __future__ import annotations

import argparse
import json
import random
import time
from dataclasses import dataclass, field
from typing import Dict, List

import redis


@dataclass
class MockPackage:
    package_id: str
    zone: str = "left_conveyor"
    barcode_face: str = "down"
    # 模擬步驟計數，用來推進狀態
    tick: int = 0
    finished: bool = False
    size: Dict[str, float] = field(
        default_factory=lambda: {"length": 0.22, "width": 0.15, "height": 0.10}
    )

    def step(self) -> None:
        """每個 tick 推進一點，模擬機器人在處理它。"""
        if self.finished:
            return
        self.tick += 1
        # zone 推進邏輯：left_conveyor → workspace → right_conveyor
        if self.tick == 2:
            self.zone = "workspace"
        elif self.tick == 5:
            # 假設此時條碼朝上
            self.barcode_face = "up"
        elif self.tick == 8:
            self.zone = "right_conveyor"
        elif self.tick == 11:
            self.finished = True

    def to_object(self) -> Dict:
        return {
            "id": self.package_id,
            "class": "package_box",
            "bbox": [120, 80, 380, 290],
            "score": 0.93,
            "pose_3d": {"position": [0.42, 0.10, 0.85], "yaw": 0.30},
            "size": self.size,
            "barcode_face": self.barcode_face,
            "barcode_score": 0.81,
            "zone": self.zone,
        }


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="Mock vision_events publisher")
    p.add_argument("--host", default="127.0.0.1")
    p.add_argument("--port", type=int, default=6379)
    p.add_argument("--db", type=int, default=0)
    p.add_argument("--password", default=None)
    p.add_argument("--channel", default="vision_events")
    p.add_argument(
        "--packages",
        default="P_mock_001",
        help="comma-separated package ids to simulate",
    )
    p.add_argument("--rate", type=float, default=1.0, help="frames per second")
    p.add_argument("--shuffle-faces", action="store_true", help="random barcode_face")
    return p.parse_args()


def main() -> None:
    args = parse_args()
    r = redis.Redis(
        host=args.host, port=args.port, db=args.db, password=args.password
    )
    pkgs: List[MockPackage] = [MockPackage(package_id=pid.strip()) for pid in args.packages.split(",") if pid.strip()]
    frame_idx = 0
    interval = 1.0 / max(args.rate, 0.1)

    print(f"[mock_vision] publishing to {args.channel} @ {args.rate} Hz, packages={[p.package_id for p in pkgs]}")
    try:
        while True:
            frame_idx += 1
            # 過濾完成的包裹
            active = [p for p in pkgs if not p.finished]
            if not active:
                print("[mock_vision] all packages finished, exiting")
                break

            if args.shuffle_faces:
                for p in active:
                    p.barcode_face = random.choice(["up", "down", "side"])
            else:
                for p in active:
                    p.step()

            payload = {
                "event": "vision_update",
                "ts": time.time(),
                "frame_id": f"img_{frame_idx:06d}",
                "camera": "head",
                "image_path": None,
                "objects": [p.to_object() for p in active],
            }
            r.publish(args.channel, json.dumps(payload))
            print(
                f"[mock_vision] frame={frame_idx} active={len(active)} "
                f"states={[(p.package_id, p.zone, p.barcode_face) for p in active]}"
            )
            time.sleep(interval)
    except KeyboardInterrupt:
        print("\n[mock_vision] interrupted")


if __name__ == "__main__":
    main()