Skip to content

O
Oil_Level_Recognition_System
Commit f8cfcff7 by Yuhaibo
Options

1

parent 2ac40d5e
Showing with 1175 additions and 146 deletions
__main__.py
......@@ -24,8 +24,6 @@ os.environ['ULTRALYTICS_CONFIG_DIR'] = os.path.join(current_dir, '.cache', 'ultr
# 修复 OpenMP 运行时冲突问题
os.environ['KMP_DUPLICATE_LIB_OK'] = 'TRUE' # 允许多个OpenMP库共存（临时解决方案）
print("[环境变量] ultralytics离线模式已启用")
print("[环境变量] OpenMP冲突已修复")
from qtpy import QtWidgets
......
app.py
......@@ -620,11 +620,10 @@ class MainWindow(
if os.path.exists(icon_path):
icon = QtGui.QIcon(icon_path)
self.setWindowIcon(icon)
print(f"[主窗口] 窗口图标已设置: {icon_path}")
else:
print(f"[主窗口] 图标文件不存在: {icon_path}")
pass
except Exception as e:
print(f"[主窗口] 设置窗口图标失败: {e}")
pass
def _loadDefaultConfig(self):
"""从 default_config.yaml 加载配置"""
......@@ -764,7 +763,6 @@ class MainWindow(
# 🔥 为每个通道面板的任务标签设置变量名（channel1mission, channel2mission, channel3mission, channel4mission）
mission_var_name = f'channel{i+1}mission'
setattr(self, mission_var_name, channelPanel.taskLabel)
print(f"[MainWindow] 已设置任务标签变量: {mission_var_name}")
if hasattr(self, '_connectChannelPanelSignals'):
self._connectChannelPanelSignals(channelPanel)
......@@ -795,7 +793,6 @@ class MainWindow(
history_panel.setObjectName(f"HistoryVideoPanel_{i+1}")
self.historyVideoPanels.append(history_panel)
print(f"[MainWindow] 已创建 {len(self.historyVideoPanels)} 个历史视频面板")
# 通过handler初始化通道面板数据
if hasattr(self, 'initializeChannelPanels'):
......@@ -822,7 +819,6 @@ class MainWindow(
# 🔥 设置曲线面板的任务选择下拉框变量名（curvemission）
self.curvemission = self.curvePanel.curvemission
print(f"[MainWindow] 已设置曲线任务变量: curvemission")
# 连接任务选择变化信号
self.curvemission.currentTextChanged.connect(self._onCurveMissionChanged)
......@@ -854,7 +850,6 @@ class MainWindow(
self.videoLayoutStack.addWidget(layout_widget)
print(f"[MainWindow] 曲线模式布局已创建：左侧子布局栈（实时/历史） + 右侧共用CurvePanel")
def _createRealtimeCurveSubLayout(self):
"""创建实时检测曲线子布局（索引0）- 左侧通道列表"""
......@@ -898,7 +893,6 @@ class MainWindow(
sublayout.addWidget(self.curve_scroll_area)
self.curveLayoutStack.addWidget(sublayout_widget)
print(f"[MainWindow] 实时检测曲线子布局已创建（索引0）- 基于CSV文件的动态通道系统")
def _createHistoryCurveSubLayout(self):
"""创建历史回放曲线子布局（索引1）- 使用历史视频面板容器"""
......@@ -942,7 +936,6 @@ class MainWindow(
sublayout.addWidget(self.history_scroll_area)
self.curveLayoutStack.addWidget(sublayout_widget)
print(f"[MainWindow] 历史回放曲线子布局已创建（索引1）- 历史视频面板容器系统")
def _onChannelCurveClicked(self, task_name):
"""
......@@ -951,15 +944,9 @@ class MainWindow(
Args:
task_name: 通道面板的任务名称
"""
print(f"🔄 [主窗口] 通道面板查看曲线按钮被点击，任务名称: {task_name}")
# 设置 curvemission 的值
if hasattr(self, 'curvePanel') and self.curvePanel:
success = self.curvePanel.setMissionFromTaskName(task_name)
if success:
print(f"✅ [主窗口] 已设置 curvemission 为: {task_name}")
else:
print(f"⚠️ [主窗口] 设置 curvemission 失败: {task_name}")
# 切换到曲线模式
self.toggleVideoPageMode()
......@@ -1260,7 +1247,6 @@ class MainWindow(
# ========== 通道管理按钮信号 ==========
# 🔥 已改为内嵌显示，由 MissionPanelHandler 处理，不再使用弹窗
# self.missionTable.channelManageClicked.connect(self.onChannelManage) # 旧的弹窗方式
print("[App] 通道管理已改为内嵌显示，不再使用弹窗")
# ========== 通道面板信号（为所有面板连接） ==========
# 注意：channelConnected, channelDisconnected, channelEdited, amplifyClicked, channelNameChanged
......@@ -1509,15 +1495,12 @@ class MainWindow(
def closeEvent(self, event):
"""窗口关闭事件"""
try:
print("[应用] 正在关闭应用...")
# 清理全局检测线程
if hasattr(self, 'view_handler') and self.view_handler:
video_handler = getattr(self.view_handler, 'video_handler', None)
if video_handler:
thread_manager = getattr(video_handler, 'thread_manager', None)
if thread_manager:
print("[应用] 清理全局检测线程...")
thread_manager.cleanup_global_detection_thread()
# 保存窗口状态
......@@ -1528,10 +1511,7 @@ class MainWindow(
# 保存当前页面索引
self.settings.setValue("window/last_page", self.getCurrentPageIndex())
print("[应用] 应用关闭清理完成")
except Exception as e:
print(f"[应用] 关闭清理失败: {e}")
import traceback
traceback.print_exc()
......
handlers/modelpage/model_set_handler.py
......@@ -519,18 +519,11 @@ class ModelSetHandler:
current_dir = Path(__file__).parent.parent.parent
model_dir = current_dir / "database" / "model" / "detection_model"
print(f"[模型扫描] 扫描目录: {model_dir}")
if not model_dir.exists():
print(f"[模型扫描] 目录不存在")
return models
print(f"[模型扫描] 目录存在: {model_dir.exists()}")
# 扫描所有子目录（数字和非数字）
all_subdirs = [d for d in model_dir.iterdir() if d.is_dir()]
print(f"[模型扫描] 找到子目录数量: {len(all_subdirs)}")
print(f"[模型扫描] 子目录列表: {[d.name for d in all_subdirs]}")
# 分离数字目录和非数字目录
digit_subdirs = [d for d in all_subdirs if d.name.isdigit()]
......@@ -542,10 +535,8 @@ class ModelSetHandler:
# 合并：数字目录在前，非数字目录在后
sorted_subdirs = sorted_digit_subdirs + sorted_non_digit_subdirs
print(f"[模型扫描] 排序后的子目录: {[d.name for d in sorted_subdirs]}")
for subdir in sorted_subdirs:
print(f"[模型扫描] 处理子目录: {subdir.name}")
# 检查是否有weights子目录（优先检查train/weights，然后weights）
train_weights_dir = subdir / "train" / "weights"
......@@ -553,15 +544,10 @@ class ModelSetHandler:
if train_weights_dir.exists():
search_dir = train_weights_dir
print(f"[模型扫描] 找到train/weights目录: {search_dir}")
elif weights_dir.exists():
search_dir = weights_dir
print(f"[模型扫描] 找到weights目录: {search_dir}")
else:
search_dir = subdir
print(f"[模型扫描] 使用根目录: {search_dir}")
print(f"[模型扫描] 搜索目录: {search_dir}")
# 按优先级查找模型文件：best > last > epoch1
# 支持的扩展名：.dat, .pt, .template_*, 无扩展名
......@@ -576,7 +562,6 @@ class ModelSetHandler:
# 检查文件名是否匹配模式
if file.name.startswith('best.') and not file.name.endswith('.pt'):
selected_model = file
print(f"[模型扫描] 找到best模型: {file.name}")
break
# 优先级2: last模型（如果没有best）
......@@ -584,7 +569,6 @@ class ModelSetHandler:
for file in search_dir.iterdir():
if file.is_file() and file.name.startswith('last.') and not file.name.endswith('.pt'):
selected_model = file
print(f"[模型扫描] 找到last模型: {file.name}")
break
# 优先级3: epoch1模型（如果没有best和last）
......@@ -592,7 +576,6 @@ class ModelSetHandler:
for file in search_dir.iterdir():
if file.is_file() and file.name.startswith('epoch1.') and not file.name.endswith('.pt'):
selected_model = file
print(f"[模型扫描] 找到epoch1模型: {file.name}")
break
# 如果都没找到，尝试查找任何非.pt文件
......@@ -600,7 +583,6 @@ class ModelSetHandler:
for file in search_dir.iterdir():
if file.is_file() and not file.name.endswith('.pt') and not file.name.endswith('.txt') and not file.name.endswith('.yaml'):
selected_model = file
print(f"[模型扫描] 找到其他模型: {file.name}")
break
# 如果找到了模型文件，添加到列表
......@@ -632,16 +614,11 @@ class ModelSetHandler:
'file_name': selected_model.name
}
models.append(model_info)
print(f"[模型扫描] 添加模型: {model_name} ({selected_model.name})")
else:
print(f"[模型扫描] 子目录 {subdir.name} 中未找到有效模型")
except Exception as e:
import traceback
traceback.print_exc()
print(f"[模型扫描] 扫描异常: {e}")
print(f"[模型扫描] 总共找到 {len(models)} 个模型")
return models
def _mergeModelInfo(self, channel_models, scanned_models):
......
handlers/videopage/AUTO_DOT_README.md 0 → 100644
# 自动标点功能模块使用说明
## 功能概述
`auto_dot.py` 模块实现了基于YOLO分割掩码的自动标点功能，可以自动检测容器的顶部和底部位置，替代人工手动标点。
## 核心特性
- **输入**: 图片 + 检测框
- **输出**: 点位置信息 + 标注后的图片
- **检测方法**:
1. **liquid底部 + air顶部** (最可靠)
2. **liquid底部 + liquid顶部** (次选)
3. **air底部 + air顶部** (备选)
## 独立调试
### 1. 准备测试数据
将测试图片放置到:
```
D:\restructure\liquid_level_line_detection_system\test_data\test_image.jpg
```
### 2. 配置检测框
编辑 `auto_dot.py` 中的 `test_auto_dot()` 函数，修改 `boxes` 参数:
```python
# 方式1: [x1, y1, x2, y2] 格式
boxes = [
[100, 200, 300, 600], # 第一个容器
[400, 200, 600, 600], # 第二个容器
]
# 方式2: [cx, cy, size] 格式
boxes = [
[200, 400, 400], # 中心点(200, 400), 尺寸400
]
```
### 3. 运行测试
```bash
cd D:\restructure\liquid_level_line_detection_system\handlers\videopage
python auto_dot.py
```
### 4. 查看结果
- **控制台输出**: 详细的检测过程和结果
- **标注图片**: `D:\restructure\liquid_level_line_detection_system\test_output\auto_dot_result.jpg`
## API 使用示例
```python
from handlers.videopage.auto_dot import AutoDotDetector
import cv2
# 1. 创建检测器
detector = AutoDotDetector(
model_path="path/to/model.dat",
device='cuda' # 或 'cpu'
)
# 2. 加载图片
image = cv2.imread("test_image.jpg")
# 3. 定义检测框
boxes = [
[100, 200, 300, 600], # [x1, y1, x2, y2]
]
# 4. 执行检测
result = detector.detect_container_boundaries(
image=image,
boxes=boxes,
conf_threshold=0.5
)
# 5. 获取结果
if result['success']:
for container in result['containers']:
print(f"容器 {container['index']}:")
print(f" 顶部: ({container['top_x']}, {container['top']})")
print(f" 底部: ({container['bottom_x']}, {container['bottom']})")
print(f" 高度: {container['height']}px")
print(f" 置信度: {container['confidence']:.3f}")
# 保存标注图片
cv2.imwrite("result.jpg", result['annotated_image'])
```
## 输出数据结构
```python
{
'success': bool, # 检测是否成功
'containers': [
{
'index': int, # 容器索引
'top': int, # 顶部y坐标
'bottom': int, # 底部y坐标
'top_x': int, # 顶部x坐标
'bottom_x': int, # 底部x坐标
'height': int, # 容器高度(像素)
'confidence': float, # 检测置信度
'method': str # 检测方法
},
...
],
'annotated_image': np.ndarray # 标注后的图片
}
```
## 检测方法说明
### 方法1: liquid_air (最可靠)
- **容器底部**: liquid掩码的最低点
- **容器顶部**: air掩码的最高点
- **适用场景**: 同时检测到液体和空气
### 方法2: liquid_only (次选)
- **容器底部**: liquid掩码的最低点
- **容器顶部**: liquid掩码的最高点
- **适用场景**: 只检测到液体，未检测到空气
### 方法3: air_only (备选)
- **容器底部**: air掩码的最低点
- **容器顶部**: air掩码的最高点
- **适用场景**: 只检测到空气，未检测到液体
## 可视化标注
标注图片包含:
- **绿色圆点**: 容器顶部
- **红色圆点**: 容器底部
- **青色连线**: 容器高度
- **水平参考线**: 顶部和底部的水平位置
- **文字标注**: Top-N, Bottom-N, 高度值
## 注意事项
1. **模型路径**: 确保模型文件存在且可访问
2. **检测框位置**: 检测框应覆盖完整的容器区域
3. **置信度阈值**: 默认0.5，可根据实际情况调整
4. **GPU加速**: 建议使用CUDA加速，提高检测速度
## 调试技巧
1. **查看控制台输出**: 详细的检测过程日志
2. **检查标注图片**: 验证检测结果的准确性
3. **调整检测框**: 如果检测失败，尝试调整检测框的位置和大小
4. **降低置信度**: 如果检测不到掩码，尝试降低 `conf_threshold`
## 接入系统
调试成功后，可以在主系统中调用:
```python
from handlers.videopage.auto_dot import AutoDotDetector
# 在标注页面添加"自动标点"按钮
# 点击后调用 detector.detect_container_boundaries()
# 将返回的 top/bottom 坐标填充到标注点位置
```
handlers/videopage/auto_dot.py 0 → 100644
# -*- coding: utf-8 -*-
"""
自动标点功能模块
通过YOLO分割掩码自动检测容器的顶部和底部位置
"""
import cv2
import numpy as np
from pathlib import Path
from typing import Dict, List, Tuple, Optional
import sys
# 添加项目根目录到路径
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
from database.config import get_temp_models_dir
class AutoDotDetector:
"""
自动标点检测器
功能：
1. 输入图片和检测框
2. 使用YOLO分割模型自动检测容器顶部和底部
3. 输出点位置信息和标注后的图片
"""
def __init__(self, model_path: str = None, device: str = 'cuda'):
"""
初始化自动标点检测器
Args:
model_path: YOLO模型路径（.pt 或 .dat 文件）
device: 计算设备 ('cuda' 或 'cpu')
"""
self.model = None
self.model_path = model_path
self.device = self._validate_device(device)
if model_path:
self.load_model(model_path)
def _validate_device(self, device: str) -> str:
"""验证并选择可用的设备"""
try:
import torch
if device in ['cuda', '0'] or device.startswith('cuda:'):
if torch.cuda.is_available():
return 'cuda' if device in ['cuda', '0'] else device
else:
print("⚠️ CUDA不可用，切换到CPU")
return 'cpu'
return device
except Exception:
return 'cpu'
def load_model(self, model_path: str) -> bool:
"""
加载YOLO模型
Args:
model_path: 模型文件路径
Returns:
bool: 加载是否成功
"""
try:
import os
if not os.path.exists(model_path):
print(f"❌ 模型文件不存在: {model_path}")
return False
# 如果是 .dat 文件，先解码
if model_path.endswith('.dat'):
decoded_path = self._decode_dat_model(model_path)
if decoded_path is None:
print(f"❌ .dat 文件解码失败: {model_path}")
return False
model_path = decoded_path
# 设置离线模式
os.environ['YOLO_VERBOSE'] = 'False'
os.environ['YOLO_OFFLINE'] = '1'
os.environ['ULTRALYTICS_OFFLINE'] = 'True'
from ultralytics import YOLO
print(f"🔄 正在加载模型: {model_path}")
self.model = YOLO(model_path)
self.model.to(self.device)
self.model_path = model_path
print(f"✅ 模型加载成功: {os.path.basename(model_path)}")
return True
except Exception as e:
print(f"❌ 模型加载失败: {e}")
return False
def _decode_dat_model(self, dat_path: str) -> Optional[str]:
"""解码 .dat 格式的模型文件"""
try:
import struct
import hashlib
SIGNATURE = b'LDS_MODEL_FILE'
VERSION = 1
ENCRYPTION_KEY = "liquid_detection_system_2024"
key_hash = hashlib.sha256(ENCRYPTION_KEY.encode('utf-8')).digest()
with open(dat_path, 'rb') as f:
signature = f.read(len(SIGNATURE))
if signature != SIGNATURE:
return None
version = struct.unpack('<I', f.read(4))[0]
if version != VERSION:
return None
filename_len = struct.unpack('<I', f.read(4))[0]
original_filename = f.read(filename_len).decode('utf-8')
data_len = struct.unpack('<Q', f.read(8))[0]
encrypted_data = f.read(data_len)
# XOR 解密
decrypted_data = bytearray()
key_len = len(key_hash)
for i, byte in enumerate(encrypted_data):
decrypted_data.append(byte ^ key_hash[i % key_len])
decrypted_data = bytes(decrypted_data)
# 保存到临时目录
temp_dir = Path(get_temp_models_dir())
temp_dir.mkdir(exist_ok=True)
path_hash = hashlib.md5(str(dat_path).encode()).hexdigest()[:8]
temp_model_path = temp_dir / f"temp_{Path(dat_path).stem}_{path_hash}.pt"
with open(temp_model_path, 'wb') as f:
f.write(decrypted_data)
return str(temp_model_path)
except Exception as e:
print(f"❌ 解码.dat文件失败: {e}")
return None
def detect_container_boundaries(
self,
image: np.ndarray,
conf_threshold: float = 0.5
) -> Dict:
"""
自动检测容器的顶部和底部边界
Args:
image: 输入图片 (numpy.ndarray, BGR格式) - 已经裁剪好的容器图像
conf_threshold: 置信度阈值
Returns:
dict: {
'success': bool,
'top': int, # 容器顶部y坐标
'bottom': int, # 容器底部y坐标
'top_x': int, # 顶部点x坐标
'bottom_x': int, # 底部点x坐标
'height': int, # 容器高度
'confidence': float, # 检测置信度
'method': str, # 检测方法
'annotated_image': np.ndarray # 标注后的图片
}
"""
if self.model is None:
return {
'success': False,
'annotated_image': image.copy(),
'error': '模型未加载'
}
if image.size == 0:
return {
'success': False,
'annotated_image': image.copy(),
'error': '图像为空'
}
print(f"\n{'='*60}")
print(f"🔍 开始检测容器边界")
print(f" 图像尺寸: {image.shape[1]}x{image.shape[0]}")
# 执行YOLO推理（直接使用输入图像，不再裁剪）
print(f" 🔄 执行YOLO推理...")
try:
mission_results = self.model.predict(
source=image,
imgsz=640,
conf=conf_threshold,
iou=0.5,
device=self.device,
save=False,
verbose=False,
half=True if self.device != 'cpu' else False,
stream=False
)
mission_result = mission_results[0]
except Exception as e:
print(f" ❌ YOLO推理失败: {e}")
return {
'success': False,
'annotated_image': image.copy(),
'error': f'YOLO推理失败: {e}'
}
# 处理检测结果
if mission_result.masks is None:
print(f" ⚠️ 未检测到任何掩码")
return {
'success': False,
'annotated_image': image.copy(),
'error': '未检测到任何掩码'
}
masks = mission_result.masks.data.cpu().numpy() > 0.5
classes = mission_result.boxes.cls.cpu().numpy().astype(int)
confidences = mission_result.boxes.conf.cpu().numpy()
print(f" ✅ 检测到 {len(masks)} 个对象")
# 收集liquid、air和foam的y坐标
liquid_y_coords = []
liquid_x_coords = []
air_y_coords = []
air_x_coords = []
foam_masks = []
max_liquid_conf = 0.0
max_air_conf = 0.0
max_foam_conf = 0.0
for i in range(len(masks)):
if confidences[i] < conf_threshold:
continue
class_name = self.model.names[classes[i]]
conf = confidences[i]
# 调整mask尺寸到输入图像大小
resized_mask = cv2.resize(
masks[i].astype(np.uint8),
(image.shape[1], image.shape[0])
) > 0.5
y_coords, x_coords = np.where(resized_mask)
print(f" - {class_name}: {len(y_coords)} 像素, 置信度: {conf:.3f}")
if class_name == 'liquid' and len(y_coords) > 0:
liquid_y_coords.extend(y_coords)
liquid_x_coords.extend(x_coords)
max_liquid_conf = max(max_liquid_conf, conf)
elif class_name == 'air' and len(y_coords) > 0:
air_y_coords.extend(y_coords)
air_x_coords.extend(x_coords)
max_air_conf = max(max_air_conf, conf)
elif class_name == 'foam' and len(y_coords) > 0:
foam_masks.append((resized_mask, y_coords, x_coords))
max_foam_conf = max(max_foam_conf, conf)
# 计算容器边界（坐标已经是图像坐标，不需要偏移）
container_info = self._calculate_boundaries(
liquid_y_coords, liquid_x_coords,
air_y_coords, air_x_coords,
foam_masks,
max_liquid_conf, max_air_conf, max_foam_conf,
0, 0, 0 # 不需要坐标偏移
)
if not container_info:
return {
'success': False,
'annotated_image': image.copy(),
'error': '无法计算容器边界'
}
# 在图片上绘制标注
annotated_image = image.copy()
self._draw_annotations(
annotated_image,
container_info,
0, image.shape[1] # left=0, right=图像宽度
)
print(f"\n{'='*60}")
print(f"✅ 自动标点完成")
return {
'success': True,
'top': container_info['top'],
'bottom': container_info['bottom'],
'top_x': container_info['top_x'],
'bottom_x': container_info['bottom_x'],
'height': container_info['height'],
'confidence': container_info['confidence'],
'method': container_info['method'],
'annotated_image': annotated_image
}
def _parse_targets(self, boxes: List) -> List[Tuple[int, int, int]]:
"""解析boxes为targets格式 [(cx, cy, size), ...]"""
targets = []
for box in boxes:
if len(box) == 3:
targets.append(tuple(box))
elif len(box) >= 4:
x1, y1, x2, y2 = box[:4]
cx = int((x1 + x2) / 2)
cy = int((y1 + y2) / 2)
size = max(abs(x2 - x1), abs(y2 - y1))
targets.append((cx, cy, size))
return targets
def _calculate_boundaries(
self,
liquid_y_coords: List[int],
liquid_x_coords: List[int],
air_y_coords: List[int],
air_x_coords: List[int],
foam_masks: List,
liquid_conf: float,
air_conf: float,
foam_conf: float,
crop_top: int,
crop_left: int,
idx: int
) -> Optional[Dict]:
"""
计算容器的顶部和底部边界
逻辑（按用户需求的6种情况）：
情况1: 仅air → air最低点 + air最高点
情况2: 仅liquid → liquid最低点 + liquid最高点
情况3: 仅foam → foam最低点 + foam最高点
情况4: liquid + air → liquid最低点 + air最高点
情况5: liquid + foam + air → liquid最低点 + air最高点
情况6: liquid + foam → liquid最低点 + foam最高点
"""
container_bottom = None
container_top = None
bottom_x = None
top_x = None
method = None
confidence = 0.0
has_liquid = len(liquid_y_coords) > 0
has_air = len(air_y_coords) > 0
has_foam = len(foam_masks) > 0
# 情况5: liquid + foam + air → liquid最低点 + air最高点
if has_liquid and has_foam and has_air:
# 容器底部 = liquid的最低点
liquid_y_array = np.array(liquid_y_coords)
liquid_x_array = np.array(liquid_x_coords)
max_y = np.max(liquid_y_array)
bottom_region_mask = liquid_y_array >= (max_y - 5)
container_bottom_in_crop = int(np.median(liquid_y_array[bottom_region_mask]))
on_bottom_line = liquid_y_array == container_bottom_in_crop
if np.sum(on_bottom_line) > 0:
bottom_x_in_crop = int(np.median(liquid_x_array[on_bottom_line]))
else:
bottom_x_in_crop = int(np.median(liquid_x_array[bottom_region_mask]))
# 容器顶部 = air的最高点
air_y_array = np.array(air_y_coords)
air_x_array = np.array(air_x_coords)
min_y = np.min(air_y_array)
top_region_mask = air_y_array <= (min_y + 5)
container_top_in_crop = int(np.median(air_y_array[top_region_mask]))
on_top_line = air_y_array == container_top_in_crop
if np.sum(on_top_line) > 0:
top_x_in_crop = int(np.median(air_x_array[on_top_line]))
else:
top_x_in_crop = int(np.median(air_x_array[top_region_mask]))
container_bottom = container_bottom_in_crop + crop_top
container_top = container_top_in_crop + crop_top
bottom_x = bottom_x_in_crop + crop_left
top_x = top_x_in_crop + crop_left
method = 'liquid_foam_air'
confidence = (liquid_conf + air_conf + foam_conf) / 3
print(f" ✅ 情况5 (liquid+foam+air): 底部={container_bottom}px (liquid最低), 顶部={container_top}px (air最高)")
# 情况4: liquid + air → liquid最低点 + air最高点
elif has_liquid and has_air:
# 容器底部 = liquid的最低点
liquid_y_array = np.array(liquid_y_coords)
liquid_x_array = np.array(liquid_x_coords)
max_y = np.max(liquid_y_array)
bottom_region_mask = liquid_y_array >= (max_y - 5)
container_bottom_in_crop = int(np.median(liquid_y_array[bottom_region_mask]))
on_bottom_line = liquid_y_array == container_bottom_in_crop
if np.sum(on_bottom_line) > 0:
bottom_x_in_crop = int(np.median(liquid_x_array[on_bottom_line]))
else:
bottom_x_in_crop = int(np.median(liquid_x_array[bottom_region_mask]))
# 容器顶部 = air的最高点
air_y_array = np.array(air_y_coords)
air_x_array = np.array(air_x_coords)
min_y = np.min(air_y_array)
top_region_mask = air_y_array <= (min_y + 5)
container_top_in_crop = int(np.median(air_y_array[top_region_mask]))
on_top_line = air_y_array == container_top_in_crop
if np.sum(on_top_line) > 0:
top_x_in_crop = int(np.median(air_x_array[on_top_line]))
else:
top_x_in_crop = int(np.median(air_x_array[top_region_mask]))
container_bottom = container_bottom_in_crop + crop_top
container_top = container_top_in_crop + crop_top
bottom_x = bottom_x_in_crop + crop_left
top_x = top_x_in_crop + crop_left
method = 'liquid_air'
confidence = (liquid_conf + air_conf) / 2
print(f" ✅ 情况4 (liquid+air): 底部={container_bottom}px (liquid最低), 顶部={container_top}px (air最高)")
# 情况6: liquid + foam → liquid最低点 + foam最高点
elif has_liquid and has_foam:
# 容器底部 = liquid的最低点
liquid_y_array = np.array(liquid_y_coords)
liquid_x_array = np.array(liquid_x_coords)
max_y = np.max(liquid_y_array)
# 取最低5像素范围的中位数y坐标
bottom_region_mask = liquid_y_array >= (max_y - 5)
container_bottom_in_crop = int(np.median(liquid_y_array[bottom_region_mask]))
# 在该y坐标上，找所有x坐标的中位数（更准确的中心点）
on_bottom_line = liquid_y_array == container_bottom_in_crop
if np.sum(on_bottom_line) > 0:
bottom_x_in_crop = int(np.median(liquid_x_array[on_bottom_line]))
else:
bottom_x_in_crop = int(np.median(liquid_x_array[bottom_region_mask]))
# 容器顶部 = 所有foam的最高点
all_foam_y = []
all_foam_x = []
for mask, y_coords, x_coords in foam_masks:
all_foam_y.extend(y_coords)
all_foam_x.extend(x_coords)
foam_y_array = np.array(all_foam_y)
foam_x_array = np.array(all_foam_x)
min_y = np.min(foam_y_array)
# 取最高5像素范围的中位数y坐标
top_region_mask = foam_y_array <= (min_y + 5)
container_top_in_crop = int(np.median(foam_y_array[top_region_mask]))
# 在该y坐标上，找所有x坐标的中位数
on_top_line = foam_y_array == container_top_in_crop
if np.sum(on_top_line) > 0:
top_x_in_crop = int(np.median(foam_x_array[on_top_line]))
else:
top_x_in_crop = int(np.median(foam_x_array[top_region_mask]))
container_bottom = container_bottom_in_crop + crop_top
container_top = container_top_in_crop + crop_top
bottom_x = bottom_x_in_crop + crop_left
top_x = top_x_in_crop + crop_left
method = 'liquid_foam'
confidence = (liquid_conf + foam_conf) / 2
print(f" ✅ 情况6 (liquid+foam): 底部={container_bottom}px (liquid最低), 顶部={container_top}px (foam最高)")
# 情况2: 仅liquid → liquid最低点 + liquid最高点
elif has_liquid:
# 容器底部 = liquid的最低点
liquid_y_array = np.array(liquid_y_coords)
liquid_x_array = np.array(liquid_x_coords)
max_y = np.max(liquid_y_array)
bottom_region_mask = liquid_y_array >= (max_y - 5)
container_bottom_in_crop = int(np.median(liquid_y_array[bottom_region_mask]))
on_bottom_line = liquid_y_array == container_bottom_in_crop
if np.sum(on_bottom_line) > 0:
bottom_x_in_crop = int(np.median(liquid_x_array[on_bottom_line]))
else:
bottom_x_in_crop = int(np.median(liquid_x_array[bottom_region_mask]))
# 容器顶部 = liquid的最高点
min_y = np.min(liquid_y_array)
top_region_mask = liquid_y_array <= (min_y + 5)
container_top_in_crop = int(np.median(liquid_y_array[top_region_mask]))
on_top_line = liquid_y_array == container_top_in_crop
if np.sum(on_top_line) > 0:
top_x_in_crop = int(np.median(liquid_x_array[on_top_line]))
else:
top_x_in_crop = int(np.median(liquid_x_array[top_region_mask]))
container_bottom = container_bottom_in_crop + crop_top
container_top = container_top_in_crop + crop_top
bottom_x = bottom_x_in_crop + crop_left
top_x = top_x_in_crop + crop_left
method = 'liquid_only'
confidence = liquid_conf
print(f" ✅ 情况2 (仅liquid): 底部={container_bottom}px (liquid最低), 顶部={container_top}px (liquid最高)")
# 情况1: 仅air → air最低点 + air最高点
elif has_air:
# 容器底部 = air的最低点
air_y_array = np.array(air_y_coords)
air_x_array = np.array(air_x_coords)
max_y = np.max(air_y_array)
bottom_region_mask = air_y_array >= (max_y - 5)
container_bottom_in_crop = int(np.median(air_y_array[bottom_region_mask]))
on_bottom_line = air_y_array == container_bottom_in_crop
if np.sum(on_bottom_line) > 0:
bottom_x_in_crop = int(np.median(air_x_array[on_bottom_line]))
else:
bottom_x_in_crop = int(np.median(air_x_array[bottom_region_mask]))
# 容器顶部 = air的最高点
min_y = np.min(air_y_array)
top_region_mask = air_y_array <= (min_y + 5)
container_top_in_crop = int(np.median(air_y_array[top_region_mask]))
on_top_line = air_y_array == container_top_in_crop
if np.sum(on_top_line) > 0:
top_x_in_crop = int(np.median(air_x_array[on_top_line]))
else:
top_x_in_crop = int(np.median(air_x_array[top_region_mask]))
container_bottom = container_bottom_in_crop + crop_top
container_top = container_top_in_crop + crop_top
bottom_x = bottom_x_in_crop + crop_left
top_x = top_x_in_crop + crop_left
method = 'air_only'
confidence = air_conf
print(f" ✅ 情况1 (仅air): 底部={container_bottom}px (air最低), 顶部={container_top}px (air最高)")
# 情况3: 仅foam → foam最低点 + foam最高点
elif has_foam:
# 收集所有foam的y坐标
all_foam_y = []
all_foam_x = []
for mask, y_coords, x_coords in foam_masks:
all_foam_y.extend(y_coords)
all_foam_x.extend(x_coords)
# 容器底部 = foam的最低点
foam_y_array = np.array(all_foam_y)
foam_x_array = np.array(all_foam_x)
max_y = np.max(foam_y_array)
bottom_region_mask = foam_y_array >= (max_y - 5)
container_bottom_in_crop = int(np.median(foam_y_array[bottom_region_mask]))
on_bottom_line = foam_y_array == container_bottom_in_crop
if np.sum(on_bottom_line) > 0:
bottom_x_in_crop = int(np.median(foam_x_array[on_bottom_line]))
else:
bottom_x_in_crop = int(np.median(foam_x_array[bottom_region_mask]))
# 容器顶部 = foam的最高点
min_y = np.min(foam_y_array)
top_region_mask = foam_y_array <= (min_y + 5)
container_top_in_crop = int(np.median(foam_y_array[top_region_mask]))
on_top_line = foam_y_array == container_top_in_crop
if np.sum(on_top_line) > 0:
top_x_in_crop = int(np.median(foam_x_array[on_top_line]))
else:
top_x_in_crop = int(np.median(foam_x_array[top_region_mask]))
container_bottom = container_bottom_in_crop + crop_top
container_top = container_top_in_crop + crop_top
bottom_x = bottom_x_in_crop + crop_left
top_x = top_x_in_crop + crop_left
method = 'foam_only'
confidence = foam_conf
print(f" ✅ 情况3 (仅foam): 底部={container_bottom}px (foam最低), 顶部={container_top}px (foam最高)")
else:
print(f" ❌ 无法检测到有效的边界（未检测到liquid、air或foam）")
return None
# 验证边界有效性
if container_bottom is None or container_top is None:
return None
if container_bottom <= container_top:
print(f" ❌ 边界无效: 底部({container_bottom}) <= 顶部({container_top})")
return None
container_height = container_bottom - container_top
print(f" 📏 容器高度: {container_height}px, 置信度: {confidence:.3f}")
return {
'index': idx,
'top': int(container_top),
'bottom': int(container_bottom),
'top_x': int(top_x),
'bottom_x': int(bottom_x),
'height': int(container_height),
'confidence': float(confidence),
'method': method
}
def _draw_annotations(
self,
image: np.ndarray,
container_info: Dict,
left: int,
right: int
):
"""在图片上绘制标注点和线"""
top_y = container_info['top']
bottom_y = container_info['bottom']
top_x = container_info['top_x']
bottom_x = container_info['bottom_x']
idx = container_info['index']
# 颜色定义
top_color = (0, 255, 0) # 绿色 - 顶部
bottom_color = (0, 0, 255) # 红色 - 底部
line_color = (255, 255, 0) # 青色 - 连接线
# 绘制顶部点
cv2.circle(image, (top_x, top_y), 8, top_color, -1)
cv2.circle(image, (top_x, top_y), 10, top_color, 2)
# 绘制底部点
cv2.circle(image, (bottom_x, bottom_y), 8, bottom_color, -1)
cv2.circle(image, (bottom_x, bottom_y), 10, bottom_color, 2)
# 绘制连接线
cv2.line(image, (top_x, top_y), (bottom_x, bottom_y), line_color, 2)
# 绘制水平参考线
cv2.line(image, (left, top_y), (right, top_y), top_color, 1, cv2.LINE_AA)
cv2.line(image, (left, bottom_y), (right, bottom_y), bottom_color, 1, cv2.LINE_AA)
# 添加文字标注
font = cv2.FONT_HERSHEY_SIMPLEX
font_scale = 0.6
thickness = 2
# 顶部标注
top_text = f"Top-{idx}"
cv2.putText(image, top_text, (top_x + 15, top_y - 10),
font, font_scale, top_color, thickness)
# 底部标注
bottom_text = f"Bottom-{idx}"
cv2.putText(image, bottom_text, (bottom_x + 15, bottom_y + 20),
font, font_scale, bottom_color, thickness)
# 高度标注
mid_y = (top_y + bottom_y) // 2
mid_x = max(top_x, bottom_x) + 20
height_text = f"{container_info['height']}px"
cv2.putText(image, height_text, (mid_x, mid_y),
font, font_scale, line_color, thickness)
# ==================== 独立调试功能 ====================
def test_auto_dot():
"""独立调试函数"""
import os
print("="*80)
print("🧪 自动标点功能测试")
print("="*80)
# 配置参数
model_path = r"D:\restructure\liquid_level_line_detection_system\database\model\detection_model\detect\best.dat"
test_image_path = r"D:\restructure\liquid_level_line_detection_system\test_image\4.jpg"
output_dir = r"D:\restructure\liquid_level_line_detection_system\test_output"
# 检查文件
if not os.path.exists(model_path):
print(f"❌ 模型文件不存在: {model_path}")
return
if not os.path.exists(test_image_path):
print(f"❌ 测试图片不存在: {test_image_path}")
print(f"💡 请将测试图片放置到: {test_image_path}")
return
# 创建输出目录
os.makedirs(output_dir, exist_ok=True)
# 加载图片
print(f"\n📷 加载测试图片: {test_image_path}")
image = cv2.imread(test_image_path)
if image is None:
print(f"❌ 无法读取图片")
return
print(f" 图片尺寸: {image.shape[1]}x{image.shape[0]}")
# 创建检测器
print(f"\n🔧 初始化自动标点检测器...")
detector = AutoDotDetector(model_path=model_path, device='cuda')
# 执行检测（直接传入整张图片）
print(f"\n🚀 开始自动标点检测...")
result = detector.detect_container_boundaries(
image=image,
conf_threshold=0.5
)
# 输出结果
print(f"\n{'='*80}")
print(f"📊 检测结果:")
print(f"{'='*80}")
if result['success']:
print(f"✅ 检测成功\n")
print(f"容器边界:")
print(f" - 顶部位置: ({result['top_x']}, {result['top']})")
print(f" - 底部位置: ({result['bottom_x']}, {result['bottom']})")
print(f" - 容器高度: {result['height']}px")
print(f" - 检测方法: {result['method']}")
print(f" - 置信度: {result['confidence']:.3f}")
print()
# 保存标注图片
output_path = os.path.join(output_dir, "auto_dot_result.jpg")
cv2.imwrite(output_path, result['annotated_image'])
print(f"💾 标注图片已保存: {output_path}")
# 显示图片（可选）
try:
cv2.imshow("Auto Dot Result", result['annotated_image'])
print(f"\n👀 按任意键关闭图片窗口...")
cv2.waitKey(0)
cv2.destroyAllWindows()
except:
print(f"⚠️ 无法显示图片（可能是无GUI环境）")
else:
print(f"❌ 检测失败")
if 'error' in result:
print(f" 错误信息: {result['error']}")
print(f"\n{'='*80}")
print(f"✅ 测试完成")
print(f"{'='*80}")
if __name__ == "__main__":
test_auto_dot()
handlers/videopage/channelpanel_handler.py
......@@ -1905,7 +1905,6 @@ class ChannelPanelHandler:
config_path = os.path.join(project_root, 'database', 'config', 'default_config.yaml')
if not os.path.exists(config_path):
print(f"[ConfigWatcher] 配置文件不存在: {config_path}")
return
# 创建文件系统监控器
......@@ -1915,61 +1914,43 @@ class ChannelPanelHandler:
# 连接文件变化信号
self._config_watcher.fileChanged.connect(self._onConfigFileChanged)
print(f"[ConfigWatcher] 已开始监控配置文件: {config_path}")
except Exception as e:
print(f"[ConfigWatcher] 初始化配置文件监控器失败: {e}")
import traceback
traceback.print_exc()
def _onConfigFileChanged(self, path):
"""配置文件变化时的回调"""
try:
print(f"🔄 [ConfigWatcher] 检测到配置文件变化: {path}")
# 延迟一小段时间，确保文件写入完成
QtCore.QTimer.singleShot(100, self._reloadChannelConfig)
except Exception as e:
print(f"[ConfigWatcher] 处理配置文件变化失败: {e}")
import traceback
traceback.print_exc()
def _reloadChannelConfig(self):
"""重新加载通道配置"""
try:
print("🔄 [ConfigWatcher] 开始重新加载通道配置...")
# 获取配置文件路径
project_root = get_project_root()
config_path = os.path.join(project_root, 'database', 'config', 'default_config.yaml')
print(f" 📂 [ConfigWatcher] 配置文件路径: {config_path}")
print(f" 📂 [ConfigWatcher] 文件是否存在: {os.path.exists(config_path)}")
if not os.path.exists(config_path):
print(f"[ConfigWatcher] 配置文件不存在: {config_path}")
return
# 读取配置文件
with open(config_path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f) or {}
print(f" 📄 [ConfigWatcher] 配置文件内容键: {list(config.keys())}")
print(f" 🗺️ [ConfigWatcher] 通道面板映射: {list(self._channel_panels_map.keys())}")
# 🔥 关键修复：更新 self._config，这样 _getChannelConfigFromFile 才能读取到最新配置
old_config = self._config
self._config = config
print(f" 🔄 [ConfigWatcher] 已更新内部配置缓存 (self._config)")
# 更新每个通道面板的名称和地址信息
for i in range(1, 5):
channel_id = f'channel{i}'
channel_key = f'channel{i}'
print(f" [ConfigWatcher] 处理 {channel_id}...")
# 获取通道面板
panel = self._channel_panels_map.get(channel_id)
if not panel:
......@@ -2015,15 +1996,10 @@ class ChannelPanelHandler:
# 重新添加监控（因为某些编辑器保存文件时会删除再创建，导致监控失效）
if hasattr(self, '_config_watcher'):
monitored_files = self._config_watcher.files()
print(f" 👀 [ConfigWatcher] 当前监控的文件: {monitored_files}")
if config_path not in monitored_files:
self._config_watcher.addPath(config_path)
print(f" 🔄 [ConfigWatcher] 重新添加文件监控: {config_path}")
print("[ConfigWatcher] 通道配置重新加载完成（包括地址配置）")
except Exception as e:
print(f"[ConfigWatcher] 重新加载通道配置失败: {e}")
import traceback
traceback.print_exc()
......
handlers/videopage/general_set_handler.py
......@@ -117,6 +117,7 @@ class GeneralSetPanelHandler:
widget.annotationEngineRequested.connect(self._handleAnnotationEngineRequest)
widget.frameLoadRequested.connect(self._handleFrameLoadRequest)
widget.annotationDataRequested.connect(self._handleAnnotationDataRequest)
widget.liveFrameRequested.connect(self._handleLiveFrameRequest)
def _handleRefreshModelList(self, model_widget=None):
"""处理刷新模型列表请求"""
......@@ -264,7 +265,6 @@ class GeneralSetPanelHandler:
if self.general_set_panel:
self.general_set_panel.setTaskIdOptions(task_ids)
print(f"[Handler] 已加载 {len(task_ids)} 个任务编号选项")
except Exception as e:
print(f"[Handler] 加载任务ID选项失败: {e}")
import traceback
......@@ -405,10 +405,6 @@ class GeneralSetPanelHandler:
model_config = default_config.get('model', {}).copy()
model_config['model_path'] = absolute_path
print(f"[Handler] 加载通道 {channel_id} 的模型配置:")
print(f" 相对路径: {channel_model_path}")
print(f" 绝对路径: {absolute_path}")
# 调用widget的方法应用配置
if self.general_set_panel:
self.general_set_panel.applyModelConfigFromHandler(
......@@ -693,26 +689,15 @@ class GeneralSetPanelHandler:
channel_frame = None
if self.general_set_panel and self.general_set_panel.channel_id:
channel_frame = self.getLatestFrame(self.general_set_panel.channel_id)
if channel_frame is not None:
pass
# 如果没有获取到通道画面，使用测试图像
# 如果没有获取到通道画面，弹出提示框并返回
if channel_frame is None:
pass
import numpy as np
channel_frame = np.zeros((720, 1280, 3), dtype=np.uint8)
channel_frame[:] = (100, 120, 140) # 灰色背景
# 添加文字说明
cv2.putText(channel_frame, "Test Annotation Frame", (50, 50),
cv2.FONT_HERSHEY_SIMPLEX, 1.5, (255, 255, 255), 2)
cv2.putText(channel_frame, "Draw detection areas and mark liquid levels", (50, 100),
cv2.FONT_HERSHEY_SIMPLEX, 0.8, (200, 200, 200), 1)
# 添加测试区域
cv2.rectangle(channel_frame, (200, 200), (400, 400), (0, 255, 0), 2)
cv2.rectangle(channel_frame, (500, 300), (700, 500), (0, 0, 255), 2)
QtWidgets.QMessageBox.warning(
self.main_window,
"获取画面失败",
"获取通道画面失败，请先连接通道"
)
return
# 2. 保存原始帧用于标注结果显示
self._annotation_source_frame = channel_frame.copy() if channel_frame is not None else None
......@@ -724,6 +709,9 @@ class GeneralSetPanelHandler:
if self.general_set_panel and self.general_set_panel.channel_name:
annotation_widget.setChannelName(self.general_set_panel.channel_name)
# 3.5. 初始化物理变焦控制器
self._initPhysicalZoomForAnnotation(annotation_widget)
# 4. 连接标注完成信号
def on_annotation_completed(boxes, bottoms, tops):
print(f"\n[DEBUG] ========== 标注完成回调 ==========")
......@@ -789,6 +777,9 @@ class GeneralSetPanelHandler:
# 5. 加载图像并显示标注界面
if annotation_widget.loadFrame(channel_frame):
# 启用实时画面预览
annotation_widget.enableLivePreview(True)
# 🔥 关键修复：延迟显示窗口，确保全屏应用后再显示
# 这样可以确保标注帧在全屏模式下立即显示
QtCore.QTimer.singleShot(150, annotation_widget.show)
......@@ -1246,6 +1237,73 @@ class GeneralSetPanelHandler:
if self.annotation_widget:
self.annotation_widget.showAnnotationError(f"获取标注数据失败: {str(e)}")
def _handleLiveFrameRequest(self):
"""处理实时画面请求"""
try:
# 获取通道最新画面
if self.general_set_panel and self.general_set_panel.channel_id:
channel_frame = self.getLatestFrame(self.general_set_panel.channel_id)
# 更新标注界面的画面
if channel_frame is not None and self.annotation_widget:
self.annotation_widget.updateLiveFrame(channel_frame)
except Exception as e:
pass
def _initPhysicalZoomForAnnotation(self, annotation_widget):
"""为标注界面初始化物理变焦控制器"""
try:
# 尝试导入物理变焦控制器
try:
from handlers.videopage.physical_zoom_controller import PhysicalZoomController
except ImportError:
try:
from physical_zoom_controller import PhysicalZoomController
except ImportError:
return
# 获取通道配置
if not self.general_set_panel or not self.general_set_panel.channel_id:
return
channel_id = self.general_set_panel.channel_id
# 从配置文件获取设备IP
config = self._getChannelConfig(channel_id)
if not config:
return
device_ip = config.get('address', '')
if not device_ip or 'rtsp://' not in device_ip:
return
# 提取IP地址
import re
match = re.search(r'@(\d+\.\d+\.\d+\.\d+)', device_ip)
if not match:
return
device_ip = match.group(1)
# 创建物理变焦控制器
physical_zoom_controller = PhysicalZoomController(
device_ip=device_ip,
username='admin',
password='cei345678',
channel=1
)
# 尝试连接设备
if physical_zoom_controller.connect_device():
# 设置到标注界面
annotation_widget.setPhysicalZoomController(physical_zoom_controller)
print(f"[标注界面] 物理变焦已启用 ({device_ip})")
else:
print(f"[标注界面] 物理变焦设备连接失败")
except Exception as e:
print(f"[标注界面] 初始化物理变焦失败: {e}")
def showGeneralSetPanel(self):
"""显示常规设置面板"""
from widgets.videopage.general_set import GeneralSetPanel
......
handlers/videopage/modelsetting_handler.py
......@@ -144,10 +144,6 @@ class ModelSettingHandler:
model_config['model_path'] = absolute_path
config_source = f"default_config.yaml → {channel_model_key} + model (全局参数)"
print(f"[Handler] 加载通道 {channel_id} 的模型配置")
print(f" 相对路径: {channel_model_path}")
print(f" 绝对路径: {absolute_path}")
print(f" model_config['model_path'] = {model_config.get('model_path', 'None')}")
else:
# 使用全局模型配置
model_config = default_config.get('model', {}).copy()
......
widgets/modelpage/modelset_page.py
......@@ -334,12 +334,11 @@ class ModelSetPage(QtWidgets.QWidget):
QTextEdit {
border: 1px solid #ccc;
background-color: white;
font-family: Consolas, Monaco, monospace;
font-size: 10pt;
}
""")
self.model_info_text.setPlaceholderText("选择模型后将显示模型文件夹内的txt文件内容...")
FontManager.applyToWidget(self.model_info_text)
# 应用全局字体管理
FontManager.applyToWidget(self.model_info_text, size=10)
right_text_layout.addWidget(self.model_info_text)
# 将左右两部分添加到中间主布局（1:1比例）
......@@ -1467,6 +1466,76 @@ class ModelSetPage(QtWidgets.QWidget):
except Exception as e:
return False
def _loadModelTxtFiles(self, model_name):
"""读取并显示模型文件夹内的txt文件"""
try:
# 清空文本显示区域
self.model_info_text.clear()
# 获取模型信息
if model_name not in self._model_params:
self.model_info_text.setPlainText(f"未找到模型 '{model_name}' 的信息")
return
model_info = self._model_params[model_name]
model_path = model_info.get('path', '')
if not model_path:
self.model_info_text.setPlainText(f"模型 '{model_name}' 没有路径信息")
return
# 获取模型所在的目录
model_dir = Path(model_path).parent
if not model_dir.exists():
self.model_info_text.setPlainText(f"模型目录不存在:\n{model_dir}")
return
# 查找目录中的所有txt文件
txt_files = list(model_dir.glob("*.txt"))
if not txt_files:
self.model_info_text.setPlainText(f"模型目录中没有找到txt文件:\n{model_dir}")
return
# 读取并显示所有txt文件的内容
content_parts = []
content_parts.append(f"模型目录: {model_dir}\n")
content_parts.append("=" * 60 + "\n\n")
for txt_file in sorted(txt_files):
content_parts.append(f"【文件: {txt_file.name}】\n")
content_parts.append("-" * 60 + "\n")
try:
# 尝试使用UTF-8编码读取
with open(txt_file, 'r', encoding='utf-8') as f:
file_content = f.read()
except UnicodeDecodeError:
# 如果UTF-8失败，尝试GBK编码
try:
with open(txt_file, 'r', encoding='gbk') as f:
file_content = f.read()
except Exception as e:
file_content = f"无法读取文件（编码错误）: {str(e)}"
except Exception as e:
file_content = f"读取文件时出错: {str(e)}"
content_parts.append(file_content)
content_parts.append("\n\n" + "=" * 60 + "\n\n")
# 显示所有内容
full_content = "".join(content_parts)
self.model_info_text.setPlainText(full_content)
# 滚动到顶部
cursor = self.model_info_text.textCursor()
cursor.movePosition(QtGui.QTextCursor.Start)
self.model_info_text.setTextCursor(cursor)
except Exception as e:
self.model_info_text.setPlainText(f"读取txt文件时出错:\n{str(e)}")
if __name__ == "__main__":
......
widgets/videopage/general_set.py
......@@ -270,18 +270,10 @@ class GeneralSetPanel(QtWidgets.QWidget):
content_layout.setContentsMargins(5, 5, 5, 5)
content_layout.setSpacing(10)
# 1. 任务信息区域
# 1. 任务信息区域（合并任务信息、模型配置、数据传输）
task_group = self._createTaskInfoGroup()
content_layout.addWidget(task_group)
# 2. 模型配置区域
model_group = self._createModelPathGroup()
content_layout.addWidget(model_group)
# 3. 数据传输区域
system_group = self._createSystemSettingsGroup()
content_layout.addWidget(system_group)
# 4. 标注区域
annotation_group = self._createAnnotationGroup()
content_layout.addWidget(annotation_group)
......@@ -368,7 +360,7 @@ class GeneralSetPanel(QtWidgets.QWidget):
return panel
def _createTaskInfoGroup(self):
"""创建任务信息组"""
"""创建任务信息组（合并任务信息、模型配置、数据传输）"""
group = QtWidgets.QGroupBox("任务信息")
layout = QtWidgets.QGridLayout()
layout.setColumnStretch(1, 1)
......@@ -376,14 +368,13 @@ class GeneralSetPanel(QtWidgets.QWidget):
layout.setHorizontalSpacing(15)
layout.setVerticalSpacing(10)
# 任务编号
# 第一行：任务编号、任务名称
task_id_label = QtWidgets.QLabel("任务编号:")
self.task_id_edit = QtWidgets.QLineEdit()
self.task_id_edit.setReadOnly(True)
self.task_id_edit.setPlaceholderText("任务编号")
self.task_id_edit.setMinimumWidth(scale_w(140)) # 响应式宽度
# 任务名称
task_name_label = QtWidgets.QLabel("任务名称:")
self.task_name_edit = QtWidgets.QLineEdit()
self.task_name_edit.setReadOnly(True)
......@@ -395,44 +386,25 @@ class GeneralSetPanel(QtWidgets.QWidget):
layout.addWidget(task_name_label, 0, 2)
layout.addWidget(self.task_name_edit, 0, 3)
group.setLayout(layout)
return group
def _createModelPathGroup(self):
"""创建模型配置展示组"""
group = QtWidgets.QGroupBox("模型配置")
layout = QtWidgets.QHBoxLayout()
layout.setSpacing(15)
model_path_label = QtWidgets.QLabel("当前模型路径:")
# 第二行：检测模型、数据推送地址
model_path_label = QtWidgets.QLabel("检测模型:")
self.model_path_display = QtWidgets.QLineEdit()
self.model_path_display.setReadOnly(True)
self.model_path_display.setPlaceholderText("未配置模型路径")
self.model_path_display.setMinimumWidth(scale_w(320)) # 响应式宽度
layout.addWidget(model_path_label)
layout.addWidget(self.model_path_display, 1)
self.model_path_display.setPlaceholderText("未配置检测模型")
self.model_path_display.setMinimumWidth(scale_w(140)) # 响应式宽度
group.setLayout(layout)
return group
def _createSystemSettingsGroup(self):
"""创建数据传输组"""
group = QtWidgets.QGroupBox("数据传输")
layout = QtWidgets.QHBoxLayout()
layout.setSpacing(15)
# 数据推送地址
push_label = QtWidgets.QLabel("数据推送地址:")
self.push_edit = QtWidgets.QLineEdit("192.168.1.234/put/push/height")
layout.addWidget(push_label)
layout.addWidget(self.push_edit, 1)
layout.addWidget(model_path_label, 1, 0)
layout.addWidget(self.model_path_display, 1, 1)
layout.addWidget(push_label, 1, 2)
layout.addWidget(self.push_edit, 1, 3)
group.setLayout(layout)
return group
def _createAnnotationGroup(self):
"""创建标注区域组"""
group = QtWidgets.QGroupBox("检测区域预览")
......@@ -580,10 +552,15 @@ class GeneralSetPanel(QtWidgets.QWidget):
return self._current_model_config.copy()
def setModelPathDisplay(self, model_path):
"""设置模型路径显示文本"""
"""设置模型路径显示文本（只显示文件名）"""
if not hasattr(self, 'model_path_display'):
return
text = model_path if model_path else ""
if model_path:
# 只显示文件名，不显示完整路径
import os
text = os.path.basename(model_path)
else:
text = ""
self.model_path_display.setText(text)
def setModelConfig(self, config):
......@@ -986,6 +963,7 @@ class AnnotationWidget(QtWidgets.QWidget):
annotationEngineRequested = QtCore.Signal() # 请求标注引擎
frameLoadRequested = QtCore.Signal() # 请求加载帧
annotationDataRequested = QtCore.Signal() # 请求标注数据
liveFrameRequested = QtCore.Signal() # 请求实时画面
def __init__(self, parent=None, annotation_engine=None):
super(AnnotationWidget, self).__init__(parent)
......@@ -1009,6 +987,20 @@ class AnnotationWidget(QtWidgets.QWidget):
self.area_states = [] # 存储区域状态列表（默认、空、满）
self.channel_name = "" # 通道名称
# 实时画面预览相关
self.live_preview_enabled = False # 是否启用实时预览
self.live_timer = None # 实时画面更新定时器
# 物理变焦相关
self.physical_zoom_controller = None # 物理变焦控制器
self.physical_zoom_enabled = False # 是否启用物理变焦
self.zoom_factor = 1.0 # 当前变焦倍数
self.min_zoom = 1.0 # 最小变焦倍数
self.max_zoom = 30.0 # 最大变焦倍数
self.zoom_step = 0.5 # 变焦步长
self.zoom_center_x = 0 # 变焦中心X坐标
self.zoom_center_y = 0 # 变焦中心Y坐标
self._initUI()
self._connectSignals()
......@@ -1045,7 +1037,10 @@ class AnnotationWidget(QtWidgets.QWidget):
def _connectSignals(self):
"""连接信号"""
pass
# 创建实时画面更新定时器
self.live_timer = QtCore.QTimer()
self.live_timer.timeout.connect(self._requestLiveFrame)
self.live_timer.setInterval(100) # 100ms更新一次，约10fps
def _applyFullScreen(self):
"""应用全屏模式（延迟调用，确保控件已初始化）"""
......@@ -1063,6 +1058,38 @@ class AnnotationWidget(QtWidgets.QWidget):
"""设置通道名称（用于生成区域默认名称）"""
self.channel_name = channel_name
def enableLivePreview(self, enabled=True):
"""启用/禁用实时画面预览"""
self.live_preview_enabled = enabled
if enabled:
self.live_timer.start()
else:
self.live_timer.stop()
def _requestLiveFrame(self):
"""请求获取最新画面（通过信号通知handler）"""
if self.live_preview_enabled:
self.liveFrameRequested.emit()
def updateLiveFrame(self, frame):
"""更新实时画面（由handler调用）"""
if frame is not None and self.live_preview_enabled:
self.current_frame = frame.copy()
self._updateDisplay()
def setPhysicalZoomController(self, controller):
"""设置物理变焦控制器"""
self.physical_zoom_controller = controller
if controller:
self.physical_zoom_enabled = True
# 获取变焦能力
capabilities = controller.get_zoom_capabilities()
if capabilities:
self.min_zoom = capabilities.get('min_zoom', 1.0)
self.max_zoom = capabilities.get('max_zoom', 30.0)
else:
self.physical_zoom_enabled = False
def _generateAreaName(self, area_index):
"""生成区域默认名称：通道name_区域1234"""
if self.channel_name:
......@@ -1883,6 +1910,25 @@ class AnnotationWidget(QtWidgets.QWidget):
else:
return
def wheelEvent(self, event):
"""鼠标滚轮事件 - 物理变焦"""
if self.physical_zoom_controller and self.physical_zoom_enabled:
delta = event.angleDelta().y()
if delta > 0:
# 向上滚动 - 放大
target_zoom = min(self.max_zoom, self.zoom_factor + self.zoom_step)
if target_zoom != self.zoom_factor:
self.zoom_factor = target_zoom
self.physical_zoom_controller.zoom_to_factor(target_zoom)
else:
# 向下滚动 - 缩小
target_zoom = max(self.min_zoom, self.zoom_factor - self.zoom_step)
if target_zoom != self.zoom_factor:
self.zoom_factor = target_zoom
self.physical_zoom_controller.zoom_to_factor(target_zoom)
else:
super().wheelEvent(event)
def keyPressEvent(self, event):
"""键盘事件处理 - 增强版快捷键绑定"""
# 如果正在编辑，ESC键取消编辑
......@@ -1892,8 +1938,12 @@ class AnnotationWidget(QtWidgets.QWidget):
# 🔥 增强版快捷键绑定
if event.key() == Qt.Key_R:
# R键：重置所有标注
self._onReset()
# R键：重置变焦（如果启用物理变焦）或重置标注
if self.physical_zoom_controller and self.physical_zoom_enabled:
self.physical_zoom_controller.zoom_to_factor(1.0)
self.zoom_factor = 1.0
else:
self._onReset()
elif event.key() == Qt.Key_C:
# C键：完成标注
self._onComplete()
......@@ -1909,6 +1959,10 @@ class AnnotationWidget(QtWidgets.QWidget):
elif event.key() == Qt.Key_H:
# H键：显示/隐藏帮助信息
self._toggleHelpDisplay()
elif event.key() == Qt.Key_F:
# F键：自动聚焦（如果启用物理变焦）
if self.physical_zoom_controller and self.physical_zoom_enabled:
self.physical_zoom_controller.auto_focus()
elif event.key() == Qt.Key_1:
# 1键：快速设置区域1高度为20mm
self._quickSetHeight(0, "20mm")
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment