python写的rtsp码流录制工具

显示全部楼层 · 发表于前天 23:57

运行环境，windows10，python3.13.2。
需要提前创建tmp和video文件夹。

运行截图

代码
[Python]

import sys
import os
import threading
import cv2
import datetime
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
QHBoxLayout, QLabel, QLineEdit, QPushButton,
QTextEdit, QFrame, QMessageBox, QSplitter)
from PyQt5.QtGui import QImage, QPixmap, QFont
from PyQt5.QtCore import Qt, QTimer, pyqtSignal, pyqtSlot
class RTSPRecorder(QMainWindow):
# 定义信号用于线程间通信
log_signal = pyqtSignal(str)
preview_signal = pyqtSignal(object)
def __init__(self):
super().__init__()
self.rtsp_url = ""
self.is_recording = False
self.cap = None
self.output_file = ""
self.log_messages = []
# 先初始化UI，确保log_text控件先创建
self.init_ui()
# 连接信号和槽
self.log_signal.connect(self.add_log_slot)
self.preview_signal.connect(self.update_preview_slot)
# 然后再创建必要的目录
self.create_directories()
# 定时器用于更新预览
self.timer = QTimer(self)
self.timer.timeout.connect(self.update_preview)
# 录制线程
self.record_thread = None
def create_directories(self):
"""创建video和tmp目录"""
try:
# 获取脚本所在目录的绝对路径，而不是依赖当前工作目录
script_dir = os.path.dirname(os.path.abspath(__file__))
self.video_dir = os.path.join(script_dir, "video")
self.tmp_dir = os.path.join(script_dir, "tmp")
self.log_signal.emit(f"脚本目录: {script_dir}")
self.log_signal.emit(f"视频目录将创建在: {self.video_dir}")
self.log_signal.emit(f"临时目录将创建在: {self.tmp_dir}")
# 确保父目录存在
os.makedirs(script_dir, exist_ok=True)
if not os.path.exists(self.video_dir):
os.makedirs(self.video_dir, exist_ok=True)
self.log_signal.emit(f"成功创建视频目录: {self.video_dir}")
else:
self.log_signal.emit(f"视频目录已存在: {self.video_dir}")
if not os.path.exists(self.tmp_dir):
os.makedirs(self.tmp_dir, exist_ok=True)
self.log_signal.emit(f"成功创建临时目录: {self.tmp_dir}")
else:
self.log_signal.emit(f"临时目录已存在: {self.tmp_dir}")
# 测试目录写入权限
test_file = os.path.join(self.video_dir, "test_permission.txt")
with open(test_file, "w") as f:
f.write("Permission test")
os.remove(test_file)
self.log_signal.emit(f"成功测试目录写入权限")
except Exception as e:
error_msg = f"无法创建或访问必要的目录: {str(e)}"
self.log_signal.emit(error_msg)
# 尝试使用系统临时目录作为备选
try:
self.video_dir = os.path.join(os.environ.get("TEMP", "/tmp"), "rtsp_video")
self.tmp_dir = os.path.join(os.environ.get("TEMP", "/tmp"), "rtsp_tmp")
os.makedirs(self.video_dir, exist_ok=True)
os.makedirs(self.tmp_dir, exist_ok=True)
self.log_signal.emit(f"使用系统临时目录作为备选: {self.video_dir}")
self.log_signal.emit(f"使用系统临时目录作为备选: {self.tmp_dir}")
except Exception as e2:
QMessageBox.critical(self, "目录创建错误",
f"无法创建必要的目录:\n{error_msg}\n\n" \
f"尝试使用系统临时目录也失败:\n{str(e2)}")
sys.exit(1)
def init_ui(self):
"""初始化用户界面"""
self.setWindowTitle("RTSP 录制工具")
self.setGeometry(100, 100, 800, 600)
# 设置中文字体
font = QFont()
font.setFamily("SimHei")
self.setFont(font)
# 主布局
main_widget = QWidget()
main_layout = QVBoxLayout(main_widget)
# RTSP地址输入区域
rtsp_layout = QHBoxLayout()
rtsp_label = QLabel("RTSP地址:")
self.rtsp_input = QLineEdit()
self.rtsp_input.setPlaceholderText("rtsp://username:password@ip:port/path")
self.rtsp_input.setFont(font)
rtsp_layout.addWidget(rtsp_label)
rtsp_layout.addWidget(self.rtsp_input)
# 按钮区域
button_layout = QHBoxLayout()
self.start_btn = QPushButton("开始录制")
self.start_btn.clicked.connect(self.start_recording)
self.start_btn.setFont(font)
self.stop_btn = QPushButton("停止录制")
self.stop_btn.clicked.connect(self.stop_recording)
self.stop_btn.setEnabled(False)
self.stop_btn.setFont(font)
button_layout.addWidget(self.start_btn)
button_layout.addWidget(self.stop_btn)
# 预览窗口
preview_frame = QFrame()
preview_frame.setFrameShape(QFrame.StyledPanel)
preview_layout = QVBoxLayout(preview_frame)
self.preview_label = QLabel("视频预览")
self.preview_label.setAlignment(Qt.AlignCenter)
self.preview_label.setMinimumHeight(300)
self.preview_label.setFont(font)
preview_layout.addWidget(self.preview_label)
# 日志区域
log_frame = QFrame()
log_frame.setFrameShape(QFrame.StyledPanel)
log_layout = QVBoxLayout(log_frame)
log_label = QLabel("运行日志")
log_label.setFont(font)
self.log_text = QTextEdit()
self.log_text.setReadOnly(True)
self.log_text.setFont(font)
log_layout.addWidget(log_label)
log_layout.addWidget(self.log_text)
# 使用分隔器布局预览和日志区域
splitter = QSplitter(Qt.Vertical)
splitter.addWidget(preview_frame)
splitter.addWidget(log_frame)
splitter.setSizes([300, 200])
# 添加所有组件到主布局
main_layout.addLayout(rtsp_layout)
main_layout.addLayout(button_layout)
main_layout.addWidget(splitter)
self.setCentralWidget(main_widget)
@pyqtSlot(str)
def add_log_slot(self, message):
"""添加日志信息的槽函数，确保在主线程中执行"""
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_entry = f"[{timestamp}] {message}"
self.log_messages.append(log_entry)
self.log_text.append(log_entry)
# 自动滚动到底部
self.log_text.verticalScrollBar().setValue(
self.log_text.verticalScrollBar().maximum()
)
@pyqtSlot(object)
def update_preview_slot(self, frame):
"""更新视频预览的槽函数，确保在主线程中执行"""
if frame is not None:
# 转换为RGB格式
rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
# 转换为QImage
h, w, c = rgb_frame.shape
q_img = QImage(rgb_frame.data, w, h, w * c, QImage.Format_RGB888)
# 缩放图像以适应预览窗口
pixmap = QPixmap.fromImage(q_img).scaled(
self.preview_label.width(),
self.preview_label.height(),
Qt.KeepAspectRatio,
Qt.SmoothTransformation
)
self.preview_label.setPixmap(pixmap)
def add_log(self, message):
"""添加日志信息（为了兼容旧代码而保留）"""
self.log_signal.emit(message)
def start_recording(self):
"""开始录制RTSP流"""
self.rtsp_url = self.rtsp_input.text().strip()
if not self.rtsp_url:
QMessageBox.warning(self, "警告", "请输入RTSP地址")
return
self.is_recording = True
self.start_btn.setEnabled(False)
self.stop_btn.setEnabled(True)
# 创建录制线程
self.record_thread = threading.Thread(target=self.record_stream)
self.record_thread.daemon = True
self.record_thread.start()
# 启动预览定时器
self.timer.start(30) # 大约33fps
self.log_signal.emit(f"开始录制RTSP流: {self.rtsp_url}")
def stop_recording(self):
"""停止录制RTSP流"""
if not self.is_recording:
return
self.is_recording = False
self.start_btn.setEnabled(True)
self.stop_btn.setEnabled(False)
# 停止定时器
self.timer.stop()
# 等待录制线程结束
if self.record_thread and self.record_thread.is_alive():
self.record_thread.join(timeout=3.0) # 设置超时，避免卡死
# 释放资源
if self.cap:
self.cap.release()
self.cap = None
# 清理tmp目录
self.clean_tmp_directory()
self.log_signal.emit(f"停止录制，文件保存至: {self.output_file}")
self.preview_label.setText("视频预览")
def record_stream(self):
"""录制RTSP流的线程函数"""
try:
# 禁用OpenCV的多线程，避免FFmpeg错误
cv2.setNumThreads(0)
# 打开RTSP流
self.cap = cv2.VideoCapture(self.rtsp_url)
if not self.cap.isOpened():
self.log_signal.emit("无法打开RTSP流，请检查地址是否正确")
# 在录制线程中使用信号停止录制
self.is_recording = False
# 通过调用主线程的stop_recording来清理资源
QApplication.instance().invokeMethod(self, "stop_recording", Qt.QueuedConnection)
return
# 获取视频信息
fps = self.cap.get(cv2.CAP_PROP_FPS)
width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
self.log_signal.emit(f"视频信息 - 分辨率: {width}x{height}, FPS: {fps:.2f}")
# 创建输出文件名
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
self.output_file = os.path.join(self.video_dir, f"recording_{timestamp}.mp4")
# 定义编码器和创建VideoWriter对象
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter(self.output_file, fourcc, fps, (width, height))
if not out.isOpened():
self.log_signal.emit("无法创建输出视频文件，请检查权限")
self.is_recording = False
QApplication.instance().invokeMethod(self, "stop_recording", Qt.QueuedConnection)
return
# 缓存文件
tmp_file = os.path.join(self.tmp_dir, f"temp_{timestamp}.mp4")
tmp_out = cv2.VideoWriter(tmp_file, fourcc, fps, (width, height))
frame_count = 0
error_count = 0
last_preview_time = datetime.datetime.now()
preview_interval = datetime.timedelta(milliseconds=100) # 每100ms更新一次预览
while self.is_recording:
ret, frame = self.cap.read()
if not ret:
error_count += 1
self.log_signal.emit(f"无法接收帧 {error_count} 次，可能是网络问题")
if error_count > 10: # 连续10次错误则停止
self.log_signal.emit("连续多次无法接收帧，停止录制")
break
else:
error_count = 0 # 重置错误计数
# 写入正式文件和缓存文件
out.write(frame)
tmp_out.write(frame)
frame_count += 1
if frame_count % 100 == 0: # 每100帧记录一次
self.log_signal.emit(f"已录制 {frame_count} 帧")
# 控制预览频率，避免UI更新太频繁
current_time = datetime.datetime.now()
if current_time - last_preview_time > preview_interval:
# 发送帧给预览槽函数
self.preview_signal.emit(frame.copy())
last_preview_time = current_time
# 释放资源
out.release()
tmp_out.release()
self.log_signal.emit(f"录制完成，共录制 {frame_count} 帧")
except Exception as e:
self.log_signal.emit(f"录制过程中发生错误: {str(e)}")
finally:
# 确保释放资源
if hasattr(self, 'cap') and self.cap:
self.cap.release()
self.cap = None
def update_preview(self):
"""更新视频预览窗口（已修改为通过信号槽机制实现）"""
# 此方法现在基本为空，实际预览逻辑已移至record_stream方法中通过preview_signal实现
pass
def clean_tmp_directory(self):
"""清理tmp目录"""
try:
for filename in os.listdir(self.tmp_dir):
file_path = os.path.join(self.tmp_dir, filename)
try:
if os.path.isfile(file_path) or os.path.islink(file_path):
os.unlink(file_path)
self.log_signal.emit(f"删除临时文件: {filename}")
elif os.path.isdir(file_path):
# 如果是目录，递归删除
import shutil
shutil.rmtree(file_path)
self.log_signal.emit(f"删除临时目录: {filename}")
except Exception as e:
self.log_signal.emit(f"清理文件 {file_path} 时出错: {str(e)}")
self.log_signal.emit("已清理临时文件目录")
except Exception as e:
self.log_signal.emit(f"清理临时目录时出错: {str(e)}")
def closeEvent(self, event):
"""窗口关闭事件处理"""
if self.is_recording:
reply = QMessageBox.question(
self, "确认", "正在录制中，确定要退出吗？",
QMessageBox.Yes | QMessageBox.No, QMessageBox.No
)
if reply == QMessageBox.Yes:
self.stop_recording()
event.accept()
else:
event.ignore()
else:
event.accept()
if __name__ == "__main__":
app = QApplication(sys.argv)
# 设置全局字体，确保中文正常显示
font = QFont("SimHei")
app.setFont(font)
window = RTSPRecorder()
window.show()
sys.exit(app.exec_())

