Upload files to 'src/unitdev01/unitdev01'

chiyu
ken910606 4 months ago
parent 14b70f6e2e
commit 00eb6b512d

@ -0,0 +1,398 @@
#!/usr/bin/env python3
from PyQt6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel,
QPushButton, QLineEdit)
from PyQt6.QtCore import pyqtSignal
class CommPanel(QWidget):
"""通讯设置面板类"""
# 定义信号
udp_connection_added = pyqtSignal(str, int) # ip, port
udp_connection_toggled = pyqtSignal(dict, QPushButton, QLabel) # conn, btn, status_label
udp_connection_removed = pyqtSignal(dict, QWidget) # conn, panel
ws_connection_added = pyqtSignal(str) # url
ws_connection_toggled = pyqtSignal(dict, QPushButton, QLabel) # conn, btn, status_label
ws_connection_removed = pyqtSignal(dict, QWidget) # conn, panel
status_message = pyqtSignal(str, int) # message, timeout
def __init__(self, parent=None):
super().__init__(parent)
self.udp_connections = []
self.ws_connections = []
self._init_ui()
def _init_ui(self):
"""初始化UI"""
layout = QVBoxLayout(self)
layout.setContentsMargins(10, 10, 10, 10)
layout.setSpacing(10)
# ========== UDP MAVLink 區域 ==========
udp_title = QLabel("UDP")
udp_title.setStyleSheet("""
color: #DDD;
font-size: 14px;
font-weight: bold;
padding: 5px;
background-color: #333;
border-radius: 4px;
""")
layout.addWidget(udp_title)
# UDP 連接列表容器
self.udp_list_widget = QWidget()
self.udp_list_layout = QVBoxLayout(self.udp_list_widget)
self.udp_list_layout.setContentsMargins(0, 0, 0, 0)
self.udp_list_layout.setSpacing(5)
layout.addWidget(self.udp_list_widget)
# UDP 添加新連接區域
add_udp_widget = QWidget()
add_udp_layout = QHBoxLayout(add_udp_widget)
add_udp_layout.setContentsMargins(0, 0, 0, 0)
self.udp_ip_input = QLineEdit()
self.udp_ip_input.setText("127.0.0.1")
self.udp_ip_input.setPlaceholderText("IP")
self.udp_ip_input.setStyleSheet("""
QLineEdit {
background-color: #333;
color: #DDD;
border: 1px solid #555;
border-radius: 4px;
padding: 5px;
}
""")
self.udp_port_input = QLineEdit()
self.udp_port_input.setText("14540")
self.udp_port_input.setPlaceholderText("Port")
self.udp_port_input.setFixedWidth(80)
self.udp_port_input.setStyleSheet("""
QLineEdit {
background-color: #333;
color: #DDD;
border: 1px solid #555;
border-radius: 4px;
padding: 5px;
}
""")
add_udp_btn = QPushButton("添加")
add_udp_btn.clicked.connect(self._handle_add_udp)
add_udp_btn.setStyleSheet("""
QPushButton {
background-color: #4CAF50;
color: white;
border: none;
padding: 6px 12px;
border-radius: 4px;
min-width: 30px;
}
QPushButton:hover { background-color: #45a049; }
""")
add_udp_layout.addWidget(QLabel("IP:", styleSheet="color: #DDD;"))
add_udp_layout.addWidget(self.udp_ip_input)
add_udp_layout.addWidget(QLabel("Port:", styleSheet="color: #DDD;"))
add_udp_layout.addWidget(self.udp_port_input)
add_udp_layout.addWidget(add_udp_btn)
layout.addWidget(add_udp_widget)
# 分隔線
separator = QWidget()
separator.setFixedHeight(20)
layout.addWidget(separator)
# ========== WebSocket 區域 ==========
ws_title = QLabel("WebSocket")
ws_title.setStyleSheet("""
color: #DDD;
font-size: 14px;
font-weight: bold;
padding: 5px;
background-color: #333;
border-radius: 4px;
""")
layout.addWidget(ws_title)
# WebSocket 連接列表容器
self.ws_list_widget = QWidget()
self.ws_list_layout = QVBoxLayout(self.ws_list_widget)
self.ws_list_layout.setContentsMargins(0, 0, 0, 0)
self.ws_list_layout.setSpacing(5)
layout.addWidget(self.ws_list_widget)
# WebSocket 添加新連接區域
add_ws_widget = QWidget()
add_ws_layout = QHBoxLayout(add_ws_widget)
add_ws_layout.setContentsMargins(0, 0, 0, 0)
self.ws_url_input = QLineEdit()
self.ws_url_input.setPlaceholderText("host")
self.ws_url_input.setStyleSheet("""
QLineEdit {
background-color: #333;
color: #DDD;
border: 1px solid #555;
border-radius: 4px;
padding: 5px;
}
""")
add_ws_btn = QPushButton("添加")
add_ws_btn.clicked.connect(self._handle_add_ws)
add_ws_btn.setStyleSheet("""
QPushButton {
background-color: #4CAF50;
color: white;
border: none;
padding: 6px 12px;
border-radius: 4px;
min-width: 30px;
}
QPushButton:hover { background-color: #45a049; }
""")
add_ws_layout.addWidget(QLabel("URL:", styleSheet="color: #DDD;"))
add_ws_layout.addWidget(self.ws_url_input)
add_ws_layout.addWidget(add_ws_btn)
layout.addWidget(add_ws_widget)
layout.addStretch()
def _handle_add_udp(self):
"""處理添加 UDP 連接"""
ip = self.udp_ip_input.text().strip()
port_text = self.udp_port_input.text().strip()
if not ip or not port_text:
self.status_message.emit("請輸入 IP 和 Port", 3000)
return
try:
port = int(port_text)
if port < 1 or port > 65535:
raise ValueError("Port 超出範圍")
except ValueError:
self.status_message.emit("Port 必須是 1-65535 的數字", 3000)
return
# 檢查是否已存在相同連接
for conn in self.udp_connections:
if conn['ip'] == ip and conn['port'] == port:
self.status_message.emit("連接已存在", 3000)
return
# 發送信號通知主窗口
self.udp_connection_added.emit(ip, port)
# 清空輸入框
self.udp_ip_input.clear()
self.udp_port_input.clear()
def _handle_add_ws(self):
"""處理添加 WebSocket 連接"""
input_url = self.ws_url_input.text().strip()
if not input_url:
self.status_message.emit("請輸入 WebSocket URL", 3000)
return
# 自動添加 ws:// 前綴
if not input_url.startswith('ws://') and not input_url.startswith('wss://'):
url = f'ws://{input_url}'
else:
url = input_url
# 基本 URL 格式驗證
try:
if '://' in url:
parts = url.split('://', 1)
if len(parts) == 2 and ':' not in parts[1]:
self.status_message.emit("URL 格式錯誤,需要包含端口號 (例如: 127.0.0.1:8756)", 3000)
return
except:
self.status_message.emit("URL 格式錯誤", 3000)
return
# 檢查是否已存在相同連接
for conn in self.ws_connections:
if conn['url'] == url:
self.status_message.emit("連接已存在", 3000)
return
# 發送信號通知主窗口
self.ws_connection_added.emit(url)
# 清空輸入框
self.ws_url_input.clear()
def create_udp_connection_panel(self, conn):
"""創建 UDP 連接面板"""
panel = QWidget()
panel.setStyleSheet("""
QWidget {
background-color: #2A2A2A;
border-radius: 6px;
padding: 8px;
border: 1px solid #444;
}
""")
layout = QHBoxLayout(panel)
layout.setContentsMargins(8, 8, 8, 8)
# 連接資訊
info_label = QLabel(f"{conn['name']} - {conn['ip']}:{conn['port']}")
info_label.setStyleSheet("color: #DDD; font-size: 12px;")
# 狀態指示器
status_label = QLabel("")
if conn.get('enabled', False):
status_label.setStyleSheet("color: #4CAF50; font-size: 16px;")
status_label.setToolTip("運行中")
else:
status_label.setStyleSheet("color: #888; font-size: 16px;")
status_label.setToolTip("已停止")
# 控制按鈕
toggle_btn = QPushButton("停止" if conn.get('enabled', False) else "啟動")
toggle_btn.setFixedWidth(60)
toggle_btn.clicked.connect(lambda: self.udp_connection_toggled.emit(conn, toggle_btn, status_label))
toggle_btn.setStyleSheet("""
QPushButton {
background-color: #444;
color: #DDD;
border: none;
padding: 4px 8px;
border-radius: 3px;
font-size: 11px;
}
QPushButton:hover { background-color: #555; }
""")
remove_btn = QPushButton("移除")
remove_btn.setFixedWidth(60)
remove_btn.clicked.connect(lambda: self.udp_connection_removed.emit(conn, panel))
remove_btn.setStyleSheet("""
QPushButton {
background-color: #d32f2f;
color: white;
border: none;
padding: 4px 8px;
border-radius: 3px;
font-size: 11px;
}
QPushButton:hover { background-color: #b71c1c; }
""")
layout.addWidget(status_label)
layout.addWidget(info_label)
layout.addStretch()
layout.addWidget(toggle_btn)
layout.addWidget(remove_btn)
# 儲存引用
panel.connection = conn
panel.toggle_btn = toggle_btn
panel.status_label = status_label
return panel
def create_ws_connection_panel(self, conn):
"""創建 WebSocket 連接面板"""
panel = QWidget()
panel.setStyleSheet("""
QWidget {
background-color: #2A2A2A;
border-radius: 6px;
padding: 8px;
border: 1px solid #444;
}
""")
layout = QHBoxLayout(panel)
layout.setContentsMargins(8, 8, 8, 8)
# 連接資訊
info_label = QLabel(f"{conn['name']} - {conn['url']}")
info_label.setStyleSheet("color: #DDD; font-size: 12px;")
# 狀態指示器
status_label = QLabel("")
if conn.get('enabled', False):
status_label.setStyleSheet("color: #4CAF50; font-size: 16px;")
status_label.setToolTip("運行中")
else:
status_label.setStyleSheet("color: #888; font-size: 16px;")
status_label.setToolTip("已停止")
# 控制按鈕
toggle_btn = QPushButton("停止" if conn.get('enabled', False) else "啟動")
toggle_btn.setFixedWidth(60)
toggle_btn.clicked.connect(lambda: self.ws_connection_toggled.emit(conn, toggle_btn, status_label))
toggle_btn.setStyleSheet("""
QPushButton {
background-color: #444;
color: #DDD;
border: none;
padding: 4px 8px;
border-radius: 3px;
font-size: 11px;
}
QPushButton:hover { background-color: #555; }
""")
remove_btn = QPushButton("移除")
remove_btn.setFixedWidth(60)
remove_btn.clicked.connect(lambda: self.ws_connection_removed.emit(conn, panel))
remove_btn.setStyleSheet("""
QPushButton {
background-color: #d32f2f;
color: white;
border: none;
padding: 4px 8px;
border-radius: 3px;
font-size: 11px;
}
QPushButton:hover { background-color: #b71c1c; }
""")
layout.addWidget(status_label)
layout.addWidget(info_label)
layout.addStretch()
layout.addWidget(toggle_btn)
layout.addWidget(remove_btn)
# 儲存引用
panel.connection = conn
panel.toggle_btn = toggle_btn
panel.status_label = status_label
return panel
def add_udp_panel(self, conn):
"""添加 UDP 連接面板到列表"""
panel = self.create_udp_connection_panel(conn)
self.udp_list_layout.addWidget(panel)
self.udp_connections.append(conn)
return panel
def add_ws_panel(self, conn):
"""添加 WebSocket 連接面板到列表"""
panel = self.create_ws_connection_panel(conn)
self.ws_list_layout.addWidget(panel)
self.ws_connections.append(conn)
return panel
def remove_udp_connection_from_list(self, conn):
"""從列表中移除 UDP 連接"""
if conn in self.udp_connections:
self.udp_connections.remove(conn)
def remove_ws_connection_from_list(self, conn):
"""從列表中移除 WebSocket 連接"""
if conn in self.ws_connections:
self.ws_connections.remove(conn)

