ray/huojv
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'main/ui' into ui

Browse Source 
# Conflicts:
#	preview.py
This commit is contained in:
Ray
2025-11-01 23:36:41 +08:00
parent 4a0549114f eb896fc31e
commit 73d1f87ec7
11 changed files with 352 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.idea/huojv.iml generated
View File

@@ -2,7 +2,7 @@
<module type="PYTHON_MODULE" version="4"> <module type="PYTHON_MODULE" version="4">
<component name="NewModuleRootManager"> <component name="NewModuleRootManager">
<content url="file://$MODULE_DIR$" /> <content url="file://$MODULE_DIR$" />
<orderEntry type="jdk" jdkName="dnf" jdkType="Python SDK" /> <orderEntry type="jdk" jdkName="Python 3.9" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" /> <orderEntry type="sourceFolder" forTests="false" />
</component> </component>
</module> </module>

2
.idea/misc.xml generated
View File

@@ -3,5 +3,5 @@
<component name="Black"> <component name="Black">
<option name="sdkName" value="D:\CONDA\anaconda3" /> <option name="sdkName" value="D:\CONDA\anaconda3" />
</component> </component>
<component name="ProjectRootManager" version="2" project-jdk-name="dnf" project-jdk-type="Python SDK" /> <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.9" project-jdk-type="Python SDK" />
</project> </project>

17
config.json
View File

@@ -2,12 +2,22 @@
"groups": [ "groups": [
{ {
"name": "配置组1", "name": "配置组1",
"serial_port": "COM6", "serial_port": "COM10",
"serial_baudrate": 9600, "serial_baudrate": 9600,
"camera_index": 0, "camera_index": 0,
"camera_width": 1920, "camera_width": 1920,
"camera_height": 1080, "camera_height": 1080,
"move_velocity": 470, "move_velocity": 470,
"active": false
},
{
"name": "火炬游戏",
"serial_port": "COM11",
"serial_baudrate": 9600,
"camera_index": 1,
"camera_width": 1920,
"camera_height": 1080,
"move_velocity": 470,
"active": true "active": true
} }
], ],
@@ -16,7 +26,8 @@
"preview_height": 700, "preview_height": 700,
"preview_columns": 2, "preview_columns": 2,
"preview_rows": 2, "preview_rows": 2,
"show_preview": true "show_preview": true,
"preview_multi_window": false,
"preview_use_all_groups": true
} }
} }

4
config.py
View File

@@ -23,7 +23,9 @@ class ConfigManager:
"preview_height": 700, "preview_height": 700,
"preview_columns": 2, "preview_columns": 2,
"preview_rows": 2, "preview_rows": 2,
"show_preview": True "show_preview": True,
"preview_multi_window": False,
"preview_use_all_groups": True
} }
} }
self.config = self.load_config() self.config = self.load_config()

44
gui_config.py
View File

@@ -14,6 +14,7 @@ class ConfigGUI:
self.root.minsize(900, 600) # 设置最小尺寸 self.root.minsize(900, 600) # 设置最小尺寸
self.selected_index = 0 self.selected_index = 0
self.preview_thread = None # 添加预览线程引用
self.setup_ui() self.setup_ui()
self.load_current_config() self.load_current_config()
@@ -97,6 +98,24 @@ class ConfigGUI:
self.create_entry(preview_frame, "preview_columns", "预览列数:", prefix="display") self.create_entry(preview_frame, "preview_columns", "预览列数:", prefix="display")
self.create_entry(preview_frame, "preview_rows", "预览行数:", prefix="display") self.create_entry(preview_frame, "preview_rows", "预览行数:", prefix="display")
# 多窗口预览开关
multi_row = ttk.Frame(preview_frame)
multi_row.pack(fill=tk.X, pady=3)
ttk.Label(multi_row, text="多窗口预览:", width=15).pack(side=tk.LEFT)
self.preview_multi_window_var = tk.BooleanVar(value=False)
multi_cb = ttk.Checkbutton(multi_row, variable=self.preview_multi_window_var)
multi_cb.pack(side=tk.LEFT, padx=5)
self.config_vars["display_preview_multi_window"] = self.preview_multi_window_var
# 预览全部配置组开关
all_row = ttk.Frame(preview_frame)
all_row.pack(fill=tk.X, pady=3)
ttk.Label(all_row, text="预览全部配置组:", width=15).pack(side=tk.LEFT)
self.preview_use_all_groups_var = tk.BooleanVar(value=True)
all_cb = ttk.Checkbutton(all_row, variable=self.preview_use_all_groups_var)
all_cb.pack(side=tk.LEFT, padx=5)
self.config_vars["display_preview_use_all_groups"] = self.preview_use_all_groups_var
# 保存按钮 # 保存按钮
save_frame = ttk.Frame(right_frame) save_frame = ttk.Frame(right_frame)
save_frame.pack(fill=tk.X, pady=10) save_frame.pack(fill=tk.X, pady=10)
@@ -145,6 +164,12 @@ class ConfigGUI:
for key in ['preview_width', 'preview_height', 'preview_columns', 'preview_rows']: for key in ['preview_width', 'preview_height', 'preview_columns', 'preview_rows']:
if key in display: if key in display:
self.config_vars[f"display_{key}"].set(str(display[key])) self.config_vars[f"display_{key}"].set(str(display[key]))
# 多窗口预览
if 'preview_multi_window' in display:
self.preview_multi_window_var.set(bool(display['preview_multi_window']))
# 预览全部配置组
if 'preview_use_all_groups' in display:
self.preview_use_all_groups_var.set(bool(display['preview_use_all_groups']))
# 初次自动扫描一次采集卡和串口 # 初次自动扫描一次采集卡和串口
try: try:
self.scan_cameras() self.scan_cameras()
@@ -384,6 +409,10 @@ class ConfigGUI:
display[key] = int(var.get()) display[key] = int(var.get())
except: except:
pass pass
# 保存多窗口预览布尔值
display['preview_multi_window'] = bool(self.preview_multi_window_var.get())
# 保存预览全部配置组布尔值
display['preview_use_all_groups'] = bool(self.preview_use_all_groups_var.get())
config_manager.config['display'] = display config_manager.config['display'] = display
# 保存到文件 # 保存到文件
@@ -397,20 +426,13 @@ class ConfigGUI:
return False return False
def start_preview(self): def start_preview(self):
"""启动预览窗口""" """启动预览窗口（主线程，弹窗模式）"""
# 保存配置（不显示消息框，静默保存）
if not self.save_config_silent(): if not self.save_config_silent():
messagebox.showerror("错误", "配置保存失败，无法启动预览") messagebox.showerror("错误", "配置保存失败，无法启动预览")
return return
# 重新加载配置（从文件读取最新配置）
config_manager.load_config() config_manager.load_config()
# 启动预览窗口
from preview import PreviewWindow from preview import PreviewWindow
preview_thread = threading.Thread(target=lambda: PreviewWindow().run()) PreviewWindow().run(self.root) # 以主窗口为父Toplevel弹出独立预览窗
preview_thread.daemon = True
preview_thread.start()
def save_config_silent(self): def save_config_silent(self):
"""静默保存配置（不显示消息框）""" """静默保存配置（不显示消息框）"""
@@ -462,6 +484,10 @@ class ConfigGUI:
display[key] = int(var.get()) display[key] = int(var.get())
except: except:
pass pass
# 保存多窗口预览布尔值（静默）
display['preview_multi_window'] = bool(self.preview_multi_window_var.get())
# 保存预览全部配置组布尔值（静默）
display['preview_use_all_groups'] = bool(self.preview_use_all_groups_var.get())
config_manager.config['display'] = display config_manager.config['display'] = display
# 保存到文件 # 保存到文件

12
main.py
View File

@@ -260,15 +260,9 @@ while True:
if shizi.fuhuo(im_opencv[0]): if shizi.fuhuo(im_opencv[0]):
print('点击复活') print('点击复活')
mouse_gui.send_data_absolute(left + 536, top + 627, may=1) mouse_gui.send_data_absolute(left + 536, top + 627, may=1)
mouse_gui.send_data_absolute(rw[0], rw[1], may=0)
continue
if detections['zhaozi'] is not None:
move_to(rw, detections['zhaozi'])
continue
if len(detections['daojv']) != 0:
move_to(rw, process_points(detections['daojv']))
for i in range(3 + len(detections['daojv'])):
keyboard.send_data("AA")
time.sleep(0.15) time.sleep(0.15)
keyboard.release() keyboard.release()
continue continue

83
main_multi.py
View File

@@ -6,6 +6,10 @@ import multiprocessing
import sys import sys
from config import config_manager from config import config_manager
from main_single import run_automation_for_group from main_single import run_automation_for_group
from utils.logger import logger, throttle
import logging
import time
def main(): def main():
"""主函数""" """主函数"""
@@ -16,44 +20,44 @@ def main():
groups = config_manager.config.get('groups', []) groups = config_manager.config.get('groups', [])
if not groups: if not groups:
print("❌ 没有找到任何配置组") logger.error("❌ 没有找到任何配置组")
print("请先运行 gui_config.py 创建配置组") logger.info("请先运行 gui_config.py 创建配置组")
return return
# 询问要启动哪些配置组 # 询问要启动哪些配置组
print("=" * 60) logger.info("=" * 60)
print("🔥 多配置组启动器") logger.info("🔥 多配置组启动器")
print("=" * 60) logger.info("=" * 60)
print("\n可用配置组:") logger.info("\n可用配置组:")
for i, group in enumerate(groups): for i, group in enumerate(groups):
active_mark = "✓" if group.get('active', False) else " " active_mark = "✓" if group.get('active', False) else " "
print(f" [{i}] {active_mark} {group['name']}") logger.info(f" [{i}] {active_mark} {group['name']}")
print(f" 串口: {group['serial_port']} | 采集卡: {group['camera_index']}") logger.info(f" 串口: {group['serial_port']} | 采集卡: {group['camera_index']}")
print("\n选择启动方式:") logger.info("\n选择启动方式:")
print(" 1. 启动所有活动配置组") logger.info(" 1. 启动所有活动配置组")
print(" 2. 启动所有配置组") logger.info(" 2. 启动所有配置组")
print(" 3. 选择特定配置组") logger.info(" 3. 选择特定配置组")
print(" 0. 退出") logger.info(" 0. 退出")
choice = input("\n请选择 (0-3): ").strip() choice = input("\n请选择 (0-3): ").strip()
selected_indices = [] selected_indices = []
if choice == "0": if choice == "0":
print("👋 退出") logger.info("👋 退出")
return return
elif choice == "1": elif choice == "1":
# 启动所有活动配置组 # 启动所有活动配置组
selected_indices = [i for i, g in enumerate(groups) if g.get('active', False)] selected_indices = [i for i, g in enumerate(groups) if g.get('active', False)]
if not selected_indices: if not selected_indices:
print("❌ 没有活动的配置组") logger.error("❌ 没有活动的配置组")
return return
print(f"\n✅ 将启动 {len(selected_indices)} 个活动配置组") logger.info(f"\n✅ 将启动 {len(selected_indices)} 个活动配置组")
elif choice == "2": elif choice == "2":
# 启动所有配置组 # 启动所有配置组
selected_indices = list(range(len(groups))) selected_indices = list(range(len(groups)))
print(f"\n✅ 将启动所有 {len(selected_indices)} 个配置组") logger.info(f"\n✅ 将启动所有 {len(selected_indices)} 个配置组")
elif choice == "3": elif choice == "3":
# 选择特定配置组 # 选择特定配置组
indices_input = input("请输入要启动的配置组索引(用逗号分隔,如: 0,1,2): ").strip() indices_input = input("请输入要启动的配置组索引(用逗号分隔,如: 0,1,2): ").strip()
@@ -62,34 +66,33 @@ def main():
# 验证索引有效性 # 验证索引有效性
selected_indices = [i for i in selected_indices if 0 <= i < len(groups)] selected_indices = [i for i in selected_indices if 0 <= i < len(groups)]
if not selected_indices: if not selected_indices:
print("❌ 没有有效的配置组索引") logger.error("❌ 没有有效的配置组索引")
return return
print(f"\n✅ 将启动 {len(selected_indices)} 个配置组") logger.info(f"\n✅ 将启动 {len(selected_indices)} 个配置组")
except ValueError: except ValueError:
print("❌ 输入格式错误") logger.error("❌ 输入格式错误")
return return
else: else:
print("❌ 无效选择") logger.error("❌ 无效选择")
return return
# 显示将要启动的配置组 # 显示将要启动的配置组
print("\n将要启动的配置组:") logger.info("\n将要启动的配置组:")
for idx in selected_indices: for idx in selected_indices:
group = groups[idx] group = groups[idx]
print(f" • {group['name']} (串口:{group['serial_port']}, 采集卡:{group['camera_index']})") logger.info(f" • {group['name']} (串口:{group['serial_port']}, 采集卡:{group['camera_index']})")
confirm = input("\n确认启动? (y/n): ").strip().lower() confirm = input("\n确认启动? (y/n): ").strip().lower()
if confirm != 'y': if confirm != 'y':
print("❌ 取消启动") logger.info("❌ 取消启动")
return return
# 启动多进程 # 启动多进程
print("\n🚀 开始启动多个配置组...")
processes = [] processes = []
for idx in selected_indices: for idx in selected_indices:
group = groups[idx] group = groups[idx]
print(f"启动进程: {group['name']}...") logger.info(f"启动进程: {group['name']}...")
process = multiprocessing.Process( process = multiprocessing.Process(
target=run_automation_for_group, target=run_automation_for_group,
args=(idx,), args=(idx,),
@@ -97,12 +100,12 @@ def main():
) )
process.start() process.start()
processes.append((idx, group['name'], process)) processes.append((idx, group['name'], process))
print(f"✅ {group['name']} 已启动 (PID: {process.pid})") logger.info(f"✅ {group['name']} 已启动 (PID: {process.pid})")
print(f"\n✅ 成功启动 {len(processes)} 个配置组进程") logger.info(f"\n✅ 成功启动 {len(processes)} 个配置组进程")
print("\n" + "=" * 60) logger.info("\n" + "=" * 60)
print("运行状态:") logger.info("运行状态:")
print("=" * 60) logger.info("=" * 60)
# 监控进程状态 # 监控进程状态
try: try:
@@ -112,32 +115,32 @@ def main():
if proc.is_alive(): if proc.is_alive():
alive_count += 1 alive_count += 1
else: else:
print(f"⚠️ {name} 进程已退出 (退出码: {proc.exitcode})") logger.warning(f"⚠️ {name} 进程已退出 (退出码: {proc.exitcode})")
if alive_count == 0: if alive_count == 0:
print("\n所有进程已退出") logger.info("\n所有进程已退出")
break break
import time
time.sleep(2) time.sleep(2)
# 打印存活状态 # 打印存活状态
alive_names = [name for idx, name, proc in processes if proc.is_alive()] alive_names = [name for idx, name, proc in processes if proc.is_alive()]
if alive_names: if alive_names:
print(f"\r📊 运行中: {', '.join(alive_names)} ({alive_count}/{len(processes)})", end='', flush=True) # 每2秒节流一次状态行
throttle("multi_alive", 2.0, logging.INFO, f"📊 运行中: {', '.join(alive_names)} ({alive_count}/{len(processes)})")
except KeyboardInterrupt: except KeyboardInterrupt:
print("\n\n🛑 收到停止信号，正在关闭所有进程...") logger.warning("\n\n🛑 收到停止信号，正在关闭所有进程...")
for idx, name, proc in processes: for idx, name, proc in processes:
if proc.is_alive(): if proc.is_alive():
print(f"正在停止 {name}...") logger.info(f"正在停止 {name}...")
proc.terminate() proc.terminate()
proc.join(timeout=5) proc.join(timeout=5)
if proc.is_alive(): if proc.is_alive():
print(f"强制停止 {name}...") logger.warning(f"强制停止 {name}...")
proc.kill() proc.kill()
print(f"✅ {name} 已停止") logger.info(f"✅ {name} 已停止")
print("\n👋 所有进程已停止") logger.info("\n👋 所有进程已停止")
if __name__ == "__main__": if __name__ == "__main__":
multiprocessing.freeze_support() # Windows下需要 multiprocessing.freeze_support() # Windows下需要

