Compare commits

..

12 Commits

Author SHA1 Message Date
Ray
c6ccf2052f 采集卡bug修复 2025-10-29 17:05:47 +08:00
Ray
735eb79e0d 采集卡bug修复 2025-10-29 17:05:38 +08:00
Ray
b0ff800826 采集卡bug修复 2025-10-29 16:53:34 +08:00
Ray
d3d1299323 采集卡bug修复 2025-10-29 16:47:30 +08:00
Ray
7af3e2353a 采集卡bug修复 2025-10-29 16:28:06 +08:00
Ray
f0fc28d827 采集卡bug修复 2025-10-29 15:37:48 +08:00
Ray
3a8873acc2 采集卡bug修复 2025-10-29 15:34:28 +08:00
Ray
bcc971d528 采集卡bug修复 2025-10-29 15:27:50 +08:00
Ray
f7dbf223cb 采集卡bug修复 2025-10-29 15:21:44 +08:00
Ray
cd43334957 采集卡bug修复 2025-10-29 14:36:45 +08:00
Ray
0e16ec99c3 采集卡bug修复 2025-10-29 14:06:05 +08:00
Ray
3f1dd4e8c1 ui界面版本 2025-10-29 10:49:38 +08:00
16 changed files with 3072 additions and 107 deletions

99
CHANGELOG.md Normal file
View File

@@ -0,0 +1,99 @@
# 更新日志
## 最新更新 - 多组配置系统
### 主要变更
#### 1. 新增文件
- **config.py** - 配置管理器,支持多组配置
- **gui_config.py** - 可视化配置界面tkinter
- **preview.py** - 采集卡预览窗口系统
- **launcher.py** - 统一启动器
- **README.md** - 使用文档
- **config.json** - 配置文件(自动生成)
#### 2. 修改的文件
- **main.py** - 完全重写,支持配置系统
- 从配置文件加载参数
- 支持动态读取v值
- 移除硬编码配置
- **utils/mouse.py** - 移除硬编码串口
- 新增 `init_mouse_keyboard()` 函数
- 支持从配置初始化
- **utils/get_image.py** - 支持从配置初始化
- GetImage改为延迟初始化
### 核心功能
1. **多组配置管理**
- 支持创建多个配置组
- 每个配置组独立的串口、采集卡、速度设置
- 可视化切换活动配置
2. **可视化配置界面**
- GUI界面管理所有配置
- 实时编辑和保存
- 支持添加/删除配置组
3. **采集卡预览系统**
- 网格方式预览多个采集卡
- 点击放大查看
- 实时画面更新
4. **灵活的参数配置**
- 串口配置(端口、波特率)
- 采集卡配置(索引、分辨率)
- 移动速度(v值)配置
- 预览窗口配置
### 使用方式
#### 旧版本(已保留 main_old.py
```bash
python main_old.py # 直接运行,使用硬编码配置
```
#### 新版本
```bash
# 方式1: 使用启动器(推荐)
python launcher.py
# 方式2: 直接运行
python main.py # 读取config.json的active配置
```
### 配置示例
```json
{
"groups": [
{
"name": "配置组1",
"serial_port": "COM6",
"serial_baudrate": 9600,
"camera_index": 0,
"camera_width": 1920,
"camera_height": 1080,
"move_velocity": 470, // ← move_to函数中的v值
"active": true
}
]
}
```
### 兼容性
- ✅ 保留所有原有功能
- ✅ YOLO模型无需修改
- ✅ 游戏逻辑完全兼容
- ⚠️ 旧代码会使用默认配置
### 下一步
如需添加更多配置:
1.`config.py``default_config` 中添加新字段
2.`gui_config.py` 中添加对应的输入框
3.`main.py` 中读取并使用新配置

161
PREVIEW_CONFIG_GUIDE.md Normal file
View File

@@ -0,0 +1,161 @@
# 预览配置说明
## 📖 预览配置参数说明
预览配置位于配置界面的**"预览配置"**区域,用于控制采集卡预览窗口的显示方式。
### 配置参数
#### 1. **预览宽度** (`preview_width`)
- **说明**:预览窗口的宽度(像素)
- **默认值**1000
- **建议值**
- 单个采集卡800-1200
- 2个采集卡1000-1600
- 4个采集卡1200-2000
- **用途**:控制预览窗口的整体宽度
#### 2. **预览高度** (`preview_height`)
- **说明**:预览窗口的高度(像素)
- **默认值**700
- **建议值**
- 单个采集卡600-900
- 2个采集卡700-1000
- 4个采集卡800-1200
- **用途**:控制预览窗口的整体高度
#### 3. **预览列数** (`preview_columns`)
- **说明**:预览窗口中每行显示的采集卡数量
- **默认值**2
- **范围**1-4推荐
- **用途**:决定采集卡预览的网格布局列数
- **示例**
- 2列适合2-4个采集卡
- 1列适合单个采集卡全屏显示
- 3列适合3个或6个采集卡
#### 4. **预览行数** (`preview_rows`)
- **说明**:预览窗口中每列显示的采集卡数量
- **默认值**2
- **范围**1-4推荐
- **用途**:决定采集卡预览的网格布局行数
- **示例**
- 2行配合2列可以显示4个采集卡
- 1行适合1-3个采集卡水平排列
- 3行适合3个或更多采集卡
### 📐 配置示例
#### 示例1单个采集卡全屏显示
```
预览宽度1200
预览高度900
预览列数1
预览行数1
```
**效果**:单个采集卡占据整个预览窗口
#### 示例22个采集卡并排显示
```
预览宽度1600
预览高度800
预览列数2
预览行数1
```
**效果**:两个采集卡水平并排显示
#### 示例34个采集卡网格显示2x2
```
预览宽度1600
预览高度1200
预览列数2
预览行数2
```
**效果**四个采集卡以2x2网格方式显示
#### 示例46个采集卡网格显示3x2
```
预览宽度1800
预览高度1200
预览列数3
预览行数2
```
**效果**六个采集卡以3x2网格方式显示
### 🎯 配置技巧
1. **根据采集卡数量调整**
- 采集卡数量 = 列数 × 行数
- 例如3个采集卡可以用 3列×1行 或 2列×2行最后一个位置为空
2. **根据屏幕尺寸调整**
- 确保预览窗口能够完整显示在你的屏幕上
- 如果屏幕较小,可以减小预览宽度和高度
3. **性能考虑**
- 预览窗口越大CPU占用可能越高
- 推荐在1000-1600像素宽度范围内
4. **视觉体验**
- 保持预览窗口的宽高比接近采集卡的宽高比
- 例如采集卡是1920×1080预览窗口可以是1600×900
### 🔧 如何配置
1. **打开配置界面**
```bash
python gui_config.py
```
2. **找到预览配置区域**
- 在配置界面右侧找到"预览配置"区域
3. **修改参数**
- 在对应的输入框中输入你想要的数值
- 点击"保存配置"按钮
4. **测试预览**
- 点击"启动预览"按钮查看效果
- 如果效果不理想,可以返回继续调整
### ⚠️ 注意事项
1. **字段名必须正确**
- 配置使用 `preview_width`、`preview_height`、`preview_columns`、`preview_rows`
- 不要使用旧版本的 `preview_window_width` 和 `preview_window_height`
2. **数值类型**
- 所有数值必须是整数
- 宽度和高度建议在 500-3000 像素之间
- 列数和行数建议在 1-4 之间
3. **配置文件位置**
- 配置文件保存在 `config.json` 文件中的 `display` 部分
- 修改后会立即生效(需要重启预览窗口)
### 📝 配置文件示例
```json
{
"groups": [...],
"display": {
"preview_width": 1000,
"preview_height": 700,
"preview_columns": 2,
"preview_rows": 2,
"show_preview": true
}
}
```
### 🚀 快速开始
1. 默认配置2x2网格1000x700窗口适合大多数场景
2. 如果有多个采集卡先尝试2列×2行的布局
3. 如果只有一个采集卡设置为1列×1行增大窗口尺寸
4. 根据实际效果微调,直到满意为止
---
**提示**预览窗口可以点击任意采集卡画面来放大查看。放大后的窗口会在新窗口中显示大小固定为1280×720。

141
README.md Normal file
View File

