平衡滚球图传录制
方案介绍：
- OpenMV Cam 开启 AP 热点模式，始终运行色块识别（找滚球），同时通过 RTSP 按需图传。
- 电脑端用 Python + OpenCV 实现 RTSP 客户端，负责播放直播画面，并支持一键录制、截图和历史录像回放。
流程：
- OpenMV 上电后立即开启 AP 热点，并且识别主循环（找小球色块）始终在跑，不受是否有人连接影响。
- 程序等待电脑或手机连接到这个 AP 热点。
- 检测到至少一个客户端连接后，才启动 RTSP 服务端，避免没人观看时也白白占用资源。
- 电脑或手机用支持 RTSP 的播放器（比如 VLC）打开
rtsp://<OpenMV的AP的IP>:554/，就能实时看到并录制识别画面了。
为了不让 RTSP 收发阻塞识别的主循环，这里没有直接用 rtsp_server.stream()，而是继承 rtsp.rtsp_server 重写了一个 rtsp_server_poll，把 TCP 连接的 accept、收数据、发送图像都改成了非阻塞的 poll() 方法，每次调用最多等待 10ms，放进 while True 主循环里跟色块识别一起执行，只有真正有客户端在播放时才会发送图像，尽量不影响识别帧率。
# 识别始终运行 + AP 模式 + RTSP 按需推流
#
# 工作流程：
# 1. OpenMV 启动 AP 热点
# 2. 等待电脑或手机连接 AP
# 3. 检测到至少一个客户端后，启动 RTSP 和图像识别主循环
import errno
import network
import rtsp
import socket
import csi
import time
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:
# accept 超时或当前没有客户端连接
error_no = getattr(e, "errno", None)
if error_no not in (
None,
errno.EAGAIN,
errno.ETIMEDOUT
):
print("RTSP accept error:", repr(e))
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()
# ============================================================
# 摄像头初始化
# ============================================================
csi0 = csi.CSI()
csi0.reset()
csi0.pixformat(csi.RGB565)
csi0.framesize(csi.QVGA)
# 保留你的窗口设置
csi0.window((320, 20))
# 根据实际小球阈值修改
thresholds = (0, 37, -128, 127, -128, 127)
# ============================================================
# AP 热点初始化
# ============================================================
SSID = "OPENMV_AP"
KEY = "1234567890"
CHANNEL = 2
network_if = network.WLAN(network.AP_IF)
network_if.config(
ssid=SSID,
key=KEY,
channel=CHANNEL
)
network_if.active(True)
# 等待 AP 接口完成启动
while not network_if.active():
time.sleep_ms(100)
ap_ip = network_if.ifconfig()[0]
print("================================")
print("AP mode started")
print("SSID:", SSID)
print("KEY :", KEY)
print("IP :", ap_ip)
print("================================")
# ============================================================
# 等待客户端连接 AP
# ============================================================
print("Waiting for a device to connect to AP...")
wait_count = 0
while True:
try:
stations = network_if.status("stations")
except OSError as e:
stations = []
print("Read AP stations error:", repr(e))
if stations and len(stations) > 0:
print("AP client connected.")
print("Connected stations:", stations)
break
wait_count += 1
if wait_count % 10 == 0:
print("Still waiting for AP client...")
time.sleep_ms(500)
# ============================================================
# 客户端连接 AP 后，再创建 RTSP 服务
# ============================================================
print("Starting RTSP server...")
print("RTSP URL: rtsp://{}:554/".format(ap_ip))
server = rtsp_server_poll(network_if)
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)
)
# ============================================================
# 主循环
# 只有 AP 被连接后才会运行到这里
# ============================================================
clock = time.clock()
frame_i = 0
while True:
clock.tick()
frame_i += 1
img = csi0.snapshot()
# --------------------------------------------------------
# 图像识别
# --------------------------------------------------------
blobs = img.find_blobs(
[thresholds],
pixels_threshold=5,
area_threshold=5,
merge=True
)
for blob in blobs:
img.draw_detection(blob)
# 识别结果的其他用途可以放在这里
#
# if blobs:
# print("found", len(blobs), "blobs")
# --------------------------------------------------------
# RTSP 按需发送
# --------------------------------------------------------
server.poll(img, quality=70)
# --------------------------------------------------------
# 帧率打印
# --------------------------------------------------------
if frame_i % 100 == 0:
print(clock.fps(), "fps")
csi0.window((320, 20))是根据实际平衡滚球平台的安装位置裁剪出来的取景窗口，需要根据自己的摄像头安装角度和位置调整。thresholds需要根据实际使用的小球颜色重新取阈值。- 只有当画面在被 RTSP 客户端播放（
PLAY）时才会推送图像，平时没人连接播放不会额外消耗发送带宽和 CPU。
电脑端：播放 + 录制回放脚本
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录像及网络推流；
配套官方机械臂、无人机、智能车、云台等套件，还有竞赛支持，一键复制代码运行！