310
preview.py
View File

@@ -11,6 +11,7 @@ from config import config_manager
# 抑制OpenCV的警告信息（兼容不同版本） # 抑制OpenCV的警告信息（兼容不同版本）
import sys import sys
import io import io
import time as _time
os.environ['OPENCV_LOG_LEVEL'] = 'SILENT' os.environ['OPENCV_LOG_LEVEL'] = 'SILENT'
os.environ['OPENCV_IO_ENABLE_OPENEXR'] = '0' os.environ['OPENCV_IO_ENABLE_OPENEXR'] = '0'
@@ -26,13 +27,25 @@ try:
except Exception: except Exception:
pass pass
# 简单日志限流
_last_log_times = {}
def log_throttle(key: str, interval_sec: float, text: str):
now = _time.time()
last = _last_log_times.get(key, 0)
if now - last >= interval_sec:
_last_log_times[key] = now
try:
print(text)
except Exception:
pass
class PreviewWindow: class PreviewWindow:
"""采集卡预览窗口""" """采集卡预览窗口"""
def __init__(self): def __init__(self):
self.config = config_manager.config self.config = config_manager.config
self.caps = {} self.caps = {}
self.frames = {} self.frames = {}
self.large_window = None self.large_windows = {} # idx -> Toplevel
self.running = True self.running = True
def init_cameras(self): def init_cameras(self):
@@ -40,11 +53,17 @@ class PreviewWindow:
print("🔧 开始初始化采集卡...") print("🔧 开始初始化采集卡...")
loaded_count = 0 loaded_count = 0
# 如果没有活动配置，加载所有配置 # 根据配置选择加载哪些组
active_groups = [g for g in self.config['groups'] if g.get('active', True)] display = self.config.get('display', {})
if not active_groups: use_all = bool(display.get('preview_use_all_groups', True))
print("⚠️ 没有活动的配置组，将尝试加载所有配置组") if use_all:
active_groups = self.config['groups'] target_groups = self.config['groups']
print(f"📷 预览全部配置组，共 {len(target_groups)} 组")
else:
target_groups = [g for g in self.config['groups'] if g.get('active', False)]
print(f"📷 仅预览活动配置组，共 {len(target_groups)} 组")
if not target_groups:
print("⚠️ 没有活动的配置组，将不显示任何采集卡。可在配置中启用‘预览全部配置组’或设置活动组")
# 重定向stderr来抑制OpenCV的错误输出 # 重定向stderr来抑制OpenCV的错误输出
old_stderr = sys.stderr old_stderr = sys.stderr
@@ -53,7 +72,7 @@ class PreviewWindow:
try: try:
sys.stderr = suppressed_output sys.stderr = suppressed_output
for i, group in enumerate(active_groups): for i, group in enumerate(target_groups):
try: try:
cam_idx = group['camera_index'] cam_idx = group['camera_index']
print(f" 尝试打开采集卡 {cam_idx} ({group['name']})...") print(f" 尝试打开采集卡 {cam_idx} ({group['name']})...")
@@ -125,6 +144,40 @@ class PreviewWindow:
else: else:
print(f"✅ 成功加载 {loaded_count} 个采集卡") print(f"✅ 成功加载 {loaded_count} 个采集卡")
def grab_once(self):
"""抓取每個采集卡當前單幀（裁剪後，存入 self.frames），抓取结果增强debug输出"""
for idx, data in self.caps.items():
try:
cap = data['cap']
frm_name = data.get('name', str(idx))
if cap is None:
print(f"[抓取] 采集卡{idx}({frm_name}) cap为None")
self.frames[idx] = None
continue
ret, frame = cap.read()
if not ret or frame is None:
print(f"[抓取] 采集卡{idx}({frm_name}) 没抓到帧 ret={ret} frame=None")
self.frames[idx] = None
continue
height, width = frame.shape[:2]
crop_top = 30
crop_bottom = min(crop_top + 720, height)
crop_width = min(1280, width)
if crop_bottom <= crop_top or crop_width <= 0:
print(f"[抓取] 采集卡{idx}({frm_name}) 尺寸异常: H={height} W={width} 裁剪失败")
self.frames[idx] = None
continue
frame = frame[crop_top:crop_bottom, 0:crop_width]
# 额外检查：像素全黑？
if np.mean(frame) < 3:
print(f"[抓取] 采集卡{idx}({frm_name}) 帧抓到但内容接近全黑 shape={frame.shape}")
else:
print(f"[抓取] 采集卡{idx}({frm_name}) 成功 shape={frame.shape} 类型={frame.dtype}")
self.frames[idx] = frame
except Exception as e:
print(f"[抓取] 采集卡{idx}({data.get('name', str(idx))}) 抓取异常: {e}")
self.frames[idx] = None
def capture_frames(self): def capture_frames(self):
"""捕获帧""" """捕获帧"""
while self.running: while self.running:
@@ -159,19 +212,21 @@ class PreviewWindow:
import time import time
time.sleep(0.01) # 避免CPU占用过高 time.sleep(0.01) # 避免CPU占用过高
def create_grid_window(self): def create_grid_window(self, master=None):
"""创建网格窗口""" """创建网格窗口，可挂载在主Tk下作为Toplevel弹窗"""
# 获取显示配置
display = self.config.get('display', {}) display = self.config.get('display', {})
preview_width = display.get('preview_width', 1000) preview_width = display.get('preview_width', 1000)
preview_height = display.get('preview_height', 700) preview_height = display.get('preview_height', 700)
columns = display.get('preview_columns', 2) columns = display.get('preview_columns', 2)
rows = display.get('preview_rows', 2) rows = display.get('preview_rows', 2)
root = tk.Tk() if master is not None:
root = tk.Toplevel(master)
else:
root = tk.Tk()
root.title("采集卡预览 - 点击放大") root.title("采集卡预览 - 点击放大")
root.geometry(f"{preview_width}x{preview_height}") root.geometry(f"{preview_width}x{preview_height}")
root.update_idletasks() # 立即更新窗口尺寸 root.update_idletasks()
canvas = Canvas(root, bg='black', width=preview_width, height=preview_height) canvas = Canvas(root, bg='black', width=preview_width, height=preview_height)
canvas.pack(fill=tk.BOTH, expand=True) canvas.pack(fill=tk.BOTH, expand=True)
@@ -183,11 +238,10 @@ class PreviewWindow:
self.debug_count = 0 self.debug_count = 0
def update_frames_once(): def update_frames_once():
"""在主线程中更新一帧（使用after循环）""" """在主线程中每5秒刷新画面（直接显示高频采集线程生成的帧，去除grab_once）"""
if not self.running: if not self.running:
return return
try: try:
# 如果没有加载任何采集卡，显示提示
if not self.caps: if not self.caps:
canvas.delete("all") canvas.delete("all")
canvas.create_text( canvas.create_text(
@@ -198,7 +252,7 @@ class PreviewWindow:
font=('Arial', 14), font=('Arial', 14),
justify=tk.CENTER justify=tk.CENTER
) )
root.after(33, update_frames_once) root.after(5000, update_frames_once)
return return
# 计算每个预览窗口的位置和大小 # 计算每个预览窗口的位置和大小
@@ -228,79 +282,74 @@ class PreviewWindow:
frame_idx = 0 frame_idx = 0
for idx in self.caps.keys(): for idx in self.caps.keys():
frame = self.frames.get(idx)
has_frame = frame is not None
print(f'[DEBUG] update UI idx={idx}, has_frame={has_frame}, frame_info={getattr(frame, "shape", None)}')
name = self.caps[idx]['name']
row = frame_idx // columns row = frame_idx // columns
col = frame_idx % columns col = frame_idx % columns
x = col * cell_width x = col * cell_width
y = row * cell_height y = row * cell_height
center_x = x + cell_width // 2 center_x = x + cell_width // 2
center_y = y + cell_height // 2 center_y = y + cell_height // 2
name = self.caps[idx]['name'] # 如果没拿到frame，临时主动再抓一次，并DEBUG
if not has_frame:
# 检查帧数据 print(f'[DEBUG] UI未获得frame，尝试临时抓取采集卡{idx}...')
has_frame = idx in self.frames and self.frames[idx] is not None try:
cap_test = self.caps[idx]['cap']
if has_frame: ret, fr2 = cap_test.read()
# 调整图像大小 print(f'[DEBUG] 临时抓取ret={ret}, shape={getattr(fr2, "shape", None)}')
if ret and fr2 is not None:
frame = fr2
has_frame = True
else:
frame = None
has_frame = False
except Exception as e:
print(f'[DEBUG] 临时抓取异常：{e}')
frame = None
has_frame = False
if has_frame:
try: try:
frame = self.frames[idx]
h, w = frame.shape[:2] h, w = frame.shape[:2]
# 确保尺寸有效
if w <= 0 or h <= 0: if w <= 0 or h <= 0:
texts_to_draw.append((center_x, center_y, f"{name}\n尺寸无效", 'red')) detail = f"采集卡{idx}:{name}\n抓到帧但图像尺寸异常 ({w}x{h})"
texts_to_draw.append((center_x, center_y, detail, 'red'))
frame_idx += 1 frame_idx += 1
continue continue
# 计算缩放比例，留一些边距
scale = min(cell_width / w, cell_height / h, 1.0) * 0.85 scale = min(cell_width / w, cell_height / h, 1.0) * 0.85
new_w = max(1, int(w * scale)) new_w = max(1, int(w * scale))
new_h = max(1, int(h * scale)) new_h = max(1, int(h * scale))
# 确保缩放后的尺寸有效
if new_w <= 0 or new_h <= 0: if new_w <= 0 or new_h <= 0:
texts_to_draw.append((center_x, center_y, f"{name}\n缩放失败", 'red')) detail = f"采集卡{idx}:{name}\n缩放计算异常 ({w}x{h})"
texts_to_draw.append((center_x, center_y, detail, 'red'))
frame_idx += 1 frame_idx += 1
continue continue
display_frame = frame
# 调试输出（仅前几次，避免刷屏） if len(display_frame.shape) == 3 and display_frame.shape[2] == 3:
if frame_idx == 0 and len(images_to_draw) == 0: display_frame = cv2.cvtColor(display_frame, cv2.COLOR_BGR2RGB)
self.debug_count += 1 resized_frame = cv2.resize(display_frame, (new_w, new_h))
if self.debug_count <= 3: # 只打印前3次
print(f"🔍 预览调试 #{self.debug_count}:")
print(f" 配置尺寸: {preview_width}x{preview_height}")
print(f" 实际画布: {canvas_width}x{canvas_height}")
print(f" 单元格大小: {cell_width}x{cell_height}")
print(f" 原始帧: {w}x{h}")
print(f" 缩放后: {new_w}x{new_h}")
resized_frame = cv2.resize(frame, (new_w, new_h))
# 转换为PIL图像
# 确保颜色通道正确（BGR -> RGB）
if len(resized_frame.shape) == 3 and resized_frame.shape[2] == 3:
resized_frame = cv2.cvtColor(resized_frame, cv2.COLOR_BGR2RGB)
pil_image = Image.fromarray(resized_frame) pil_image = Image.fromarray(resized_frame)
photo = ImageTk.PhotoImage(image=pil_image) photo = ImageTk.PhotoImage(image=pil_image)
# 保持引用，避免被GC if not hasattr(self, 'photo_refs'):
self.photo_objects[idx] = photo self.photo_refs = {}
self._grid_photos.append(photo) self.photo_refs[idx] = photo
canvas.create_image(center_x, center_y, image=photo, anchor='center')
canvas.update() # 保证实时刷新
print(f'[DEBUG] UI idx={idx}, 绘制photo id={id(photo)}, photo_refs={len(self.photo_refs)}')
images_to_draw.append((photo, center_x, center_y)) images_to_draw.append((photo, center_x, center_y))
texts_to_draw.append((center_x, y + 15, name, 'white')) texts_to_draw.append((center_x, y + 15, name, 'white'))
frame_idx += 1 frame_idx += 1
except Exception as e: except Exception as e:
# 忽略pyimage相关错误，避免刷屏 errstr = str(e)
if "pyimage" not in str(e).lower(): detail = f"采集卡{idx}:{name}\n帧处理异常：{errstr}"
print(f"处理帧 {idx} 错误: {e}") print(f'[DEBUG] idx={idx}, 帧处理异常：{errstr}')
import traceback texts_to_draw.append((center_x, center_y, detail, 'red'))
traceback.print_exc()
# 如果处理失败，显示等待提示
texts_to_draw.append((center_x, center_y, f"{name}\n处理失败", 'red'))
frame_idx += 1 frame_idx += 1
else: else:
# 显示等待提示 detail = f"采集卡{idx}:{name}\n抓取失败/暂无画面"
texts_to_draw.append((center_x, center_y, f"{name}\n等待画面...", 'gray')) print(f'[DEBUG] idx={idx} 依然无可用帧，最终显示错误提示')
texts_to_draw.append((center_x, center_y, detail, 'red'))
frame_idx += 1 frame_idx += 1
# 清空画布 # 清空画布
@@ -308,14 +357,20 @@ class PreviewWindow:
# 先绘制所有图像（底层） # 先绘制所有图像（底层）
if images_to_draw and self.debug_count <= 3: if images_to_draw and self.debug_count <= 3:
print(f"✅ 准备绘制 {len(images_to_draw)} 个图像到画布 ({canvas_width}x{canvas_height})") log_throttle("draw_prepare", 2.0, f"✅ 准备绘制 {len(images_to_draw)} 个图像到画布 ({canvas_width}x{canvas_height})")
for photo, x, y in images_to_draw: for photo, x, y in images_to_draw:
try: try:
# PhotoImage强引用，防止被GC
if not hasattr(self, 'photo_refs'):
self.photo_refs = {}
self.photo_refs[idx] = photo
canvas.create_image(x, y, image=photo, anchor='center') canvas.create_image(x, y, image=photo, anchor='center')
canvas.update() # 保证实时刷新
print(f'[DEBUG] idx={idx}, photo id={id(photo)}, photo refs={len(self.photo_refs)}')
except Exception as e: except Exception as e:
if "pyimage" not in str(e).lower(): if "pyimage" not in str(e).lower():
print(f"绘制图像错误: {e}") log_throttle("draw_image_error", 2.0, f"绘制图像错误: {e}")
# 再绘制所有文本（上层） # 再绘制所有文本（上层）
for x, y, text, color in texts_to_draw: for x, y, text, color in texts_to_draw:
@@ -325,7 +380,7 @@ class PreviewWindow:
else: else:
canvas.create_text(x, y, text=text, fill=color, font=('Arial', 10)) canvas.create_text(x, y, text=text, fill=color, font=('Arial', 10))
except Exception as e: except Exception as e:
print(f"绘制文本错误: {e}") log_throttle("draw_text_error", 2.0, f"绘制文本错误: {e}")
# 绘制分割线，区分不同的采集卡窗口 # 绘制分割线，区分不同的采集卡窗口
try: try:
@@ -351,14 +406,14 @@ class PreviewWindow:
dash=(5, 5) # 虚线效果 dash=(5, 5) # 虚线效果
) )
except Exception as e: except Exception as e:
print(f"绘制分割线错误: {e}") log_throttle("draw_grid_error", 2.0, f"绘制分割线错误: {e}")
except Exception as e: except Exception as e:
print(f"更新帧错误: {e}") log_throttle("update_frame_error", 1.0, f"更新帧错误: {e}")
import traceback import traceback
traceback.print_exc() traceback.print_exc()
# 约30fps # 每5秒更新一次
root.after(33, update_frames_once) root.after(5000, update_frames_once)
def on_canvas_click(event): def on_canvas_click(event):
"""点击画布事件""" """点击画布事件"""
@@ -402,89 +457,128 @@ class PreviewWindow:
root.destroy() root.destroy()
root.protocol("WM_DELETE_WINDOW", on_closing) root.protocol("WM_DELETE_WINDOW", on_closing)
root.mainloop() # 仅在独立窗口（无master）时进入事件循环
if master is None:
root.mainloop()
def show_large_window(self, idx): def show_large_window(self, idx):
"""显示大窗口""" """显示指定采集卡的大窗口（可多实例）"""
if self.large_window is not None and self.large_window.winfo_exists(): # 如果已存在该索引窗口，先销毁再创建，确保刷新
self.large_window.destroy() if idx in self.large_windows and self.large_windows[idx].winfo_exists():
try:
self.large_windows[idx].destroy()
except Exception:
pass
self.large_window = tk.Toplevel() win = tk.Toplevel()
self.large_window.title(f"放大视图 - {self.caps[idx]['name']}") win.title(f"放大视图 - {self.caps[idx]['name']}")
self.large_window.geometry("1280x720") win.geometry("1280x720")
self.large_windows[idx] = win
canvas = Canvas(self.large_window, bg='black') canvas = Canvas(win, bg='black')
canvas.pack(fill=tk.BOTH, expand=True) canvas.pack(fill=tk.BOTH, expand=True)
photo_obj = {} photo_obj = {}
# 持久引用大窗当前帧 PhotoImage，避免GC
self._large_photos = []
def update_large_once(): def update_large_once():
if not self.running or not self.large_window.winfo_exists(): if not self.running or not win.winfo_exists():
return return
try: try:
# 获取窗口大小 window_width = win.winfo_width() if win.winfo_width() > 1 else 1280
window_width = self.large_window.winfo_width() if self.large_window.winfo_width() > 1 else 1280 window_height = win.winfo_height() if win.winfo_height() > 1 else 720
window_height = self.large_window.winfo_height() if self.large_window.winfo_height() > 1 else 720
if idx in self.frames and self.frames[idx] is not None: if idx in self.frames and self.frames[idx] is not None:
# 先处理图像，再清空画布
frame = self.frames[idx] frame = self.frames[idx]
h, w = frame.shape[:2] h, w = frame.shape[:2]
# 调整到窗口大小
scale = min(window_width / w, window_height / h) scale = min(window_width / w, window_height / h)
new_w = int(w * scale) new_w = int(w * scale)
new_h = int(h * scale) new_h = int(h * scale)
resized_frame = cv2.resize(frame, (new_w, new_h)) resized_frame = cv2.resize(frame, (new_w, new_h))
# 确保颜色通道正确（BGR -> RGB）
if len(resized_frame.shape) == 3 and resized_frame.shape[2] == 3: if len(resized_frame.shape) == 3 and resized_frame.shape[2] == 3:
resized_frame = cv2.cvtColor(resized_frame, cv2.COLOR_BGR2RGB) resized_frame = cv2.cvtColor(resized_frame, cv2.COLOR_BGR2RGB)
pil_image = Image.fromarray(resized_frame) pil_image = Image.fromarray(resized_frame)
photo = ImageTk.PhotoImage(image=pil_image) photo = ImageTk.PhotoImage(image=pil_image)
# 保存引用到字典和列表，确保不被GC
photo_obj['img'] = photo photo_obj['img'] = photo
self._large_photos = [photo]
# 清空并绘制新图像
canvas.delete("all") canvas.delete("all")
canvas.create_image(window_width // 2, window_height // 2, image=photo, anchor='center') canvas.create_image(window_width // 2, window_height // 2, image=photo, anchor='center')
else: else:
# 显示等待提示
canvas.delete("all") canvas.delete("all")
# 显示更详细的调试信息
debug_text = f"等待画面...\n采集卡: {idx}\n帧状态: {idx in self.frames}\n"
if idx in self.frames:
debug_text += f"帧数据: {self.frames[idx] is not None}"
else:
debug_text += "帧数据: 未初始化"
canvas.create_text( canvas.create_text(
window_width // 2, window_width // 2,
window_height // 2, window_height // 2,
text="等待画面...", text=debug_text,
fill='gray', fill='gray',
font=('Arial', 16) font=('Arial', 12),
justify='center'
) )
except Exception as e: except Exception as e:
if "pyimage" not in str(e).lower(): # 忽略pyimage错误，避免刷屏 if "pyimage" not in str(e).lower():
print(f"更新大窗口错误: {e}") log_throttle("large_update_error", 1.0, f"更新大窗口错误: {e}")
self.large_window.after(33, update_large_once) # 放大視窗維持原本高頻刷新
win.after(33, update_large_once)
self.large_window.after(33, update_large_once) win.after(33, update_large_once)
def run(self): def on_close():
"""运行预览""" try:
if idx in self.large_windows:
del self.large_windows[idx]
finally:
win.destroy()
win.protocol("WM_DELETE_WINDOW", on_close)
def run(self, master=None):
"""运行预览，支持主Tk弹窗模式"""
self.init_cameras() self.init_cameras()
if not self.caps: if not self.caps:
# 如果没有加载任何采集卡，仍然显示窗口并显示错误信息
print("⚠️ 没有可用的采集卡，将显示错误提示窗口") print("⚠️ 没有可用的采集卡，将显示错误提示窗口")
# 启动捕获线程 display = self.config.get('display', {})
multi_window = bool(display.get('preview_multi_window', False))
if self.caps: if self.caps:
capture_thread = threading.Thread(target=self.capture_frames, daemon=True) capture_thread = threading.Thread(target=self.capture_frames, daemon=True)
capture_thread.start() capture_thread.start()
import time
time.sleep(0.5)
# 创建并显示网格窗口 if multi_window:
import time root = tk.Toplevel(master) if master else tk.Tk()
time.sleep(0.5) # 等待几帧 root.title("采集卡预览 - 多窗口模式")
self.create_grid_window() root.geometry("300x80")
tk.Label(root, text="已打开多窗口预览，可单独拖动/调整大小").pack(pady=20)
print("等待采集卡初始化...")
max_wait = 50
wait_count = 0
while wait_count < max_wait and not any(frame is not None for frame in self.frames.values()):
time.sleep(0.1)
wait_count += 1
if wait_count >= max_wait:
print("⚠️ 采集卡初始化超时，可能无法显示画面")
for idx in list(self.caps.keys()):
print(f"打开采集卡 {idx} 的预览窗口...")
self.show_large_window(idx)
def on_root_close():
self.running = False
for data in self.caps.values():
data['cap'].release()
try:
root.destroy()
except Exception:
pass
root.protocol("WM_DELETE_WINDOW", on_root_close)
# 仅在独立窗口（无master）时进入事件循环
if master is None:
root.mainloop()
else:
self.create_grid_window(master)
if __name__ == "__main__": if __name__ == "__main__":
preview = PreviewWindow() preview = PreviewWindow()

29
utils/get_image.py
View File

@@ -2,6 +2,8 @@ import time
from PIL import Image from PIL import Image
import cv2 import cv2
from utils.logger import logger, throttle
import logging
# class GetImage: # class GetImage:
# def __init__(self, cam_index=0, width=1920, height=1080): # def __init__(self, cam_index=0, width=1920, height=1080):
# self.cap = cv2.VideoCapture(cam_index,cv2.CAP_DSHOW) # self.cap = cv2.VideoCapture(cam_index,cv2.CAP_DSHOW)
@@ -63,7 +65,7 @@ except Exception:
class GetImage: class GetImage:
def __init__(self, cam_index=0, width=1920, height=1080): def __init__(self, cam_index=0, width=1920, height=1080):
print(f"🔧 正在初始化采集卡 {cam_index}...") logger.info(f"🔧 正在初始化采集卡 {cam_index}...")
self.cap = None self.cap = None
self.frame = None self.frame = None
self.running = True self.running = True
@@ -96,7 +98,7 @@ class GetImage:
# 测试读取一帧 # 测试读取一帧
ret, test_frame = self.cap.read() ret, test_frame = self.cap.read()
if ret and test_frame is not None: if ret and test_frame is not None:
print(f"✅ 采集卡 {cam_index} 打开成功") logger.info(f"✅ 采集卡 {cam_index} 打开成功")
break break
else: else:
self.cap.release() self.cap.release()
@@ -114,12 +116,8 @@ class GetImage:
sys.stderr = old_stderr sys.stderr = old_stderr
if self.cap is None or not self.cap.isOpened(): if self.cap is None or not self.cap.isOpened():
print(f"❌ 无法打开采集卡 {cam_index}") logger.error(f"❌ 无法打开采集卡 {cam_index}")
print("请检查：") logger.error("请检查：\n 1. 采集卡是否正确连接\n 2. 采集卡索引是否正确（尝试扫描采集卡）\n 3. 采集卡驱动是否安装\n 4. 采集卡是否被其他程序占用")
print(" 1. 采集卡是否正确连接")
print(" 2. 采集卡索引是否正确（尝试扫描采集卡）")
print(" 3. 采集卡驱动是否安装")
print(" 4. 采集卡是否被其他程序占用")
self.cap = None self.cap = None
return return
@@ -130,9 +128,9 @@ class GetImage:
# 实际获取设置后的分辨率 # 实际获取设置后的分辨率
actual_width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH)) actual_width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
actual_height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) actual_height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
print(f" 分辨率设置: {width}x{height} -> 实际: {actual_width}x{actual_height}") logger.info(f" 分辨率设置: {width}x{height} -> 实际: {actual_width}x{actual_height}")
except Exception as e: except Exception as e:
print(f"⚠️ 设置分辨率失败: {e}") logger.warning(f"⚠️ 设置分辨率失败: {e}")
# 启动更新线程 # 启动更新线程
threading.Thread(target=self.update, daemon=True).start() threading.Thread(target=self.update, daemon=True).start()
@@ -140,7 +138,7 @@ class GetImage:
# 等待几帧确保采集卡正常工作 # 等待几帧确保采集卡正常工作
import time import time
time.sleep(1.0) time.sleep(1.0)
print(f"✅ 采集卡 {cam_index} 初始化完成") logger.info(f"✅ 采集卡 {cam_index} 初始化完成")
def update(self): def update(self):
while self.running and self.cap is not None: while self.running and self.cap is not None:
@@ -148,13 +146,14 @@ class GetImage:
ret, frame = self.cap.read() ret, frame = self.cap.read()
if ret and frame is not None: if ret and frame is not None:
self.frame = frame self.frame = frame
# 限制读取频率，避免占满CPU
time.sleep(0.008)
else: else:
# 读取失败时不打印，避免刷屏 # 读取失败时不打印，避免刷屏
pass time.sleep(0.02)
except Exception as e: except Exception as e:
# 只在异常时打印错误 # 只在异常时打印错误
print(f"⚠️ 采集卡 {self.cam_index} 读取异常: {e}") throttle(f"cap_read_err_{self.cam_index}", 2.0, logging.WARNING, f"⚠️ 采集卡 {self.cam_index} 读取异常: {e}")
import time
time.sleep(0.1) # 出错时短暂延迟 time.sleep(0.1) # 出错时短暂延迟
def get_frame(self): def get_frame(self):
@@ -166,7 +165,7 @@ class GetImage:
im_PIL = Image.fromarray(im_opencv) im_PIL = Image.fromarray(im_opencv)
return [im_opencv, im_PIL] return [im_opencv, im_PIL]
except Exception as e: except Exception as e:
print(f"⚠️ 图像处理错误: {e}") throttle(f"img_proc_err_{self.cam_index}", 2.0, logging.WARNING, f"⚠️ 图像处理错误: {e}")
return None return None
def release(self): def release(self):

