lgv/grpc_client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add camera client

Browse Source
This commit is contained in:
linbo 2026-03-18 09:19:23 +08:00
parent 3ab4fbc3e0
commit cf033e421b
3 changed files with 1232 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

503
clients/dexhand_client.py Normal file
View File

@ -0,0 +1,503 @@
import time
import threading
from typing import Optional, List
import numpy as np
import grpc
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
QHBoxLayout, QLabel, QPushButton, QComboBox,
QSpinBox, QGroupBox, QGridLayout, QLineEdit, QMessageBox)
from PyQt5.QtCore import QTimer, Qt
from PyQt5.QtGui import QFont
import pyqtgraph as pg
from clients._path_setup import ensure_paths
ensure_paths()
from cmvr.api import dexhand_service_pb2_grpc
from cmvr.api import dexhand_command_pb2
from google.protobuf.timestamp_pb2 import Timestamp
class DexHandSensorClient:
"""灵巧手传感器数据客户端"""
def __init__(self, ip="192.168.0.222", port=50052, device_id="dexhand_001", timeout=2, retries=1):
self.ip = ip
self.port = port
self.address = f"{ip}:{port}"
self.device_id = device_id
self.timeout = timeout
self.retries = retries
self.channel = None
self.stub = None
self._connect_with_retry()
def _connect_with_retry(self):
"""连接gRPC服务器"""
attempt = 0
while attempt <= self.retries:
print(f"Connecting to DexHand gRPC server at {self.address}... (Attempt {attempt + 1})")
try:
self.channel = grpc.insecure_channel(self.address)
grpc.channel_ready_future(self.channel).result(timeout=self.timeout)
self.stub = dexhand_service_pb2_grpc.DexHandServiceStub(self.channel)
print(f"Successfully connected to DexHand gRPC server: {self.address}")
print(f"Device ID: {self.device_id}")
return True
except grpc.FutureTimeoutError:
attempt += 1
print(f"Connection timed out, attempt {attempt} failed")
if attempt > self.retries:
return False
print(f"Waiting {self.timeout}s before retrying...")
time.sleep(self.timeout)
except Exception as e:
print(f"Connection error: {e}")
return False
return False
def get_sensor_data(self) -> Optional[dexhand_command_pb2.GetSensorDataCommand.Feedback]:
"""获取传感器数据"""
if not self.stub:
print("No gRPC connection available")
return None
try:
# 创建请求
timestamp = Timestamp()
timestamp.GetCurrentTime()
request = dexhand_command_pb2.GetSensorDataCommand.Request()
request.header.device_id = self.device_id
request.header.timestamp.CopyFrom(timestamp)
# 发送请求
feedback = self.stub.GetSensorData(request)
return feedback
except grpc.RpcError as e:
print(f"gRPC error: {e.code()} - {e.details()}")
return None
except Exception as e:
print(f"Error getting sensor data: {e}")
return None
def close(self):
"""关闭连接"""
if self.channel:
self.channel.close()
print(f"Closed connection to DexHand gRPC server {self.address}")
def is_connected(self):
"""检查是否已连接"""
return self.channel is not None and self.stub is not None
class SensorDataPlotWidget(QWidget):
"""传感器数据绘制窗口"""
def __init__(self, parent=None):
super().__init__(parent)
self.client = None
self.update_timer = QTimer()
self.data_buffer = {
'sum': [], # sum数据缓冲区
'peak': [] # peak数据缓冲区
}
self.time_buffer = [] # 时间戳缓冲区
self.max_points = 100 # 最大显示点数
self.current_finger = dexhand_command_pb2.SensorData.INDEX # 默认食指
self.current_part = dexhand_command_pb2.SensorData.TIP # 默认指端
self.setup_ui()
self.setup_plots()
def setup_ui(self):
"""设置UI"""
layout = QVBoxLayout()
self.setLayout(layout)
# 标题
title = QLabel("灵巧手传感器数据监控 - 食指指端")
title_font = QFont()
title_font.setPointSize(14)
title_font.setBold(True)
title.setFont(title_font)
title.setAlignment(Qt.AlignCenter)
layout.addWidget(title)
# 控制面板
control_group = QGroupBox("连接配置")
control_layout = QGridLayout()
control_group.setLayout(control_layout)
# IP地址
control_layout.addWidget(QLabel("服务器IP:"), 0, 0)
self.ip_edit = QLineEdit("192.168.0.222")
control_layout.addWidget(self.ip_edit, 0, 1)
# 端口
control_layout.addWidget(QLabel("端口:"), 1, 0)
self.port_edit = QLineEdit("50052")
control_layout.addWidget(self.port_edit, 1, 1)
# 设备ID
control_layout.addWidget(QLabel("设备ID:"), 2, 0)
self.device_id_edit = QLineEdit("dexhand_001")
control_layout.addWidget(self.device_id_edit, 2, 1)
# 完整地址显示
control_layout.addWidget(QLabel("完整地址:"), 3, 0)
self.address_label = QLabel("192.168.0.222:50052")
self.address_label.setStyleSheet("font-weight: bold;")
control_layout.addWidget(self.address_label, 3, 1)
# 连接按钮
self.connect_btn = QPushButton("连接")
self.connect_btn.clicked.connect(self.toggle_connection)
self.connect_btn.setMinimumHeight(40)
self.connect_btn.setStyleSheet("font-weight: bold;")
control_layout.addWidget(self.connect_btn, 4, 0, 1, 2)
# 连接状态
self.status_label = QLabel("未连接")
self.status_label.setAlignment(Qt.AlignCenter)
self.status_label.setStyleSheet("color: red; font-weight: bold; padding: 5px;")
control_layout.addWidget(self.status_label, 5, 0, 1, 2)
layout.addWidget(control_group)
# 数据配置面板
data_config_group = QGroupBox("数据配置")
data_config_layout = QGridLayout()
data_config_group.setLayout(data_config_layout)
# 更新频率
data_config_layout.addWidget(QLabel("更新频率(ms):"), 0, 0)
self.freq_spin = QSpinBox()
self.freq_spin.setRange(10, 1000)
self.freq_spin.setValue(50)
self.freq_spin.valueChanged.connect(self.set_update_interval)
data_config_layout.addWidget(self.freq_spin, 0, 1)
# 显示点数
data_config_layout.addWidget(QLabel("显示点数:"), 1, 0)
self.points_spin = QSpinBox()
self.points_spin.setRange(10, 500)
self.points_spin.setValue(100)
self.points_spin.valueChanged.connect(self.set_max_points)
data_config_layout.addWidget(self.points_spin, 1, 1)
# 清空数据按钮
self.clear_btn = QPushButton("清空数据")
self.clear_btn.clicked.connect(self.clear_data)
self.clear_btn.setEnabled(False)
data_config_layout.addWidget(self.clear_btn, 2, 0)
# 测试连接按钮
self.test_btn = QPushButton("测试连接")
self.test_btn.clicked.connect(self.test_connection)
data_config_layout.addWidget(self.test_btn, 2, 1)
layout.addWidget(data_config_group)
# 创建绘图区域
plots_widget = QWidget()
plots_layout = QVBoxLayout()
plots_widget.setLayout(plots_layout)
# SUM曲线图
self.sum_plot_widget = pg.PlotWidget(title="食指指端传感器数据 - SUM值")
self.sum_plot_widget.setLabel('left', 'SUM值')
self.sum_plot_widget.setLabel('bottom', '时间 (秒)')
self.sum_plot_widget.showGrid(x=True, y=True, alpha=0.3)
self.sum_plot_widget.addLegend()
self.sum_curve = self.sum_plot_widget.plot(pen='y', name='SUM值', width=2)
plots_layout.addWidget(self.sum_plot_widget)
# PEAK曲线图
self.peak_plot_widget = pg.PlotWidget(title="食指指端传感器数据 - PEAK值")
self.peak_plot_widget.setLabel('left', 'PEAK值')
self.peak_plot_widget.setLabel('bottom', '时间 (秒)')
self.peak_plot_widget.showGrid(x=True, y=True, alpha=0.3)
self.peak_plot_widget.addLegend()
self.peak_curve = self.peak_plot_widget.plot(pen='r', name='PEAK值', width=2)
plots_layout.addWidget(self.peak_plot_widget)
layout.addWidget(plots_widget)
# 实时数据展示
data_group = QGroupBox("当前数据")
data_layout = QHBoxLayout()
data_group.setLayout(data_layout)
self.sum_label = QLabel("SUM: --")
self.sum_label.setStyleSheet("font-weight: bold; color: orange; font-size: 12pt;")
data_layout.addWidget(self.sum_label)
self.peak_label = QLabel("PEAK: --")
self.peak_label.setStyleSheet("font-weight: bold; color: red; font-size: 12pt;")
data_layout.addWidget(self.peak_label)
self.timestamp_label = QLabel("时间: --")
self.timestamp_label.setStyleSheet("font-size: 10pt;")
data_layout.addWidget(self.timestamp_label)
layout.addWidget(data_group)
# 连接输入变化信号
self.ip_edit.textChanged.connect(self.update_address_label)
self.port_edit.textChanged.connect(self.update_address_label)
def update_address_label(self):
"""更新地址标签"""
ip = self.ip_edit.text().strip()
port = self.port_edit.text().strip()
if ip and port:
self.address_label.setText(f"{ip}:{port}")
else:
self.address_label.setText("--")
def setup_plots(self):
"""初始化绘图"""
self.sum_curve.setData([], [])
self.peak_curve.setData([], [])
def test_connection(self):
"""测试连接"""
ip = self.ip_edit.text().strip()
port = self.port_edit.text().strip()
device_id = self.device_id_edit.text().strip()
if not ip or not port:
QMessageBox.warning(self, "警告", "请输入IP地址和端口")
return
try:
port = int(port)
except ValueError:
QMessageBox.warning(self, "警告", "端口必须是数字")
return
# 创建临时客户端测试连接
test_client = DexHandSensorClient(ip, port, device_id, timeout=2, retries=0)
if test_client.is_connected():
QMessageBox.information(self, "成功", "连接测试成功！")
test_client.close()
else:
QMessageBox.critical(self, "失败", "连接测试失败，请检查服务器地址和网络")
def set_client(self, client: DexHandSensorClient):
"""设置客户端"""
self.client = client
self.status_label.setText(f"已连接 (设备: {client.device_id})")
self.status_label.setStyleSheet("color: green; font-weight: bold; padding: 5px;")
self.connect_btn.setText("断开连接")
self.clear_btn.setEnabled(True)
self.ip_edit.setEnabled(False)
self.port_edit.setEnabled(False)
self.device_id_edit.setEnabled(False)
self.test_btn.setEnabled(False)
def clear_client(self):
"""清除客户端"""
self.client = None
self.status_label.setText("未连接")
self.status_label.setStyleSheet("color: red; font-weight: bold; padding: 5px;")
self.connect_btn.setText("连接")
self.clear_btn.setEnabled(False)
self.ip_edit.setEnabled(True)
self.port_edit.setEnabled(True)
self.device_id_edit.setEnabled(True)
self.test_btn.setEnabled(True)
def toggle_connection(self):
"""切换连接状态"""
if self.client is None or not self.client.is_connected():
# 尝试连接
ip = self.ip_edit.text().strip()
port = self.port_edit.text().strip()
device_id = self.device_id_edit.text().strip()
if not ip or not port:
QMessageBox.warning(self, "警告", "请输入IP地址和端口")
return
try:
port = int(port)
except ValueError:
QMessageBox.warning(self, "警告", "端口必须是数字")
return
# 显示连接中状态
self.status_label.setText("连接中...")
self.status_label.setStyleSheet("color: orange; font-weight: bold; padding: 5px;")
QApplication.processEvents()
# 创建客户端
try:
client = DexHandSensorClient(ip, port, device_id)
if client.is_connected():
self.set_client(client)
self.start_updating()
else:
self.clear_client()
QMessageBox.critical(self, "错误", f"无法连接到服务器 {ip}:{port}")
except Exception as e:
self.clear_client()
QMessageBox.critical(self, "错误", f"连接失败: {str(e)}")
else:
# 断开连接
self.stop_updating()
self.client.close()
self.clear_client()
self.clear_data()
def start_updating(self):
"""开始更新数据"""
self.update_timer.timeout.connect(self.update_data)
self.update_timer.start(self.freq_spin.value())
def stop_updating(self):
"""停止更新数据"""
self.update_timer.stop()
try:
self.update_timer.timeout.disconnect()
except:
pass
def set_update_interval(self, ms: int):
"""设置更新间隔"""
if self.update_timer.isActive():
self.update_timer.setInterval(ms)
def set_max_points(self, points: int):
"""设置最大显示点数"""
self.max_points = points
# 如果当前数据超过新限制，截断数据
if len(self.time_buffer) > self.max_points:
self.time_buffer = self.time_buffer[-self.max_points:]
self.data_buffer['sum'] = self.data_buffer['sum'][-self.max_points:]
self.data_buffer['peak'] = self.data_buffer['peak'][-self.max_points:]
def clear_data(self):
"""清空数据"""
self.data_buffer['sum'].clear()
self.data_buffer['peak'].clear()
self.time_buffer.clear()
self.sum_curve.setData([], [])
self.peak_curve.setData([], [])
self.sum_label.setText("SUM: --")
self.peak_label.setText("PEAK: --")
def extract_finger_data(self, feedback):
"""提取食指指端的传感器数据"""
if not feedback or not feedback.sensor:
return None, None
sum_values = []
peak_values = []
for sensor in feedback.sensor:
# 检查是否为食指指端
if (sensor.finger_type == self.current_finger and
sensor.part_type == self.current_part):
# 处理传感器数据
if sensor.data:
for row in sensor.data:
if row.values:
row_array = np.array(row.values)
sum_values.append(np.sum(row_array))
peak_values.append(np.max(row_array))
if sum_values and peak_values:
return np.mean(sum_values), np.mean(peak_values)
return None, None
def update_data(self):
"""更新数据"""
if not self.client or not self.client.is_connected():
return
feedback = self.client.get_sensor_data()
if not feedback:
# 如果获取数据失败，可能连接已断开
if self.client and not self.client.is_connected():
self.toggle_connection() # 自动断开
return
# 提取食指指端数据
sum_val, peak_val = self.extract_finger_data(feedback)
if sum_val is not None and peak_val is not None:
# 更新时间戳
current_time = time.time()
self.time_buffer.append(current_time)
self.data_buffer['sum'].append(sum_val)
self.data_buffer['peak'].append(peak_val)
# 限制缓冲区大小
if len(self.time_buffer) > self.max_points:
self.time_buffer.pop(0)
self.data_buffer['sum'].pop(0)
self.data_buffer['peak'].pop(0)
# 更新曲线
if len(self.time_buffer) > 1:
# 归一化时间显示（相对于第一个点）
t0 = self.time_buffer[0]
plot_time = [t - t0 for t in self.time_buffer]
self.sum_curve.setData(plot_time, self.data_buffer['sum'])
self.peak_curve.setData(plot_time, self.data_buffer['peak'])
# 自动调整Y轴范围
self.sum_plot_widget.autoRange()
self.peak_plot_widget.autoRange()
# 更新标签
self.sum_label.setText(f"SUM: {sum_val:.2f}")
self.peak_label.setText(f"PEAK: {peak_val:.2f}")
self.timestamp_label.setText(f"时间: {time.strftime('%H:%M:%S')}")
class SensorDataWindow(QMainWindow):
"""传感器数据监控主窗口"""
def __init__(self):
super().__init__()
self.setWindowTitle("灵巧手传感器数据监控 - 可配置版")
self.setGeometry(100, 100, 900, 1000)
# 设置中央部件
central_widget = SensorDataPlotWidget()
self.setCentralWidget(central_widget)
def closeEvent(self, event):
"""关闭窗口时清理资源"""
central_widget = self.centralWidget()
if central_widget:
central_widget.stop_updating()
if central_widget.client:
central_widget.client.close()
event.accept()
def main():
"""主函数"""
import sys
app = QApplication(sys.argv)
app.setStyle('Fusion') # 使用Fusion风格，看起来更现代
window = SensorDataWindow()
window.show()
sys.exit(app.exec_())
if __name__ == "__main__":
main()