@@ -0,0 +1,141 @@
# 火炬之光自动化脚本
## 功能特点
-**多配置组管理** - 支持多组配置,可同时管理多套采集卡和串口设置
- 🖥️ **可视化配置** - 提供图形化界面管理所有配置
- 👁️ **采集卡预览** - 多窗口网格预览,点击可放大查看
- ⚙️ **灵活配置** - 所有参数(串口、采集卡、移动速度等)均可配置
- 🤖 **智能自动化** - 基于YOLO目标检测的游戏自动化
## 快速开始
### 1. 安装依赖
```bash
pip install opencv-python ultralytics ddddocr ch9329Comm serial PIL numpy
```
### 2. 启动程序
```bash
python launcher.py
```
启动器提供三个选项:
- **⚙️ 配置管理** - 配置串口、采集卡等参数
- **👁️ 采集卡预览** - 实时预览采集卡画面
- **🚀 启动自动化** - 启动游戏自动化脚本
### 3. 配置设置
首次使用请先点击"配置管理"
1. **添加配置组** - 点击"添加组"创建新配置
2. **设置参数**
- 串口:选择设备串口号(如 COM6
- 波特率设置串口波特率默认9600
- 采集卡索引:选择采集卡设备号
- 采集宽度/高度:设置采集分辨率
- 移动速度设置角色移动速度v值
3. **设为活动** - 选择要使用的配置组并设为活动
4. **保存配置** - 点击"保存配置"
### 4. 预览采集卡
点击"采集卡预览"可以:
- 网格方式查看所有活动的采集卡
- 点击任意窗口可放大查看
- 实时显示采集卡画面
### 5. 启动自动化
配置完成后,点击"启动自动化"开始游戏自动化。
## 配置说明
### 配置组结构
```json
{
"groups": [
{
"name": "配置组1",
"serial_port": "COM6",
"serial_baudrate": 9600,
"camera_index": 0,
"camera_width": 1920,
"camera_height": 1080,
"move_velocity": 470,
"active": false
}
],
"display": {
"preview_window_width": 640,
"preview_window_height": 360,
"preview_columns": 2,
"preview_rows": 2,
"show_preview": true
}
}
```
### 关键参数说明
- **serial_port**: 串口号,如 COM1, COM6
- **serial_baudrate**: 波特率,通常为 9600
- **camera_index**: 采集卡索引0/1/2...
- **camera_width/height**: 采集分辨率
- **move_velocity**: 角色移动速度v值数值越大移动越快
## 文件结构
```
huojv/
├── launcher.py # 启动器
├── main.py # 主程序(自动游戏)
├── config.py # 配置管理器
├── gui_config.py # 配置GUI界面
├── preview.py # 预览窗口
├── utils/
│ ├── get_image.py # 采集卡管理
│ ├── mouse.py # 鼠标控制
│ ├── shizi.py # OCR识别
│ └── ...
├── best.pt # YOLO模型战斗
├── best0.pt # YOLO模型城镇
└── config.json # 配置文件(自动生成)
```
## 使用流程
1. 运行 `python launcher.py`
2. 点击"配置管理"设置参数
3. 点击"采集卡预览"测试采集
4. 点击"启动自动化"开始游戏
5. 享受自动化!
## 注意事项
- 确保采集卡和串口设备已正确连接
- 串口号和采集卡索引需要根据实际情况设置
- 移动速度(v值)需要根据游戏角色属性调整
- 配置文件保存在 `config.json`
## 问题排查
### 无法打开串口
- 检查串口号是否正确
- 确认串口设备已连接
- 检查其他程序是否占用串口
### 无法打开采集卡
- 检查采集卡索引是否正确0, 1, 2...
- 确认采集卡驱动已安装
- 尝试不同的采集卡索引
### 识别不准确
- 调整移动速度(v值)
- 检查采集卡画面是否清晰
- 确认游戏窗口位置正确

22
config.json Normal file
View File

@@ -0,0 +1,22 @@
{
"groups": [
{
"name": "配置组1",
"serial_port": "COM6",
"serial_baudrate": 9600,
"camera_index": 0,
"camera_width": 1920,
"camera_height": 1080,
"move_velocity": 470,
"active": true
}
],
"display": {
"preview_width": 1000,
"preview_height": 700,
"preview_columns": 2,
"preview_rows": 2,
"show_preview": true
}
}

120
config.py Normal file
View File

@@ -0,0 +1,120 @@
import json
import os
class ConfigManager:
"""配置管理器"""
def __init__(self, config_file='config.json'):
self.config_file = config_file
self.default_config = {
"groups": [
{
"name": "配置组1",
"serial_port": "COM6",
"serial_baudrate": 9600,
"camera_index": 0,
"camera_width": 1920,
"camera_height": 1080,
"move_velocity": 470,
"active": False
}
],
"display": {
"preview_width": 1000,
"preview_height": 700,
"preview_columns": 2,
"preview_rows": 2,
"show_preview": True
}
}
self.config = self.load_config()
def load_config(self):
"""加载配置文件"""
if os.path.exists(self.config_file):
try:
with open(self.config_file, 'r', encoding='utf-8') as f:
config = json.load(f)
# 确保有默认结构
if 'groups' not in config:
config['groups'] = self.default_config['groups']
if 'display' not in config:
config['display'] = self.default_config['display']
return config
except Exception as e:
print(f"加载配置文件失败: {e}")
return self.default_config.copy()
return self.default_config.copy()
def save_config(self):
"""保存配置文件"""
try:
import os
# 确保目录存在
config_dir = os.path.dirname(os.path.abspath(self.config_file))
if config_dir and not os.path.exists(config_dir):
os.makedirs(config_dir, exist_ok=True)
with open(self.config_file, 'w', encoding='utf-8') as f:
json.dump(self.config, f, ensure_ascii=False, indent=4)
print(f"✅ 配置文件已保存到: {os.path.abspath(self.config_file)}")
return True
except Exception as e:
print(f"❌ 保存配置文件失败: {e}")
import traceback
traceback.print_exc()
return False
def add_group(self, name=None):
"""添加配置组"""
if name is None:
name = f"配置组{len(self.config['groups']) + 1}"
new_group = {
"name": name,
"serial_port": "COM6",
"serial_baudrate": 9600,
"camera_index": len(self.config['groups']),
"camera_width": 1920,
"camera_height": 1080,
"move_velocity": 470,
"active": False
}
self.config['groups'].append(new_group)
return new_group
def delete_group(self, index):
"""删除配置组"""
if 0 <= index < len(self.config['groups']):
del self.config['groups'][index]
return True
return False
def update_group(self, index, **kwargs):
"""更新配置组"""
if 0 <= index < len(self.config['groups']):
self.config['groups'][index].update(kwargs)
return True
return False
def get_active_group(self):
"""获取当前活动的配置组"""
for group in self.config['groups']:
if group.get('active', False):
return group
return None
def set_active_group(self, index):
"""设置活动的配置组"""
for i, group in enumerate(self.config['groups']):
group['active'] = (i == index)
return True
def get_group_by_index(self, index):
"""根据索引获取配置组"""
if 0 <= index < len(self.config['groups']):
return self.config['groups'][index]
return None
# 全局配置管理器实例
config_manager = ConfigManager()

534
gui_config.py Normal file
View File

@@ -0,0 +1,534 @@
import tkinter as tk
from tkinter import ttk, messagebox, simpledialog
import cv2
from config import config_manager
import threading
import serial.tools.list_ports
class ConfigGUI:
"""配置GUI界面"""
def __init__(self):
self.root = tk.Tk()
self.root.title("配置管理 - 火炬之光自动化")
self.root.geometry("1000x720")
self.root.minsize(900, 600) # 设置最小尺寸
self.selected_index = 0
self.setup_ui()
self.load_current_config()
def setup_ui(self):
"""设置UI界面"""
# 左侧:配置组列表
left_frame = ttk.Frame(self.root, padding="10")
left_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=False)
ttk.Label(left_frame, text="配置组列表", font=('Arial', 12, 'bold')).pack()
# 配置组列表
self.group_listbox = tk.Listbox(left_frame, width=20, height=25)
self.group_listbox.pack(fill=tk.BOTH, expand=True, pady=5)
self.group_listbox.bind('<<ListboxSelect>>', self.on_group_select)
# 按钮
btn_frame = ttk.Frame(left_frame)
btn_frame.pack(fill=tk.X)
ttk.Button(btn_frame, text="添加组", command=self.add_group).pack(fill=tk.X, pady=2)
ttk.Button(btn_frame, text="删除组", command=self.delete_group).pack(fill=tk.X, pady=2)
ttk.Button(btn_frame, text="设为活动", command=self.set_active).pack(fill=tk.X, pady=2)
# 右侧:配置详情
right_frame = ttk.Frame(self.root, padding="10")
right_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True)
ttk.Label(right_frame, text="配置详情", font=('Arial', 12, 'bold')).pack()
# 配置项
self.config_vars = {}
# 基本配置
basic_frame = ttk.LabelFrame(right_frame, text="基本配置", padding="10")
basic_frame.pack(fill=tk.X, pady=5)
self.create_entry(basic_frame, "name", "配置组名称:")
# 采集卡索引:使用下拉+扫描
cam_row = ttk.Frame(basic_frame)
cam_row.pack(fill=tk.X, pady=3)
ttk.Label(cam_row, text="采集卡索引:", width=15).pack(side=tk.LEFT)
self.camera_index_var = tk.StringVar()
self.camera_index_cb = ttk.Combobox(cam_row, textvariable=self.camera_index_var, width=23, state="readonly")
self.camera_index_cb.pack(side=tk.LEFT, padx=5)
ttk.Button(cam_row, text="扫描采集卡", command=self.scan_cameras).pack(side=tk.LEFT, padx=5)
# 将变量放进config_vars统一管理
self.config_vars["camera_index"] = self.camera_index_var
self.create_entry(basic_frame, "camera_width", "采集宽度:")
self.create_entry(basic_frame, "camera_height", "采集高度:")
# 串口配置
serial_frame = ttk.LabelFrame(right_frame, text="串口配置", padding="10")
serial_frame.pack(fill=tk.X, pady=5)
# 串口:使用下拉+扫描
port_row = ttk.Frame(serial_frame)
port_row.pack(fill=tk.X, pady=3)
ttk.Label(port_row, text="串口:", width=15).pack(side=tk.LEFT)
self.serial_port_var = tk.StringVar()
self.serial_port_cb = ttk.Combobox(port_row, textvariable=self.serial_port_var, width=23, state="readonly")
self.serial_port_cb.pack(side=tk.LEFT, padx=5)
ttk.Button(port_row, text="扫描串口", command=self.scan_ports).pack(side=tk.LEFT, padx=5)
# 将变量放进config_vars统一管理
self.config_vars["serial_port"] = self.serial_port_var
self.create_entry(serial_frame, "serial_baudrate", "波特率:")
# 游戏配置
game_frame = ttk.LabelFrame(right_frame, text="游戏配置", padding="10")
game_frame.pack(fill=tk.X, pady=5)
self.create_entry(game_frame, "move_velocity", "移动速度(v值):")
# 预览配置
preview_frame = ttk.LabelFrame(right_frame, text="预览配置", padding="10")
preview_frame.pack(fill=tk.X, pady=5)
self.create_entry(preview_frame, "preview_width", "预览宽度:", prefix="display")
self.create_entry(preview_frame, "preview_height", "预览高度:", prefix="display")
self.create_entry(preview_frame, "preview_columns", "预览列数:", prefix="display")
self.create_entry(preview_frame, "preview_rows", "预览行数:", prefix="display")
# 保存按钮
save_frame = ttk.Frame(right_frame)
save_frame.pack(fill=tk.X, pady=10)
ttk.Button(save_frame, text="保存配置", command=self.save_config).pack(side=tk.LEFT, padx=5)
ttk.Button(save_frame, text="启动预览", command=self.start_preview).pack(side=tk.LEFT, padx=5)
# 启动程序按钮组
start_frame = ttk.Frame(right_frame)
start_frame.pack(fill=tk.X, pady=5)
ttk.Button(start_frame, text="启动单个配置组", command=self.start_program).pack(side=tk.LEFT, padx=5)
ttk.Button(start_frame, text="启动多个配置组", command=self.start_multi_program).pack(side=tk.LEFT, padx=5)
def create_entry(self, parent, key, label, prefix=None):
"""创建输入框"""
frame = ttk.Frame(parent)
frame.pack(fill=tk.X, pady=3)
ttk.Label(frame, text=label, width=15).pack(side=tk.LEFT)
var = tk.StringVar()
entry = ttk.Entry(frame, textvariable=var, width=25)
entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5)
# 保存变量引用
if prefix:
full_key = f"{prefix}_{key}"
else:
full_key = key
self.config_vars[full_key] = var
def load_current_config(self):
"""加载当前配置到界面"""
# 加载配置组列表
self.update_group_list()
# 如果有配置组,默认选择第一个
if len(config_manager.config['groups']) > 0:
self.selected_index = 0
self.group_listbox.selection_set(0)
self.load_group_config(0)
# 加载预览配置
display = config_manager.config.get('display', {})
for key in ['preview_width', 'preview_height', 'preview_columns', 'preview_rows']:
if key in display:
self.config_vars[f"display_{key}"].set(str(display[key]))
# 初次自动扫描一次采集卡和串口
try:
self.scan_cameras()
except Exception:
pass
try:
self.scan_ports()
except Exception:
pass
def update_group_list(self):
"""更新配置组列表"""
self.group_listbox.delete(0, tk.END)
for i, group in enumerate(config_manager.config['groups']):
status = "" if group.get('active', False) else " "
self.group_listbox.insert(tk.END, f"{status} {group['name']}")
def on_group_select(self, event):
"""选择配置组"""
selection = self.group_listbox.curselection()
if selection:
self.selected_index = selection[0]
self.load_group_config(self.selected_index)
def load_group_config(self, index):
"""加载配置组详情"""
group = config_manager.get_group_by_index(index)
if group:
for key, var in self.config_vars.items():
if not key.startswith('display_'):
if key in group:
var.set(str(group[key]))
# 确保下拉框列表与当前值一致
if group and 'camera_index' in group:
# 如果当前索引不在选项里,追加
values = list(self.camera_index_cb.cget('values')) if self.camera_index_cb.cget('values') else []
display_value = f"{group['camera_index']}"
if display_value not in values:
values.append(display_value)
self.camera_index_cb['values'] = values
self.camera_index_cb.set(display_value)
# 确保串口下拉框列表与当前值一致
if group and 'serial_port' in group:
# 如果当前串口不在选项里,追加
values = list(self.serial_port_cb.cget('values')) if self.serial_port_cb.cget('values') else []
port_value = group['serial_port']
if port_value not in values:
values.append(port_value)
self.serial_port_cb['values'] = values
self.serial_port_cb.set(port_value)
def scan_cameras(self, max_index: int = 10):
"""扫描系统可用的采集卡索引,并填充下拉框"""
import warnings
import sys
import io
found = []
# 临时设置OpenCV日志级别兼容不同版本
import os
old_level = os.environ.get('OPENCV_LOG_LEVEL', '')
os.environ['OPENCV_LOG_LEVEL'] = 'SILENT' # 尝试更严格的级别
os.environ['OPENCV_IO_ENABLE_OPENEXR'] = '0'
# 尝试设置日志级别不同版本的OpenCV API不同
try:
if hasattr(cv2, 'setLogLevel'):
# OpenCV 4.x
if hasattr(cv2, 'LOG_LEVEL_SILENT'):
cv2.setLogLevel(cv2.LOG_LEVEL_SILENT)
elif hasattr(cv2, 'LOG_LEVEL_ERROR'):
cv2.setLogLevel(cv2.LOG_LEVEL_ERROR)
elif hasattr(cv2, 'utils'):
cv2.utils.setLogLevel(0) # 0=SILENT/ERROR级别
except Exception:
pass
# 重定向stderr来捕获OpenCV的错误输出
old_stderr = sys.stderr
suppressed_output = io.StringIO()
try:
# 暂时重定向stderr以抑制OpenCV的错误消息
sys.stderr = suppressed_output
for idx in range(max_index + 1):
cap = None
try:
with warnings.catch_warnings():
warnings.filterwarnings('ignore')
# 尝试DSHOW后端
cap = cv2.VideoCapture(idx, cv2.CAP_DSHOW)
if not cap.isOpened():
# 回退默认后端再试
cap = cv2.VideoCapture(idx)
if cap.isOpened():
# 测试读取一帧,确保真正可用
ret, test_frame = cap.read()
if ret and test_frame is not None:
found.append(str(idx))
cap.release()
except Exception:
if cap:
try:
cap.release()
except:
pass
continue
finally:
# 恢复stderr
sys.stderr = old_stderr
# 恢复原来的日志级别
if old_level:
os.environ['OPENCV_LOG_LEVEL'] = old_level
else:
os.environ.pop('OPENCV_LOG_LEVEL', None)
os.environ.pop('OPENCV_IO_ENABLE_OPENEXR', None)
if not found:
found = ["0"] # 至少给一个默认项,避免为空
messagebox.showwarning("扫描完成", "未发现可用采集卡,已添加默认选项 0")
else:
messagebox.showinfo("扫描完成", f"发现可用采集卡索引: {', '.join(found)}")
self.camera_index_cb['values'] = found
# 若当前无选择,则选择第一项
if not self.camera_index_var.get() and found:
self.camera_index_cb.set(found[0])
def scan_ports(self):
"""扫描系统可用的串口,并填充下拉框"""
found_real = []
try:
ports = serial.tools.list_ports.comports()
found_real = [port.device for port in ports]
# 按端口名排序
found_real.sort(key=lambda x: int(x.replace('COM', '')) if x.replace('COM', '').isdigit() else 999)
except Exception as e:
print(f"扫描串口错误: {e}")
# 如果没有发现实际端口,使用默认端口列表
if not found_real:
found = ["COM1", "COM2", "COM3", "COM4", "COM5", "COM6"] # 默认给一些常见端口
messagebox.showwarning("警告", "未发现可用串口设备,已添加常用默认选项")
else:
found = found_real
messagebox.showinfo("扫描完成", f"发现可用串口: {', '.join(found)}")
self.serial_port_cb['values'] = found
# 若当前无选择,则选择第一项
if not self.serial_port_var.get() and found:
self.serial_port_cb.set(found[0])
def add_group(self):
"""添加配置组"""
name = simpledialog.askstring("添加配置组", "请输入配置组名称:", initialvalue=f"配置组{len(config_manager.config['groups'])+1}")
if name:
config_manager.add_group(name)
self.update_group_list()
messagebox.showinfo("成功", f"已添加配置组: {name}")
def delete_group(self):
"""删除配置组"""
selection = self.group_listbox.curselection()
if not selection:
messagebox.showwarning("警告", "请先选择要删除的配置组")
return
if messagebox.askyesno("确认", "确定要删除选中的配置组吗?"):
config_manager.delete_group(selection[0])
self.update_group_list()
messagebox.showinfo("成功", "配置组已删除")
def set_active(self):
"""设置为活动配置组"""
selection = self.group_listbox.curselection()
if not selection:
messagebox.showwarning("警告", "请先选择要设为活动的配置组")
return
config_manager.set_active_group(selection[0])
self.update_group_list()
messagebox.showinfo("成功", "已设置为活动配置组")
def save_config(self):
"""保存当前编辑的配置"""
# 检查索引有效性
if self.selected_index < 0 or self.selected_index >= len(config_manager.config['groups']):
messagebox.showerror("错误", f"请先选择一个有效的配置组")
return False
# 保存当前组的配置
group = config_manager.get_group_by_index(self.selected_index)
if not group:
messagebox.showerror("错误", f"配置组不存在")
return False
for key, var in self.config_vars.items():
if not key.startswith('display_'):
try:
value_str = var.get().strip() if var.get() else ""
# 特殊处理:某些字段需要转换为整数
if key in ['camera_index', 'camera_width', 'camera_height', 'serial_baudrate', 'move_velocity']:
try:
# 如果为空,使用当前值或默认值
if value_str:
value = int(value_str)
else:
# 如果下拉框为空,尝试从当前配置获取
value = group.get(key, 0)
except ValueError:
# 转换失败,使用当前值
value = group.get(key, 0)
print(f"⚠️ 字段 {key} 的值 '{value_str}' 无效,使用当前值 {value}")
else:
# 字符串字段
value = value_str if value_str else group.get(key, '')
group[key] = value
except Exception as e:
print(f"保存字段 {key} 时出错: {e}")
import traceback
traceback.print_exc()
# 保存失败时使用当前值
if key in group:
pass # 保持原值不变
# 保存预览配置
display = config_manager.config.get('display', {})
for key in ['preview_width', 'preview_height', 'preview_columns', 'preview_rows']:
var = self.config_vars.get(f"display_{key}")
if var:
try:
display[key] = int(var.get())
except:
pass
config_manager.config['display'] = display
# 保存到文件
if config_manager.save_config():
# 更新左侧列表显示
self.update_group_list()
messagebox.showinfo("成功", "配置已保存")
return True
else:
messagebox.showerror("错误", "配置保存失败")
return False
def start_preview(self):
"""启动预览窗口"""
# 保存配置(不显示消息框,静默保存)
if not self.save_config_silent():
messagebox.showerror("错误", "配置保存失败,无法启动预览")
return
# 重新加载配置(从文件读取最新配置)
config_manager.load_config()
# 启动预览窗口
from preview import PreviewWindow
preview_thread = threading.Thread(target=lambda: PreviewWindow().run())
preview_thread.daemon = True
preview_thread.start()
def save_config_silent(self):
"""静默保存配置(不显示消息框)"""
# 保存当前组的配置
if self.selected_index < 0 or self.selected_index >= len(config_manager.config['groups']):
print(f"⚠️ 选中索引 {self.selected_index} 无效")
return False
group = config_manager.get_group_by_index(self.selected_index)
if not group:
print(f"⚠️ 配置组不存在")
return False
for key, var in self.config_vars.items():
if not key.startswith('display_'):
try:
value_str = var.get().strip() if var.get() else ""
# 特殊处理:某些字段需要转换为整数
if key in ['camera_index', 'camera_width', 'camera_height', 'serial_baudrate', 'move_velocity']:
try:
# 如果为空,使用当前值或默认值
if value_str:
value = int(value_str)
else:
# 如果下拉框为空,尝试从当前配置获取
value = group.get(key, 0)
except ValueError:
# 转换失败,使用当前值
value = group.get(key, 0)
print(f"⚠️ 字段 {key} 的值 '{value_str}' 无效,使用当前值 {value}")
else:
# 字符串字段
value = value_str if value_str else group.get(key, '')
group[key] = value
except Exception as e:
print(f"保存字段 {key} 时出错: {e}")
import traceback
traceback.print_exc()
# 保存失败时保持原值不变
# 保存预览配置
display = config_manager.config.get('display', {})
for key in ['preview_width', 'preview_height', 'preview_columns', 'preview_rows']:
var = self.config_vars.get(f"display_{key}")
if var:
try:
display[key] = int(var.get())
except:
pass
config_manager.config['display'] = display
# 保存到文件
result = config_manager.save_config()
# 更新左侧列表显示
if result:
self.update_group_list()
return result
def start_program(self):
"""启动单个配置组的主程序"""
# 保存配置(静默)
if not self.save_config_silent():
messagebox.showerror("错误", "配置保存失败")
return
# 检查是否有活动配置组
active_groups = [g for g in config_manager.config['groups'] if g.get('active', False)]
if not active_groups:
messagebox.showwarning("警告", "没有活动的配置组\n\n请先选择一个配置组并设置为活动")
return
# 启动单个配置组
import subprocess
import sys
try:
# 找到活动配置组的索引
active_group = active_groups[0]
group_index = config_manager.config['groups'].index(active_group)
subprocess.Popen([
sys.executable,
"main_single.py",
str(group_index)
], creationflags=subprocess.CREATE_NEW_CONSOLE if sys.platform == 'win32' else 0)
messagebox.showinfo("成功", f"已启动配置组: {active_group['name']}\n\n请在控制台查看运行状态")
except Exception as e:
messagebox.showerror("错误", f"启动失败: {e}")
def start_multi_program(self):
"""启动多个配置组的主程序"""
# 保存配置(静默)
if not self.save_config_silent():
messagebox.showerror("错误", "配置保存失败")
return
# 启动多配置组管理器
import subprocess
import sys
try:
subprocess.Popen([
sys.executable,
"main_multi.py"
], creationflags=subprocess.CREATE_NEW_CONSOLE if sys.platform == 'win32' else 0)
messagebox.showinfo("提示", "多配置组启动器已打开\n\n请在控制台中选择要启动的配置组")
except Exception as e:
messagebox.showerror("错误", f"启动失败: {e}")
def run(self):
"""运行GUI"""
self.root.mainloop()
if __name__ == "__main__":
app = ConfigGUI()
app.run()

79
launcher.py Normal file
View File

@@ -0,0 +1,79 @@
"""
启动器 - 选择启动配置界面或主程序
"""
import tkinter as tk
from tkinter import ttk
import subprocess
import sys
def start_config_gui():
"""启动配置界面"""
subprocess.Popen([sys.executable, "gui_config.py"])
def start_preview():
"""启动预览窗口"""
subprocess.Popen([sys.executable, "preview.py"])
def start_main_program():
"""启动主程序"""
subprocess.Popen([sys.executable, "main.py"])
def create_launcher():
"""创建启动器界面"""
root = tk.Tk()
root.title("火炬之光自动化 - 启动器")
root.geometry("400x380")
root.resizable(False, False)
# 标题
title_frame = tk.Frame(root)
title_frame.pack(fill=tk.X, pady=20)
tk.Label(title_frame, text="火炬之光自动化脚本", font=('Arial', 16, 'bold')).pack()
tk.Label(title_frame, text="请选择要启动的功能", font=('Arial', 10)).pack()
# 按钮区域
button_frame = tk.Frame(root)
button_frame.pack(fill=tk.BOTH, expand=True, padx=40, pady=20)
# 配置界面按钮
config_btn = tk.Button(button_frame, text="⚙️ 配置管理",
command=start_config_gui,
width=30, height=3, font=('Arial', 12))
config_btn.pack(pady=10)
# 预览窗口按钮
preview_btn = tk.Button(button_frame, text="👁️ 采集卡预览",
command=start_preview,
width=30, height=3, font=('Arial', 12))
preview_btn.pack(pady=10)
# 主程序按钮
main_btn = tk.Button(button_frame, text="🚀 启动单个配置组",
command=start_main_program,
width=30, height=2, font=('Arial', 11))
main_btn.pack(pady=5)
def start_multi_program():
"""启动多配置组程序"""
subprocess.Popen([sys.executable, "main_multi.py"],
creationflags=subprocess.CREATE_NEW_CONSOLE if sys.platform == 'win32' else 0)
# 多配置组启动按钮
multi_btn = tk.Button(button_frame, text="🔥 启动多个配置组",
command=start_multi_program,
width=30, height=2, font=('Arial', 11))
multi_btn.pack(pady=5)
# 说明
info_label = tk.Label(root,
text="提示:首次使用请先点击"配置管理"设置参数",
font=('Arial', 9),
fg='gray')
info_label.pack(side=tk.BOTTOM, pady=10)
root.mainloop()
if __name__ == "__main__":
create_launcher()

