diff --git a/actpred/README.md b/actpred/README.md index f6e7032..aa9b05c 100644 --- a/actpred/README.md +++ b/actpred/README.md @@ -152,3 +152,33 @@ print(predictor.action_dim, predictor.history_len, predictor.frame_len) # 公 action = predictor.predict(images, action_history) # np.ndarray (action_dim,) # action_history 形状须为 (history_len, action_dim) ``` + +## Hailo 部署（AI HAT 2 / Hailo-10H） + +树莓派 AI HAT 2 上通过 `hailo_infer.py` 的 `HailoPredictor` 走 NPU 推理， +接口与 `ActionPredictorInference` 对齐。模型需先经 Hailo DFC 编译为 HEF： + +```bash +# 1. 导出 ONNX +bash _run.sh export --model checkpoints/best_inference.pt --output model.onnx +# 2. 在 x86 主机（或 Hailo 云端 DFC）上编译 + 量化校准 → model.hef，拷回树莓派 +# 3. 推理（零 torch 依赖，仅需 hailo_platform + numpy + PIL） +``` + +```python +from hailo_infer import HailoPredictor + +with HailoPredictor("model.hef") as predictor: # stats_path= 仅归一化模型需要 + action = predictor.predict(images, action_history) +``` + +```bash +python3 hailo_infer.py --hef model.hef --images f1.jpg f2.jpg f3.jpg \ + --action-history "45,0,...:45,0,..." [--stats action_stats.npy] +``` + +HEF 的 vstream 名称/形状/量化参数全部从 `HEF.get_input/output_vstream_infos()` +动态推导，不硬编码；图像输入按 Hailo 惯例为 uint8 原始像素 +（ImageNet 归一化由 DFC 编译时折叠进网络）。 +环境调研与 API 踩坑记录见 +[docs/reports/actpred_hailo_hat2_inference_research_20260813.md](../docs/reports/actpred_hailo_hat2_inference_research_20260813.md)。 diff --git a/actpred/hailo_infer.py b/actpred/hailo_infer.py new file mode 100644 index 0000000..8254d58 --- /dev/null +++ b/actpred/hailo_infer.py @@ -0,0 +1,250 @@ +"""actpred HEF（Hailo）推理封装 —— AI HAT 2 / Hailo-10H 部署后端 + +接口对齐 predict.ActionPredictorInference：predict(images, action_history)。 + +HEF 由 actpred ONNX 经 Hailo DFC 编译生成（量化校准后）。本封装不硬编码 +vstream 名称/形状/量化参数，全部从 HEF 的 input/output vstream info 动态推导： + + 图像输入: (1, F, 3, S, S) → NHWC uint8 (1, F, S, S, 3) + 或 (1, S, S, F*3) → 帧沿通道拼接 + 动作输入: (1, H, D) float32，uint8 边缘时按 quant_info 量化 + 动作输出: (1, D)，uint8 边缘时按 quant_info 反量化 + +背景与踩坑记录见 +docs/reports/actpred_hailo_hat2_inference_research_20260813.md： + - hailort 5.x 对 Hailo-10 仅支持新 API（create_infer_model）， + 旧版 VDevice.configure(hef) 报 HAILO_NOT_IMPLEMENTED； + - 绑定缓冲区 dtype/大小必须匹配 HEF 边缘格式（量化 uint8）； + - 图像输入按 Hailo 惯例为 uint8 原始像素（ImageNet 归一化由 + DFC 编译时折叠进网络）。 +""" + +import argparse +import contextlib +import json +import os + +import numpy as np +from PIL import Image + +# 与 predict.parse_action_history 相同逻辑；此处内联以保持 +# 本模块零 torch 依赖（HAT 部署环境只装 hailo_platform + numpy + PIL） + + +def parse_action_history(s: str, action_dim: int) -> np.ndarray: + steps = s.split(":") + history = [] + for step in steps: + values = [float(v) for v in step.split(",")] + if len(values) != action_dim: + raise ValueError( + f"Expected {action_dim} values per step, got {len(values)}") + history.append(values) + return np.array(history, dtype=np.float32) + + +class HailoPredictor: + """HEF 动作预测推理器（Hailo NPU 后端） + + 用法： + with HailoPredictor("actpred.hef") as predictor: + action = predictor.predict(images, action_history) + """ + + def __init__(self, hef_path: str, stats_path: str = None, vdevice=None): + from hailo_platform import HEF, VDevice + + self._stack = contextlib.ExitStack() + if vdevice is None: + vdevice = self._stack.enter_context(VDevice()) + self._vdevice = vdevice + self._infer_model = self._vdevice.create_infer_model(str(hef_path)) + self._configured = self._stack.enter_context( + self._infer_model.configure()) + try: + hef = HEF(str(hef_path)) + self._in_infos = hef.get_input_vstream_infos() + self._out_infos = hef.get_output_vstream_infos() + if len(self._out_infos) != 1: + raise ValueError( + f"HEF 应只有一个输出 vstream，实际 {len(self._out_infos)}") + self._image_info = self._find_image_input() + self._history_info = self._find_history_input() + self._out_info = self._out_infos[0] + self._derive_dims() + + self._action_mean = None + self._action_std = None + if stats_path is not None: + if not os.path.exists(stats_path): + raise FileNotFoundError(f"统计文件不存在: {stats_path}") + stats = np.load(stats_path) + if stats.shape != (2, self.action_dim): + raise ValueError( + f"action_stats.npy 形状应为 (2, {self.action_dim})，" + f"实际为 {stats.shape}: {stats_path}") + self._action_mean = stats[0].astype(np.float32) + self._action_std = stats[1].astype(np.float32) + except Exception: + self._stack.close() # 初始化失败也释放设备，避免 fd 泄漏 + raise + + # ---------------- 形状推导 ---------------- + + def _find_image_input(self): + cands = [i for i in self._in_infos if len(i.shape) in (4, 5)] + if len(cands) != 1: + raise ValueError( + "HEF 输入中无法唯一识别图像 vstream（期望一个 4/5 维输入），" + f"实际: {[(i.name, tuple(i.shape)) for i in self._in_infos]}") + return cands[0] + + def _find_history_input(self): + cands = [i for i in self._in_infos + if len(i.shape) == 3 and i is not self._image_info] + if len(cands) != 1: + raise ValueError( + "HEF 输入中无法唯一识别动作历史 vstream（期望一个 3 维输入），" + f"实际: {[(i.name, tuple(i.shape)) for i in self._in_infos]}") + return cands[0] + + def _derive_dims(self): + shape = tuple(self._image_info.shape) + if len(shape) == 5: + _, self.frame_len, c, self.image_size, _ = shape + if c != 3: + raise ValueError(f"图像输入通道维应为 3，实际形状 {shape}") + self._image_layout = "ncfhw" + else: # (1, S, S, F*3) NHWC 帧拼通道 + _, self.image_size, s2, c = shape + if s2 != self.image_size or c % 3 != 0 or c == 3: + raise ValueError( + f"无法从 4 维图像输入推导帧数（期望 (1,S,S,F*3)），" + f"实际形状 {shape}") + self.frame_len = c // 3 + self._image_layout = "nhwc_fused" + + h_shape = tuple(self._history_info.shape) + self.history_len, self.action_dim = h_shape[1], h_shape[2] + + out_shape = tuple(self._out_info.shape) + out_dim = out_shape[-1] + if out_dim != self.action_dim: + raise ValueError( + f"输出动作维度 {out_dim} 与动作历史维度 {self.action_dim} 不一致") + + # ---------------- 生命周期 ---------------- + + def close(self): + self._stack.close() + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + self.close() + return False + + # ---------------- 推理 ---------------- + + def predict(self, images, action_history: np.ndarray) -> np.ndarray: + """images: 长度为 frame_len 的 RGB 图像列表（旧→新，末帧为当前帧） + + 返回 (action_dim,) float32 动作向量。 + """ + if not isinstance(images, (list, tuple)): + images = [images] + if len(images) != self.frame_len: + raise ValueError( + f"images 应为 {self.frame_len} 帧，实际为 {len(images)} 帧。" + "冷启动时可重复传入同一帧。") + action_history = np.asarray(action_history, dtype=np.float32) + expected = (self.history_len, self.action_dim) + if action_history.shape != expected: + raise ValueError( + f"action_history 形状应为 {expected}，实际为 {action_history.shape}") + + if self._action_mean is not None: + action_history = (action_history - self._action_mean) / self._action_std + + from hailo_platform import FormatType + bindings = self._configured.create_bindings() + + frames = np.stack([self._preprocess(img) for img in images]) + bindings.input(self._image_info.name).set_buffer( + self._image_buffer(frames)) + + h_shape = tuple(self._history_info.shape) + if self._history_info.format.type == FormatType.UINT8: + q = self._history_info.quant_info + buf = np.clip(np.round(action_history / q.qp_scale + q.qp_zp), + 0, 255).astype(np.uint8).reshape(h_shape) + else: + buf = action_history.astype(np.float32).reshape(h_shape) + bindings.input(self._history_info.name).set_buffer(buf) + + out_shape = tuple(self._out_info.shape) + if self._out_info.format.type == FormatType.UINT8: + out_buf = np.zeros(out_shape, dtype=np.uint8) + else: + out_buf = np.zeros(out_shape, dtype=np.float32) + bindings.output(self._out_info.name).set_buffer(out_buf) + + self._configured.run([bindings], timeout=10000) + + result = np.asarray( + bindings.output(self._out_info.name).get_buffer(), + dtype=np.float32) + if self._out_info.format.type == FormatType.UINT8: + q = self._out_info.quant_info + result = (result - q.qp_zp) * q.qp_scale + result = result.reshape(-1) + if self._action_mean is not None: + result = result * self._action_std + self._action_mean + return result + + def _preprocess(self, image) -> np.ndarray: + img = Image.fromarray(image) if isinstance(image, np.ndarray) else image + img = img.convert("RGB").resize( + (self.image_size, self.image_size), Image.BILINEAR) + return np.asarray(img, dtype=np.uint8) # HWC uint8 + + def _image_buffer(self, frames: np.ndarray) -> np.ndarray: + """(F, S, S, 3) → HEF 图像输入缓冲区（按编译后布局）""" + if self._image_layout == "ncfhw": + return frames.transpose(0, 3, 1, 2)[np.newaxis] # (1,F,3,S,S) + # nhwc_fused: 帧沿通道拼接 (1, S, S, F*3) + return frames.reshape(1, self.image_size, self.image_size, + self.frame_len * 3) + + +def main(): + parser = argparse.ArgumentParser(description="actpred HEF (Hailo) inference") + parser.add_argument("--hef", required=True, help="HEF 模型路径") + parser.add_argument("--images", required=True, nargs="+", + help="按时间顺序（旧→新）传入 frame_len 个图像路径") + parser.add_argument("--action-history", required=True, + help="冒号分隔各步，每步 action_dim 个值") + parser.add_argument("--stats", default=None, + help="action_stats.npy 路径（训练时启用归一化才需要）") + parser.add_argument("--output-format", choices=["json", "raw"], + default="json") + args = parser.parse_args() + + with HailoPredictor(args.hef, stats_path=args.stats) as predictor: + if len(args.images) != predictor.frame_len: + raise SystemExit( + f"--images 需传入 {predictor.frame_len} 个路径（旧→新），" + f"实际为 {len(args.images)} 个") + images = [np.array(Image.open(p).convert("RGB")) for p in args.images] + history = parse_action_history(args.action_history, predictor.action_dim) + predicted = predictor.predict(images, history) + + if args.output_format == "json": + print(json.dumps({"predicted_action": predicted.tolist()}, indent=2)) + else: + print(" ".join(f"{v:.4f}" for v in predicted)) + + +if __name__ == "__main__": + main() diff --git a/actpred/test_actpred.py b/actpred/test_actpred.py index e5a66c5..ca281b6 100644 --- a/actpred/test_actpred.py +++ b/actpred/test_actpred.py @@ -10,7 +10,9 @@ import contextlib import importlib.util import io import os +import sys import tempfile +import types import unittest from pathlib import Path from unittest import mock @@ -808,5 +810,226 @@ class TestResumeCheckpoint(unittest.TestCase): +# ---------------- HailoPredictor 测试用 fake hailo_platform ---------------- + +class _FakeFormatType: + UINT8 = "UINT8" + FLOAT32 = "FLOAT32" + + +class _FakeFormat: + def __init__(self, fmt_type): + self.type = fmt_type + + +class _FakeQuantInfo: + def __init__(self, qp_zp=0.0, qp_scale=1.0): + self.qp_zp = qp_zp + self.qp_scale = qp_scale + + +class _FakeVStreamInfo: + def __init__(self, name, shape, fmt_type, quant=None): + self.name = name + self.shape = tuple(shape) + self.format = _FakeFormat(fmt_type) + self.quant_info = quant or _FakeQuantInfo() + + +class _FakeStream: + def __init__(self): + self._buffer = None + + def set_buffer(self, buf): + self._buffer = buf + + def get_buffer(self): + return self._buffer + + +class _FakeBindings: + def __init__(self): + self._streams = {} + + def input(self, name=""): + return self._streams.setdefault(("in", name), _FakeStream()) + + def output(self, name=""): + return self._streams.setdefault(("out", name), _FakeStream()) + + +class _FakeConfigured: + def __init__(self, transform): + self._transform = transform + + def create_bindings(self): + return _FakeBindings() + + def run(self, bindings_list, timeout=0): + for b in bindings_list: + self._transform(b) + + +class _FakeInferModel: + def __init__(self, transform): + self._configured = _FakeConfigured(transform) + + @contextlib.contextmanager + def configure(self): + yield self._configured + + +class _FakeVDevice: + def __init__(self, transform): + self._transform = transform + + def create_infer_model(self, path): + return _FakeInferModel(self._transform) + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + +def _install_fake_hailo(test, infos_in, infos_out, transform): + """向 sys.modules 注入 fake hailo_platform，无需硬件即可测试""" + fake = types.ModuleType("hailo_platform") + fake.FormatType = _FakeFormatType + fake.HEF = lambda path: types.SimpleNamespace( + get_input_vstream_infos=lambda: list(infos_in), + get_output_vstream_infos=lambda: list(infos_out)) + fake.VDevice = lambda: _FakeVDevice(transform) + patcher = mock.patch.dict(sys.modules, {"hailo_platform": fake}) + test.addCleanup(patcher.stop) + patcher.start() + + +def _first_row_transform(hist_name, out_name, quant_in=None, quant_out=None): + """transform：输出 = 历史动作首行（量化边缘时先反量化再量化）""" + def transform(b): + hist = np.asarray(b.input(hist_name).get_buffer(), dtype=np.float32) + if quant_in is not None: + hist = (hist - quant_in.qp_zp) * quant_in.qp_scale + first = hist.reshape(-1, hist.shape[-1])[0] + if quant_out is not None: + first = np.clip(np.round(first / quant_out.qp_scale + quant_out.qp_zp), + 0, 255).astype(np.uint8) + b.output(out_name).get_buffer()[...] = first + return transform + + +class TestHailoPredictor(unittest.TestCase): + """HEF 推理封装测试（fake hailo_platform 注入，无需 Hailo 硬件）""" + + def _frames(self, n, size=8): + return [np.random.randint(0, 255, (size, size, 3), dtype=np.uint8) + for _ in range(n)] + + def test_float_ncfhw_layout(self): + from hailo_infer import HailoPredictor + infos_in = [ + _FakeVStreamInfo("net/images", (1, 2, 3, 4, 4), _FakeFormatType.FLOAT32), + _FakeVStreamInfo("net/history", (1, 2, 3), _FakeFormatType.FLOAT32), + ] + infos_out = [_FakeVStreamInfo("net/action", (1, 3), _FakeFormatType.FLOAT32)] + seen_shapes = {} + + def transform(b): + seen_shapes["image"] = np.asarray( + b.input("net/images").get_buffer()).shape + _first_row_transform("net/history", "net/action")(b) + + _install_fake_hailo(self, infos_in, infos_out, transform) + history = np.array([[1, 2, 3], [4, 5, 6]], dtype=np.float32) + with HailoPredictor("fake.hef") as p: + self.assertEqual((p.frame_len, p.history_len, p.action_dim), (2, 2, 3)) + self.assertEqual(p.image_size, 4) + result = p.predict(self._frames(2), history) + np.testing.assert_allclose(result, [1, 2, 3]) + self.assertEqual(seen_shapes["image"], (1, 2, 3, 4, 4)) + + def test_uint8_quantized_roundtrip(self): + from hailo_infer import HailoPredictor + q_in = _FakeQuantInfo(qp_zp=128.0, qp_scale=0.01) + q_out = _FakeQuantInfo(qp_zp=10.0, qp_scale=0.5) + infos_in = [ + _FakeVStreamInfo("net/images", (1, 2, 3, 4, 4), _FakeFormatType.UINT8), + _FakeVStreamInfo("net/history", (1, 2, 3), _FakeFormatType.UINT8, q_in), + ] + infos_out = [_FakeVStreamInfo("net/action", (1, 3), _FakeFormatType.UINT8, q_out)] + _install_fake_hailo( + self, infos_in, infos_out, + _first_row_transform("net/history", "net/action", q_in, q_out)) + history = np.array([[0.5, -0.25, 0.0], [1, 1, 1]], dtype=np.float32) + with HailoPredictor("fake.hef") as p: + result = p.predict(self._frames(2), history) + np.testing.assert_allclose(result, [0.5, -0.25, 0.0], atol=0.51) + + def test_nhwc_fused_layout(self): + from hailo_infer import HailoPredictor + infos_in = [ + _FakeVStreamInfo("net/images", (1, 4, 4, 6), _FakeFormatType.UINT8), + _FakeVStreamInfo("net/history", (1, 2, 3), _FakeFormatType.FLOAT32), + ] + infos_out = [_FakeVStreamInfo("net/action", (1, 3), _FakeFormatType.FLOAT32)] + seen_shapes = {} + + def transform(b): + seen_shapes["image"] = np.asarray( + b.input("net/images").get_buffer()).shape + _first_row_transform("net/history", "net/action")(b) + + _install_fake_hailo(self, infos_in, infos_out, transform) + history = np.zeros((2, 3), dtype=np.float32) + with HailoPredictor("fake.hef") as p: + self.assertEqual(p.frame_len, 2) # 6 通道 = 2 帧 × 3 + result = p.predict(self._frames(2), history) + np.testing.assert_allclose(result, [0, 0, 0]) + self.assertEqual(seen_shapes["image"], (1, 4, 4, 6)) + + def test_frame_count_mismatch_raises(self): + from hailo_infer import HailoPredictor + infos_in = [ + _FakeVStreamInfo("net/images", (1, 2, 3, 4, 4), _FakeFormatType.FLOAT32), + _FakeVStreamInfo("net/history", (1, 2, 3), _FakeFormatType.FLOAT32), + ] + infos_out = [_FakeVStreamInfo("net/action", (1, 3), _FakeFormatType.FLOAT32)] + _install_fake_hailo(self, infos_in, infos_out, + _first_row_transform("net/history", "net/action")) + with HailoPredictor("fake.hef") as p: + with self.assertRaises(ValueError): + p.predict(self._frames(1), np.zeros((2, 3), dtype=np.float32)) + + def test_output_dim_mismatch_raises(self): + from hailo_infer import HailoPredictor + infos_in = [ + _FakeVStreamInfo("net/images", (1, 2, 3, 4, 4), _FakeFormatType.FLOAT32), + _FakeVStreamInfo("net/history", (1, 2, 3), _FakeFormatType.FLOAT32), + ] + infos_out = [_FakeVStreamInfo("net/action", (1, 4), _FakeFormatType.FLOAT32)] + _install_fake_hailo(self, infos_in, infos_out, lambda b: None) + with self.assertRaises(ValueError): + HailoPredictor("fake.hef") + + def test_stats_normalization_roundtrip(self): + from hailo_infer import HailoPredictor + infos_in = [ + _FakeVStreamInfo("net/images", (1, 2, 3, 4, 4), _FakeFormatType.FLOAT32), + _FakeVStreamInfo("net/history", (1, 2, 3), _FakeFormatType.FLOAT32), + ] + infos_out = [_FakeVStreamInfo("net/action", (1, 3), _FakeFormatType.FLOAT32)] + _install_fake_hailo(self, infos_in, infos_out, + _first_row_transform("net/history", "net/action")) + with tempfile.TemporaryDirectory() as tmpdir: + stats_path = os.path.join(tmpdir, "action_stats.npy") + np.save(stats_path, np.stack([np.full(3, 1.0), np.full(3, 2.0)])) + history = np.array([[3, 3, 3], [1, 1, 1]], dtype=np.float32) + with HailoPredictor("fake.hef", stats_path=stats_path) as p: + result = p.predict(self._frames(2), history) + np.testing.assert_allclose(result, [3, 3, 3], atol=1e-6) + + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/docs/reports/actpred_hailo_hat2_inference_research_20260813.md b/docs/reports/actpred_hailo_hat2_inference_research_20260813.md new file mode 100644 index 0000000..fd14c95 --- /dev/null +++ b/docs/reports/actpred_hailo_hat2_inference_research_20260813.md @@ -0,0 +1,130 @@ +# actpred × 树莓派 AI HAT 2（Hailo-10H）推理调研报告 + +日期：2026-08-13 +设备：Raspberry Pi 5 + Raspberry Pi AI HAT 2（Hailo-10H，8GB RAM，40 TOPS） + +## 1. 目标 + +调研如何利用 AI HAT 2 扩展板资源对 actpred 动作预测模型做推理加速， +验证板载软件栈可用性，并给出部署路径。 + +## 2. 环境实测结果 + +### 2.1 已安装的软件栈（apt 包） + +| 包 | 版本 | 说明 | +|---|---|---| +| h10-hailort | 5.1.1 | Hailo 运行时库 | +| h10-hailort-pcie-driver | 5.1.1 | PCIe 驱动与固件 | +| hailo-h10-all | 5.1.1 | Hailo-10 元包 | +| hailo-tappas-core | 5.1.0 | TAPPAS 核心 | +| python3-h10-hailort | 5.1.1 | HailoRT Python API（`hailo_platform` 模块） | +| python3-hailo-tappas | 5.1.0 | TAPPAS Python 绑定 | +| rpicam-apps-hailo-postprocess | 1.12.0 | rpicam-apps 的 AI HAT 后处理插件 | +| hailo-models | 1.0.0 | 预装 HEF 模型集 | + +> 注意：Python 模块名是 `hailo_platform`，**不是** `hailort` +> （`import hailort` 会 ModuleNotFoundError）。 + +### 2.2 硬件识别与基准 + +```bash +hailortcli scan # Device: 0001:01:00.0（正常识别） +hailortcli benchmark /usr/share/hailo-models/resnet_v1_50_h10.hef +# → resnet_v1_50: FPS: 307.59（温度均值 47.65°C） +``` + +### 2.3 预装 HEF 模型 + +`/usr/share/hailo-models/` 下预装了 `_h8l`/`_h10` 两套 HEF。 +本机为 Hailo-10H，应使用 `_h10` 后缀模型： +resnet_v1_50、yolov8m、yolov8m_pose、yolov8s_pose、yolov5n_seg、yolov11m 等。 + +## 3. Python 推理 API 实测（关键发现） + +hailort 5.x 对 Hailo-10 设备只支持**新版 infer API**： + +- ❌ 旧版 `VDevice().configure(hef)` / `InferVStreams(hef, vdev)` + 在 Hailo-10 上报 `HAILO_NOT_IMPLEMENTED (error 7)`（实测） +- ✅ 新版 `VDevice().create_infer_model(hef_path)` → `configure()` → + `create_bindings()` → `run(bindings, timeout)`（实测通过） + +实测打通的最小推理流程： + +```python +import numpy as np +from hailo_platform import VDevice, HEF + +hef_path = "/usr/share/hailo-models/resnet_v1_50_h10.hef" +hef = HEF(hef_path) +in_info = hef.get_input_vstream_infos()[0] # (224,224,3) uint8 +out_info = hef.get_output_vstream_infos()[0] # (1,1,1000) 量化 uint8 + +with VDevice() as vdev: + im = vdev.create_infer_model(hef_path) + with im.configure() as c: + b = c.create_bindings() + b.input(in_info.name).set_buffer(img_uint8_hwc) + b.output(out_info.name).set_buffer( + np.zeros(tuple(out_info.shape), dtype=np.uint8)) + c.run([b], timeout=10000) + q = np.asarray(b.output(out_info.name).get_buffer()).astype(np.float32) + res = (q - out_info.quant_info.qp_zp) * out_info.quant_info.qp_scale +``` + +### 踩坑记录 + +1. `bindings.input(name)` / `bindings.output(name)` 不支持下标访问； + `get_buffer()` 在未 `set_buffer` 前返回 None，需先按 + `HEF.get_input/output_vstream_infos()` 的 shape/dtype 自建 numpy 缓冲区。 +2. 输出缓冲区 dtype/大小必须匹配 HEF 边缘格式（量化 uint8）， + 传 float32 会报 `Output buffer size ... different than expected`； + 需手动用 `quant_info.qp_zp/qp_scale` 反量化。 +3. `VStreamInfo` 无 `format_type` 属性，shape 与量化参数从 + `get_input/output_vstream_infos()` 获取。 +4. 摄像头实时推理走 `rpicam-hello/rpicam-vid --post-process-file ...` + （rpicam-apps-hailo-postprocess 提供）。 +5. `hailo_platform.genai` 提供 LLM / VLM / Speech2Text 接口—— + AI HAT 2（Hailo-10H + 8GB RAM）支持本地大模型，是与 + AI HAT+（Hailo-8L）相比的最大增量。 + +## 4. actpred 部署路径 + +Hailo 板**不能直接运行 ONNX/PyTorch**，必须编译为 HEF（含量化）： + +``` +actpred checkpoint (best_inference.pt) + → export_onnx.py → model.onnx + → Hailo DFC（Dataflow Compiler，需 x86 Linux 主机或云端 DFC） + 量化校准（用 data/ 中的帧做 calibration set） + → model.hef（拷回树莓派） + → hailo_platform 推理（封装见 actpred/hailo_infer.py） +``` + +风险与注意点： + +- **DFC 不在树莓派本机**（无 hailo/dfc 命令），需在 x86 主机安装或 + 使用 Hailo 云端 DFC 服务。 +- actpred 是 Transformer 结构（多头自注意力），编译前需确认所用 DFC + 版本对 attention 类算子的支持；遇不支持算子可拆分模型或换 op。 +- actpred 仅 ~450K 参数，Pi 5 CPU（4×A76）直接 PyTorch/ONNX Runtime + 推理也在可接受延迟范围；HAT 迁移的主要收益是释放 CPU 给 pybot + 主控循环与图像采集。 +- HEF 的输入/输出名与量化参数由 DFC 编译决定，推理封装按 + `get_input/output_vstream_infos()` 动态适配，不硬编码名称。 + +## 5. 结论与产出 + +1. AI HAT 2 软件栈完整可用，新 API 推理链路实测打通（resnet50 307 FPS）。 +2. actpred HEF 推理封装：`actpred/hailo_infer.py`（`HailoPredictor`）， + 接口对齐 `predict.ActionPredictorInference`。 +3. 待办：在 x86 主机上用 DFC 将 actpred ONNX 编译为 HEF 并做量化校准， + 随后用 `HailoPredictor` 替换 CPU 推理端做端到端验证。 + +## 参考 + +- [Raspberry-Pi-AI-HAT+ 使用教程](https://spotpear.cn/index.php/wiki/Raspberry-Pi-5-AI-HAT-Plus-Hailo-8L-13-26-Tops-PCIe-M.2.html) +- [树莓派5 Raspberry Pi AI HAT+2 官方 Hailo-10H 8GBRAM 40TOPS](https://spotpear.cn/index.php/shop/Raspberry-Pi-5-AI-HAT-Plus-2-Hailo-10H-40-Tops.html) +- [AI HAT+ 文档](https://shumeipai.nxez.com/documentation/accessories/ai-hat-plus.html) +- [Raspberry Pi AI Kit](https://www.waveshare.net/wiki/Raspberry_Pi_AI_Kit) +- [onnx 转为 hef 用树莓派 hailo 使用](https://wenku.csdn.net/answer/1e0n2rmysd) diff --git a/docs/reports/actpred_inference_speed_benchmark_20260813.md b/docs/reports/actpred_inference_speed_benchmark_20260813.md new file mode 100644 index 0000000..9b087e5 --- /dev/null +++ b/docs/reports/actpred_inference_speed_benchmark_20260813.md @@ -0,0 +1,52 @@ +# actpred 推理速度基准测试报告（Pi 5 CPU vs Hailo-10 NPU） + +日期：2026-08-13 +设备：Raspberry Pi 5（4×A76）+ AI HAT 2（Hailo-10H） +模型：actpred（~450K 参数，frame_len=3，history_len=5，action_dim=18） + +## 1. 测试方法 + +- **PyTorch CPU**：`ActionPredictorInference(checkpoints/best_inference.pt)`， + 输入 3 帧 640×480 随机图 + 全零动作历史，预热 3 次后取 20 次均值（含 PIL 预处理） +- **ONNX Runtime CPU**：`export_onnx.py` 导出后 ORT 加载，纯模型推理， + 预热 3 次后取 50 次均值 +- **Hailo-10 NPU**：板载 `resnet_v1_50_h10.hef`（25M 参数，比 actpred 大 ~55 倍）， + 新版 `create_infer_model` API 串行单帧，预热 10 次后取 200 次均值； + 另有 `hailortcli benchmark` pipelined 吞吐 + +## 2. 结果 + +| 方案 | 延迟 | 吞吐 | 备注 | +|---|---|---|---| +| PyTorch CPU | 28.8 ms | 34.8 FPS | 含 PIL 图像预处理 | +| ONNX Runtime CPU | 5.9 ms | 168.7 FPS | 纯推理，float32 输入 | +| Hailo-10 NPU（resnet50，串行单帧） | 4.29 ms | 233 FPS | 含 Python/PCIe 开销 | +| Hailo-10 NPU（resnet50，pipelined） | — | 307 FPS | hailortcli benchmark | + +## 3. 解读 + +1. resnet50 参数量约为 actpred 的 55 倍，NPU 上 actpred HEF 推理预计 + **亚毫秒级**，NPU 不会成为瓶颈；串行单帧 4.29ms 中 Python/PCIe + 开销占比高，闭环可用 `run_async` 流水线进一步压低。 +2. 不迁移 NPU 也完全够用：ONNX Runtime 比 PyTorch 快近 5 倍（169 FPS）， + 是短期最优 CPU 方案。 +3. HAT 的真实价值在于**把推理从 CPU 卸载**，让 4×A76 专注 pybot + 主控循环与图像采集，而非绝对延迟。 + +## 4. 部署建议（按优先级） + +1. **立即可用**：给 actpred 加 ONNX Runtime 后端（改动小，收益 ~5 倍） +2. **有 x86 环境后**：DFC 编译 HEF，走 `actpred/hailo_infer.py` + 的 `HailoPredictor` 端到端验证 +3. DFC 编译环境的说明：官方仅提供 x86_64 Linux wheel + （`hailo_dataflow_compiler-*-linux_x86_64.whl`，无 arm64 构建）， + 但 HEF 产物与主机架构无关，x86 上编译一次即可在树莓派长期使用； + 无本机 x86 时可用云服务器临时实例或 QEMU 模拟（后者极慢，不推荐）。 + 编译环境调研详见 + [actpred_hailo_hat2_inference_research_20260813.md](actpred_hailo_hat2_inference_research_20260813.md) + +## 参考 + +- [张高兴的 Raspberry Pi AI 开发指南：将自定义模型编译为 Hailo NPU 的 .hef 模型](https://www.cnblogs.com/zhanggaoxing/articles/18597244.html) +- [Hailo Dataflow 编译器安装](https://blog.csdn.net/qq_40843129/article/details/148557199) +- [Hailo-8 系列——Hailo Dataflow Compiler 使用经验分享](https://blog.csdn.net/stoat04/article/details/137569246)