Merge branch 'main' of https://git.cancerai.top/qing/blackbird

actpred/README.md

@@ -9,7 +9,7 @@ Transformer（pre-LN）自注意力融合，取 [PRED] token 回归下一步动
 ## 架构
 
 ```
-F 帧图像 (F×3×64×64) ─► 共享 CNN 编码器 ─► F 个 token ─► proj → d_model ┐
+F 帧图像 (F×3×224×224) ─► 共享 CNN 编码器 ─► F 个 token ─► proj → d_model ┐
                                                                           ├─► [img×F, act×H, PRED] + 位置编码
 动作历史 (H×D)      ─► Linear proj → d_model ─► H 个 token ────────┤        │
 可学习 [PRED] token ────────────────────────────────────────┘        ▼
@@ -110,7 +110,7 @@ recorder.flush()
 
 | 参数 | 默认值 | 说明 |
 |------|--------|------|
-| `--image-size` | 64 | 图像缩放尺寸 |
+| `--image-size` | 224 | 图像缩放尺寸 |
 | `--action-dim` | 18 | 动作向量维度（对齐 bot4cj 18 列：10 舵机 + 2 云台 + 扬声器频率 + 延时 + 4 预留）|
 | `--history-len` | 5 | 历史动作窗口长度 |
 | `--frame-len` | 3 | 输入图像帧数（含当前帧，旧帧在前）|

actpred/config.py

@@ -5,7 +5,7 @@ from typing import Optional
 
 @dataclass
 class DataConfig:
-    image_size: int = 64
+    image_size: int = 224
     action_dim: int = 18
     history_len: int = 5
     frame_len: int = 3

actpred/dataset.py

@@ -16,7 +16,7 @@ def _natural_key(p):
 
 class ActionDataset(Dataset):
     def __init__(self, data_dir: str, history_len: int, action_dim: int,
-                 image_size: int = 64, normalize_actions: bool = False,
+                 image_size: int = 224, normalize_actions: bool = False,
                  action_stats_path: str = None, episode_indices: list = None,
                  frame_len: int = 3):
         self.history_len = history_len

actpred/mt_parser.py

@@ -1,7 +1,9 @@
 """轻量级 .mt 脚本 @Action 矩阵解析器
 
 从 .mt 文件中提取 @Action 块，解析为 (n_rows, n_cols) 的 float32 矩阵。
-支持 `-` 占位符（沿用上一行同列的值）；首行 `-` 从 @Servo 块的初始角度填充。
+支持 `-` 占位符（沿用上一行同列的值）；首行 `-` 填 0，
+与 pybot/bot4cj 运行时语义一致（2026-08 统一，见
+docs/reports/bot4cj_python_migration_dev_doc.md §2.3）。
 """
 
 import re
@@ -15,27 +17,6 @@ def _natural_key(p: Path):
     return [int(t) if t.isdigit() else t for t in re.split(r"(\d+)", p.name)]
 
 
-def _parse_servo_init(text: str) -> dict:
-    """从 @Servo 块提取 {servo_name: initial_angle}（max 列）"""
-    match = re.search(r"@Servo\s+(.*?)\s*@", text, re.DOTALL)
-    if not match:
-        return {}
-    servo_init = {}
-    for line in match.group(1).split("\n"):
-        line = re.sub(r"#.*$", "", line).strip()
-        if line.endswith(";"):
-            line = line[:-1].strip()
-        if not line:
-            continue
-        cols = re.split(r"\s+", line)
-        if len(cols) >= 2 and cols[0] != "name":
-            try:
-                servo_init[cols[0]] = float(cols[1])
-            except ValueError:
-                pass
-    return servo_init
-
-
 def parse_mt_action(path) -> np.ndarray:
     """解析 .mt 文件中的 @Action 矩阵
 
@@ -44,7 +25,7 @@ def parse_mt_action(path) -> np.ndarray:
 
     Returns:
         (n_data_rows, n_cols) float32 矩阵，表头行已跳过，`-` 已前向填充。
-        首行 `-` 从 @Servo 块的初始角度填充。
+        首行 `-` 填 0（与 pybot/bot4cj 运行时语义一致）。
 
     Raises:
         ValueError: 文件中无 @Action 块、数据行不足、行列不一致或数值非法
@@ -53,9 +34,6 @@ def parse_mt_action(path) -> np.ndarray:
     # 先去除整行注释（# 开头），避免注释中的 @Action 干扰正则匹配
     text = re.sub(r"(?m)^[ \t]*#.*$", "", text)
 
-    # 解析 @Servo 块的初始角度，用于首行 `-` 前向填充
-    servo_init = _parse_servo_init(text)
-
     match = re.search(r"@Action\s+(.*?)\s*@", text, re.DOTALL)
     if not match:
         raise ValueError(f"未找到 @Action 块: {path}")
@@ -84,12 +62,6 @@ def parse_mt_action(path) -> np.ndarray:
     data_rows = rows[1:]
     n_cols = len(header)
 
-    # 为每列计算初始值：s1.A → @Servo 中 s1 的 max 角度，其余为 0.0
-    col_init = []
-    for col_name in header:
-        servo_name = col_name.split(".")[0] if "." in col_name else None
-        col_init.append(servo_init.get(servo_name, 0.0))
-
     matrix = np.zeros((len(data_rows), n_cols), dtype=np.float32)
     for i, row in enumerate(data_rows):
         if len(row) != n_cols:
@@ -98,7 +70,7 @@ def parse_mt_action(path) -> np.ndarray:
                 f"不一致: {path}")
         for j, val in enumerate(row):
             if val == "-":
-                matrix[i, j] = matrix[i - 1, j] if i > 0 else col_init[j]
+                matrix[i, j] = matrix[i - 1, j] if i > 0 else 0.0
             else:
                 try:
                     matrix[i, j] = float(val)

actpred/test_actpred.py

@@ -263,7 +263,7 @@ class TestConfig(unittest.TestCase):
 
     def test_default_config(self):
         cfg = Config()
-        self.assertEqual(cfg.data.image_size, 64)
+        self.assertEqual(cfg.data.image_size, 224)
         self.assertEqual(cfg.data.action_dim, 18)
         self.assertEqual(cfg.data.history_len, 5)
         self.assertEqual(cfg.data.frame_len, 3)
@@ -321,8 +321,8 @@ class TestMtParser(unittest.TestCase):
         np.testing.assert_array_equal(matrix[0], [10, 20, 100])
         np.testing.assert_array_equal(matrix[2], [11, 22, 101])
 
-    def test_first_row_dash_uses_servo_init(self):
-        """首行 `-` 从 @Servo 初始角度（max 列）填充，无映射列填 0"""
+    def test_first_row_dash_fills_zero(self):
+        """首行 `-` 填 0（与 pybot/bot4cj 运行时语义一致），后续行 `-` 沿用"""
         content = (
             "@Servo\n"
             " name   max  min;\n"
@@ -339,8 +339,34 @@ class TestMtParser(unittest.TestCase):
         with tempfile.TemporaryDirectory() as tmpdir:
             path = _write_mt(tmpdir, "servo.mt", content)
             matrix = parse_mt_action(path)
-        np.testing.assert_array_equal(matrix[0], [90, 180, 0])
-        np.testing.assert_array_equal(matrix[1], [45, 180, 10])
+        np.testing.assert_array_equal(matrix[0], [0, 0, 0])
+        np.testing.assert_array_equal(matrix[1], [45, 0, 10])
+
+    def test_std_mt_first_row_matches_pybot(self):
+        """跨模块回归锚点：std.mt 首行解析结果与 pybot 运行时执行值一致
+
+        与 pybot/tests/test_hardvars.py 的 journal 期望值互为对照：
+        首行 `-`（s2/s4/s6/s8）解析为 0，而非 @Servo max 列。
+        """
+        content = (
+            "@Servo\n"
+            " name   max  min  offset  controller  channel;\n"
+            "  s1    90     -       -   pca9685-0        0;\n"
+            "  s2   180     -       -           -        1;\n"
+            "  s3   -90     -       -           -        2;\n"
+            "  s4  -180     -       -           -        3;\n"
+            "@\n"
+            "\n"
+            "@Action\n"
+            "s1.A  s2.A  s3.A  s4.A  s5.A  s6.A  s7.A  s8.A  s9.A  s10.A  spk.F  spk.delay;\n"
+            "  45     -    45     -    45     -    45     -    90     90     10       200;\n"
+            "@\n"
+        )
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = _write_mt(tmpdir, "std.mt", content)
+            matrix = parse_mt_action(path)
+        np.testing.assert_array_equal(
+            matrix[0], [45, 0, 45, 0, 45, 0, 45, 0, 90, 90, 10, 200])
 
     def test_no_action_block_raises(self):
         """无 @Action 块应抛 ValueError"""

docs/reports/bot4cj_python_migration_dev_doc.md

@@ -81,7 +81,7 @@ RobotStruct（r.cj）createHardVars()：按 @块名分发到硬件变量工厂
 ### 2.3 关键语义（必须逐项对齐）
 
 1. **占位符 `-`**：递归沿用上一行同列值；首行为 `-` 时，`parseMt` 解析为 `0`（数值列），`getMt` 返回字面 `-`。
-   ⚠️ 注意与 actpred `mt_parser.py` 的差异：后者首行 `-` 用 `@Servo` 初始角度填充（这是 ML 数据管线的语义）。**pybot 运行时语义以仓颉版为准（填 0）**，因为运行时 Servo 有 max/min 钳位保护；数据管线语义留在 actpred 不变。
+   actpred `mt_parser.py` 曾与此不同（首行 `-` 用 `@Servo` max 列填充），2026-08 已统一为本语义（填 0）：运行时 Servo 有 max/min 钳位保护，且 max 列是钳位上限而非初始位姿（反装舵机 max 为负值，会填出指令空间 0~180 之外的非法值）。
 2. **@Action 列名 `dev.param`**：点号拆分为硬件名 + 参数名；参数名经 `@Alias` 表转换（`A`→`angle`、`F`→`frequency`）后传入硬件。
 3. **每行执行**：同一行按硬件聚合参数字典，逐硬件调用一次 `play()`；行间时延由 `spk.delay` 参数承担。
 4. **硬件生命周期**：`active()` 按创建顺序 build，`deactive()` 倒序 destroy，且必须在 `try/finally` 中保证执行。
@@ -276,7 +276,7 @@ class I2CBus:
 | 用例组 | 覆盖 |
 |--------|------|
 | test_parser | 全部 `scripts/*.mt` 解析成功；列数不一致抛 DoesNotSync；漏分号跨行；注释/空行/CRLF；roundtrip（解析→save→再解析等价） |
-| test_placeholder | `-` 递归沿用；首行 `-` 数值列=0；与 actpred mt_parser 语义差异的显式断言（防止两边被误"统一"） |
+| test_placeholder | `-` 递归沿用；首行 `-` 数值列=0；actpred mt_parser 已统一为同一语义（2026-08），两侧测试互为语义锚点 |
 | test_servo | 正装/反装翻转/offset/钳位边界 |
 | test_hardvars | @Alias 转换；@Action 列名拆分；同行参数聚合；重名覆盖警告 |
 | test_plugin | 插件三级路径查找；PluginNotFound 报错内容；PCA9685.py 用 Mock I2C 断言寄存器写序列（prescale=121、reg=0x06+4*ch、offval 字节序） |
@@ -315,7 +315,7 @@ class I2CBus:
 | Python GIL + 解释器抖动影响动作时序 | 舵机动作节奏轻微不均 | 动作行间延时 ≥ 20ms，抖动 <1ms 不可感知；MPU6050 采样用 Event.wait 定时，不忙等；量化依据见附录 A |
 | aplay 子进程首次启动延迟 | 首个音符滞后 | build() 时预热（播放 0 长度静音）；或后续可选接 alsaaudio |
 | 仓颉版行为的未记录细节 | 等价性缺口 | §6.3 写序列逐字节 diff 兜底 |
-| .mt 脚本方言分裂（pybot vs actpred 解析器） | 语义漂移 | test_placeholder 显式断言两者差异；长期可评估 actpred 复用 pybot parser 的只读模式 |
+| .mt 脚本方言分裂（pybot vs actpred 解析器） | 语义漂移 | 占位符语义已统一（2026-08，首行 `-` 均填 0）；长期可评估 actpred 复用 pybot parser 的只读模式 |
 | 树莓派 Python 版本参差 | 语法兼容 | 目标 Python ≥ 3.9（树莓派 OS Bullseye 起自带 3.9） |
 
 ---

pybot/README.md

@@ -12,6 +12,7 @@ PCA9685 PWM 控制器 + MPU6050 姿态传感器）。
 
 - **零第三方依赖**：纯标准库（fcntl / struct / threading / wave / argparse），
   树莓派上 `git pull` 即部署，无构建步骤。
+  （picamera2 为可选依赖，仅 `@Camera` 摄像头块需要，树莓派 OS 预装，勿 pip 安装）
 - **与仓颉版逐语义对齐**：FSM 解析器、占位符递归、Servo 校准算法、
   PCA9685 寄存器序列均与 bot4cj 一致（Mock 总线 journal 可逐字节 diff）。
 - **Python 插件替代 Lua**：`modules/<块名>.py` 定义
@@ -35,7 +36,7 @@ cat scripts/std.mt | ./_run.sh              # 与 bot4cj 相同的管道用法
 
 ```
 pybot/              # 包：parser / robot / hardvars / plugin / main
-pybot/hd/           # 硬件层：i2c / spi / servo / kalman / mpu6050 / speaker
+pybot/hd/           # 硬件层：i2c / spi / servo / kalman / mpu6050 / speaker / camera
 modules/            # 硬件驱动插件：PCA9685 / SPI / ADS1115 / ADS1256
 scripts -> ../bot4cj/scripts    # .mt 脚本与仓颉版共享
 tests/              # pytest 单元测试（Mock 总线，无需硬件）
@@ -55,3 +56,15 @@ venv/bin/python -m pytest tests/ -v
 见 [bot4cj/README.md](../bot4cj/README.md) 与 `scripts/std.mt`。
 格式要点：`@块名` 开块、行 0 为表头、`;` 结行、`#` 注释、
 `-` 占位符（非首行取上一行同列值；首行数值列解析为 0）、`@` 收块。
+
+新增块（无仓颉版对应）：
+
+```
+@Camera
+ name  width  height  framerate  capacity;
+ cam0    640      480        30         4;
+@
+```
+
+`build` 启动后台抓帧线程，维护容量为 `capacity` 的有界帧队列（满丢最旧帧）；
+`play` 返回最新帧 `(时间戳, frame)`。

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/parser.py

@@ -213,9 +213,9 @@ class MatrixSaver:
 
 # ---------------- 占位符语义（与仓颉版 MatrixVars_Hardware.getMt/parseMt 对齐） ----------------
 #
-# ⚠️ 与 actpred/mt_parser.py 的语义差异（不得误统一）：
-#    actpred 数据管线首行 '-' 用 @Servo 初始角度填充；
-#    pybot 运行时首行 '-' 数值列解析为 0（Servo 有 max/min 钳位保护）。
+# 首行 '-' 数值列解析为 0（Servo 有 max/min 钳位保护）。
+# actpred/mt_parser.py 曾用 @Servo max 列填充首行 '-'，2026-08 已统一
+# 为本语义（填 0），见 docs/reports/bot4cj_python_migration_dev_doc.md §2.3。
 
 def get_mt(obj, row, col):
     """获得项目值：原始值是占位符且非首行（row>1），递归取上一行同列值"""

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,
 }
 
 

pybot/requirements.txt

@@ -1,3 +1,4 @@
 # 运行时零第三方依赖（纯标准库：fcntl/struct/threading/wave/argparse）
+# picamera2 为可选运行时依赖（仅 @Camera 需要，树莓派 OS 预装，勿 pip 安装）
 # 以下仅开发测试用
 pytest>=7.0

pybot/tests/test_camera.py

@@ -0,0 +1,100 @@
+"""Camera 测试：有界帧队列（满丢最旧）、守护线程生命周期、dry_run、.mt 接入
+
+全部经注入 grabber 运行，不碰 picamera2 / 真实摄像头。
+"""
+
+import time
+
+from pybot.hd.camera import Camera
+from pybot.robot import RobotStruct
+
+
+class FakeGrabber:
+    """假帧源：产出递增序号帧"""
+
+    def __init__(self):
+        self.count = 0
+
+    def __call__(self):
+        self.count += 1
+        return self.count
+
+
+CAMERA_MT = """
+@Camera
+ name  width  height  framerate  capacity;
+ cam0    320      240        15         2;
+@
+
+@Action
+cam0.frame;
+0;
+@
+"""
+
+
+def test_queue_capacity_drops_oldest():
+    cam = Camera(capacity=3, grabber=FakeGrabber())
+    for _ in range(5):
+        cam._grab_once()
+    assert [f for _, f in cam.snapshot()] == [3, 4, 5]  # 容量 3，丢最旧
+    assert cam.latest()[1] == 5
+    assert cam.size == 3
+
+
+def test_latest_empty():
+    assert Camera(grabber=FakeGrabber()).latest() is None
+    assert Camera(grabber=FakeGrabber()).snapshot() == []
+
+
+def test_daemon_lifecycle():
+    cam = Camera(framerate=200.0, capacity=10, grabber=FakeGrabber())
+    cam.setup()                     # 注入 grabber 时不碰硬件
+    cam.start_daemon()
+    time.sleep(0.1)
+    cam.stop_daemon()
+    cam.wait_daemon()
+    assert cam.size > 0
+    n = cam.size
+    time.sleep(0.05)
+    assert cam.size == n            # 停止后队列不再增长
+
+
+def test_grab_failure_keeps_daemon_alive(caplog):
+    def bad_grabber():
+        raise RuntimeError("boom")
+    cam = Camera(framerate=200.0, capacity=2, grabber=bad_grabber)
+    cam.start_daemon()
+    time.sleep(0.05)
+    cam.stop_daemon()
+    cam.wait_daemon()
+    assert cam.size == 0
+    assert any("grab failed" in r.message for r in caplog.records)
+
+
+def test_dry_run_no_hardware():
+    cam = Camera(dry_run=True)
+    cam.setup()
+    cam.start_daemon()              # 空操作
+    cam.stop_daemon()
+    cam.wait_daemon()
+    cam.close()
+    assert cam.latest() is None
+
+
+def test_camera_var_from_matrix():
+    robot = RobotStruct(CAMERA_MT, dry_run=True)
+    cam = robot.M["cam0"].camera
+    assert (cam.width, cam.height) == (320, 240)
+    assert cam.framerate == 15.0
+    assert cam.capacity == 2
+    robot.deactive()
+
+
+def test_camera_full_lifecycle_in_robot():
+    """dry_run 下 build/run/destroy 全流程不碰硬件"""
+    robot = RobotStruct(CAMERA_MT, dry_run=True)
+    robot.active()                  # build：dry_run 空操作
+    assert robot.M["cam0"].play({}) is None
+    robot.run()                     # @Action 引用 cam0.frame 不报错
+    robot.deactive()

pybot/tests/test_parser.py

@@ -91,7 +91,7 @@ def test_placeholder_recursion():
 
 def test_placeholder_first_row():
     """首行 '-'：get_mt 返回字面 '-'；parse_mt 解析为 typ("0")
-    ⚠️ 与 actpred/mt_parser.py 不同（后者用 @Servo 初始角度填充），不得误统一"""
+    actpred/mt_parser.py 已统一为本语义（首行 '-' 填 0），本测试即两侧共同的语义锚点"""
     m = _matrix("@T\nname v;\na -;\n@\n")
     assert get_mt(m, 1, 1) == "-"
     assert parse_mt(m, 1, 1) == 0.0