95
main.py
View File

@@ -1,37 +1,66 @@
import cv2
from utils.get_image import get_image
from utils.mouse import mouse_gui
from utils.get_image import GetImage
from utils.mouse import init_mouse_keyboard, Mouse_guiji
from ultralytics import YOLO
import time
import serial
import ch9329Comm
import time
import random
import math
from utils import shizi
from config import config_manager
# 加载YOLO模型
model = YOLO(r"best.pt").to('cuda')
model0 = YOLO(r"best0.pt").to('cuda')
# 从配置加载
active_group = config_manager.get_active_group()
if active_group is None:
print("❌ 没有活动的配置组请在gui_config.py中设置")
exit(1)
print(f"📋 使用配置组: {active_group['name']}")
# 初始化串口和鼠标
init_mouse_keyboard(active_group)
# 初始化键盘和鼠标
keyboard = ch9329Comm.keyboard.DataComm()
mouse = ch9329Comm.mouse.DataComm(1920, 1080)
kong_detections = {
'center': None,
'next': None,
'npc1': None,
'npc2': None,
'npc3': None,
'npc4': None,
'boss': None,
'daojv': [],
'gw': [],
'zhaozi': None
}
from utils.mouse import mouse, mouse_gui # 导入已初始化的mouse和mouse_gui
# 创建全局的mouse_gui实例
mouse_gui = Mouse_guiji()
# 初始化采集卡
get_image = GetImage(
cam_index=active_group['camera_index'],
width=active_group['camera_width'],
height=active_group['camera_height']
)
# 检查采集卡是否初始化成功
if get_image.cap is None:
print(f"❌ 采集卡 {active_group['camera_index']} 初始化失败")
print("请检查:")
print("1. 采集卡是否正确连接")
print("2. 采集卡索引是否正确")
print("3. 采集卡驱动是否安装")
exit(1)
print(f"✅ 初始化完成 - 串口:{active_group['serial_port']} 采集卡:{active_group['camera_index']}")
# 全局变量
left = 0
top = 30
k = 0 # 控制转圈的方向
panduan = False # 是否在图内
boss_pd = False # 是否到boss关卡
rw = (632, 378)
# 从配置读取移动速度
v = active_group['move_velocity']
def yolo_shibie(im_PIL, detections, model):
results = model(im_PIL) # 目标检测
for result in results:
@@ -56,9 +85,11 @@ def yolo_shibie(im_PIL,detections,model):
RW = [player_x, player_y]
detections[label] = RW
return detections
def sq(p1, p2):
"""计算两点之间的欧式距离"""
return math.sqrt((p1[0] - p2[0]) ** 2 + (p1[1] - p2[1]) ** 2)
def process_points(points):
if not points:
return None # 空列表情况
@@ -89,6 +120,7 @@ def process_points(points):
best_point = p
return best_point
def move_randomly(rw, k):
k = k % 4
suiji_t = float(random.randint(10, 13) / 10)
@@ -109,8 +141,11 @@ def move_randomly(rw, k):
time.sleep(suiji_t)
keyboard.release() # Release
return k + 1
def move_to(rw, mb):
v=470
"""使用配置的v值"""
global v
v = active_group['move_velocity'] # 每次都从配置读取最新值
if rw[0] >= mb[0]:
keyboard.send_data("44")
time.sleep(float(abs(rw[0] - mb[0]) / v))
@@ -128,7 +163,9 @@ def move_to(rw,mb):
keyboard.send_data("22")
time.sleep(float(abs(rw[1] - mb[1]) / v))
keyboard.release()
i = 0
print("🚀 开始自动化...")
while True:
detections = {
'center': None,
@@ -141,12 +178,20 @@ while True:
'daojv': [],
'gw': [],
'zhaozi': None
}
im_opencv = get_image.get_frame() # [RGB,PIL]
if im_opencv is None:
print("⚠️ 无法获取图像帧,跳过本次循环")
time.sleep(0.1)
continue
detections = yolo_shibie(im_opencv[1], detections, model)
if shizi.tuwai(im_opencv[0]): # 进图算法
im_opencv = get_image.get_frame() # [RGB,PIL]
if im_opencv is None:
print("⚠️ 无法获取图像帧,跳过本次循环")
time.sleep(0.1)
continue
detections = yolo_shibie(im_opencv[1], detections, model0)
print('当前在城镇中')
if detections['npc1'] is not None and sq(rw, detections['npc1']) > 80:
@@ -175,16 +220,21 @@ while True:
keyboard.release()
time.sleep(1)
im_opencv = get_image.get_frame() # [RGB,PIL]
if im_opencv is None:
print("⚠️ 无法获取图像帧")
continue
if shizi.daoying(im_opencv[0]):
mouse_gui.send_data_absolute(rw[0], rw[1] - 110, may=1)
time.sleep(1)
continue
else:
print("离npc4有点远 点击进入")
move_to(rw, detections['npc4'])
time.sleep(1)
im_opencv = get_image.get_frame() # [RGB,PIL]
if im_opencv is None:
print("⚠️ 无法获取图像帧")
continue
if shizi.daoying(im_opencv[0]):
mouse_gui.send_data_absolute(rw[0], rw[1] - 110, may=1)
time.sleep(1)
@@ -228,6 +278,9 @@ while True:
time.sleep(0.5)
im_opencv = get_image.get_frame() # [RGB,PIL]
if im_opencv is None:
print("⚠️ 无法获取图像帧")
continue
detections = {
'center': None,
'next': None,
@@ -239,7 +292,6 @@ while True:
'daojv': [],
'gw': [],
'zhaozi': None
}
detections = yolo_shibie(im_opencv[1], detections, model)
if detections['next'] is not None or len(detections['daojv']) != 0 or len(detections['gw']) != 0 or detections['boss'] is not None:
@@ -273,8 +325,6 @@ while True:
mb = (rw[0], rw[1] + 100)
move_to(rw, mb)
time.sleep(0.25)
continue
elif boss_pd == True and detections['center'] is None and detections['next'] is None: # boss出现了 但是没有中心
@@ -298,3 +348,4 @@ while True:
mouse_gui.send_data_absolute(rw[0], rw[1] - 110, may=1)
time.sleep(1)
continue

145
main_multi.py Normal file
View File

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

325
main_new.py Normal file
View File