26
utils/logger.py Normal file
View File

@@ -0,0 +1,26 @@
import logging
import sys
import time
from typing import Dict
# 基础logger配置（控制台输出）
_logger = logging.getLogger("huojv")
if not _logger.handlers:
_logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
fmt = logging.Formatter(fmt='[%(asctime)s] %(levelname)s %(message)s', datefmt='%H:%M:%S')
handler.setFormatter(fmt)
_logger.addHandler(handler)
# 简单的节流打印：同一个key在interval秒内只打印一次
_last_log_times: Dict[str, float] = {}
def throttle(key: str, interval_sec: float, level: int, msg: str):
now = time.time()
last = _last_log_times.get(key, 0.0)
if now - last >= interval_sec:
_last_log_times[key] = now
_logger.log(level, msg)
# 对外暴露
logger = _logger

2
utils/shizi.py
View File

@@ -45,7 +45,7 @@ def tiaozhan(image):
def tuichu(image): def tuichu(image):
image = image[24:58, 569:669] image = image[36:58, 560:669]
image = cv2.cvtColor(image, cv2.COLOR_RGB2BGR) image = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
# 将裁剪后的图像编码成二进制格式 # 将裁剪后的图像编码成二进制格式
_, img_encoded = cv2.imencode('.png', image) _, img_encoded = cv2.imencode('.png', image)