feat: pybot 新增 @Camera 摄像头块——picamera2 惰性导入，后台守护线程维护有界帧队列（满丢最旧）

pybot/pybot/hardvars.py

@@ -12,6 +12,7 @@ JsonValue 通信协议改为原生 dict/str（无跨语言边界后无需序列
 import logging
 
 from .hd import i2c as hd_i2c
+from .hd.camera import Camera
 from .hd.mpu6050 import MPU6050
 from .hd.servo import Servo
 from .hd.speaker import Speaker
@@ -186,3 +187,41 @@ class SpeakerVar(HardwareVar):
                           volumn=volumn, max_f=max_f, min_f=min_f,
                           dry_run=dry_run)
         return cls(name, speaker)
+
+
+# ---------------- Camera 摄像头变量（pybot 新增，无仓颉版对应） ----------------
+
+class CameraVar(HardwareVar):
+    """摄像头：build 启动后台抓帧线程维护有界队列；play() 返回最新帧"""
+
+    def __init__(self, name, camera):
+        super().__init__(name)
+        self.camera = camera
+
+    def build(self):
+        log.info("Camera HVar: %s is starting now", self.name)
+        self.camera.setup()
+        self.camera.start_daemon()
+
+    def play(self, params):
+        """返回 (时间戳, frame) 或 None（队列空 / dry_run）"""
+        return self.camera.latest()
+
+    def destroy(self):
+        log.info("Camera HVar: %s is closing now", self.name)
+        self.camera.stop_daemon()
+        self.camera.wait_daemon()
+        self.camera.close()
+
+    @classmethod
+    def from_matrix(cls, matrix, i, M, dry_run=False, **kwargs):
+        name = matrix[i, matrix.index_of_col(0, "name")]
+        width = parse_mt(matrix, i, matrix.index_of_col(0, "width"), int)
+        height = parse_mt(matrix, i, matrix.index_of_col(0, "height"), int)
+        framerate = parse_mt(matrix, i, matrix.index_of_col(0, "framerate"))
+        capacity = parse_mt(matrix, i, matrix.index_of_col(0, "capacity"), int)
+        log.info("Create a Camera hard-Var: %s, %dx%d @ %.1ffps, capacity: %s",
+                 name, width, height, framerate, capacity)
+        camera = Camera(width=width, height=height, framerate=framerate,
+                        capacity=capacity, dry_run=dry_run)
+        return cls(name, camera)

pybot/pybot/hd/camera.py