@@ -0,0 +1,325 @@
import cv2
from utils.get_image import GetImage
from utils.mouse import init_mouse_keyboard, Mouse_guiji
from ultralytics import YOLO
import time
import serial
import ch9329Comm
import random
import math
from utils import shizi
from config import config_manager
# 加载YOLO模型
model = YOLO(r"best.pt").to('cuda')
model0 = YOLO(r"best0.pt").to('cuda')
# 从配置加载
active_group = config_manager.get_active_group()
if active_group is None:
print("❌ 没有活动的配置组请在gui_config.py中设置")
exit(1)
print(f"📋 使用配置组: {active_group['name']}")
# 初始化串口和鼠标
init_mouse_keyboard(active_group)
# 初始化键盘和鼠标
keyboard = ch9329Comm.keyboard.DataComm()
from utils.mouse import mouse, mouse_gui # 导入已初始化的mouse和mouse_gui
# 创建全局的mouse_gui实例
mouse_gui = Mouse_guiji()
# 初始化采集卡
get_image = GetImage(
cam_index=active_group['camera_index'],
width=active_group['camera_width'],
height=active_group['camera_height']
)
print(f"✅ 初始化完成 - 串口:{active_group['serial_port']} 采集卡:{active_group['camera_index']}")
# 全局变量
left = 0
top = 30
k = 0 # 控制转圈的方向
panduan = False # 是否在图内
boss_pd = False # 是否到boss关卡
rw = (632, 378)
# 从配置读取移动速度
v = active_group['move_velocity']
def yolo_shibie(im_PIL, detections, model):
results = model(im_PIL) # 目标检测
for result in results:
for i in range(len(result.boxes.xyxy)):
left, top, right, bottom = result.boxes.xyxy[i]
scalar_tensor = result.boxes.cls[i]
value = scalar_tensor.item()
label = result.names[int(value)]
if label == 'center' or label == 'next' or label == 'boss' or label == 'zhaozi':
player_x = int(left + (right - left) / 2)
player_y = int(top + (bottom - top) / 2) + 30
RW = [player_x, player_y]
detections[label] = RW
elif label == 'daojv' or label == 'gw':
player_x = int(left + (right - left) / 2)
player_y = int(top + (bottom - top) / 2) + 30
RW = [player_x, player_y]
detections[label].append(RW)
elif label == 'npc1' or label == 'npc2' or label == 'npc3' or label == 'npc4':
player_x = int(left + (right - left) / 2)
player_y = int(bottom) + 30
RW = [player_x, player_y]
detections[label] = RW
return detections
def sq(p1, p2):
"""计算两点之间的欧式距离"""
return math.sqrt((p1[0] - p2[0]) ** 2 + (p1[1] - p2[1]) ** 2)
def process_points(points):
if not points:
return None # 空列表情况
n = len(points)
if n == 1:
# 只有一个点,直接返回
return points[0]
elif n == 2:
# 两个点取中点
x = (points[0][0] + points[1][0]) / 2
y = (points[0][1] + points[1][1]) / 2
return [x, y]
else:
# 随机选3个点
sample_points = random.sample(points, 3)
# 对每个点计算到这3个点的总距离
min_sum = float('inf')
best_point = None
for p in points:
dist_sum = sum(sq(p, sp) for sp in sample_points)
if dist_sum < min_sum:
min_sum = dist_sum
best_point = p
return best_point
def move_randomly(rw, k):
k = k % 4
suiji_t = float(random.randint(10, 13) / 10)
if k == 0:
keyboard.send_data("66")
time.sleep(suiji_t)
keyboard.release() # Release
elif k == 1:
keyboard.send_data("88")
time.sleep(suiji_t)
keyboard.release() # Release
elif k == 2:
keyboard.send_data("44")
time.sleep(suiji_t)
keyboard.release() # Release
elif k == 3:
keyboard.send_data("22")
time.sleep(suiji_t)
keyboard.release() # Release
return k + 1
def move_to(rw, mb):
"""使用配置的v值"""
global v
v = active_group['move_velocity'] # 每次都从配置读取最新值
if rw[0] >= mb[0]:
keyboard.send_data("44")
time.sleep(float(abs(rw[0] - mb[0]) / v))
keyboard.release() # Release
else:
keyboard.send_data("66")
time.sleep(float(abs(rw[0] - mb[0]) / v))
keyboard.release() # Release
if rw[1] >= mb[1]:
keyboard.send_data("88")
time.sleep(float(abs(rw[1] - mb[1]) / v))
keyboard.release() # Release
else:
keyboard.send_data("22")
time.sleep(float(abs(rw[1] - mb[1]) / v))
keyboard.release()
i = 0
print("🚀 开始自动化...")
while True:
detections = {
'center': None,
'next': None,
'npc1': None,
'npc2': None,
'npc3': None,
'npc4': None,
'boss': None,
'daojv': [],
'gw': [],
'zhaozi': None
}
im_opencv = get_image.get_frame() # [RGB,PIL]
detections = yolo_shibie(im_opencv[1], detections, model)
if shizi.tuwai(im_opencv[0]): # 进图算法
im_opencv = get_image.get_frame() # [RGB,PIL]
detections = yolo_shibie(im_opencv[1], detections, model0)
print('当前在城镇中')
if detections['npc1'] is not None and sq(rw, detections['npc1']) > 80:
print("向npc1靠近")
print(sq(rw, detections['npc1']))
move_to(rw, detections['npc1'])
continue
elif detections['npc1'] is not None and sq(rw, detections['npc1']) <= 80:
print("在npc旁边向上走")
print(sq(rw, detections['npc1']))
mb = (detections['npc1'][0], detections['npc1'][1] - 1010)
move_to(rw, mb)
continue
elif detections['npc3'] is not None and detections['npc4'] is None:
print("在npc3旁边向右走")
mb = (rw[0], detections['npc3'][1] - 50)
move_to(rw, mb)
mb = (rw[0] + 700, rw[1])
move_to(rw, mb)
continue
elif detections['npc4'] is not None:
if sq(detections['npc4'], rw) < 50:
print("离npc4很近 直接进入")
keyboard.send_data("DD")
time.sleep(0.15)
keyboard.release()
time.sleep(1)
im_opencv = get_image.get_frame() # [RGB,PIL]
if shizi.daoying(im_opencv[0]):
mouse_gui.send_data_absolute(rw[0], rw[1] - 110, may=1)
time.sleep(1)
continue
else:
print("离npc4有点远 点击进入")
move_to(rw, detections['npc4'])
time.sleep(1)
im_opencv = get_image.get_frame() # [RGB,PIL]
if shizi.daoying(im_opencv[0]):
mouse_gui.send_data_absolute(rw[0], rw[1] - 110, may=1)
time.sleep(1)
continue
elif shizi.tiaozhan(im_opencv[0]): # 开启挑战
print('进入塔4')
mouse_gui.send_data_absolute(left + 1100, top + 600, may=1)
time.sleep(0.3)
mouse_gui.send_data_absolute(left + 433, top + 455, may=1)
panduan = True
continue
elif shizi.jieshu(im_opencv[0]): # 结束挑战
print('结束挑战')
mouse_gui.send_data_absolute(left + 542, top + 644, may=1)
time.sleep(0.8)
mouse_gui.send_data_absolute(left + 706, top + 454, may=1)
continue
elif panduan: # 图内情况
print("在图内")
if shizi.shuzi(im_opencv[0]):
boss_pd = True
print("进入到boss")
if shizi.fuhuo(im_opencv[0]):
print('点击复活')
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)
keyboard.release()
continue
if shizi.tuichu(im_opencv[0]) and detections['next'] is None and len(detections['daojv']) == 0 and len(detections['gw']) == 0 and boss_pd:
print("识别到可以退出挑战!!!!!!!!!!!!!!!!!!")
for i in range(3):
time.sleep(0.5)
im_opencv = get_image.get_frame() # [RGB,PIL]
detections = {
'center': None,
'next': None,
'npc1': None,
'npc2': None,
'npc3': None,
'npc4': None,
'boss': None,
'daojv': [],
'gw': [],
'zhaozi': None
}
detections = yolo_shibie(im_opencv[1], detections, model)
if detections['next'] is not None or len(detections['daojv']) != 0 or len(detections['gw']) != 0 or detections['boss'] is not None:
break
else:
mouse_gui.send_data_absolute(left + 640, top + 40, may=1) # 点击退出
panduan = False # 退出挑战
boss_pd = False
time.sleep(2.0)
continue
if detections['center'] is None and detections['next'] is None and (len(detections['gw']) != 0 or detections["boss"] is not None): # 识别不到中心情况 但是有怪物
print("未检测到中心点,但是有怪物")
if len(detections['gw']) != 0:
move_to(rw, detections['gw'][0])
time.sleep(0.26)
elif detections['boss'] is not None: # 跟随boss
boss_suiji1 = random.randint(-30, 30)
boss_suiji2 = random.randint(-30, 30)
detections['boss'] = (detections['boss'][0] + boss_suiji1, detections['boss'][1] + boss_suiji2)
move_to(rw, detections['boss'])
time.sleep(0.7)
continue
elif (detections['center'] is not None and detections['next'] is None and len(detections['gw']) != 0) or (boss_pd == True and detections['center'] is not None and detections['next'] is None): # 识别到中心 但是有怪物
if detections['center'][0] >= rw[0] and detections['center'][1] < rw[1]: # 3
mb = (rw[0] + 100, rw[1])
elif detections['center'][0] <= rw[0] and detections['center'][1] < rw[1]: # 4
mb = (rw[0], rw[1] - 100)
elif detections['center'][0] <= rw[0] and detections['center'][1] > rw[1]: # 1
mb = (rw[0] - 100, rw[1])
elif detections['center'][0] >= rw[0] and detections['center'][1] > rw[1]: # 2
mb = (rw[0], rw[1] + 100)
move_to(rw, mb)
time.sleep(0.25)
continue
elif boss_pd == True and detections['center'] is None and detections['next'] is None: # boss出现了 但是没有中心
k = move_randomly(rw, k)
continue
elif detections['next'] is not None:
print('进入下一层啦啦啦啦啦啦')
panduan = True
move_to(rw, detections['next'])
for i in range(2):
keyboard.send_data("DD")
time.sleep(0.15)
keyboard.release()
continue
else:
k = move_randomly(rw, k)
continue
elif shizi.daoying(im_opencv[0]):
mouse_gui.send_data_absolute(rw[0], rw[1] - 110, may=1)
time.sleep(1)
continue

300
main_old.py Normal file
View File

@@ -0,0 +1,300 @@
import cv2
from utils.get_image import get_image
from utils.mouse import mouse_gui
from ultralytics import YOLO
import time
import serial
import ch9329Comm
import time
import random
import math
from utils import shizi
model = YOLO(r"best.pt").to('cuda')
model0 = YOLO(r"best0.pt").to('cuda')
keyboard = ch9329Comm.keyboard.DataComm()
mouse = ch9329Comm.mouse.DataComm(1920, 1080)
kong_detections = {
'center': None,
'next': None,
'npc1': None,
'npc2': None,
'npc3': None,
'npc4': None,
'boss': None,
'daojv': [],
'gw': [],
'zhaozi': None
}
left=0
top=30
k=0#控制转圈的方向
panduan=False#是否在图内
boss_pd=False#是否到boss关卡
rw=(632,378)
def yolo_shibie(im_PIL,detections,model):
results = model(im_PIL)#目标检测
for result in results:
for i in range(len(result.boxes.xyxy)):
left, top, right, bottom = result.boxes.xyxy[i]
scalar_tensor = result.boxes.cls[i]
value = scalar_tensor.item()
label = result.names[int(value)]
if label=='center'or label=='next' or label=='boss' or label=='zhaozi':
player_x = int(left+(right-left)/2)
player_y = int(top+(bottom-top)/2)+30
RW = [player_x, player_y]
detections[label] = RW
elif label=='daojv' or label=='gw':
player_x = int(left + (right - left) / 2)
player_y = int(top + (bottom - top) / 2) + 30
RW = [player_x, player_y]
detections[label].append(RW)
elif label=='npc1' or label=='npc2' or label=='npc3' or label=='npc4':
player_x = int(left+(right-left)/2)
player_y = int(bottom)+30
RW = [player_x, player_y]
detections[label] = RW
return detections
def sq(p1, p2):
"""计算两点之间的欧式距离"""
return math.sqrt((p1[0] - p2[0]) ** 2 + (p1[1] - p2[1]) ** 2)
def process_points(points):
if not points:
return None # 空列表情况
n = len(points)
if n == 1:
# 只有一个点,直接返回
return points[0]
elif n == 2:
# 两个点取中点
x = (points[0][0] + points[1][0]) / 2
y = (points[0][1] + points[1][1]) / 2
return [x, y]
else:
# 随机选3个点
sample_points = random.sample(points, 3)
# 对每个点计算到这3个点的总距离
min_sum = float('inf')
best_point = None
for p in points:
dist_sum = sum(sq(p, sp) for sp in sample_points)
if dist_sum < min_sum:
min_sum = dist_sum
best_point = p
return best_point
def move_randomly(rw, k):
k = k % 4
suiji_t=float(random.randint(10,13)/10)
if k == 0:
keyboard.send_data("66")
time.sleep(suiji_t)
keyboard.release() # Release
elif k == 1:
keyboard.send_data("88")
time.sleep(suiji_t)
keyboard.release() # Release
elif k == 2:
keyboard.send_data("44")
time.sleep(suiji_t)
keyboard.release() # Release
elif k == 3:
keyboard.send_data("22")
time.sleep(suiji_t)
keyboard.release() # Release
return k + 1
def move_to(rw,mb):
v=470
if rw[0]>=mb[0]:
keyboard.send_data("44")
time.sleep(float(abs(rw[0]-mb[0])/v))
keyboard.release() # Release
else:
keyboard.send_data("66")
time.sleep(float(abs(rw[0] - mb[0]) / v))
keyboard.release() # Release
if rw[1]>=mb[1]:
keyboard.send_data("88")
time.sleep(float(abs(rw[1] - mb[1]) / v))
keyboard.release() # Release
else:
keyboard.send_data("22")
time.sleep(float(abs(rw[1] - mb[1]) / v))
keyboard.release()
i=0
while True:
detections = {
'center': None,
'next': None,
'npc1': None,
'npc2': None,
'npc3': None,
'npc4': None,
'boss': None,
'daojv': [],
'gw': [],
'zhaozi':None
}
im_opencv = get_image.get_frame()#[RGB,PIL]
detections=yolo_shibie(im_opencv[1],detections,model)
if shizi.tuwai(im_opencv[0]): # 进图算法
im_opencv = get_image.get_frame() # [RGB,PIL]
detections = yolo_shibie(im_opencv[1], detections, model0)
print('当前在城镇中')
if detections['npc1'] is not None and sq(rw, detections['npc1']) > 80:
print("向npc1靠近")
print(sq(rw, detections['npc1']))
move_to(rw, detections['npc1'])
continue
elif detections['npc1'] is not None and sq(rw, detections['npc1']) <= 80:
print("在npc旁边向上走")
print(sq(rw, detections['npc1']))
mb = (detections['npc1'][0], detections['npc1'][1] - 1010)
move_to(rw, mb)
continue
elif detections['npc3'] is not None and detections['npc4'] is None:
print("在npc3旁边向右走")
mb = (rw[0], detections['npc3'][1]-50)
move_to(rw, mb)
mb = (rw[0] + 700, rw[1])
move_to(rw, mb)
continue
elif detections['npc4'] is not None:
if sq(detections['npc4'], rw) < 50:
print("离npc4很近 直接进入")
keyboard.send_data("DD")
time.sleep(0.15)
keyboard.release()
time.sleep(1)
im_opencv = get_image.get_frame() # [RGB,PIL]
if shizi.daoying(im_opencv[0]):
mouse_gui.send_data_absolute(rw[0], rw[1] - 110, may=1)
time.sleep(1)
continue
else:
print("离npc4有点远 点击进入")
move_to(rw, detections['npc4'])
time.sleep(1)
im_opencv = get_image.get_frame() # [RGB,PIL]
if shizi.daoying(im_opencv[0]):
mouse_gui.send_data_absolute(rw[0], rw[1] - 110, may=1)
time.sleep(1)
continue
elif shizi.tiaozhan(im_opencv[0]):#开启挑战
print('进入塔4')
mouse_gui.send_data_absolute(left+1100,top+600,may=1)
time.sleep(0.3)
mouse_gui.send_data_absolute(left + 433, top + 455,may=1)
panduan = True
continue
elif shizi.jieshu(im_opencv[0]):#结束挑战
print('结束挑战')
mouse_gui.send_data_absolute(left+542,top+644,may=1)
time.sleep(0.8)
mouse_gui.send_data_absolute(left + 706, top + 454,may=1)
continue
elif panduan :#图内情况
print("在图内")
if shizi.shuzi(im_opencv[0]) :
boss_pd = True
print("进入到boss")
if shizi.fuhuo(im_opencv[0]):
print('点击复活')
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)
keyboard.release()
continue
if shizi.tuichu(im_opencv[0]) and detections['next'] is None and len(detections['daojv'])==0 and len(detections['gw'])==0 and boss_pd:
print("识别到可以退出挑战!!!!!!!!!!!!!!!!!!")
for i in range(3):
time.sleep(0.5)
im_opencv = get_image.get_frame()#[RGB,PIL]
detections = {
'center': None,
'next': None,
'npc1': None,
'npc2': None,
'npc3': None,
'npc4': None,
'boss': None,
'daojv': [],
'gw': [],
'zhaozi': None
}
detections = yolo_shibie(im_opencv[1], detections,model)
if detections['next'] is not None or len(detections['daojv'])!=0 or len(detections['gw'])!=0 or detections['boss'] is not None:
break
else:
mouse_gui.send_data_absolute(left + 640, top + 40,may=1)#点击退出
panduan = False#退出挑战
boss_pd = False
time.sleep(2.0)
continue
if detections['center'] is None and detections['next'] is None and (len(detections['gw'])!=0 or detections["boss"] is not None):#识别不到中心情况 但是有怪物
print("未检测到中心点,但是有怪物")
if len(detections['gw'])!=0:
move_to(rw,detections['gw'][0])
time.sleep(0.26)
elif detections['boss'] is not None:#跟随boss
boss_suiji1=random.randint(-30,30)
boss_suiji2 = random.randint(-30, 30)
detections['boss']=(detections['boss'][0]+boss_suiji1,detections['boss'][1]+boss_suiji2)
move_to(rw,detections['boss'])
time.sleep(0.7)
continue
elif (detections['center'] is not None and detections['next'] is None and len(detections['gw'])!=0) or (boss_pd==True and detections['center'] is not None and detections['next'] is None) :#识别到中心 但是有怪物
if detections['center'][0]>=rw[0] and detections['center'][1]<rw[1]:#3
mb=(rw[0]+100,rw[1])
elif detections['center'][0]<=rw[0] and detections['center'][1]<rw[1]:#4
mb=(rw[0], rw[1]-100)
elif detections['center'][0]<=rw[0] and detections['center'][1]>rw[1]:#1
mb=(rw[0]-100, rw[1])
elif detections['center'][0]>=rw[0] and detections['center'][1]>rw[1]:#2
mb=(rw[0], rw[1]+100)
move_to(rw, mb)
time.sleep(0.25)
continue
elif boss_pd==True and detections['center'] is None and detections['next'] is None:#boss出现了 但是没有中心
k=move_randomly(rw,k)
continue
elif detections['next'] is not None:
print('进入下一层啦啦啦啦啦啦')
panduan = True
move_to(rw,detections['next'])
for i in range(2):
keyboard.send_data("DD")
time.sleep(0.15)
keyboard.release()
continue
else:
k = move_randomly(rw, k)
continue
elif shizi.daoying(im_opencv[0]):
mouse_gui.send_data_absolute(rw[0], rw[1] - 110, may=1)
time.sleep(1)
continue