@ -130,6 +130,129 @@ class UDPMavlinkReceiver(threading.Thread):
"""停止接收器""" """停止接收器"""
self.running = False self.running = False
class SerialMavlinkReceiver(threading.Thread):
"""串口 MAVLink 接收器"""
def __init__(self, port, baudrate, signals, connection_name):
super().__init__(daemon=True)
self.port = port
self.baudrate = baudrate
self.signals = signals
self.connection_name = connection_name
self.running = False
self.mav = None
def run(self):
"""執行串口接收循環"""
self.running = True
try:
print(f"Serial MAVLink receiver started on {self.port} at {self.baudrate} baud")
# 創建 MAVLink 串口連接
self.mav = mavutil.mavlink_connection(
self.port,
baud=self.baudrate,
source_system=255
)
print(f"Waiting for heartbeat from {self.port}...")
self.mav.wait_heartbeat()
print(f"Heartbeat received from system {self.mav.target_system}, component {self.mav.target_component}")
while self.running:
try:
msg = self.mav.recv_match(blocking=True, timeout=1.0)
if msg is None:
continue
self.process_mavlink_message(msg)
except Exception as e:
if self.running:
print(f"Error receiving MAVLink message from serial: {e}")
except Exception as e:
print(f"Serial receiver error: {e}")
finally:
if self.mav:
try:
self.mav.close()
except:
pass
def process_mavlink_message(self, msg):
"""處理 MAVLink 訊息"""
try:
msg_type = msg.get_type()
system_id = msg.get_srcSystem()
drone_id = f"s5_{system_id}" # 使用 serial_ 前綴表示串口來源
if msg_type == "HEARTBEAT":
mode = mavutil.mode_string_v10(msg)
armed = bool(msg.base_mode & 128)
self.signals.update_signal.emit('state', drone_id, {
'mode': mode,
'armed': armed
})
elif msg_type == "BATTERY_STATUS":
voltage = msg.voltages[0] / 1000
self.signals.update_signal.emit('battery', drone_id, {
'voltage': voltage
})
elif msg_type == "GLOBAL_POSITION_INT":
latitude = msg.lat / 1e7
longitude = msg.lon / 1e7
self.signals.update_signal.emit('gps', drone_id, {
'lat': latitude,
'lon': longitude
})
elif msg_type == "GPS_RAW_INT":
fix_type = msg.fix_type
elif msg_type == "LOCAL_POSITION_NED":
x = msg.y
y = msg.x
z = -msg.z
self.signals.update_signal.emit('local_pose', drone_id, {
'x': x,
'y': y,
'z': z
})
self.signals.update_signal.emit('altitude', drone_id, {
'altitude': z
})
self.signals.update_signal.emit('velocity', drone_id, {
'vx': msg.vx,
'vy': msg.vy,
'vz': msg.vz
})
elif msg_type == "ATTITUDE":
pitch = math.degrees(msg.pitch)
self.signals.update_signal.emit('attitude', drone_id, {
'pitch': pitch,
'roll': 0,
'yaw': 0,
'rates': (0, 0, 0)
})
elif msg_type == "VFR_HUD":
groundspeed = msg.groundspeed
heading = msg.heading
self.signals.update_signal.emit('hud', drone_id, {
'heading': heading,
'groundspeed': groundspeed
})
except Exception as e:
print(f"Error processing MAVLink message from serial: {e}")
def stop(self):
"""停止接收器"""
self.running = False
class WebSocketMavlinkReceiver(threading.Thread): class WebSocketMavlinkReceiver(threading.Thread):
"""WebSocket MAVLink 接收器""" """WebSocket MAVLink 接收器"""
def __init__(self, url, signals, connection_name): def __init__(self, url, signals, connection_name):
@ -266,6 +389,9 @@ class DroneMonitor(Node):
# WebSocket 接收器列表 # WebSocket 接收器列表
self.ws_receivers = [] self.ws_receivers = []
# 串口接收器列表
self.serial_receivers = []
# 主题检测定时器 # 主题检测定时器
self.create_timer(1.0, self.scan_topics) self.create_timer(1.0, self.scan_topics)
@ -508,4 +634,13 @@ class DroneMonitor(Node):
def ping_callback(self, drone_id, msg): def ping_callback(self, drone_id, msg):
self.latest_data[(drone_id, 'ping')] = { self.latest_data[(drone_id, 'ping')] = {
'ping': msg.data 'ping': msg.data
} }
def start_serial_connection(self, port='/dev/ttyUSB0', baudrate=57600):
"""啟動串口 MAVLink 連接"""
connection_name = f"Serial_{port.replace('/', '_')}"
receiver = SerialMavlinkReceiver(port, baudrate, self.signals, connection_name)
receiver.start()
self.serial_receivers.append(receiver)
print(f"Started serial connection on {port} at {baudrate} baud")
return receiver

