diff --git a/pybot/README.md b/pybot/README.md index 985b8f2..5bc5664 100644 --- a/pybot/README.md +++ b/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)`。 diff --git a/pybot/pybot/hardvars.py b/pybot/pybot/hardvars.py index 3f236a7..d855fe0 100644 --- a/pybot/pybot/hardvars.py +++ b/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) diff --git a/pybot/pybot/hd/camera.py b/pybot/pybot/hd/camera.py new file mode 100644 index 0000000..d8b5287 --- /dev/null +++ b/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") diff --git a/pybot/pybot/robot.py b/pybot/pybot/robot.py index fb0c588..d94852a 100644 --- a/pybot/pybot/robot.py +++ b/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, } diff --git a/pybot/requirements.txt b/pybot/requirements.txt index ac94880..9ca75b3 100644 --- a/pybot/requirements.txt +++ b/pybot/requirements.txt @@ -1,3 +1,4 @@ # 运行时零第三方依赖（纯标准库：fcntl/struct/threading/wave/argparse） +# picamera2 为可选运行时依赖（仅 @Camera 需要，树莓派 OS 预装，勿 pip 安装） # 以下仅开发测试用 pytest>=7.0 diff --git a/pybot/tests/test_camera.py b/pybot/tests/test_camera.py new file mode 100644 index 0000000..5d0a932 --- /dev/null +++ b/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()