349
main_single.py Normal file
View File

@@ -0,0 +1,349 @@
"""
单配置组运行的主程序
用于在独立进程中运行单个配置组的自动化
"""
import cv2
from utils.get_image import GetImage
from utils.mouse import init_mouse_keyboard, Mouse_guiji
from ultralytics import YOLO
import time
import serial
import ch9329Comm
import random
import math
from utils import shizi
from config import config_manager
import sys
def run_automation_for_group(group_index):
"""为单个配置组运行自动化"""
# 重新加载配置
config_manager.load_config()
group = config_manager.get_group_by_index(group_index)
if group is None:
print(f"❌ 配置组索引 {group_index} 不存在")
return
print(f"🚀 启动配置组: {group['name']} (索引: {group_index})")
print(f" 串口: {group['serial_port']}")
print(f" 采集卡: {group['camera_index']}")
# 加载YOLO模型
model = YOLO(r"best.pt").to('cuda')
model0 = YOLO(r"best0.pt").to('cuda')
# 初始化串口和鼠标
try:
init_mouse_keyboard(group)
except Exception as e:
print(f"❌ 初始化串口失败: {e}")
return
# 初始化键盘和鼠标
keyboard = ch9329Comm.keyboard.DataComm()
from utils.mouse import mouse
mouse_gui = Mouse_guiji()
# 初始化采集卡
get_image = GetImage(
cam_index=group['camera_index'],
width=group['camera_width'],
height=group['camera_height']
)
# 检查采集卡是否初始化成功
if get_image.cap is None:
print(f"❌ 采集卡 {group['camera_index']} 初始化失败")
return
print(f"✅ 配置组 {group['name']} 初始化完成")
# 全局变量
left = 0
top = 30
k = 0 # 控制转圈的方向
panduan = False # 是否在图内
boss_pd = False # 是否到boss关卡
rw = (632, 378)
v = group['move_velocity'] # 从配置读取移动速度
def yolo_shibie(im_PIL, detections, model):
results = model(im_PIL)
for result in results:
for i in range(len(result.boxes.xyxy)):
left, top, right, bottom = result.boxes.xyxy[i]
scalar_tensor = result.boxes.cls[i]
value = scalar_tensor.item()
label = result.names[int(value)]
if label == 'center' or label == 'next' or label == 'boss' or label == 'zhaozi':
player_x = int(left + (right - left) / 2)
player_y = int(top + (bottom - top) / 2) + 30
RW = [player_x, player_y]
detections[label] = RW
elif label == 'daojv' or label == 'gw':
player_x = int(left + (right - left) / 2)
player_y = int(top + (bottom - top) / 2) + 30
RW = [player_x, player_y]
detections[label].append(RW)
elif label == 'npc1' or label == 'npc2' or label == 'npc3' or label == 'npc4':
player_x = int(left + (right - left) / 2)
player_y = int(bottom) + 30
RW = [player_x, player_y]
detections[label] = RW
return detections
def sq(p1, p2):
"""计算两点之间的欧式距离"""
return math.sqrt((p1[0] - p2[0]) ** 2 + (p1[1] - p2[1]) ** 2)
def process_points(points):
if not points:
return None
n = len(points)
if n == 1:
return points[0]
elif n == 2:
x = (points[0][0] + points[1][0]) / 2
y = (points[0][1] + points[1][1]) / 2
return [x, y]
else:
sample_points = random.sample(points, 3)
min_sum = float('inf')
best_point = None
for p in points:
dist_sum = sum(sq(p, sp) for sp in sample_points)
if dist_sum < min_sum:
min_sum = dist_sum
best_point = p
return best_point
def move_randomly(rw, k):
k = k % 4
suiji_t = float(random.randint(10, 13) / 10)
if k == 0:
keyboard.send_data("66")
time.sleep(suiji_t)
keyboard.release()
elif k == 1:
keyboard.send_data("88")
time.sleep(suiji_t)
keyboard.release()
elif k == 2:
keyboard.send_data("44")
time.sleep(suiji_t)
keyboard.release()
elif k == 3:
keyboard.send_data("22")
time.sleep(suiji_t)
keyboard.release()
return k + 1
def move_to(rw, mb):
"""使用配置的v值"""
nonlocal v
v = group['move_velocity']
if rw[0] >= mb[0]:
keyboard.send_data("44")
time.sleep(float(abs(rw[0] - mb[0]) / v))
keyboard.release()
else:
keyboard.send_data("66")
time.sleep(float(abs(rw[0] - mb[0]) / v))
keyboard.release()
if rw[1] >= mb[1]:
keyboard.send_data("88")
time.sleep(float(abs(rw[1] - mb[1]) / v))
keyboard.release()
else:
keyboard.send_data("22")
time.sleep(float(abs(rw[1] - mb[1]) / v))
keyboard.release()
# 主循环
print(f"🔄 配置组 {group['name']} 开始自动化循环...")
while True:
try:
detections = {
'center': None,
'next': None,
'npc1': None,
'npc2': None,
'npc3': None,
'npc4': None,
'boss': None,
'daojv': [],
'gw': [],
'zhaozi': None
}
im_opencv = get_image.get_frame()
if im_opencv is None:
time.sleep(0.1)
continue
detections = yolo_shibie(im_opencv[1], detections, model)
if shizi.tuwai(im_opencv[0]):
im_opencv = get_image.get_frame()
if im_opencv is None:
time.sleep(0.1)
continue
detections = yolo_shibie(im_opencv[1], detections, model0)
print(f'[{group["name"]}] 当前在城镇中')
if detections['npc1'] is not None and sq(rw, detections['npc1']) > 80:
print(f"[{group['name']}] 向npc1靠近")
move_to(rw, detections['npc1'])
continue
elif detections['npc1'] is not None and sq(rw, detections['npc1']) <= 80:
print(f"[{group['name']}] 在npc旁边向上走")
mb = (detections['npc1'][0], detections['npc1'][1] - 1010)
move_to(rw, mb)
continue
elif detections['npc3'] is not None and detections['npc4'] is None:
print(f"[{group['name']}] 在npc3旁边向右走")
mb = (rw[0], detections['npc3'][1] - 50)
move_to(rw, mb)
mb = (rw[0] + 700, rw[1])
move_to(rw, mb)
continue
elif detections['npc4'] is not None:
if sq(detections['npc4'], rw) < 50:
print(f"[{group['name']}] 离npc4很近 直接进入")
keyboard.send_data("DD")
time.sleep(0.15)
keyboard.release()
time.sleep(1)
im_opencv = get_image.get_frame()
if im_opencv and shizi.daoying(im_opencv[0]):
mouse_gui.send_data_absolute(rw[0], rw[1] - 110, may=1)
time.sleep(1)
continue
else:
print(f"[{group['name']}] 离npc4有点远 点击进入")
move_to(rw, detections['npc4'])
time.sleep(1)
im_opencv = get_image.get_frame()
if im_opencv and shizi.daoying(im_opencv[0]):
mouse_gui.send_data_absolute(rw[0], rw[1] - 110, may=1)
time.sleep(1)
continue
elif shizi.tiaozhan(im_opencv[0]):
print(f'[{group["name"]}] 进入塔4')
mouse_gui.send_data_absolute(left + 1100, top + 600, may=1)
time.sleep(0.3)
mouse_gui.send_data_absolute(left + 433, top + 455, may=1)
panduan = True
continue
elif shizi.jieshu(im_opencv[0]):
print(f'[{group["name"]}] 结束挑战')
mouse_gui.send_data_absolute(left + 542, top + 644, may=1)
time.sleep(0.8)
mouse_gui.send_data_absolute(left + 706, top + 454, may=1)
continue
elif panduan:
if shizi.shuzi(im_opencv[0]):
boss_pd = True
print(f"[{group['name']}] 进入到boss")
if shizi.fuhuo(im_opencv[0]):
print(f'[{group["name"]}] 点击复活')
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)
keyboard.release()
continue
if shizi.tuichu(im_opencv[0]) and detections['next'] is None and len(detections['daojv']) == 0 and len(detections['gw']) == 0 and boss_pd:
print(f"[{group['name']}] 识别到可以退出挑战!!!!!!!!!!!!!!!!!!")
for i in range(3):
time.sleep(0.5)
im_opencv = get_image.get_frame()
if im_opencv is None:
continue
detections = {
'center': None, 'next': None,
'npc1': None, 'npc2': None, 'npc3': None, 'npc4': None,
'boss': None, 'zhaozi': None,
'daojv': [], 'gw': []
}
detections = yolo_shibie(im_opencv[1], detections, model)
if detections['next'] is not None or len(detections['daojv']) != 0 or len(detections['gw']) != 0 or detections['boss'] is not None:
break
else:
mouse_gui.send_data_absolute(left + 640, top + 40, may=1)
panduan = False
boss_pd = False
time.sleep(2.0)
continue
if detections['center'] is None and detections['next'] is None and (len(detections['gw']) != 0 or detections["boss"] is not None):
if len(detections['gw']) != 0:
move_to(rw, detections['gw'][0])
time.sleep(0.26)
elif detections['boss'] is not None:
boss_suiji1 = random.randint(-30, 30)
boss_suiji2 = random.randint(-30, 30)
detections['boss'] = (detections['boss'][0] + boss_suiji1, detections['boss'][1] + boss_suiji2)
move_to(rw, detections['boss'])
time.sleep(0.7)
continue
elif (detections['center'] is not None and detections['next'] is None and len(detections['gw']) != 0) or (boss_pd == True and detections['center'] is not None and detections['next'] is None):
if detections['center'][0] >= rw[0] and detections['center'][1] < rw[1]:
mb = (rw[0] + 100, rw[1])
elif detections['center'][0] <= rw[0] and detections['center'][1] < rw[1]:
mb = (rw[0], rw[1] - 100)
elif detections['center'][0] <= rw[0] and detections['center'][1] > rw[1]:
mb = (rw[0] - 100, rw[1])
elif detections['center'][0] >= rw[0] and detections['center'][1] > rw[1]:
mb = (rw[0], rw[1] + 100)
move_to(rw, mb)
time.sleep(0.25)
continue
elif boss_pd == True and detections['center'] is None and detections['next'] is None:
k = move_randomly(rw, k)
continue
elif detections['next'] is not None:
print(f'[{group["name"]}] 进入下一层啦啦啦啦啦啦')
panduan = True
move_to(rw, detections['next'])
for i in range(2):
keyboard.send_data("DD")
time.sleep(0.15)
keyboard.release()
continue
else:
k = move_randomly(rw, k)
continue
elif shizi.daoying(im_opencv[0]):
mouse_gui.send_data_absolute(rw[0], rw[1] - 110, may=1)
time.sleep(1)
continue
except KeyboardInterrupt:
print(f"🛑 配置组 {group['name']} 停止运行")
break
except Exception as e:
print(f"❌ 配置组 {group['name']} 运行错误: {e}")
import traceback
traceback.print_exc()
time.sleep(1)
if __name__ == "__main__":
if len(sys.argv) > 1:
group_index = int(sys.argv[1])
run_automation_for_group(group_index)
else:
# 默认运行活动配置组
active_group = config_manager.get_active_group()
if active_group is None:
print("❌ 没有活动的配置组")
sys.exit(1)
group_index = config_manager.config['groups'].index(active_group)
run_automation_for_group(group_index)

