平衡滚球图传录制
方案介绍：
- OpenMV Cam 开启 AP 热点模式，始终运行色块识别（找滚球），同时通过 RTSP 按需图传。
- 电脑端用 Python + OpenCV 实现 RTSP 客户端，负责播放直播画面，并支持一键录制、截图和历史录像回放。
流程：
- OpenMV 上电后立即开启 AP 热点，色块识别（找小球）+ 位置式 PID 主循环全程运行，不受是否有人连接影响。
- RTSP 服务随 AP 一起启动，不阻塞等待客户端连接；
poll()内部会透明处理客户端连入/断开，只有真正在播放（PLAY）时才会发送图像。 - 电脑或手机连接热点后，用支持 RTSP 的播放器（比如 VLC）或前面提到的电脑端脚本打开
rtsp://<OpenMV的AP的IP>:554/，就能实时看到并录制识别画面了。
为了不让 RTSP 收发阻塞识别的主循环，这里没有直接用 rtsp_server.stream()，而是继承 rtsp.rtsp_server 重写了一个 rtsp_server_poll，把 TCP 连接的 accept、收数据、发送图像都改成了非阻塞的 poll() 方法，每次调用最多等待 10ms，放进 while True 主循环里跟色块识别一起执行，只有真正有客户端在播放时才会发送图像，尽量不影响识别帧率。
# 小球居中控制程序（位置式 PID + RTSP 推流）
# OpenMV 在狭长的轨道区域内识别深色小球，用位置式 PID
# 调节俯仰轴，使小球的横坐标靠近画面中心，并通过 RTSP 实时预览。
#
# 安全警告：使用官方云台时禁止执行 G28，回零动作会损坏设备。
# 程序启动时只读取当前位置作为中立位置，不进行机械回零。
#
# 【用户设备说明】
# 请根据你实际使用的设备修改以下配置：
# 1. THRESHOLD —— 深色小球在 LAB 颜色空间中的阈值，按你的小球颜色标定。
# 2. TRACK_LENGTH_MM —— 轨道实际长度（毫米），请按实物测量。
# 3. TARGET_POSITION_MM —— 小球的目标停止位置（毫米），0=左端，TRACK_LENGTH_MM=右端。
# 4. window((x,y,w,h)) —— 摄像头识别窗口，请对准你的轨道画面区域。
# 5. KP/KI/KD 等 PID 参数 —— 按你的电机与轨道实际响应整定。
# 6. MOVE_SPEED / CMD_INTERVAL_MS —— 按你的电机速度与串口刷新能力调整。
# 7. USE_WINC1500 —— 选择 WiFi 硬件：True=WINC1500 扩展板，False=板载 WiFi。
# 8. SSID / KEY / CHANNEL —— 热点的名称、密码与信道。
#
# 【RTSP 使用】
# 电脑/手机连接热点后，用 VLC 等播放器打开：
# rtsp://<终端打印的IP>:554/
import errno
import network
import rtsp
import socket
import csi
import time
from pid import PID # PID 控制器，见 pid.py
# import PanTilt # 云台相关：连接云台后取消注释
class rtsp_server_poll(rtsp.rtsp_server):
# ------------------------------------------------------------
# 持久监听 RTSP TCP 端口。
# 每次调用最多阻塞 10ms，不会影响识别主循环。
# ------------------------------------------------------------
def __valid_tcp_socket(self):
if self.__tcp__socket is None:
try:
if getattr(self, "_listener", None) is None:
s = socket.socket(
socket.AF_INET,
socket.SOCK_STREAM,
)
if hasattr(socket, "SO_REUSEADDR"):
s.setsockopt(
socket.SOL_SOCKET,
socket.SO_REUSEADDR,
1,
)
s.bind(self.__myaddr)
s.listen(1)
s.settimeout(0.01)
self._listener = s
self.__tcp__socket, self.__client_addr = \
self._listener.accept()
print("RTSP client connected:", self.__client_addr)
except OSError as e:
error_no = getattr(e, "errno", None)
if error_no not in (
None,
errno.EAGAIN,
errno.ETIMEDOUT,
):
# 监听 socket 失效时重建
print("RTSP listen socket rebuild:", repr(e))
if getattr(self, "_listener", None) is not None:
try:
self._listener.close()
except OSError:
pass
self._listener = None
return self.__tcp__socket is not None
# ------------------------------------------------------------
# stream() 的单次循环体。
# 仅当客户端正在播放时才发送当前图像帧。
# ------------------------------------------------------------
def poll(self, img, quality=90):
if self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(0.001)
try:
data = self.__tcp__socket.recv(1400)
if data and len(data):
self.__parse_rtsp_request(data)
else:
# recv 返回空数据，说明客户端主动断开
raise OSError
except OSError as e:
error_no = getattr(e, "errno", None)
if error_no not in (
errno.EAGAIN,
errno.ETIMEDOUT,
):
raise e
if self.__playing:
self.__send_rtp(
lambda pathname, session: img,
quality,
)
except OSError as e:
print("poll socket error:", repr(e))
self.__close_tcp_socket()
self.__close_udp_socket()
# ================================================================
# 用户配置
# ================================================================
# WiFi 硬件选择：
# True = WINC1500 扩展板（network.WINC，仅 OPEN/WEP 安全，无需密码）
# False = 板载 WiFi（network.WLAN，可配置密码）
USE_WINC1500 = False
SSID = "OPENMV_AP"
KEY = "1234567890"
CHANNEL = 2
# LAB 颜色阈值：主要利用 L 通道筛选较暗的小球，a/b 通道保持全范围。
# 【用户】按你实际的小球颜色重新标定。
THRESHOLD = (0, 42, -128, 127, -128, 127)
# 轨道实际长度和小球的指定停止位置（单位：毫米）。
# 0 mm 对应 x=10，250 mm 对应 x=309，125 mm 对应画面中心。
# 【用户】按你的轨道实物测量值修改。
TRACK_LENGTH_MM = 250.0
TARGET_POSITION_MM = 125.0
POSITION_MARGIN_PX = 10
# PID 参数。
# 按你的电机与轨道实际响应整定。
KP_ANGLE = 0.8 # 比例项：主要纠偏力度。
KI_ANGLE = 0.3 # 积分项：消除静态偏差。
KD_ANGLE = 0.15 # 微分项：抑制快速摆动。
INTEGRAL_LIMIT = 400 # 积分限幅，防止积分饱和。
MAX_TILT_OFFSET = 100 # 输出限幅，防止动作过大。
MOVE_SPEED = 400 # PG1 位置指令使用的电机速度。
CMD_INTERVAL_MS = 10 # PID 计算和电机指令周期（10ms = 100Hz）。
# ================================================================
# 摄像头初始化
# ================================================================
# 使用 QVGA 彩色图像，只保留轨道所在的 320×24 狭长区域。
# 【用户】window 参数请对准你轨道上的画面区域。
csi0 = csi.CSI()
csi0.reset()
csi0.pixformat(csi.RGB565)
csi0.framesize(csi.QVGA)
csi0.window((0, 110, 320, 24))
# 等待曝光稳定后关闭自动增益和自动白平衡，避免颜色阈值随画面变化漂移。
csi0.snapshot(time=2000)
csi0.auto_gain(False)
csi0.auto_whitebal(False)
# ================================================================
# WiFi 初始化（两种方式二选一，由 USE_WINC1500 决定）
# ================================================================
if USE_WINC1500:
# WINC1500 扩展板 AP 模式（仅支持 OPEN/WEP 安全，无需密码）。
wlan = network.WINC(mode=network.WINC.MODE_AP)
wlan.start_ap(SSID, security=network.WINC.OPEN, channel=CHANNEL)
else:
# 板载 WiFi AP 模式。
wlan = network.WLAN(network.AP_IF)
wlan.config(ssid=SSID, key=KEY, channel=CHANNEL)
wlan.active(True)
while not wlan.active():
time.sleep_ms(100)
ap_ip = wlan.ifconfig()[0]
print("================================")
print("WiFi AP started")
print("SSID:", SSID)
print("IP :", ap_ip)
print("RTSP URL: rtsp://{}:554/".format(ap_ip))
print("Waiting for device to connect AP (RTSP server already running)...")
print("Connect with VLC: rtsp://{}:554/".format(ap_ip))
print("================================")
# 注意：此处不阻塞等待客户端连接。
# RTSP 服务的 poll() 会透明处理客户端连入/断开，客户端随时可以播放。
# ================================================================
# RTSP 服务
# ================================================================
server = rtsp_server_poll(wlan)
server.register_setup_cb(
lambda pathname, session:
print("RTSP SETUP", pathname, session),
)
server.register_play_cb(
lambda pathname, session:
print("RTSP PLAY", pathname, session),
)
server.register_teardown_cb(
lambda pathname, session:
print("RTSP TEARDOWN", pathname, session),
)
# ================================================================
# PID 控制器
# ================================================================
pid = PID(
kp=KP_ANGLE,
ki=KI_ANGLE,
kd=KD_ANGLE,
integral_limit=INTEGRAL_LIMIT,
output_limit=MAX_TILT_OFFSET,
)
last_cmd_ms = 0 # 上一次执行 PID 并发送位置指令的时刻。
neutral_tilt = 0 # 程序启动时读取的当前位置，作为安全中立位置。
# ================================================================
# 云台初始化（云台相关：连接云台后取消注释）
# ================================================================
# pt = PanTilt.PanTilt(uart_port=3)
# pt.cmd("M17") # 使能电机
# pt.cmd("G90") # 绝对位置模式
# ================================================================
# 主循环：色块识别 + 位置式 PID + RTSP 推流
# ================================================================
clock = time.clock()
frame_i = 0
while True:
clock.tick()
frame_i += 1
# 获取一帧图像。先把毫米目标限制在轨道范围内，再将轨道的
# 0～250 mm 线性映射到画面的 x=10～309，忽略左右各 10 像素。
img = csi0.snapshot()
target_mm = max(
0.0,
min(TRACK_LENGTH_MM, TARGET_POSITION_MM),
)
map_left_x = POSITION_MARGIN_PX
map_right_x = img.width() - 1 - POSITION_MARGIN_PX
setpoint_x = int(
map_left_x
+ target_mm
* (map_right_x - map_left_x)
/ TRACK_LENGTH_MM
+ 0.5
)
# 找出满足深色阈值的连通区域（merge=True 合并相邻区域），
# 选择像素数最多的色块作为目标，并在预览画面中标出。
blobs = img.find_blobs(
[THRESHOLD],
pixels_threshold=5,
area_threshold=5,
merge=True,
)
target = None
if blobs:
target = max(blobs, key=lambda blob: blob.pixels)
img.draw_detection(target)
now = time.ticks_ms()
if target is not None:
# 横向误差定义为“指定停止位置 - 小球位置”：
# 正值表示小球在指定位置左侧，负值表示小球在指定位置右侧。
error_x = setpoint_x - target.cx
# 每 CMD_INTERVAL_MS(10ms) 触发一次，即固定 100Hz 周期；
# ticks_diff 可正确处理毫秒计数器回绕。
if time.ticks_diff(now, last_cmd_ms) >= CMD_INTERVAL_MS:
# 固定使用 100Hz 周期，dt 恒定。
dt = 0.01
# 位置式 PID 输出的是相对中立位置的俯仰偏移量。
tilt_offset = pid.compute(error_x, dt)
tilt_target = neutral_tilt + int(tilt_offset)
print("error_x=%d, tilt_offset=%d, tilt_target=%d" %
(error_x, tilt_offset, tilt_target))
# ========== 用户在此添加 PID 输出到电机的代码 ==========
# tilt_target : 本周期 PID 计算出的俯仰目标位置
# （已叠加中立位置）。
# MOVE_SPEED : 电机运动速度。
# 示例（连接云台后取消下面注释）：
# pt.cmd("PG1 T{} S{}".format(tilt_target, MOVE_SPEED))
# =====================================================
last_cmd_ms = now
# ------------------------------------------------------------
# RTSP 按需推流：仅当客户端 PLAY 时才发送当前帧。
# ------------------------------------------------------------
server.poll(img, quality=70)
# ================================================================
# 停止流程（云台相关：连接云台后取消注释）
# ================================================================
# pt.cmd("PG1 T{} S{}".format(neutral_tilt, MOVE_SPEED))
# time.sleep_ms(400)
# pt.cmd("V T0")
# time.sleep_ms(50)
# pt.cmd("M18")
# time.sleep_ms(50)
# print("POS_DONE MOTOR_STOPPED")
⚠️ 安全警告：如果连接了官方云台，禁止执行 G28，回零动作会损坏设备。程序启动时只读取云台当前位置作为中立位置，不做机械回零。
这份代码在原来"识别 + RTSP 推流"的基础上，加了一个位置式 PID：识别轨道内的深色小球，让小球的横坐标靠近指定停止位置，再把 PID 输出的俯仰偏移量交给云台（PanTilt，示例中默认注释掉，接好云台再打开）。用到自己的设备时需要按需修改：
THRESHOLD—— 深色小球在 LAB 颜色空间中的阈值，按你的小球颜色标定。TRACK_LENGTH_MM/TARGET_POSITION_MM—— 轨道实际长度和小球目标停止位置（毫米），按实物测量，0 对应轨道左端。csi0.window((0, 110, 320, 24))—— 摄像头识别窗口，请对准你实际轨道所在的画面区域。KP_ANGLE/KI_ANGLE/KD_ANGLE/INTEGRAL_LIMIT/MAX_TILT_OFFSET—— PID 参数，按电机与轨道实际响应整定。MOVE_SPEED/CMD_INTERVAL_MS—— 电机运动速度和 PID 计算/下发指令的周期。USE_WINC1500—— 选择 WiFi 硬件：True用 WINC1500 扩展板（network.WINC，仅 OPEN/WEP，无需密码），False用板载 WiFi（network.WLAN，可配置密码）。SSID/KEY/CHANNEL—— 热点名称、密码与信道。- 这一版不再等待 AP 客户端连接才启动 RTSP，而是热点一开就创建好 RTSP 服务，
poll()内部会透明处理客户端连入/断开，只有真正在播放时才会发送图像，平时没人连接播放不会额外消耗发送带宽和 CPU。 - PID 控制器
from pid import PID需要额外准备一个pid.py模块放在 OpenMV 存储上，模块的写法/引入方式可参考模块的使用一节。
pid.py（PID 控制器模块）
把下面的代码保存成 pid.py，和主程序一起放到 OpenMV 存储上（保存方式见模块的使用），主程序里 from pid import PID 就能直接引入。
# PID 控制器（位置式 PID）
#
# 用法：
# from pid import PID
# pid = PID(
# kp=0.8,
# ki=0.3,
# kd=0.15,
# integral_limit=400, # 积分限幅（可选）
# output_limit=100, # 输出限幅（可选）
# )
# output = pid.compute(error, dt) # dt 单位：秒
class PID:
"""位置式 PID 控制器。"""
def __init__(
self,
kp,
ki,
kd,
integral_limit=None,
output_limit=None,
):
self.kp = kp
self.ki = ki
self.kd = kd
self.integral_limit = integral_limit
self.output_limit = output_limit
self.reset()
def reset(self):
"""清空积分项和上一时刻误差。"""
self._integral = 0.0
self._last_error = 0.0
def compute(self, error, dt):
"""位置式 PID：输出 = kp*e + ki*∫e dt + kd*de/dt。
:param error: 当前误差（设定值 - 测量值）。
:param dt: 距上一次计算的时间间隔（秒）。
:return: PID 输出值。
"""
# 积分项
self._integral += error * dt
if self.integral_limit is not None:
self._integral = max(
-self.integral_limit,
min(self.integral_limit, self._integral),
)
# 微分项
derivative = (error - self._last_error) / dt if dt > 0 else 0.0
self._last_error = error
# 位置式输出
output = (
self.kp * error
+ self.ki * self._integral
+ self.kd * derivative
)
# 输出限幅
if self.output_limit is not None:
output = max(
-self.output_limit,
min(self.output_limit, output),
)
return output
电脑端：播放 + 录制回放脚本
OpenMV 端只负责按需推流，真正的"看直播、点按钮录像、回看录像"是在电脑端用 Python（OpenCV + tkinter）做的一个小工具，功能：
- 实时预览 OpenMV 推来的 RTSP 画面，断流会自动重连。
- 「开始录制/停止录制」按钮，一键把当前画面存成 MP4（默认存在
~/Movies/OpenMV/rec_时间戳.mp4）。 - 截图按钮，存一张当前画面的 JPG。
- 录像列表 + 内嵌回放：双击列表里的文件即可播放，支持暂停/继续、拖动进度条、回到直播。
- 拉流放在独立线程里，录制和界面刷新都从这一份最新帧读取，所以回放历史录像的时候，OpenMV 那边的直播录制并不会被打断。
- RTSP 走 UDP 会丢包、到达速率不稳定，所以录制时按"墙钟时间"而不是"收到的帧数"来写文件：这一秒该有 30 帧，就补足 30 帧（不够就复制上一帧），这样录出来的视频时长永远等于真实时长，丢包只会表现为画面短暂定格，而不会导致回放整体快进或变速。
依赖 opencv-python，运行方式：
.venv-rtsp/bin/python rtsp_recorder.py # 默认连接相机
.venv-rtsp/bin/python rtsp_recorder.py rtsp://<ip>:554/
.venv-rtsp/bin/python rtsp_recorder.py --test # 无窗口自检
#!/usr/bin/env python3
"""OpenMV RTSP 播放 + 录制器（tkinter 图形界面）。
功能:
- 实时预览相机 RTSP 流（断流自动重连）
- 「开始录制/停止录制」按钮，录像存 ~/Movies/OpenMV/rec_时间戳.mp4
- 截图按钮
- 录像列表 + 内嵌回放（播放/暂停、进度条拖动、回到直播）
- 回放期间录制不中断（拉流线程独立于显示）
用法:
.venv-rtsp/bin/python rtsp_recorder.py # 默认连相机
.venv-rtsp/bin/python rtsp_recorder.py rtsp://<ip>:554/
.venv-rtsp/bin/python rtsp_recorder.py --test # 无窗口自检
注意: OpenMV 的 rtsp 库是单连接服务器，看和录必须共用同一路流，
本程序用同一个连接同时做显示和录制。
"""
import argparse
import os
import subprocess
import sys
import threading
import time
from datetime import datetime
from pathlib import Path
# 必须在创建 VideoCapture 前设置：走 UDP。
# 注意不要加 fflags;nobuffer——MJPEG 流不带尺寸信息，探测阶段会因数据不足而失败。
os.environ.setdefault("OPENCV_FFMPEG_CAPTURE_OPTIONS", "rtsp_transport;udp")
import cv2
DEFAULT_URL = "rtsp://192.168.4.1:554/"
OUT_DIR = Path.home() / "Movies" / "OpenMV"
PLAY_FPS = 30.0 # 录制文件的时间轴帧率
class Recorder:
"""按固定 30fps 墙钟时间轴写 MP4。
RTSP over UDP 的到达帧率会因丢包波动，若按到达帧数直接写文件，
回放速度就不对。这里以墙钟时间为准：帧不足就复制上一帧补齐，
录像时长始终等于真实时长（丢包表现为短暂定格而非快进）。
"""
def __init__(self, outdir):
self.outdir = Path(outdir)
self.writer = None
self.path = None
self.frames = 0
self.t0 = 0.0
self.lock = threading.Lock()
@property
def active(self):
return self.writer is not None
def start(self, frame):
with self.lock:
self.outdir.mkdir(parents=True, exist_ok=True)
self.path = self.outdir / datetime.now().strftime("rec_%Y%m%d_%H%M%S.mp4")
h, w = frame.shape[:2]
self.writer = cv2.VideoWriter(
str(self.path), cv2.VideoWriter_fourcc(*"mp4v"), PLAY_FPS, (w, h)
)
self.frames, self.t0 = 0, time.time()
print("开始录制:", self.path)
def write(self, frame):
with self.lock:
if self.writer is None:
return
target = int((time.time() - self.t0) * PLAY_FPS) + 1
while self.frames < target:
self.writer.write(frame)
self.frames += 1
def stop(self):
with self.lock:
if self.writer is None:
return None
self.writer.release()
self.writer = None
elapsed = time.time() - self.t0
print("录制完成: %s (%.1f 秒)" % (self.path, elapsed))
return self.path
def open_stream(url):
return cv2.VideoCapture(url, cv2.CAP_FFMPEG)
class App:
def __init__(self, root, url, outdir):
import tkinter as tk
from tkinter import ttk
self.tk = tk
self.root = root
self.url = url
self.outdir = Path(outdir)
self.rec = Recorder(outdir)
self.latest = None # 最新一帧（BGR）
self.lock = threading.Lock()
self.status = "连接中"
self.fps = 0.0
self.stop_flag = False
self.mode = "live" # live | playback
self.play_cap = None
self.play_total = 0
self.play_paused = False
self._seek_guard = False # 区分程序设置滑块与用户拖动
root.title("OpenMV RTSP 录制器")
root.protocol("WM_DELETE_WINDOW", self.on_close)
# ---- 地址栏 ----
addr = tk.Frame(root)
addr.pack(fill="x", padx=6, pady=(6, 0))
tk.Label(addr, text="相机地址:").pack(side="left")
self.entry_url = tk.Entry(addr)
self.entry_url.insert(0, url)
self.entry_url.pack(side="left", fill="x", expand=True, padx=4)
self.entry_url.bind("<Return>", lambda e: self.connect_clicked())
tk.Button(addr, text="连接", width=8,
command=self.connect_clicked).pack(side="left")
self._gen = 0 # 连接代数：变化时拉流线程重开连接
# ---- 视频画面 ----
self.canvas = tk.Label(root, bg="black", width=64, height=20,
text="连接中: " + url, fg="white")
self.canvas.pack(fill="both", expand=True, padx=6, pady=6)
# ---- 控制按钮行 ----
bar = tk.Frame(root)
bar.pack(fill="x", padx=6)
self.btn_rec = tk.Button(bar, text="● 开始录制", width=12,
command=self.toggle_record)
self.btn_rec.pack(side="left", padx=2)
tk.Button(bar, text="截图", width=8,
command=self.snapshot).pack(side="left", padx=2)
tk.Button(bar, text="打开录像文件夹", width=14,
command=self.open_folder).pack(side="left", padx=2)
self.lbl_status = tk.Label(bar, text="", anchor="e")
self.lbl_status.pack(side="right", padx=4)
# ---- 回放控制行 ----
pb = tk.Frame(root)
pb.pack(fill="x", padx=6, pady=(4, 0))
self.btn_pause = tk.Button(pb, text="暂停", width=8, state="disabled",
command=self.toggle_pause)
self.btn_pause.pack(side="left", padx=2)
self.btn_live = tk.Button(pb, text="回到直播", width=10, state="disabled",
command=self.back_to_live)
self.btn_live.pack(side="left", padx=2)
self.seek = ttk.Scale(pb, from_=0, to=100, orient="horizontal",
command=self.on_seek, state="disabled")
self.seek.pack(side="left", fill="x", expand=True, padx=8)
# ---- 录像列表 ----
lf = tk.LabelFrame(root, text="录像列表（双击回放）")
lf.pack(fill="x", padx=6, pady=6)
inner = tk.Frame(lf)
inner.pack(fill="x", padx=4, pady=4)
sb = tk.Scrollbar(inner)
self.listbox = tk.Listbox(inner, height=5, yscrollcommand=sb.set)
sb.config(command=self.listbox.yview)
self.listbox.pack(side="left", fill="x", expand=True)
sb.pack(side="left", fill="y")
tk.Button(inner, text="▶ 回放选中", width=10,
command=self.play_selected).pack(side="left", padx=6)
self.listbox.bind("<Double-Button-1>", lambda e: self.play_selected())
self.refresh_list()
threading.Thread(target=self.reader, daemon=True).start()
root.after(33, self.tick)
def connect_clicked(self):
"""从输入框取地址并重连。支持裸 IP 或完整 rtsp:// 地址。"""
text = self.entry_url.get().strip()
if not text:
return
if not text.startswith("rtsp://"):
text = "rtsp://%s:554/" % text
self.entry_url.delete(0, "end")
self.entry_url.insert(0, text)
self.url = text
self._gen += 1 # 通知拉流线程换地址重连
self.status = "连接中"
# ---------- 拉流线程：显示与录制的共同来源，回放时也不停 ----------
def reader(self):
cap, gen = None, -1
n, t0 = 0, time.time()
while not self.stop_flag:
if cap is None or gen != self._gen:
if cap is not None:
cap.release()
gen = self._gen
self.status = "连接中"
self.fps = 0.0
cap = open_stream(self.url)
n, t0 = 0, time.time()
continue
ok, frame = cap.read()
if not ok:
if gen != self._gen: # 用户刚换了地址，立即重连不等待
continue
self.status = "断线重连中"
self.fps = 0.0
cap.release()
cap = None
time.sleep(2)
continue
self.status = "直播中"
with self.lock:
self.latest = frame
self.rec.write(frame) # 录制独立于显示模式
n += 1
dt = time.time() - t0
if dt >= 1.0:
self.fps, n, t0 = n / dt, 0, time.time()
if cap is not None:
cap.release()
# ---------- 定时刷新：按模式渲染画面 ----------
def tick(self):
if self.stop_flag:
return
if self.mode == "live":
with self.lock:
frame = self.latest
if frame is not None:
self.show(frame)
else:
self.playback_step()
text = "%s %.1f fps" % (self.status, self.fps)
if self.rec.active:
text += " ● REC %ds" % int(time.time() - self.rec.t0)
self.lbl_status.config(fg="red")
else:
self.lbl_status.config(fg="black")
self.lbl_status.config(text=text)
self.root.after(33, self.tick)
def show(self, frame):
ok, buf = cv2.imencode(".ppm", frame)
if not ok:
return
img = self.tk.PhotoImage(data=buf.tobytes())
self.canvas.configure(image=img, width=img.width(), height=img.height())
self.canvas.image = img # 防止被 GC
# ---------- 录制 ----------
def toggle_record(self):
if self.rec.active:
self.rec.stop()
self.btn_rec.config(text="● 开始录制", fg="black")
self.refresh_list()
else:
with self.lock:
frame = self.latest
if frame is None:
self.status = "未连接，无法录制"
return
self.rec.start(frame)
self.btn_rec.config(text="■ 停止录制", fg="red")
def snapshot(self):
with self.lock:
frame = self.latest
if frame is None:
return
self.outdir.mkdir(parents=True, exist_ok=True)
p = self.outdir / datetime.now().strftime("snap_%Y%m%d_%H%M%S.jpg")
cv2.imwrite(str(p), frame)
print("截图:", p)
def open_folder(self):
self.outdir.mkdir(parents=True, exist_ok=True)
if sys.platform == "darwin":
subprocess.Popen(["open", str(self.outdir)])
elif os.name == "nt":
os.startfile(str(self.outdir)) # noqa
else:
subprocess.Popen(["xdg-open", str(self.outdir)])
# ---------- 录像列表 / 回放 ----------
def refresh_list(self):
self.listbox.delete(0, "end")
if self.outdir.exists():
for p in sorted(self.outdir.glob("rec_*.mp4"), reverse=True):
self.listbox.insert("end", p.name)
def play_selected(self):
sel = self.listbox.curselection()
if not sel:
return
path = self.outdir / self.listbox.get(sel[0])
cap = cv2.VideoCapture(str(path))
if not cap.isOpened():
self.status = "无法打开 " + path.name
return
if self.play_cap is not None:
self.play_cap.release()
self.play_cap = cap
self.play_total = max(int(cap.get(cv2.CAP_PROP_FRAME_COUNT)), 1)
self.mode = "playback"
self.play_paused = False
self.status = "回放 " + path.name
self.btn_pause.config(state="normal", text="暂停")
self.btn_live.config(state="normal")
self.seek.config(state="normal", to=self.play_total - 1)
def playback_step(self):
if self.play_cap is None or self.play_paused:
return
ok, frame = self.play_cap.read()
if not ok: # 播完：停在结尾，可拖回或重播
self.play_paused = True
self.btn_pause.config(text="重新播放")
return
self.show(frame)
pos = self.play_cap.get(cv2.CAP_PROP_POS_FRAMES)
self._seek_guard = True
self.seek.set(pos)
self._seek_guard = False
def toggle_pause(self):
if self.play_cap is None:
return
if self.btn_pause["text"] == "重新播放":
self.play_cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
self.play_paused = False
self.btn_pause.config(text="暂停")
return
self.play_paused = not self.play_paused
self.btn_pause.config(text="继续" if self.play_paused else "暂停")
def on_seek(self, value):
if self._seek_guard or self.play_cap is None:
return
self.play_cap.set(cv2.CAP_PROP_POS_FRAMES, float(value))
# 拖动时立即显示目标帧（即使处于暂停）
ok, frame = self.play_cap.read()
if ok:
self.show(frame)
self.play_cap.set(cv2.CAP_PROP_POS_FRAMES, float(value))
def back_to_live(self):
self.mode = "live"
self.status = "直播中"
if self.play_cap is not None:
self.play_cap.release()
self.play_cap = None
self.btn_pause.config(state="disabled", text="暂停")
self.btn_live.config(state="disabled")
self.seek.config(state="disabled")
def on_close(self):
self.stop_flag = True
self.rec.stop()
self.root.destroy()
def run_test(url, outdir, seconds=3):
"""无窗口自检：连流、测 fps、录 seconds 秒、验证文件。"""
rec = Recorder(outdir)
cap = open_stream(url)
ok, frame = cap.read()
if not ok:
print("FAIL: 无法从流中读到帧")
return 1
print("连接成功，帧尺寸 %dx%d" % (frame.shape[1], frame.shape[0]))
n, t0 = 0, time.time()
while time.time() - t0 < 1.0:
ok, frame = cap.read()
if not ok:
print("FAIL: 测速阶段断流")
return 1
n += 1
print("实测 %.1f fps" % (n / (time.time() - t0)))
rec.start(frame)
t0 = time.time()
while time.time() - t0 < seconds:
ok, frame = cap.read()
if not ok:
print("FAIL: 录制阶段断流")
rec.stop()
return 1
rec.write(frame)
path = rec.stop()
cap.release()
size = path.stat().st_size
print("OK: %s (%.1f KB)" % (path, size / 1024))
return 0 if size > 0 else 1
def main():
ap = argparse.ArgumentParser(description="OpenMV RTSP player + recorder GUI")
ap.add_argument("url", nargs="?", default=DEFAULT_URL)
ap.add_argument("--outdir", default=str(OUT_DIR), help="录像/截图保存目录")
ap.add_argument("--test", action="store_true", help="无窗口自检后退出")
args = ap.parse_args()
if args.test:
sys.exit(run_test(args.url, args.outdir))
# uv/python-build-standalone 的 Tcl 路径是编译机路径，需手动指向真实位置
for var, name in (("TCL_LIBRARY", "tcl8.6"), ("TK_LIBRARY", "tk8.6")):
lib = Path(sys.base_prefix) / "lib" / name
if var not in os.environ and lib.is_dir():
os.environ[var] = str(lib)
import tkinter as tk
root = tk.Tk()
App(root, args.url, args.outdir)
root.mainloop()
if __name__ == "__main__":
main()
实际运行效果（电脑端界面正在直播预览、并已有历史录像）：
600倍AI算力提升，480+FPS彩色全局快门，120+FPS YOLO目标检测，200+FPS FOMO目标检测！
星瞳科技2026最新款OpenMV N6高性能AI智能图像识别摄像头正式开售啦！
快来了解详情～：
淘宝 星瞳科技OpenMV品牌店：https://item.taobao.com/item.htm?ft=t&id=1029138529714
天猫 OpenMV旗舰店：https://detail.tmall.com/item.htm?id=1028359546575
京东 星瞳旗舰店: https://item.jd.com/10212563443894.html
支持AI大模型及Agent协同；
标配480FPS百万像素彩色全局快门，可拍摄识别超高速运动物体；
120+FPS YOLO目标检测，高速运行复杂AI算法，支持语音识别；
内置WiFi、蓝牙5.1、以太网、麦克风、IMU；
内置H264和JPEG编码硬件加速，支持MP4录像及网络推流；
配套官方机械臂、无人机、智能车、云台等套件，还有竞赛支持，一键复制代码运行！