@ -0,0 +1,378 @@
#!/usr/bin/env python3
from PyQt6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel,
QPushButton, QSizePolicy, QCheckBox)
from PyQt6.QtCore import pyqtSignal
class DronePanel(QWidget):
"""單個無人機面板類別"""
# 定義信號
mode_change_requested = pyqtSignal(str) # drone_id
arm_requested = pyqtSignal(str) # drone_id
takeoff_requested = pyqtSignal(str) # drone_id
setpoint_requested = pyqtSignal(str) # drone_id
selection_changed = pyqtSignal(str, int) # drone_id, state
def __init__(self, drone_id, parent=None):
super().__init__(parent)
self.drone_id = drone_id
self.display_id = 's' + drone_id.split('_')[1]
self._init_ui()
def _init_ui(self):
"""初始化UI"""
self.setObjectName(f"panel_{self.drone_id}")
self.setFixedHeight(140)
self.setStyleSheet("""
background-color: #2A2A2A;
border-radius: 8px;
""")
# 主佈局
main_layout = QHBoxLayout(self)
main_layout.setContentsMargins(8, 8, 8, 8)
main_layout.setSpacing(0)
# 創建內容容器(包含 info 和 control
content_widget = QWidget()
content_widget.setStyleSheet("background-color: #333; border-radius: 6px;")
content_layout = QHBoxLayout(content_widget)
content_layout.setContentsMargins(8, 8, 8, 8)
content_layout.setSpacing(8)
# 左側資訊區域
info_widget = self._create_info_section()
# 右側控制按鈕區域
control_widget = self._create_control_section()
# 將 info 和 control 加入內容容器
content_layout.addWidget(info_widget)
content_layout.addWidget(control_widget)
# 將內容容器加入主佈局
main_layout.addWidget(content_widget)
def _create_info_section(self):
"""創建資訊顯示區域"""
info_widget = QWidget()
info_layout = QVBoxLayout(info_widget)
info_layout.setContentsMargins(0, 0, 0, 0)
info_layout.setSpacing(4)
# 頂部標題欄
header = QWidget()
header_layout = QHBoxLayout(header)
header_layout.setContentsMargins(0, 0, 0, 0)
# 勾選框
self.checkbox = QCheckBox()
self.checkbox.setObjectName(f"{self.drone_id}_checkbox")
self.checkbox.setStyleSheet("QCheckBox { color: #DDD; }")
self.checkbox.stateChanged.connect(
lambda state: self.selection_changed.emit(self.drone_id, state)
)
# ID 顯示
id_label = QLabel(self.display_id)
id_label.setStyleSheet("""
font-weight: bold;
font-size: 14px;
color: #7FFFD4;
min-width: 80px;
""")
header_layout.addWidget(self.checkbox)
header_layout.addWidget(id_label)
header_layout.addStretch()
info_layout.addWidget(header)
# 第一行:狀態 (模式 + ARM狀態)
status_row = self._create_status_row()
info_layout.addWidget(status_row)
# 第二行:電池
battery_row = self._create_battery_row()
info_layout.addWidget(battery_row)
# 第三行:位置 + 高度
position_row = self._create_position_row()
info_layout.addWidget(position_row)
# 第四行:航向 + 速度
nav_row = self._create_nav_row()
info_layout.addWidget(nav_row)
return info_widget
def _create_status_row(self):
"""創建狀態行"""
status_row = QWidget()
status_layout = QHBoxLayout(status_row)
status_layout.setContentsMargins(0, 0, 0, 0)
status_title = QLabel("狀態:")
status_title.setStyleSheet("color: #888; min-width: 50px;")
self.mode_label = QLabel("--")
self.mode_label.setObjectName(f"{self.drone_id}_mode")
self.mode_label.setStyleSheet("color: #DDD;")
self.armed_label = QLabel("--")
self.armed_label.setObjectName(f"{self.drone_id}_armed")
self.armed_label.setStyleSheet("color: #DDD;")
status_layout.addWidget(status_title)
status_layout.addWidget(self.mode_label)
status_layout.addWidget(self.armed_label)
status_layout.addStretch()
return status_row
def _create_battery_row(self):
"""創建電池行"""
battery_row = QWidget()
battery_layout = QHBoxLayout(battery_row)
battery_layout.setContentsMargins(0, 0, 0, 0)
battery_title = QLabel("電池:")
battery_title.setStyleSheet("color: #888; min-width: 50px;")
self.battery_label = QLabel("--")
self.battery_label.setObjectName(f"{self.drone_id}_battery")
self.battery_label.setStyleSheet("color: #DDD;")
battery_layout.addWidget(battery_title)
battery_layout.addWidget(self.battery_label)
battery_layout.addStretch()
return battery_row
def _create_position_row(self):
"""創建位置行"""
position_row = QWidget()
position_layout = QHBoxLayout(position_row)
position_layout.setContentsMargins(0, 0, 0, 0)
position_title = QLabel("位置:")
position_title.setStyleSheet("color: #888; min-width: 50px;")
self.local_label = QLabel("--")
self.local_label.setObjectName(f"{self.drone_id}_local")
self.local_label.setStyleSheet("color: #DDD;")
altitude_title = QLabel("高度:")
altitude_title.setStyleSheet("color: #888; margin-left: 10px;")
self.altitude_label = QLabel("--")
self.altitude_label.setObjectName(f"{self.drone_id}_altitude")
self.altitude_label.setStyleSheet("color: #DDD;")
position_layout.addWidget(position_title)
position_layout.addWidget(self.local_label)
position_layout.addWidget(altitude_title)
position_layout.addWidget(self.altitude_label)
position_layout.addStretch()
return position_row
def _create_nav_row(self):
"""創建導航行"""
nav_row = QWidget()
nav_layout = QHBoxLayout(nav_row)
nav_layout.setContentsMargins(0, 0, 0, 0)
heading_title = QLabel("航向:")
heading_title.setStyleSheet("color: #888; min-width: 50px;")
self.heading_label = QLabel("--")
self.heading_label.setObjectName(f"{self.drone_id}_heading")
self.heading_label.setStyleSheet("color: #DDD;")
speed_title = QLabel("速度:")
speed_title.setStyleSheet("color: #888; margin-left: 10px;")
self.groundspeed_label = QLabel("--")
self.groundspeed_label.setObjectName(f"{self.drone_id}_groundspeed")
self.groundspeed_label.setStyleSheet("color: #DDD;")
nav_layout.addWidget(heading_title)
nav_layout.addWidget(self.heading_label)
nav_layout.addWidget(speed_title)
nav_layout.addWidget(self.groundspeed_label)
nav_layout.addStretch()
return nav_row
def _create_control_section(self):
"""創建控制按鈕區域"""
control_widget = QWidget()
control_layout = QVBoxLayout(control_widget)
control_layout.setContentsMargins(0, 0, 0, 0)
control_layout.setSpacing(6)
control_widget.setFixedWidth(80)
btn_style = """
QPushButton {
background-color: #444;
color: #DDD;
border: none;
border-radius: 4px;
font-size: 11px;
}
QPushButton:hover {
background-color: #555;
}
"""
# 模式切換按鈕
mode_btn = QPushButton("切換模式")
mode_btn.setStyleSheet(btn_style)
mode_btn.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Expanding)
mode_btn.clicked.connect(lambda: self.mode_change_requested.emit(self.drone_id))
# 解鎖按鈕
arm_btn = QPushButton("解鎖")
arm_btn.setStyleSheet(btn_style)
arm_btn.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Expanding)
arm_btn.clicked.connect(lambda: self.arm_requested.emit(self.drone_id))
# 起飛按鈕
takeoff_btn = QPushButton("起飛")
takeoff_btn.setStyleSheet(btn_style)
takeoff_btn.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Expanding)
takeoff_btn.clicked.connect(lambda: self.takeoff_requested.emit(self.drone_id))
# Setpoint 按鈕
setpoint_btn = QPushButton("Setpoint")
setpoint_btn.setStyleSheet(btn_style)
setpoint_btn.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Expanding)
setpoint_btn.clicked.connect(lambda: self.setpoint_requested.emit(self.drone_id))
control_layout.addWidget(mode_btn)
control_layout.addWidget(arm_btn)
control_layout.addWidget(takeoff_btn)
control_layout.addWidget(setpoint_btn)
return control_widget
def update_field(self, field, text, color=None):
"""更新指定欄位的值"""
label = self.findChild(QLabel, f"{self.drone_id}_{field}")
if label and label.text() != text:
label.setText(text)
if color:
label.setStyleSheet(f"color: {color};")
def get_checkbox(self):
"""獲取勾選框"""
return self.checkbox
def set_checked(self, checked):
"""設置勾選狀態"""
self.checkbox.setChecked(checked)
def is_checked(self):
"""獲取勾選狀態"""
return self.checkbox.isChecked()
class SocketGroupPanel(QWidget):
# 定義信號
group_selection_changed = pyqtSignal(str, int) # socket_id, state
def __init__(self, socket_id, color='#AAAAAA', parent=None):
super().__init__(parent)
self.socket_id = socket_id
self.color = color
self._init_ui()
def _init_ui(self):
"""初始化UI"""
self.setObjectName(f"socket_group_{self.socket_id}")
self.setStyleSheet("""
background-color: #1E1E1E;
border-radius: 12px;
""")
layout = QVBoxLayout(self)
layout.setContentsMargins(10, 10, 10, 10)
layout.setSpacing(6)
# Socket 分組標題行 - 包含勾選框
title_row = QWidget()
title_layout = QHBoxLayout(title_row)
title_layout.setContentsMargins(0, 0, 0, 0)
# 分組勾選框
self.group_checkbox = QCheckBox()
self.group_checkbox.setObjectName(f"socket_{self.socket_id}_checkbox")
self.group_checkbox.setStyleSheet(f"""
QCheckBox {{ color: #DDD; }}
QCheckBox::indicator {{
width: 14px;
height: 14px;
border: 2px solid #888888;
border-radius: 3px;
background: transparent;
}}
QCheckBox::indicator:checked {{
background-color: {self.color};
border: 2px solid #888888;
}}
QCheckBox::indicator:indeterminate {{
background-color: #666;
border: 2px solid #888888;
}}
""")
self.group_checkbox.stateChanged.connect(
lambda state: self.group_selection_changed.emit(self.socket_id, state)
)
# Socket 分組標題
title_label = QLabel(f"Socket {self.socket_id}")
title_label.setStyleSheet(f"""
font-weight: bold;
font-size: 16px;
color: {self.color};
margin-bottom: 8px;
padding: 4px 8px;
border-radius: 6px;
""")
title_layout.addWidget(self.group_checkbox)
title_layout.addWidget(title_label)
title_layout.addStretch()
layout.addWidget(title_row)
# 創建子容器用於放置該 socket 下的所有無人機面板
self.drones_container = QWidget()
self.drones_layout = QVBoxLayout(self.drones_container)
self.drones_layout.setContentsMargins(0, 0, 0, 0)
self.drones_layout.setSpacing(4)
layout.addWidget(self.drones_container)
def add_drone_panel(self, panel):
"""添加無人機面板到分組"""
self.drones_layout.addWidget(panel)
def clear_drones(self):
"""清空所有無人機面板"""
while self.drones_layout.count():
item = self.drones_layout.takeAt(0)
if item.widget():
item.widget().setParent(None)
def get_checkbox(self):
"""獲取分組勾選框"""
return self.group_checkbox
def set_checked(self, checked):
"""設置分組勾選狀態"""
self.group_checkbox.setChecked(checked)
def set_check_state(self, state):
"""設置分組勾選狀態(支持半選)"""
self.group_checkbox.setCheckState(state)