447
preview.py Normal file
View File

@@ -0,0 +1,447 @@
import cv2
import tkinter as tk
from tkinter import Canvas
from PIL import Image, ImageTk
import threading
import numpy as np
import warnings
import os
from config import config_manager
# 抑制OpenCV的警告信息兼容不同版本
import sys
import io
os.environ['OPENCV_LOG_LEVEL'] = 'SILENT'
os.environ['OPENCV_IO_ENABLE_OPENEXR'] = '0'
try:
if hasattr(cv2, 'setLogLevel'):
if hasattr(cv2, 'LOG_LEVEL_SILENT'):
cv2.setLogLevel(cv2.LOG_LEVEL_SILENT)
elif hasattr(cv2, 'LOG_LEVEL_ERROR'):
cv2.setLogLevel(cv2.LOG_LEVEL_ERROR)
elif hasattr(cv2, 'utils'):
cv2.utils.setLogLevel(0)
except Exception:
pass
class PreviewWindow:
"""采集卡预览窗口"""
def __init__(self):
self.config = config_manager.config
self.caps = {}
self.frames = {}
self.large_window = None
self.running = True
def init_cameras(self):
"""初始化所有相机"""
print("🔧 开始初始化采集卡...")
loaded_count = 0
# 如果没有活动配置,加载所有配置
active_groups = [g for g in self.config['groups'] if g.get('active', True)]
if not active_groups:
print("⚠️ 没有活动的配置组,将尝试加载所有配置组")
active_groups = self.config['groups']
# 重定向stderr来抑制OpenCV的错误输出
old_stderr = sys.stderr
suppressed_output = io.StringIO()
try:
sys.stderr = suppressed_output
for i, group in enumerate(active_groups):
try:
cam_idx = group['camera_index']
print(f" 尝试打开采集卡 {cam_idx} ({group['name']})...")
cap = None
# 尝试多种后端打开
backends_to_try = [
(int(cam_idx), cv2.CAP_DSHOW),
(int(cam_idx), cv2.CAP_ANY),
(int(cam_idx), None),
]
for idx, backend in backends_to_try:
try:
with warnings.catch_warnings():
warnings.filterwarnings('ignore')
if backend is not None:
cap = cv2.VideoCapture(idx, backend)
else:
cap = cv2.VideoCapture(idx)
if cap.isOpened():
# 测试读取一帧
ret, test_frame = cap.read()
if ret and test_frame is not None:
break
else:
cap.release()
cap = None
except Exception:
if cap:
try:
cap.release()
except:
pass
cap = None
continue
if cap and cap.isOpened():
try:
cap.set(cv2.CAP_PROP_FRAME_WIDTH, group['camera_width'])
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, group['camera_height'])
except Exception as e:
print(f" ⚠️ 设置分辨率失败: {e}")
self.caps[i] = {
'cap': cap,
'group': group,
'name': group['name']
}
loaded_count += 1
print(f" ✅ 采集卡 {cam_idx} 初始化成功")
else:
print(f" ❌ 采集卡 {cam_idx} 无法打开")
except Exception as e:
print(f" ❌ 采集卡 {group.get('camera_index', '?')} 初始化失败: {e}")
import traceback
traceback.print_exc()
finally:
# 恢复stderr
sys.stderr = old_stderr
if loaded_count == 0:
print("⚠️ 警告:没有成功加载任何采集卡!")
print("请检查:")
print("1. 采集卡是否正确连接")
print("2. 采集卡索引是否正确")
print("3. 是否有活动的配置组")
else:
print(f"✅ 成功加载 {loaded_count} 个采集卡")
def capture_frames(self):
"""捕获帧"""
while self.running:
for idx, data in self.caps.items():
try:
ret, frame = data['cap'].read()
if ret and frame is not None:
# 裁剪到配置的区域
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 and crop_width > 0:
frame = frame[crop_top:crop_bottom, 0:crop_width]
# 转换颜色空间
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
self.frames[idx] = frame_rgb
else:
print(f"⚠️ 采集卡 {idx} 裁剪参数无效")
else:
# 读取失败,清除旧帧
if idx in self.frames:
self.frames[idx] = None
except Exception as e:
print(f"捕获帧 {idx} 错误: {e}")
import traceback
traceback.print_exc()
import time
time.sleep(0.01) # 避免CPU占用过高
def create_grid_window(self):
"""创建网格窗口"""
# 获取显示配置
display = self.config.get('display', {})
preview_width = display.get('preview_width', 1000)
preview_height = display.get('preview_height', 700)
columns = display.get('preview_columns', 2)
rows = display.get('preview_rows', 2)
root = tk.Tk()
root.title("采集卡预览 - 点击放大")
root.geometry(f"{preview_width}x{preview_height}")
canvas = Canvas(root, bg='black')
canvas.pack(fill=tk.BOTH, expand=True)
# 存储图像对象
self.photo_objects = {}
def update_frames_once():
"""在主线程中更新一帧使用after循环"""
if not self.running:
return
try:
# 如果没有加载任何采集卡,显示提示
if not self.caps:
canvas.delete("all")
canvas.create_text(
root.winfo_width() // 2,
root.winfo_height() // 2,
text="未找到可用的采集卡\n\n请检查:\n1. 采集卡是否正确连接\n2. 配置组是否有活动的采集卡\n3. 采集卡索引是否正确",
fill='yellow',
font=('Arial', 14),
justify=tk.CENTER
)
root.after(33, update_frames_once)
return
# 计算每个预览窗口的位置和大小
# 确保画布已完全初始化
root.update_idletasks()
canvas.update_idletasks()
# 获取画布实际尺寸
canvas_width = canvas.winfo_width()
canvas_height = canvas.winfo_height()
# 如果画布还没初始化,使用默认值
if canvas_width <= 1 or canvas_height <= 1:
canvas_width = 1000
canvas_height = 700
# 确保尺寸有效
if canvas_width < 100 or canvas_height < 100:
canvas_width = 1000
canvas_height = 700
cell_width = max(10, canvas_width // columns)
cell_height = max(10, canvas_height // rows)
# 先收集所有需要显示的图像
images_to_draw = []
texts_to_draw = []
frame_idx = 0
for idx in self.caps.keys():
row = frame_idx // columns
col = frame_idx % columns
x = col * cell_width
y = row * cell_height
center_x = x + cell_width // 2
center_y = y + cell_height // 2
name = self.caps[idx]['name']
# 检查帧数据
has_frame = idx in self.frames and self.frames[idx] is not None
if has_frame:
# 调整图像大小
try:
frame = self.frames[idx]
h, w = frame.shape[:2]
# 确保尺寸有效
if w <= 0 or h <= 0:
texts_to_draw.append((center_x, center_y, f"{name}\n尺寸无效", 'red'))
frame_idx += 1
continue
# 计算缩放比例,留一些边距
scale = min(cell_width / w, cell_height / h, 1.0) * 0.85
new_w = max(1, int(w * scale))
new_h = max(1, int(h * scale))
# 确保缩放后的尺寸有效
if new_w <= 0 or new_h <= 0:
texts_to_draw.append((center_x, center_y, f"{name}\n缩放失败", 'red'))
frame_idx += 1
continue
# 调试输出(仅前几次)
if frame_idx == 0 and len(images_to_draw) == 0:
print(f"🔍 预览调试: 画布={canvas_width}x{canvas_height}, 单元格={cell_width}x{cell_height}, 原始帧={w}x{h}, 缩放后={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)
photo = ImageTk.PhotoImage(image=pil_image)
# 保持引用避免被GC
self.photo_objects[idx] = photo
images_to_draw.append((photo, center_x, center_y))
texts_to_draw.append((center_x, y + 15, name, 'white'))
frame_idx += 1
except Exception as e:
# 忽略pyimage相关错误避免刷屏
if "pyimage" not in str(e).lower():
print(f"处理帧 {idx} 错误: {e}")
import traceback
traceback.print_exc()
# 如果处理失败,显示等待提示
texts_to_draw.append((center_x, center_y, f"{name}\n处理失败", 'red'))
frame_idx += 1
else:
# 显示等待提示
texts_to_draw.append((center_x, center_y, f"{name}\n等待画面...", 'gray'))
frame_idx += 1
# 清空画布
canvas.delete("all")
# 先绘制所有图像(底层)
if images_to_draw:
if frame_idx == len(images_to_draw):
print(f"✅ 准备绘制 {len(images_to_draw)} 个图像")
for photo, x, y in images_to_draw:
try:
canvas.create_image(x, y, image=photo, anchor='center')
except Exception as e:
if "pyimage" not in str(e).lower():
print(f"绘制图像错误: {e}")
# 再绘制所有文本(上层)
for x, y, text, color in texts_to_draw:
try:
if color == 'white':
canvas.create_text(x, y, text=text, fill=color, font=('Arial', 10, 'bold'))
else:
canvas.create_text(x, y, text=text, fill=color, font=('Arial', 10))
except Exception as e:
print(f"绘制文本错误: {e}")
except Exception as e:
print(f"更新帧错误: {e}")
import traceback
traceback.print_exc()
# 约30fps
root.after(33, update_frames_once)
def on_canvas_click(event):
"""点击画布事件"""
window_width = root.winfo_width()
window_height = root.winfo_height()
cell_width = window_width // columns
cell_height = window_height // rows
col = int(event.x // cell_width)
row = int(event.y // cell_height)
index = row * columns + col
# 找到对应的配置
if index < len(self.caps):
idx = list(self.caps.keys())[index]
self.show_large_window(idx)
canvas.bind('<Button-1>', on_canvas_click)
# 等待窗口完全初始化后再开始更新
def start_updates():
"""延迟启动更新,确保窗口已完全显示"""
root.update_idletasks()
root.update()
update_frames_once()
# 使用after在主线程中循环刷新延迟启动
root.after(100, start_updates)
def on_closing():
"""关闭窗口"""
self.running = False
for data in self.caps.values():
data['cap'].release()
root.destroy()
root.protocol("WM_DELETE_WINDOW", on_closing)
root.mainloop()
def show_large_window(self, idx):
"""显示大窗口"""
if self.large_window is not None and self.large_window.winfo_exists():
self.large_window.destroy()
self.large_window = tk.Toplevel()
self.large_window.title(f"放大视图 - {self.caps[idx]['name']}")
self.large_window.geometry("1280x720")
canvas = Canvas(self.large_window, bg='black')
canvas.pack(fill=tk.BOTH, expand=True)
photo_obj = {}
def update_large_once():
if not self.running or not self.large_window.winfo_exists():
return
try:
# 获取窗口大小
window_width = self.large_window.winfo_width() if self.large_window.winfo_width() > 1 else 1280
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:
# 先处理图像,再清空画布
frame = self.frames[idx]
h, w = frame.shape[:2]
# 调整到窗口大小
scale = min(window_width / w, window_height / h)
new_w = int(w * scale)
new_h = int(h * scale)
resized_frame = cv2.resize(frame, (new_w, new_h))
# 确保颜色通道正确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)
photo = ImageTk.PhotoImage(image=pil_image)
# 保存引用到字典确保不被GC
photo_obj['img'] = photo
# 清空并绘制新图像
canvas.delete("all")
canvas.create_image(window_width // 2, window_height // 2, image=photo, anchor='center')
else:
# 显示等待提示
canvas.delete("all")
canvas.create_text(
window_width // 2,
window_height // 2,
text="等待画面...",
fill='gray',
font=('Arial', 16)
)
except Exception as e:
if "pyimage" not in str(e).lower(): # 忽略pyimage错误避免刷屏
print(f"更新大窗口错误: {e}")
self.large_window.after(33, update_large_once)
self.large_window.after(33, update_large_once)
def run(self):
"""运行预览"""
self.init_cameras()
if not self.caps:
# 如果没有加载任何采集卡,仍然显示窗口并显示错误信息
print("⚠️ 没有可用的采集卡,将显示错误提示窗口")
# 启动捕获线程
if self.caps:
capture_thread = threading.Thread(target=self.capture_frames, daemon=True)
capture_thread.start()
# 创建并显示网格窗口
import time
time.sleep(0.5) # 等待几帧
self.create_grid_window()
if __name__ == "__main__":
preview = PreviewWindow()
preview.run()

67
test_camera.py Normal file
View File

@@ -0,0 +1,67 @@
"""
采集卡测试脚本
用于测试采集卡是否能正常工作
"""
import cv2
from utils.get_image import GetImage
from config import config_manager
def test_camera():
"""测试采集卡"""
print("🔧 采集卡测试工具")
print("=" * 50)
# 获取活动配置
active_group = config_manager.get_active_group()
if active_group is None:
print("❌ 没有活动的配置组")
print("请先运行 python gui_config.py 设置配置")
return
print(f"📋 使用配置: {active_group['name']}")
print(f" 采集卡索引: {active_group['camera_index']}")
print(f" 分辨率: {active_group['camera_width']}x{active_group['camera_height']}")
print()
# 初始化采集卡
get_image = GetImage(
cam_index=active_group['camera_index'],
width=active_group['camera_width'],
height=active_group['camera_height']
)
if get_image.cap is None:
print("❌ 采集卡初始化失败")
return
print("✅ 采集卡初始化成功")
print("'q' 退出测试")
print()
# 测试循环
frame_count = 0
while True:
frame_data = get_image.get_frame()
if frame_data is None:
print("⚠️ 无法获取帧")
continue
frame_count += 1
if frame_count % 30 == 0: # 每30帧显示一次状态
print(f"📊 已获取 {frame_count}")
# 显示图像
cv2.imshow('采集卡测试', frame_data[0])
# 按 'q' 退出
if cv2.waitKey(1) & 0xFF == ord('q'):
break
# 清理
get_image.release()
cv2.destroyAllWindows()
print("🔚 测试结束")
if __name__ == "__main__":
test_camera()

View File

@@ -41,33 +41,140 @@ import cv2
# cv2.destroyAllWindows()
import threading
import warnings
# 抑制OpenCV的警告信息兼容不同版本
import os
import sys
import io
os.environ['OPENCV_LOG_LEVEL'] = 'SILENT'
os.environ['OPENCV_IO_ENABLE_OPENEXR'] = '0'
try:
if hasattr(cv2, 'setLogLevel'):
if hasattr(cv2, 'LOG_LEVEL_SILENT'):
cv2.setLogLevel(cv2.LOG_LEVEL_SILENT)
elif hasattr(cv2, 'LOG_LEVEL_ERROR'):
cv2.setLogLevel(cv2.LOG_LEVEL_ERROR)
elif hasattr(cv2, 'utils'):
cv2.utils.setLogLevel(0)
except Exception:
pass
class GetImage:
def __init__(self, cam_index=0, width=1920, height=1080):
self.cap = cv2.VideoCapture(cam_index, cv2.CAP_DSHOW)
self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, width)
self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height)
print(f"🔧 正在初始化采集卡 {cam_index}...")
self.cap = None
self.frame = None
self.running = True
self.cam_index = cam_index
# 尝试多种方式打开采集卡
backends_to_try = [
(cam_index, cv2.CAP_DSHOW),
(cam_index, cv2.CAP_ANY),
(cam_index, None), # 默认后端
]
# 重定向stderr来抑制OpenCV的错误输出
old_stderr = sys.stderr
suppressed_output = io.StringIO()
try:
sys.stderr = suppressed_output
for idx, backend in backends_to_try:
try:
with warnings.catch_warnings():
warnings.filterwarnings('ignore', category=UserWarning)
if backend is not None:
self.cap = cv2.VideoCapture(idx, backend)
else:
self.cap = cv2.VideoCapture(idx)
if self.cap.isOpened():
# 测试读取一帧
ret, test_frame = self.cap.read()
if ret and test_frame is not None:
print(f"✅ 采集卡 {cam_index} 打开成功")
break
else:
self.cap.release()
self.cap = None
except Exception as e:
if self.cap:
try:
self.cap.release()
except:
pass
self.cap = None
continue
finally:
# 恢复stderr
sys.stderr = old_stderr
if self.cap is None or not self.cap.isOpened():
print(f"❌ 无法打开采集卡 {cam_index}")
print("请检查:")
print(" 1. 采集卡是否正确连接")
print(" 2. 采集卡索引是否正确(尝试扫描采集卡)")
print(" 3. 采集卡驱动是否安装")
print(" 4. 采集卡是否被其他程序占用")
self.cap = None
return
# 设置分辨率
try:
self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, width)
self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height)
# 实际获取设置后的分辨率
actual_width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
actual_height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
print(f" 分辨率设置: {width}x{height} -> 实际: {actual_width}x{actual_height}")
except Exception as e:
print(f"⚠️ 设置分辨率失败: {e}")
# 启动更新线程
threading.Thread(target=self.update, daemon=True).start()
# 等待几帧确保采集卡正常工作
import time
time.sleep(1.0)
print(f"✅ 采集卡 {cam_index} 初始化完成")
def update(self):
while self.running:
while self.running and self.cap is not None:
try:
ret, frame = self.cap.read()
if ret:
if ret and frame is not None:
self.frame = frame
else:
# 读取失败时不打印,避免刷屏
pass
except Exception as e:
# 只在异常时打印错误
print(f"⚠️ 采集卡 {self.cam_index} 读取异常: {e}")
import time
time.sleep(0.1) # 出错时短暂延迟
def get_frame(self):
if self.frame is None:
if self.cap is None or self.frame is None:
return None
try:
im_opencv = cv2.cvtColor(self.frame, cv2.COLOR_BGR2RGB)
im_opencv = im_opencv[30:30+720, 0:1280]
im_PIL = Image.fromarray(im_opencv)
return [im_opencv, im_PIL]
except Exception as e:
print(f"⚠️ 图像处理错误: {e}")
return None
def release(self):
self.running = False
time.sleep(0.2)
if self.cap is not None:
self.cap.release()
cv2.destroyAllWindows()
get_image = GetImage()
# get_image 将在main.py中初始化
get_image = None

View File

@@ -1,17 +1,34 @@
import random
import time
import ch9329Comm
import time
import random
import serial
serial.ser = serial.Serial('COM6', 9600) # 开启串口
mouse = ch9329Comm.mouse.DataComm(1920, 1080)
# 全局变量由main.py初始化
serial.ser = None
mouse = None
def init_mouse_keyboard(config_group):
"""初始化鼠标和串口"""
global serial, mouse
# 初始化串口
serial.ser = serial.Serial(
config_group['serial_port'],
config_group['serial_baudrate']
)
# 初始化鼠标
mouse = ch9329Comm.mouse.DataComm(
config_group['camera_width'],
config_group['camera_height']
)
print(f"✅ 串口已打开: {config_group['serial_port']} {config_group['serial_baudrate']}")
print(f"✅ 鼠标已初始化: {config_group['camera_width']}x{config_group['camera_height']}")
def bezier_point(t, p0, p1, p2, p3):
"""计算三次贝塞尔曲线上的点"""
x = (1-t)**3 * p0[0] + 3*(1-t)**2*t*p1[0] + 3*(1-t)*t**2*p2[0] + t**3*p3[0]
y = (1-t)**3 * p0[1] + 3*(1-t)**2*t*p1[1] + 3*(1-t)*t**2*p2[1] + t**3*p3[1]
return (x, y)
def move_mouse_bezier(mouse, start, end, duration=1, steps=120):
"""
用贝塞尔曲线模拟鼠标移动(安全版)
@@ -53,5 +70,6 @@ class Mouse_guiji():
mouse.click1() # 点击右
self.point=(x,y)
mouse_gui = Mouse_guiji()
# mouse_gui 将在main.py中初始化
mouse_gui = None