复制代码

附上代码常见报错解决办法，感谢cxqdly 的分享
运行脚本遇到三个问题以及解决办法：
1. ModuleNotFoundError: No module named 'cv2'
错误原因：OpenCV 库未安装
解决办法：pip install opencv-python -i https://pypi.tuna.tsinghua.edu.cn/simple

2. ModuleNotFoundError: No module named 'PyQt5'
错误原因：PyQt5 未安装
解决办法：pip install PyQt5 -i https://pypi.tuna.tsinghua.edu.cn/simple

3. qt.qpa.plugin: Could not find the Qt platform plugin "windows" in ""
错误原因：未配置环境变量
解决办法：新建系统变量QT_QPA_PLATFORM_PLUGIN_PATH，路径为PyQt5/Qt/plugins/platforms 目录中qwindows.dll对应的地址

4. 如果第三步配置变量后运行还报一样的错误，关闭cmd窗口后重新打开运行即可显示rtsp运行的界面

发表于前天 23:58

运行脚本遇到三个问题以及解决办法：
1. ModuleNotFoundError: No module named 'cv2'
错误原因：OpenCV 库未安装
解决办法：pip install opencv-python -i https://pypi.tuna.tsinghua.edu.cn/simple

2. ModuleNotFoundError: No module named 'PyQt5'
错误原因：PyQt5 未安装
解决办法：pip install PyQt5 -i https://pypi.tuna.tsinghua.edu.cn/simple

3. qt.qpa.plugin: Could not find the Qt platform plugin "windows" in ""
错误原因：未配置环境变量
解决办法：新建系统变量QT_QPA_PLATFORM_PLUGIN_PATH，路径为PyQt5/Qt/plugins/platforms 目录中qwindows.dll对应的地址

4. 如果第三步配置变量后运行还报一样的错误，关闭cmd窗口后重新打开运行即可显示rtsp运行的界面

发表于前天 23:58

rtsp地址一般从摄像机获取，各厂家的摄像机码流格式都不一样的，在网上搜，找到了可以放VLC里面先播放试试。

登录 · 发表于前天 23:59

这样的话，电脑可以作为录播来使用了。楼主高手

发表于前天 23:59

666，摄像头电脑客户端了，搞个4画面的，直接电脑监控、录像了

发表于昨天 00:00

不错哦，相当于录像机了

发表于昨天 00:01

不错啊，代码整齐，是ai辅助的吧

发表于昨天 00:01

感谢楼主的热心分享

发表于昨天 00:02

大神；可以运行了；RTSP地址怎么获取啊？

发表于昨天 00:02

PIP 下我也是出现这问题

python写的rtsp码流录制工具

本帖子中包含更多资源

本帖子中包含更多资源