File diff suppressed because it is too large Load Diff

@ -0,0 +1,292 @@
#!/usr/bin/env python3
from PyQt6.QtWebEngineWidgets import QWebEngineView
from PyQt6.QtCore import QTimer, pyqtSignal, QObject, pyqtSlot
from PyQt6.QtWebChannel import QWebChannel
class DroneMap:
"""無人機地圖類別 - 負責管理 Leaflet 地圖顯示"""
def __init__(self):
"""初始化地圖"""
self.map_view = QWebEngineView()
self.map_loaded = False
self.pending_map_updates = {}
# 創建橋接對象
self.bridge = MapBridge()
# 設置 QWebChannel
self.channel = QWebChannel()
self.channel.registerObject('bridge', self.bridge)
self.map_view.page().setWebChannel(self.channel)
# 設置地圖 HTML
inline_html = '''
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8" />
<link rel="stylesheet" href="https://unpkg.com/leaflet/dist/leaflet.css" />
<script src="https://unpkg.com/leaflet/dist/leaflet.js"></script>
<script src="https://unpkg.com/leaflet-rotatedmarker/leaflet.rotatedMarker.js"></script>
<script src="qrc:///qtwebchannel/qwebchannel.js"></script>
<style>
html, body, #map { height: 100%; margin: 0; }
.map-controls {
position: absolute;
top: 10px;
right: 10px;
z-index: 1000;
}
.control-button {
padding: 8px 12px;
background-color: #f44336;
color: white;
border: none;
border-radius: 4px;
cursor: pointer;
font-size: 12px;
box-shadow: 0 2px 5px rgba(0,0,0,0.2);
}
.control-button:hover {
background-color: #d32f2f;
}
</style>
</head>
<body>
<div id="map"></div>
<div class="map-controls">
<button class="control-button" onclick="clearAllTrajectories()">清除軌跡</button>
</div>
<script>
var bridge;
new QWebChannel(qt.webChannelTransport, function(channel) {
bridge = channel.objects.bridge;
});
var map = L.map('map').setView([0, 0], 19);
L.tileLayer('https://{s}.tile.openstreetmap.org/{z}/{x}/{y}.png', {
maxZoom: 19,
attribution: '© OpenStreetMap contributors'
}).addTo(map);
// 地圖點擊事件
map.on('click', function(e) {
if (bridge) {
bridge.emitGpsSignal(e.latlng.lat, e.latlng.lng);
console.log('點擊位置:', e.latlng.lat, e.latlng.lng);
}
});
var arrowIcon = L.icon({
iconUrl: 'https://cdn-icons-png.flaticon.com/512/399/399308.png',
iconSize: [40, 40],
iconAnchor: [20, 20]
});
function getColorBySocketId(id) {
if (id.startsWith("s0_")) return "#00BFFF"; // 天藍色
if (id.startsWith("s1_")) return "#FFD700"; // 金色
if (id.startsWith("s2_")) return "#FF6969"; // 淺紅色
if (id.startsWith("s3_")) return "#FF69B4"; // 熱粉紅
if (id.startsWith("s4_")) return "#00FA9A"; // 中春綠
if (id.startsWith("s5_")) return "#9370DB"; // 中紫色 (串口)
if (id.startsWith("s6_")) return "#FFA500"; // 橙色
if (id.startsWith("s7_")) return "#20B2AA"; // 淺海綠
if (id.startsWith("s8_")) return "#7CFC00"; // 草綠色
if (id.startsWith("s9_")) return "#FF8C00"; // 深橙色
return "#AAAAAA"; // 灰色 (預設)
}
function createIdIcon(id) {
const color = getColorBySocketId(id);
const sysid = id.split('_')[1];
return L.divIcon({
className: 'drone-id',
html: `<div style="
position: relative;
left: 2px;
background-color: ${color};
color: black;
width: 16px;
height: 16px;
border-radius: 50%;
display: flex;
align-items: center;
justify-content: center;
font-weight: bold;
font-size: 10px;
">${sysid}</div>`,
iconSize: [20, 20],
iconAnchor: [10, 10]
});
}
var markers = {};
var idLabels = {};
var trajectories = {};
var trajectoryLines = {};
var focusedId = null;
var initialized = false;
var trajectoryGroup = L.layerGroup().addTo(map);
function initTrajectory(id) {
if (!trajectories[id]) {
trajectories[id] = [];
const color = getColorBySocketId(id);
trajectoryLines[id] = L.polyline([], {
color: color,
weight: 3,
opacity: 0.7,
smoothFactor: 1
}).addTo(trajectoryGroup);
}
}
function addTrajectoryPoint(id, lat, lon) {
initTrajectory(id);
const point = [lat, lon];
trajectories[id].push(point);
if (trajectories[id].length > 1000) {
trajectories[id].shift();
}
trajectoryLines[id].setLatLngs([...trajectories[id]]);
}
function focusOn(id) {
if (!markers[id]) return;
focusedId = id;
var latlng = markers[id].getLatLng();
map.flyTo(latlng, map.getZoom());
}
setInterval(() => {
if (focusedId && markers[focusedId]) {
var latlng = markers[focusedId].getLatLng();
map.panTo(latlng);
}
}, 1000);
function updateDrone(lat, lon, id, heading) {
if (markers[id]) {
const lastPos = markers[id].getLatLng();
const distance = lastPos.distanceTo([lat, lon]);
if (distance > 1) {
addTrajectoryPoint(id, lat, lon);
}
markers[id]
.setLatLng([lat, lon])
.setRotationAngle(heading);
idLabels[id].setLatLng([lat, lon]);
} else {
initTrajectory(id);
addTrajectoryPoint(id, lat, lon);
markers[id] = L.marker([lat, lon], {
icon: arrowIcon,
rotationAngle: heading,
rotationOrigin: 'center'
})
.on('click', function () {
focusOn(id);
})
.addTo(map);
idLabels[id] = L.marker([lat, lon], {
icon: createIdIcon(id),
zIndexOffset: 1000
})
.on('click', function() {
focusOn(id);
})
.addTo(map);
if (!initialized || id < focusedId) {
focusOn(id);
initialized = true;
}
}
}
function clearAllTrajectories() {
trajectories = {};
Object.values(trajectoryLines).forEach(line => {
trajectoryGroup.removeLayer(line);
});
trajectoryLines = {};
console.log('所有軌跡已清除');
}
</script>
</body>
</html>
'''
self.map_view.setHtml(inline_html)
self.map_view.loadFinished.connect(self._on_map_loaded)
# 設置地圖更新計時器
self.map_update_timer = QTimer()
self.map_update_timer.timeout.connect(self.update_map_positions)
self.map_update_timer.start(200) # 每 200ms 更新一次
def _on_map_loaded(self, ok: bool):
"""地圖加載完成回調"""
if ok:
self.map_loaded = True
else:
print("⚠️ 地圖加載失敗")
def update_drone_position(self, drone_id, lat, lon, heading):
"""更新無人機位置(加入待處理隊列)"""
self.pending_map_updates[drone_id] = (lat, lon, heading)
def update_map_positions(self):
"""批量更新地圖上的無人機位置"""
if not self.map_loaded or not self.pending_map_updates:
return
# 批量執行所有待更新的位置
js_commands = []
for drone_id, (lat, lon, heading) in self.pending_map_updates.items():
js_commands.append(f"updateDrone({lat:.6f}, {lon:.6f}, '{drone_id}', {heading:.1f});")
if js_commands:
combined_js = "\n".join(js_commands)
self.map_view.page().runJavaScript(combined_js)
# 清空待更新緩存
self.pending_map_updates.clear()
def clear_trajectories(self):
"""清除所有軌跡"""
if self.map_loaded:
self.map_view.page().runJavaScript("clearAllTrajectories();")
def focus_on_drone(self, drone_id):
"""聚焦到指定無人機"""
if self.map_loaded:
self.map_view.page().runJavaScript(f"focusOn('{drone_id}');")
def get_widget(self):
"""獲取地圖 widget"""
return self.map_view
def get_gps_signal(self):
"""獲取 GPS 信號"""
return self.bridge.gps_signal
class MapBridge(QObject):
"""JavaScript 和 Python 之間的橋接類"""
gps_signal = pyqtSignal(float, float) # lat, lon
def __init__(self):
super().__init__()
@pyqtSlot(float, float)
def emitGpsSignal(self, lat, lon):
"""供 JavaScript 調用的方法"""
self.gps_signal.emit(lat, lon)
Loading…
Cancel
Save