@@ -0,0 +1,126 @@
+"""摄像头帧队列 —— pybot 新增（无仓颉版对应）
+
+后台守护线程按固定帧率抓帧，维护一个有界帧队列
+（deque(maxlen=capacity)，满了丢最旧帧）——实时控制场景永远消费最新帧。
+生命周期模式对齐 MPU6050 采样线程：
+    setup() → start_daemon() → latest()/snapshot() → stop_daemon()/wait_daemon() → close()
+
+帧来源 picamera2（树莓派 OS 预装）是可选依赖：仅在 setup() 内惰性导入，
+不使用 @Camera 的脚本 / 单测 / --dry-run 都不需要它，零依赖承诺不变。
+测试可注入 grabber（callable() -> frame）完全脱离硬件。
+
+dry_run 模式：setup/start 均为空操作（不碰硬件），latest() 恒为 None。
+"""
+
+import logging
+import threading
+import time
+from collections import deque
+
+log = logging.getLogger("pybot.camera")
+
+
+class Camera:
+    """picamera2 摄像头 + 后台有界帧队列
+
+    width/height: 帧尺寸；framerate: 抓帧帧率；capacity: 队列容量（满丢最旧）；
+    grabber: 可注入的抓帧回调（测试用），None = 用 picamera2 抓帧。
+    """
+
+    def __init__(self, width=640, height=480, framerate=30.0, capacity=4,
+                 grabber=None, dry_run=False):
+        self.width = width
+        self.height = height
+        self.framerate = framerate
+        self.capacity = capacity
+        self.dry_run = dry_run
+        self._grabber = grabber
+        self._picam = None
+        self._queue = deque(maxlen=capacity)
+        self._lock = threading.Lock()
+        self._stop = threading.Event()
+        self._thread = None
+
+    # ---------------- 初始化 ----------------
+
+    def setup(self):
+        """打开摄像头（dry_run / 已初始化 / 注入 grabber 时不碰硬件，幂等）"""
+        if self.dry_run or self._grabber is not None:
+            return
+        try:
+            from picamera2 import Picamera2     # 可选依赖，惰性导入
+        except ImportError:
+            raise RuntimeError(
+                "@Camera 需要可选依赖 picamera2（树莓派 OS 预装，勿 pip 安装）；"
+                "无硬件环境请用 --dry-run 或注入 grabber") from None
+        self._picam = Picamera2()
+        self._picam.configure(self._picam.create_video_configuration(
+            main={"size": (self.width, self.height)}))
+        self._picam.start()
+        self._grabber = self._picam.capture_array
+        time.sleep(0.1)     # 等传感器稳定（对齐 MPU6050 setup）
+        log.debug("camera setup done (%dx%d @ %.1ffps)",
+                  self.width, self.height, self.framerate)
+
+    # ---------------- 帧访问（线程安全） ----------------
+
+    def latest(self):
+        """最新一帧 (monotonic 时间戳, frame)；队列空返回 None"""
+        with self._lock:
+            return self._queue[-1] if self._queue else None
+
+    def snapshot(self):
+        """队列快照 list（旧→新），供消费方批量取帧"""
+        with self._lock:
+            return list(self._queue)
+
+    @property
+    def size(self):
+        """当前队列内帧数"""
+        with self._lock:
+            return len(self._queue)
+
+    # ---------------- 抓帧线程 ----------------
+
+    def start_daemon(self):
+        if self.dry_run or self._thread is not None:
+            return
+        self._stop.clear()
+        self._thread = threading.Thread(target=self._loop,
+                                        name="camera-grabber", daemon=True)
+        self._thread.start()
+        log.debug("camera grabber started")
+
+    def _grab_once(self):
+        """抓一帧入队（满则挤掉最旧帧）；单独成方法便于无线程单测"""
+        frame = self._grabber()
+        with self._lock:
+            self._queue.append((time.monotonic(), frame))
+
+    def _loop(self):
+        interval = 1.0 / self.framerate
+        while not self._stop.wait(interval):
+            try:
+                self._grab_once()
+            except Exception:
+                # grabber 异常类型取决于帧源实现，守护线程不许死
+                log.warning("camera grab failed", exc_info=True)
+
+    def stop_daemon(self):
+        self._stop.set()
+
+    def wait_daemon(self):
+        if self._thread is not None:
+            self._thread.join()
+            self._thread = None
+            log.debug("camera grabber stopped")
+
+    # ---------------- 释放 ----------------
+
+    def close(self):
+        """关闭摄像头；幂等"""
+        if self._picam is not None:
+            self._picam.stop()
+            self._picam.close()
+            self._picam = None
+            log.debug("camera closed")

pybot/pybot/robot.py

@@ -13,7 +13,8 @@ import atexit
 import logging
 
 from . import plugin
-from .hardvars import (AliasVar, I2CVar, MPU6050Var, ServoVar, SpeakerVar)
+from .hardvars import (AliasVar, CameraVar, I2CVar, MPU6050Var, ServoVar,
+                       SpeakerVar)
 from .parser import MatrixParser, MatrixSaver, get_mt, parse_mt
 
 log = logging.getLogger("pybot.robot")
@@ -25,6 +26,7 @@ _FACTORIES = {
     "Servo": ServoVar,
     "Speaker": SpeakerVar,
     "MPU6050_BIND": MPU6050Var,
+    "Camera": CameraVar,
 }