727
clients/gui_camera_client..py Normal file
View File

@ -0,0 +1,727 @@
# camera_player_qt.py
import sys
import cv2
import av
import numpy as np
import grpc
import threading
import time
from datetime import datetime
from pathlib import Path
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
QHBoxLayout, QLabel, QPushButton, QLineEdit,
QFileDialog, QMessageBox, QStatusBar, QGroupBox,
QGridLayout, QComboBox, QSpinBox, QCheckBox, QFrame)
from PyQt5.QtCore import Qt, QThread, pyqtSignal, QTimer
from PyQt5.QtGui import QImage, QPixmap
# 导入生成的protobuf模块
from cmvr.api import camera_service_pb2_grpc
from cmvr.api import camera_command_pb2
from cmvr.api.common_pb2 import CommandHeader
from google.protobuf.timestamp_pb2 import Timestamp
class StreamThread(QThread):
"""流接收线程"""
frame_received = pyqtSignal(object, int, int) # frame, width, height
status_update = pyqtSignal(dict)
error_occurred = pyqtSignal(str)
keyframe_received = pyqtSignal()
# 支持的编码格式
CODECS = {
'H.265 (HEVC)': 'hevc',
'H.264 (AVC)': 'h264'
}
def __init__(self, device_id, server_addr, codec='hevc', wait_for_keyframe=True):
super().__init__()
self.device_id = device_id
self.server_addr = server_addr
self.codec_name = codec
self.wait_for_keyframe = wait_for_keyframe
self.running = False
self.streaming = False
# gRPC
self.channel = None
self.stub = None
# 解码器
self.codec = None
self._init_decoder()
# 统计
self.frame_count = 0
self.fps_counter = 0
self.fps_time = time.time()
self.fps = 0
self.keyframe_count = 0
def _init_decoder(self):
"""初始化解码器"""
try:
# 根据选择的编码创建解码器
self.codec = av.CodecContext.create(self.codec_name, "r")
self.codec.thread_count = 1
# 设置低延迟模式
if hasattr(self.codec, 'flags'):
try:
self.codec.flags |= av.codec.context.AV_CODEC_FLAG_LOW_DELAY
except:
pass
print(f"✓ {self.codec_name.upper()}解码器初始化成功")
except Exception as e:
print(f"✗ 解码器初始化失败: {e}")
self.codec = None
def _create_header(self):
"""创建请求头"""
timestamp = Timestamp()
timestamp.GetCurrentTime()
return CommandHeader.Request(
device_id=self.device_id,
timestamp=timestamp
)
def connect(self):
"""连接服务器"""
try:
self.channel = grpc.insecure_channel(self.server_addr)
grpc.channel_ready_future(self.channel).result(timeout=3)
self.stub = camera_service_pb2_grpc.CameraServiceStub(self.channel)
return True
except Exception as e:
self.error_occurred.emit(f"连接失败: {e}")
return False
def start_camera(self):
"""启动摄像头"""
try:
request = camera_command_pb2.StartCameraCommand.Request(
header=self._create_header()
)
response = self.stub.StartCamera(request, timeout=3)
return response.header.success
except Exception as e:
self.error_occurred.emit(f"启动摄像头失败: {e}")
return False
def stop_camera(self):
"""停止摄像头"""
try:
request = camera_command_pb2.StopCameraCommand.Request(
header=self._create_header()
)
self.stub.StopCamera(request, timeout=3)
except:
pass
def run(self):
"""线程主函数"""
if not self.connect():
return
if not self.start_camera():
return
self.running = True
self.streaming = True
self.frame_count = 0
self.fps_counter = 0
self.fps_time = time.time()
keyframe_received = False
def request_gen():
while self.streaming:
yield camera_command_pb2.GetRGBImageStreamCommand.Request(
header=self._create_header(),
eof=False
)
time.sleep(0.001)
yield camera_command_pb2.GetRGBImageStreamCommand.Request(
header=self._create_header(),
eof=True
)
try:
responses = self.stub.GetRGBImageStream(request_gen())
for response in responses:
if not self.streaming:
break
if response.header.success:
frame_data = response.color_frame.data
is_keyframe = response.color_frame.is_key_frame
# 等待首个关键帧
if self.wait_for_keyframe and not keyframe_received and not is_keyframe:
continue
if is_keyframe and not keyframe_received:
keyframe_received = True
self.keyframe_count += 1
self.keyframe_received.emit()
# 解码
if self.codec and frame_data:
packet = av.Packet(frame_data)
frames = self.codec.decode(packet)
if frames:
# 转换为RGB (已经是RGB格式)
img = frames[0].to_ndarray(format='rgb24')
# 更新统计
self.fps_counter += 1
self.frame_count += 1
# 计算FPS
now = time.time()
if now - self.fps_time >= 1.0:
self.fps = self.fps_counter
self.fps_counter = 0
self.fps_time = now
# 发送状态更新
self.status_update.emit({
'fps': self.fps,
'frames': self.frame_count,
'keyframes': self.keyframe_count
})
# 发送帧到主线程
h, w = img.shape[:2]
self.frame_received.emit(img, w, h)
else:
print(f"流错误: {response.header.error_message}")
except Exception as e:
self.error_occurred.emit(f"流错误: {e}")
finally:
self.stop_camera()
if self.channel:
self.channel.close()
self.running = False
self.streaming = False
def stop(self):
"""停止线程"""
self.streaming = False
self.wait(2000) # 等待最多2秒
class CameraPlayerQt(QMainWindow):
"""基于Qt的摄像头拉流播放器"""
def __init__(self):
super().__init__()
self.setWindowTitle("摄像头拉流播放器 (Qt版)")
self.setMinimumSize(1000, 800)
# 状态变量
self.connected = False
self.streaming = False
self.recording = False
self.stream_thread = None
self.video_writer = None
# 录制目录
self.record_path = Path("./recordings")
self.record_path.mkdir(exist_ok=True)
# 创建UI
self._create_ui()
# 状态栏
self.status_bar = QStatusBar()
self.setStatusBar(self.status_bar)
self.status_bar.showMessage("就绪")
# 定时器用于更新显示
self.display_timer = QTimer()
self.display_timer.timeout.connect(self._update_display)
self.display_timer.start(30) # 约33fps
# 当前帧缓存
self.current_frame = None
self.frame_width = 0
self.frame_height = 0
def _create_ui(self):
"""创建UI"""
central_widget = QWidget()
self.setCentralWidget(central_widget)
main_layout = QVBoxLayout(central_widget)
# ========== 连接设置区域 ==========
conn_group = QGroupBox("连接设置")
conn_layout = QGridLayout()
# 第一行：设备ID和编码格式
conn_layout.addWidget(QLabel("设备ID:"), 0, 0)
self.device_id = QLineEdit("right_hand_cam")
conn_layout.addWidget(self.device_id, 0, 1)
conn_layout.addWidget(QLabel("编码格式:"), 0, 2)
self.codec_combo = QComboBox()
self.codec_combo.addItems(['H.265 (HEVC)', 'H.264 (AVC)'])
self.codec_combo.setCurrentIndex(0)
self.codec_combo.setMinimumWidth(120)
conn_layout.addWidget(self.codec_combo, 0, 3)
# 等待关键帧选项
self.wait_keyframe = QCheckBox("等待关键帧")
self.wait_keyframe.setChecked(True)
conn_layout.addWidget(self.wait_keyframe, 0, 4)
# 第二行：服务器地址
conn_layout.addWidget(QLabel("服务器IP:"), 1, 0)
self.server_ip = QLineEdit("192.168.0.222")
conn_layout.addWidget(self.server_ip, 1, 1)
conn_layout.addWidget(QLabel("端口:"), 1, 2)
self.server_port = QLineEdit("50052")
self.server_port.setMaximumWidth(80)
conn_layout.addWidget(self.server_port, 1, 3)
# 超时设置
conn_layout.addWidget(QLabel("超时(秒):"), 1, 4)
self.timeout_spin = QSpinBox()
self.timeout_spin.setRange(1, 10)
self.timeout_spin.setValue(3)
self.timeout_spin.setMaximumWidth(60)
conn_layout.addWidget(self.timeout_spin, 1, 5)
# 第三行：录制目录
conn_layout.addWidget(QLabel("录制目录:"), 2, 0)
self.record_dir = QLineEdit(str(self.record_path))
conn_layout.addWidget(self.record_dir, 2, 1, 1, 4)
self.browse_btn = QPushButton("浏览")
self.browse_btn.clicked.connect(self._browse_dir)
conn_layout.addWidget(self.browse_btn, 2, 5)
# 第四行：控制按钮
btn_layout = QHBoxLayout()
self.connect_btn = QPushButton("连接")
self.connect_btn.clicked.connect(self._connect)
self.connect_btn.setMinimumWidth(80)
btn_layout.addWidget(self.connect_btn)
self.preview_btn = QPushButton("预览")
self.preview_btn.clicked.connect(self._toggle_preview)
self.preview_btn.setEnabled(False)
self.preview_btn.setMinimumWidth(80)
btn_layout.addWidget(self.preview_btn)
self.record_btn = QPushButton("录制")
self.record_btn.clicked.connect(self._toggle_recording)
self.record_btn.setEnabled(False)
self.record_btn.setMinimumWidth(80)
btn_layout.addWidget(self.record_btn)
self.disconnect_btn = QPushButton("断开")
self.disconnect_btn.clicked.connect(self._disconnect)
self.disconnect_btn.setEnabled(False)
self.disconnect_btn.setMinimumWidth(80)
btn_layout.addWidget(self.disconnect_btn)
btn_layout.addStretch()
conn_layout.addLayout(btn_layout, 3, 0, 1, 6)
conn_group.setLayout(conn_layout)
main_layout.addWidget(conn_group)
# ========== 状态显示区域 ==========
status_group = QGroupBox("状态信息")
status_layout = QHBoxLayout()
self.conn_status = QLabel("● 未连接")
self.conn_status.setStyleSheet("color: red;")
status_layout.addWidget(self.conn_status)
status_layout.addWidget(QLabel("|"))
self.codec_status = QLabel("编码: H.265")
status_layout.addWidget(self.codec_status)
status_layout.addWidget(QLabel("|"))
self.fps_status = QLabel("FPS: 0")
status_layout.addWidget(self.fps_status)
status_layout.addWidget(QLabel("|"))
self.frame_status = QLabel("帧数: 0")
status_layout.addWidget(self.frame_status)
status_layout.addWidget(QLabel("|"))
self.resolution_status = QLabel("分辨率: -")
status_layout.addWidget(self.resolution_status)
status_layout.addWidget(QLabel("|"))
self.keyframe_status = QLabel("关键帧: 等待")
self.keyframe_status.setStyleSheet("color: orange;")
status_layout.addWidget(self.keyframe_status)
status_layout.addWidget(QLabel("|"))
self.record_status = QLabel("录制: 未开始")
self.record_status.setStyleSheet("color: gray;")
status_layout.addWidget(self.record_status)
status_layout.addStretch()
status_group.setLayout(status_layout)
main_layout.addWidget(status_group)
# ========== 视频预览区域 ==========
video_group = QGroupBox("视频预览")
video_layout = QVBoxLayout()
self.video_label = QLabel()
self.video_label.setAlignment(Qt.AlignCenter)
self.video_label.setMinimumSize(640, 480)
self.video_label.setStyleSheet("background-color: black; border: 1px solid gray;")
video_layout.addWidget(self.video_label)
video_group.setLayout(video_layout)
main_layout.addWidget(video_group, 1) # 1表示拉伸因子
# ========== 操作说明 ==========
info_frame = QFrame()
info_frame.setFrameStyle(QFrame.StyledPanel | QFrame.Sunken)
info_layout = QHBoxLayout(info_frame)
info_label = QLabel("操作说明: ①选择编码格式 ②连接 ③预览 ④录制 (按'停止预览'结束)")
info_label.setStyleSheet("color: gray; padding: 5px;")
info_layout.addWidget(info_label)
main_layout.addWidget(info_frame)
def _browse_dir(self):
"""浏览目录"""
directory = QFileDialog.getExistingDirectory(self, "选择录制目录",
self.record_dir.text())
if directory:
self.record_dir.setText(directory)
self.record_path = Path(directory)
self.record_path.mkdir(exist_ok=True)
def _get_server_addr(self):
"""获取服务器地址"""
return f"{self.server_ip.text()}:{self.server_port.text()}"
def _get_codec(self):
"""获取编码格式"""
codec_map = {
'H.265 (HEVC)': 'hevc',
'H.264 (AVC)': 'h264'
}
return codec_map[self.codec_combo.currentText()]
def _connect(self):
"""连接服务器"""
codec = self._get_codec()
self.status_bar.showMessage(f"正在连接服务器 (编码: {codec.upper()})...")
# 更新编码状态显示
self.codec_status.setText(f"编码: {self.codec_combo.currentText()}")
# 创建流线程
self.stream_thread = StreamThread(
self.device_id.text(),
self._get_server_addr(),
codec=codec,
wait_for_keyframe=self.wait_keyframe.isChecked()
)
# 连接信号
self.stream_thread.frame_received.connect(self._on_frame_received)
self.stream_thread.status_update.connect(self._on_status_update)
self.stream_thread.error_occurred.connect(self._on_error)
self.stream_thread.keyframe_received.connect(self._on_keyframe_received)
# 连接测试
if self.stream_thread.connect():
self.connected = True
self.conn_status.setText("● 已连接")
self.conn_status.setStyleSheet("color: green;")
self.connect_btn.setEnabled(False)
self.preview_btn.setEnabled(True)
self.disconnect_btn.setEnabled(True)
self.codec_combo.setEnabled(False)
self.wait_keyframe.setEnabled(False)
self.status_bar.showMessage("服务器连接成功", 3000)
else:
self.stream_thread = None
def _disconnect(self):
"""断开连接"""
# 停止预览
if self.streaming:
self._stop_preview()
# 停止录制
if self.recording:
self._stop_recording()
# 清理线程
if self.stream_thread:
self.stream_thread.stop()
self.stream_thread = None
self.connected = False
self.conn_status.setText("● 未连接")
self.conn_status.setStyleSheet("color: red;")
self.connect_btn.setEnabled(True)
self.preview_btn.setEnabled(False)
self.preview_btn.setText("预览")
self.record_btn.setEnabled(False)
self.record_btn.setText("录制")
self.disconnect_btn.setEnabled(False)
self.codec_combo.setEnabled(True)
self.wait_keyframe.setEnabled(True)
# 清空画面
self.video_label.clear()
self.current_frame = None
# 重置状态
self.keyframe_status.setText("关键帧: 等待")
self.keyframe_status.setStyleSheet("color: orange;")
self.fps_status.setText("FPS: 0")
self.frame_status.setText("帧数: 0")
self.resolution_status.setText("分辨率: -")
self.status_bar.showMessage("已断开连接", 3000)
def _toggle_preview(self):
"""切换预览"""
if not self.streaming:
self._start_preview()
else:
self._stop_preview()
def _start_preview(self):
"""开始预览"""
if not self.stream_thread:
return
self.streaming = True
self.stream_thread.streaming = True
self.stream_thread.start()
self.preview_btn.setText("停止")
self.record_btn.setEnabled(True)
if self.wait_keyframe.isChecked():
self.keyframe_status.setText("关键帧: 等待中...")
self.keyframe_status.setStyleSheet("color: orange;")
else:
self.keyframe_status.setText("关键帧: 不等待")
self.keyframe_status.setStyleSheet("color: gray;")
self.status_bar.showMessage("预览已启动...")
def _stop_preview(self):
"""停止预览"""
if self.stream_thread:
self.stream_thread.streaming = False
self.stream_thread.wait(2000)
self.streaming = False
self.preview_btn.setText("预览")
self.record_btn.setEnabled(False)
# 停止录制
if self.recording:
self._stop_recording()
self.keyframe_status.setText("关键帧: 等待")
self.keyframe_status.setStyleSheet("color: orange;")
# 清空画面
self.video_label.clear()
self.current_frame = None
self.status_bar.showMessage("预览已停止", 3000)
def _toggle_recording(self):
"""切换录制"""
if not self.recording:
self._start_recording()
else:
self._stop_recording()
def _start_recording(self):
"""开始录制"""
if not self.streaming:
QMessageBox.warning(self, "警告", "请先启动预览")
return
# 更新录制目录
self.record_path = Path(self.record_dir.text())
self.record_path.mkdir(parents=True, exist_ok=True)
# 生成文件名
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
codec_name = self.codec_combo.currentText().replace(' ', '_')
filename = self.record_path / f"rec_{self.device_id.text()}_{codec_name}_{timestamp}.mp4"
# 获取画面尺寸
if self.current_frame is not None:
height, width = self.current_frame.shape[:2]
else:
width, height = 640, 480
# 创建VideoWriter
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
self.video_writer = cv2.VideoWriter(str(filename), fourcc, 30.0, (width, height))
if self.video_writer.isOpened():
self.recording = True
self.record_btn.setText("停止")
self.record_status.setText("● 录制中")
self.record_status.setStyleSheet("color: red;")
self.status_bar.showMessage(f"开始录制: {filename.name}", 3000)
else:
QMessageBox.critical(self, "错误", "无法创建视频文件")
def _stop_recording(self):
"""停止录制"""
if self.video_writer:
self.video_writer.release()
self.video_writer = None
self.recording = False
self.record_btn.setText("录制")
self.record_status.setText("录制: 未开始")
self.record_status.setStyleSheet("color: gray;")
self.status_bar.showMessage("录制已停止", 3000)
def _on_frame_received(self, frame, width, height):
"""收到新帧"""
self.current_frame = frame
self.frame_width = width
self.frame_height = height
# 录制
if self.recording and self.video_writer:
# 转换为BGR用于录制
frame_bgr = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
self.video_writer.write(frame_bgr)
def _update_display(self):
"""更新显示"""
if self.current_frame is not None:
# 直接使用QImage，避免转换
bytes_per_line = 3 * self.frame_width
qimage = QImage(self.current_frame.data, self.frame_width,
self.frame_height, bytes_per_line, QImage.Format_RGB888)
# 缩放以适应标签大小
label_size = self.video_label.size()
pixmap = QPixmap.fromImage(qimage).scaled(
label_size, Qt.KeepAspectRatio, Qt.SmoothTransformation)
self.video_label.setPixmap(pixmap)
# 更新分辨率显示
self.resolution_status.setText(f"分辨率: {self.frame_width}x{self.frame_height}")
def _on_status_update(self, stats):
"""状态更新"""
self.fps_status.setText(f"FPS: {stats['fps']}")
self.frame_status.setText(f"帧数: {stats['frames']}")
def _on_keyframe_received(self):
"""收到关键帧"""
if self.wait_keyframe.isChecked():
self.keyframe_status.setText("关键帧: 已接收")
self.keyframe_status.setStyleSheet("color: green;")
def _on_error(self, msg):
"""错误处理"""
QMessageBox.critical(self, "错误", msg)
self.status_bar.showMessage(f"错误: {msg}")
def closeEvent(self, event):
"""窗口关闭事件"""
# 停止预览
if self.streaming:
self._stop_preview()
# 停止录制
if self.recording:
self._stop_recording()
# 断开连接
if self.connected:
self._disconnect()
event.accept()
def main():
app = QApplication(sys.argv)
# 设置应用样式
app.setStyle('Fusion')
# 设置全局样式表
app.setStyleSheet("""
QGroupBox {
font-weight: bold;
border: 1px solid gray;
border-radius: 5px;
margin-top: 10px;
padding-top: 10px;
}
QGroupBox::title {
subcontrol-origin: margin;
left: 10px;
padding: 0 5px 0 5px;
}
QPushButton {
padding: 5px 10px;
}
QLineEdit {
padding: 3px;
}
QComboBox {
padding: 3px;
}
""")
window = CameraPlayerQt()
window.show()
print("\n" + "=" * 60)
print("摄像头拉流播放器 (Qt版) 启动")
print("=" * 60)
print("支持编码:")
print(" • H.265 (HEVC) - 高压缩率")
print(" • H.264 (AVC) - 兼容性好")
print("\n操作步骤:")
print("1. 选择编码格式")
print("2. 输入设备ID、服务器IP和端口")
print("3. 点击【连接】按钮")
print("4. 点击【预览】按钮开始观看")
print("5. 点击【录制】按钮开始/停止录制")
print("=" * 60 + "\n")
sys.exit(app.exec_())
if __name__ == "__main__":
main()

2
requirements.txt
View File

@ -8,6 +8,8 @@ PyQt5==5.15.11
requests==2.32.5
yourdfpy
uvicorn==0.38.0
av>=10.0.0
opencv-python>=4.0.0
numpy==1.26.4
opencv-contrib-python==4.10.0.84