问题描述
新手求教
在使用K230做简单视觉算法时发现，无论视频流帧速率是多少，经sensor.snapshot都只有约33FPS出帧。请问应该如何提高帧数？需要换镜像吗？
谢谢
代码如下
import time
import image
from media.sensor import (
Sensor,
CAM_CHN_ID_0,
CAM_CHN_ID_1,
)
from media.display import Display
from media.media import MediaManager
import config
from vision.debug_overlay import DebugOverlay
from vision.interrupts import reraise_if_stop
_PIXFORMAT_NAME_TO_ENUM = {
"YUV420SP": Sensor.YUV420SP,
"RGB888": Sensor.RGB888,
"RGB565": Sensor.RGB565,
"GRAYSCALE": Sensor.GRAYSCALE,
}
_CHN_ID = {0: CAM_CHN_ID_0, 1: CAM_CHN_ID_1}
class Camera:
def __init__(self, cfg=None):
self.cfg = cfg if cfg is not None else config
self._sensor = None
self._osd = None
self._media_inited = False
self._display_inited = False
self._sensor_running = False
self._algo_chn = _CHN_ID[self.cfg.ALGO_CHN]
self._display_chn = _CHN_ID[self.cfg.DISPLAY_CHN]
# 帧统计
self._frame_count = 0
self._debug_overlay = DebugOverlay(self.cfg)
# sensor.snapshot() 在 K230 上偶发 RuntimeError（如 failed(3)）时按丢帧处理。
self._snapshot_error_log_remaining = 5
# ------------------------------------------------------------------ #
# 诊断
# ------------------------------------------------------------------ #
def log_configured_modes(self):
"""打印本次请求的 sensor mode 与各通道实际输出尺寸。
通道编号取自 config（``DISPLAY_CHN``/``ALGO_CHN``），不是 hardcode 的
"CHN0/CHN1"——swap 时也不会失真。
"""
if self._sensor is None:
return
print(
"[camera] request: CSI%d sensor=%dx%d@%d, "
"display→CHN%d(%dx%d)=%s, algo→CHN%d=%s"
% (
self.cfg.SENSOR_ID,
self.cfg.SENSOR_REQ_WIDTH,
self.cfg.SENSOR_REQ_HEIGHT,
self.cfg.SENSOR_NOMINAL_FPS,
self.cfg.DISPLAY_CHN,
self.cfg.SENSOR_DISPLAY_W,
self.cfg.SENSOR_DISPLAY_H,
self.cfg.DISPLAY_PIXFORMAT,
self.cfg.ALGO_CHN,
self.cfg.ALGO_PIXFORMAT,
)
)
for role, chn in (("display", self._display_chn),
("algo ", self._algo_chn)):
try:
print(
"[camera] CHN%d (%s): %dx%d"
% (
self.cfg.DISPLAY_CHN if role.strip() == "display"
else self.cfg.ALGO_CHN,
role,
self._sensor.width(chn=chn),
self._sensor.height(chn=chn),
)
)
except Exception as e:
print("[camera] query %s failed: %s" % (role, e))
# ------------------------------------------------------------------ #
# 生命周期
# ------------------------------------------------------------------ #
def init(self, pre_media_hook=None):
"""构造 sensor、绑定 CHN0 到 LCD、配置 CHN1 算法通道、初始化 Display + Media。
CHN0 显示通道输出不得超过传感器原生分辨率（SENSOR_DISPLAY_W/H），
与 LCD 物理尺寸（DISPLAY_WIDTH/HEIGHT）解耦，由 config 单独控制。
:param pre_media_hook: 可选回调，在 sensor0 配置完成后、Display.init /
MediaManager.init 之前调用。用于双摄场景下在此时机初始化 sensor1
（K230 SDK 要求所有 sensor 在 MediaManager.init 之前配置完毕）。
"""
self._sensor = Sensor(
id=self.cfg.SENSOR_ID,
width=self.cfg.SENSOR_REQ_WIDTH,
height=self.cfg.SENSOR_REQ_HEIGHT,
fps=self.cfg.SENSOR_NOMINAL_FPS,
)
self._sensor.reset()
self._sensor.set_hmirror(False)
self._sensor.set_vflip(False)
# ---- CHN0：显示通道 ----
# 输出尺寸上限为传感器原生分辨率（SENSOR_DISPLAY_W/H），不能超过
# SENSOR_REQ_WIDTH/HEIGHT；LCD 物理尺寸（DISPLAY_WIDTH/HEIGHT）
# 用于 Display.init 和 OSD，两者独立。
lcd_src = getattr(self.cfg, "LCD_VIDEO_SOURCE", "csi0")
if lcd_src == "csi0":
self._sensor.set_framesize(
chn=self._display_chn,
width=self.cfg.SENSOR_DISPLAY_W,
height=self.cfg.SENSOR_DISPLAY_H,
)
self._sensor.set_pixformat(
_PIXFORMAT_NAME_TO_ENUM[self.cfg.DISPLAY_PIXFORMAT],
chn=self._display_chn,
)
bind_info = self._sensor.bind_info(chn=self._display_chn)
Display.bind_layer(**bind_info, layer=Display.LAYER_VIDEO1)
else:
self._sensor.set_framesize(
chn=self._display_chn,
width=self.cfg.SENSOR_DISPLAY_W,
height=self.cfg.SENSOR_DISPLAY_H,
)
self._sensor.set_pixformat(
_PIXFORMAT_NAME_TO_ENUM[self.cfg.DISPLAY_PIXFORMAT],
chn=self._display_chn,
)
print(
"[camera] LCD_VIDEO_SOURCE=%s, CSI0 CHN0 configured but not "
"bound (no display drain)" % lcd_src
)
# ---- CHN1：算法通道 ----
self._sensor.set_framesize(
chn=self._algo_chn,
width=self.cfg.ALGO_WIDTH,
height=self.cfg.ALGO_HEIGHT,
)
self._sensor.set_pixformat(
_PIXFORMAT_NAME_TO_ENUM[self.cfg.ALGO_PIXFORMAT],
chn=self._algo_chn,
)
# 加深输出 buffer 队列（变更 C），默认 K230D 为 3，增至 6 消除 p95 长尾
trk_buf = getattr(self.cfg, "TRK_OUTPUT_BUFFER_NUM", 0)
if trk_buf > 0:
chn_idx = self.cfg.ALGO_CHN
self._sensor._chn_attr[chn_idx].buffer_num = trk_buf
# ---- 双摄挂接点 ----
# K230 SDK 要求所有 sensor 在 Display.init / MediaManager.init 之前完成
# 构造与通道配置。此 hook 供 AimCamera 等第二路 sensor 在此时机初始化。
if pre_media_hook is not None:
pre_media_hook()
# ---- Display ----
display_type = getattr(Display, self.cfg.DISPLAY_TYPE)
Display.init(
display_type,
width=self.cfg.DISPLAY_WIDTH,
height=self.cfg.DISPLAY_HEIGHT,
to_ide=self.cfg.DISPLAY_TO_IDE,
)
self._display_inited = True
MediaManager.init()
self._media_inited = True
# ---- OSD 缓冲 ----
if self.cfg.DEBUG_DISPLAY:
self._osd = image.Image(
self.cfg.DISPLAY_WIDTH,
self.cfg.DISPLAY_HEIGHT,
image.ARGB8888,
)
self._osd.clear()
self._debug_overlay.set_osd(self._osd)
# ---- 调试 overlay：计算坐标（不再预分配 blit/mask 中间图）---- #
self._debug_overlay.setup()
return self
def _apply_focus(self):
"""在 sensor.run() 之后尝试软件调焦。
"""
try:
caps = self._sensor.focus_caps()
except Exception as e:
print("[camera] focus_caps() error (non-fatal):", e)
return
# caps = (isSupport, minPos, maxPos)
if not caps or not caps[0]:
print(
"[camera] focus: VCM not supported by firmware "
"(caps=%s) — adjust lens barrel physically" % str(caps)
)
return
min_pos, max_pos = int(caps[1]), int(caps[2])
if self.cfg.SENSOR_AUTOFOCUS:
try:
self._sensor.auto_focus(True)
print("[camera] focus: auto")
except Exception as e:
print("[camera] auto_focus failed (non-fatal):", e)
else:
pos = max(min_pos, min(max_pos, self.cfg.SENSOR_FOCUS_POS))
try:
self._sensor.focus_pos(pos)
print(
"[camera] focus: manual pos=%d (range %d~%d)"
% (pos, min_pos, max_pos)
)
except Exception as e:
print("[camera] focus_pos failed (non-fatal):", e)
def start(self):
"""启动 sensor 数据流。MediaManager.init 之后调用（参见 Sensor 文档）。
焦点配置在 sensor.run() **之后**应用：sensor fd 由内核在 run() 启动
数据流时才真正打开，MediaManager.init() 只建立媒体图，fd 尚未就绪。
"""
if self._sensor is None:
raise RuntimeError("Camera.init() must be called before start()")
self._sensor.run()
self._sensor_running = True
self._apply_focus()
self._debug_overlay.reset_fps_window(time.ticks_ms())
return self
def stop(self):
"""严格逆序释放：sensor.stop → Display.deinit → MediaManager.deinit。
每个步骤都被保护，确保即使前一步抛异常也不会导致后续资源泄漏。
"""
try:
if self._sensor_running and isinstance(self._sensor, Sensor):
self._sensor.stop()
except Exception as e:
print("[camera] sensor.stop failed:", e)
finally:
self._sensor_running = False
if self._display_inited:
try:
Display.deinit()
except Exception as e:
print("[camera] Display.deinit failed:", e)
finally:
self._display_inited = False
# 让底层有时间归还 buffer
time.sleep_ms(50)
if self._media_inited:
try:
MediaManager.deinit()
except Exception as e:
print("[camera] MediaManager.deinit failed:", e)
finally:
self._media_inited = False
# ------------------------------------------------------------------ #
# 帧读取
# ------------------------------------------------------------------ #
def read_algo_frame(self, timeout=None):
"""从 CHN1 拉一帧算法输入。返回 ``image.Image`` 或 ``None``。
:param timeout: ``None`` 时使用 ``config.SNAPSHOT_TIMEOUT_MS``。
"""
if timeout is None:
timeout = self.cfg.SNAPSHOT_TIMEOUT_MS
try:
img = self._sensor.snapshot(chn=self._algo_chn, timeout=timeout)
except RuntimeError as e:
reraise_if_stop(e)
if self._snapshot_error_log_remaining > 0:
self._snapshot_error_log_remaining -= 1
print("[camera] snapshot RuntimeError:", e)
return None
except Exception as e:
reraise_if_stop(e)
if self._snapshot_error_log_remaining > 0:
self._snapshot_error_log_remaining -= 1
print("[camera] snapshot failed:", e)
return None
if img is not None:
self._frame_count += 1
return img
def read_algo_frame_fast(self, timeout=None):
"""变更 D：``dump_frame=True`` 跳过 sensor.py 格式检测，手动构造 Image。
预设 GRAYSCALE 格式（``ALGO_PIXFORMAT="GRAYSCALE"``，ISP 输出
YUV420 → Image 仅取 Y 平面）。若 py_video_frame_info 属性访问
失败则自动 fallback 到 :meth:`read_algo_frame`。
"""
if self._sensor is None:
return None
if timeout is None:
timeout = self.cfg.SNAPSHOT_TIMEOUT_MS
try:
vf = self._sensor.snapshot(
chn=self._algo_chn, timeout=timeout, dump_frame=True,
)
except RuntimeError as e:
reraise_if_stop(e)
if self._snapshot_error_log_remaining > 0:
self._snapshot_error_log_remaining -= 1
print("[camera] fast snapshot RuntimeError:", e)
return None
except Exception as e:
reraise_if_stop(e)
if self._snapshot_error_log_remaining > 0:
self._snapshot_error_log_remaining -= 1
print("[camera] fast snapshot failed:", e)
return None
if vf is None:
return None
try:
img = image.Image(
self.cfg.ALGO_WIDTH, self.cfg.ALGO_HEIGHT,
image.GRAYSCALE,
alloc=image.ALLOC_VB,
phyaddr=vf.v_frame.phys_addr[0],
virtaddr=vf.v_frame.virt_addr[0],
poolid=vf.pool_id,
)
except Exception as e:
if self._snapshot_error_log_remaining > 0:
self._snapshot_error_log_remaining -= 1
print("[camera] fast Image wrap failed, fallback:", e)
return self.read_algo_frame(timeout)
if img is not None:
self._frame_count += 1
return img
def release_algo_frame(self):
"""提前释放 CHN1 的 VB buffer（变更 B）。
受 ``config.TRK_EARLY_VB_RELEASE`` 开关守护（默认 False）：
12 fps 回归排查显示此操作可能在线程化场景下引入 VB 状态不一致。
"""
if not getattr(self.cfg, "TRK_EARLY_VB_RELEASE", False):
return
if self._sensor is None:
return
try:
chn_idx = self.cfg.ALGO_CHN
img_ref = self._sensor._imgs[chn_idx]
if img_ref is not None:
Sensor._release_image(img_ref)
self._sensor._imgs[chn_idx] = None
except Exception:
pass
def notify_track_solved(self):
"""循迹解算有效（如 ``detection.n_valid > 0``）时由主循环调用，计入 track_fps。"""
self._debug_overlay.on_track_frame()
def notify_aim_processed(self):
"""CSI1 取帧成功且 find_blobs 已跑完（不要求检出）时调用，计入 aim_proc_fps。"""
self._debug_overlay.on_aim_proc_frame()
def notify_aim_solved(self):
"""瞄准解算有效（检出至少 1 个目标）时调用，计入 aim_hit_fps。"""
self._debug_overlay.on_aim_hit_frame()
# ------------------------------------------------------------------ #
# FPS 统计（循迹 CSI0 / 瞄准 CSI1 独立）
# ------------------------------------------------------------------ #
def track_fps(self):
"""CSI0 循迹每秒解算有效帧数（Hz），非裸 snapshot 速率。"""
return self._debug_overlay.track_fps()
def aim_hit_fps(self):
"""CSI1 每秒检出有效帧数（Hz）。"""
return self._debug_overlay.aim_hit_fps()
def aim_proc_fps(self):
"""CSI1 每秒解算跑通帧数（Hz）；用于判断节流/取帧 vs 算力瓶颈。"""
return self._debug_overlay.aim_proc_fps()
def aim_fps(self):
"""兼容别名：等同 :meth:`aim_hit_fps`。"""
return self._debug_overlay.aim_fps()
def track_period_ms(self):
"""循迹平均帧周期（ms）= 1000 / track_fps。"""
return self._debug_overlay.track_period_ms()
def aim_hit_period_ms(self):
return self._debug_overlay.aim_hit_period_ms()
def aim_proc_period_ms(self):
return self._debug_overlay.aim_proc_period_ms()
def aim_period_ms(self):
return self._debug_overlay.aim_period_ms()
def frame_count(self):
return self._frame_count
# ------------------------------------------------------------------ #
# 启动期诊断：raw snapshot 计时
# ------------------------------------------------------------------ #
def probe_snapshot_timing(self, n_frames=60):
"""连续阻塞拉 ``n_frames`` 帧，统计每次 snapshot 的微秒耗时分布。
用于把"snapshot 自身阻塞时间"和"主循环其他开销（exitpoint / Python /
IDE 通讯）"分开诊断：
探针不计入 ``frame_count`` / ``track_fps`` 统计；执行完毕主循环再开始。
"""
if self._sensor is None or n_frames <= 0:
return
samples = []
# 先取 1 帧丢弃，避免首帧冷启动偏差污染统计。
try:
self._sensor.snapshot(chn=self._algo_chn,
timeout=self.cfg.SNAPSHOT_TIMEOUT_MS)
except Exception as e:
reraise_if_stop(e)
pass
for _ in range(n_frames):
t0 = time.ticks_us()
img = self._sensor.snapshot(
chn=self._algo_chn,
timeout=self.cfg.SNAPSHOT_TIMEOUT_MS,
)
dt = time.ticks_diff(time.ticks_us(), t0)
if img is not None:
samples.append(dt)
img = None
if not samples:
print("[camera.probe] no samples collected")
return
samples.sort()
n = len(samples)
avg = sum(samples) / n
p50 = samples[n // 2]
p95 = samples[min(n - 1, int(n * 0.95))]
print(
"[camera.probe] snapshot x%d min=%dus p50=%dus avg=%.0fus "
"p95=%dus max=%dus → fps_eq=%.1f"
% (n, samples[0], p50, avg, p95, samples[-1], 1e6 / avg)
)
def maybe_update_fps(self, now_ms=None):
"""1 秒内累计的算法帧数换算为 FPS。每 ``OSD_REFRESH_INTERVAL_MS`` 触发一次。"""
return self._debug_overlay.maybe_update_fps(now_ms)
def maybe_update_binary(self, now_ms=None):
"""二值 overlay 是否到了刷新时刻（独立于 FPS 文字行）。
``OSD_BINARY_REFRESH_MS=0`` 时永远返回 True（每帧刷新）；正数时按
该 ms 间隔节流。返回 True 表示主循环应当调用 :meth:`render_overlay`
把最新 ``detection`` 画上 OSD。
"""
return self._debug_overlay.maybe_update_binary(now_ms)
def maybe_update_path(self, now_ms=None):
"""路径标记是否到了刷新时刻（OSD_PATH_REFRESH_MS 控制）。
0 = 每帧；正数 = 节流到指定 ms；负数 = 不独立触发（跟随 1Hz 文字行）。
"""
return self._debug_overlay.maybe_update_path(now_ms)
def maybe_update_aim(self, now_ms=None):
"""瞄准标记是否到了刷新时刻（OSD_AIM_REFRESH_MS 控制）。
0 = 每帧；正数 = 节流到指定 ms；负数 = 不独立触发（跟随 1Hz 文字行）。
"""
return self._debug_overlay.maybe_update_aim(now_ms)
def binary_overlay_enabled(self):
"""返回主画面 ROI 红色二值 overlay 当前运行期状态。"""
return self._debug_overlay.binary_overlay_enabled()
def set_binary_overlay_enabled(self, enabled):
"""运行期切换主画面 ROI 红色二值 overlay。"""
return self._debug_overlay.set_binary_overlay_enabled(enabled)
# ------------------------------------------------------------------ #
# 调试叠加
# ------------------------------------------------------------------ #
def render_overlay(self, lines=None, detection=None, path=None,
aim_targets=None):
"""重绘 OSD：3 段 ROI + 总 ROI 外框 + 若干行文本 + (可选) 检测 / 路径 / 瞄准可视化。
"""
if not self.cfg.DEBUG_DISPLAY or self._osd is None:
return
if self._debug_overlay.draw(lines, detection, path,
aim_targets=aim_targets):
Display.show_image(self._osd, x=0, y=0, layer=Display.LAYER_OSD0)
# ------------------------------------------------------------------ #
# 采样（plan §12 阶段 A："静态赛道样张 ≥ 100 张"）
# ------------------------------------------------------------------ #
def save_algo_frame(self, img, path, quality=None):
"""把 CHN1 帧落盘为 JPEG。返回布尔成功标志。
K230 文件 IO 在 SD 卡上较慢，单帧约 30~80 ms；调用方应控制频率。
"""
if img is None:
return False
if quality is None:
quality = self.cfg.CAPTURE_JPEG_QUALITY
try:
img.save(path, quality=quality)
return True
except Exception as e:
print("[camera] save frame failed:", path, e)
return False
# ------------------------------------------------------------------ #
# 自检（plan §11.2 强制项）
# ------------------------------------------------------------------ #
def self_test(self):
"""阶段 A 自检：抓 5 帧算法输入，确认尺寸与像素格式符合配置。"""
ok = True
for _ in range(5):
img = self.read_algo_frame()
if img is None:
print("[camera.self_test] snapshot failed")
ok = False
break
if img.width() != self.cfg.ALGO_WIDTH or img.height() != self.cfg.ALGO_HEIGHT:
print(
"[camera.self_test] size mismatch %dx%d vs %dx%d"
% (
img.width(),
img.height(),
self.cfg.ALGO_WIDTH,
self.cfg.ALGO_HEIGHT,
)
)
ok = False
break
img = None
return ok
软件版本
CanMV_K230_V3P0_micropython_v1.6-0-ge4441cc_nncase_v2.9.0.img
