6 Commits
14 changed files with 392 additions and 3290 deletions
+2
View File
@@ -1,5 +1,7 @@
這是天雷系統的專案 這是天雷系統的專案
=== ===
## 功能簡介 ## 功能簡介
1. mavlink 多對多支援平台 1. mavlink 多對多支援平台
+41 -43
View File
@@ -419,8 +419,13 @@ class WebSocketMavlinkReceiver(threading.Thread):
class DroneMonitor(Node): class DroneMonitor(Node):
# Subscribe to drone ROS2 topics # Subscribe to drone ROS2 topics
_instance = None # Singleton pattern to prevent duplicate nodes
def __init__(self): def __init__(self):
super().__init__('drone_monitor') # Use a unique node name with timestamp to avoid conflicts on restart
import time
node_name = f'drone_monitor_{int(time.time() * 1000) % 100000}'
super().__init__(node_name)
self.signals = DroneSignals() self.signals = DroneSignals()
self.drone_topics = {} self.drone_topics = {}
self.lock = Lock() self.lock = Lock()
@@ -461,6 +466,17 @@ class DroneMonitor(Node):
# ================================================================================ # ================================================================================
self.serial_receivers = [] self.serial_receivers = []
# ================================================================================
# 【新增】初始化 CommandLongClient(持久化,不會每次調用都創建/銷毀)
# ================================================================================
self.command_long_client = None
try:
self.command_long_client = CommandLongClient()
except Exception as e:
print(f"⚠️ 警告: 無法初始化 CommandLongClient: {e}")
self.command_long_client = None
# ================================================================================
# 主题检测定时器 # 主题检测定时器
self.create_timer(1.0, self.scan_topics) self.create_timer(1.0, self.scan_topics)
@@ -586,7 +602,7 @@ class DroneMonitor(Node):
# ================================================================================ # ================================================================================
async def set_mode(self, drone_id, mode_name): async def set_mode(self, drone_id, mode_name):
"""使用 CommandLongClient 切換無人機飛行模式""" """使用 CommandLongClient 切換無人機飛行模式(使用非阻塞的 async 方法)"""
try: try:
# 解析 drone_id 提取 sysid # 解析 drone_id 提取 sysid
parts = drone_id.split('_') parts = drone_id.split('_')
@@ -603,30 +619,24 @@ class DroneMonitor(Node):
print(f"\n📢 [SET_MODE] {drone_id} → {mode_name} (custom_mode={custom_mode})") print(f"\n📢 [SET_MODE] {drone_id} → {mode_name} (custom_mode={custom_mode})")
# 在線程池中調用 CommandLongClient if not self.command_long_client:
loop = asyncio.get_event_loop() print(f"❌ [SET_MODE] CommandLongClient 未初始化")
executor = ThreadPoolExecutor(max_workers=1)
def _call_set_mode():
client = CommandLongClient() if CommandLongClient else None
if not client:
return False return False
result = client.change_mode(
# 直接調用 async 方法,無需線程池(避免嵌套執行器衝突)
result = await self.command_long_client.change_mode_async(
target_sysid=sysid, target_sysid=sysid,
custom_mode=float(custom_mode), custom_mode=float(custom_mode),
target_compid=0, target_compid=0,
base_mode=1.0, base_mode=1.0,
timeout_sec=2.0, timeout_sec=2.0,
) )
client.destroy_node()
return result.success if result else False
result = await loop.run_in_executor(executor, _call_set_mode) if result and result.success:
if result:
print(f"✅ [SET_MODE] 模式切換成功") print(f"✅ [SET_MODE] 模式切換成功")
return True return True
else: else:
print(f"❌ [SET_MODE] 模式切換失敗") print(f"❌ [SET_MODE] 模式切換失敗 (message={result.message if result else 'None'})")
return False return False
except Exception as e: except Exception as e:
print(f"❌ [SET_MODE] 例外錯誤: {e}") print(f"❌ [SET_MODE] 例外錯誤: {e}")
@@ -634,7 +644,7 @@ class DroneMonitor(Node):
return False return False
async def arm_drone(self, drone_id, arm): async def arm_drone(self, drone_id, arm):
"""使用 CommandLongClient 執行 ARM/DISARM""" """使用 CommandLongClient 執行 ARM/DISARM(使用非阻塞的 async 方法)"""
try: try:
# 解析 drone_id 提取 sysid # 解析 drone_id 提取 sysid
parts = drone_id.split('_') parts = drone_id.split('_')
@@ -646,29 +656,23 @@ class DroneMonitor(Node):
action_name = "解鎖" if arm else "上鎖" action_name = "解鎖" if arm else "上鎖"
print(f"\n📢 [ARM] {drone_id} → {action_name}") print(f"\n📢 [ARM] {drone_id} → {action_name}")
# 在線程池中調用 CommandLongClient if not self.command_long_client:
loop = asyncio.get_event_loop() print(f"❌ [ARM] CommandLongClient 未初始化")
executor = ThreadPoolExecutor(max_workers=1)
def _call_arm_disarm():
client = CommandLongClient() if CommandLongClient else None
if not client:
return False return False
result = client.arm_disarm(
# 直接調用 async 方法,無需線程池(避免嵌套執行器衝突)
result = await self.command_long_client.arm_disarm_async(
target_sysid=sysid, target_sysid=sysid,
arm=arm, arm=arm,
target_compid=0, target_compid=0,
timeout_sec=2.0, timeout_sec=2.0,
) )
client.destroy_node()
return result.success if result else False
result = await loop.run_in_executor(executor, _call_arm_disarm) if result and result.success:
if result:
print(f"✅ [ARM] {action_name}成功") print(f"✅ [ARM] {action_name}成功")
return True return True
else: else:
print(f"❌ [ARM] {action_name}失敗") print(f"❌ [ARM] {action_name}失敗 (message={result.message if result else 'None'})")
return False return False
except Exception as e: except Exception as e:
print(f"❌ [ARM] 例外錯誤: {e}") print(f"❌ [ARM] 例外錯誤: {e}")
@@ -676,7 +680,7 @@ class DroneMonitor(Node):
return False return False
async def takeoff_drone(self, drone_id, altitude): async def takeoff_drone(self, drone_id, altitude):
"""使用 CommandLongClient 執行無人機起飛""" """使用 CommandLongClient 執行無人機起飛(使用非阻塞的 async 方法)"""
try: try:
# 解析 drone_id 提取 sysid # 解析 drone_id 提取 sysid
parts = drone_id.split('_') parts = drone_id.split('_')
@@ -687,29 +691,23 @@ class DroneMonitor(Node):
print(f"\n📢 [TAKEOFF] {drone_id} → 起飛 (高度={altitude}m)") print(f"\n📢 [TAKEOFF] {drone_id} → 起飛 (高度={altitude}m)")
# 在線程池中調用 CommandLongClient if not self.command_long_client:
loop = asyncio.get_event_loop() print(f"❌ [TAKEOFF] CommandLongClient 未初始化")
executor = ThreadPoolExecutor(max_workers=1)
def _call_takeoff():
client = CommandLongClient() if CommandLongClient else None
if not client:
return False return False
result = client.takeoff(
# 直接調用 async 方法,無需線程池(避免嵌套執行器衝突)
result = await self.command_long_client.takeoff_async(
target_sysid=sysid, target_sysid=sysid,
altitude_m=float(altitude), altitude_m=float(altitude),
target_compid=0, target_compid=0,
timeout_sec=2.0, timeout_sec=2.0,
) )
client.destroy_node()
return result.success if result else False
result = await loop.run_in_executor(executor, _call_takeoff) if result and result.success:
if result:
print(f"✅ [TAKEOFF] 起飛成功") print(f"✅ [TAKEOFF] 起飛成功")
return True return True
else: else:
print(f"❌ [TAKEOFF] 起飛失敗") print(f"❌ [TAKEOFF] 起飛失敗 (message={result.message if result else 'None'})")
return False return False
except Exception as e: except Exception as e:
print(f"❌ [TAKEOFF] 例外錯誤: {e}") print(f"❌ [TAKEOFF] 例外錯誤: {e}")
+14 -1
View File
@@ -33,7 +33,7 @@ from mission_group import (
# ================================================================================ # ================================================================================
class ControlStationUI(QMainWindow): class ControlStationUI(QMainWindow):
VERSION = '2.0.3' VERSION = '2.0.6'
def __init__(self): def __init__(self):
super().__init__() super().__init__()
@@ -52,6 +52,10 @@ class ControlStationUI(QMainWindow):
self.executor = rclpy.executors.SingleThreadedExecutor() self.executor = rclpy.executors.SingleThreadedExecutor()
self.executor.add_node(self.monitor) self.executor.add_node(self.monitor)
# 添加 CommandLongClient 到執行器(這樣它的回調才能被處理)
if self.monitor.command_long_client:
self.executor.add_node(self.monitor.command_long_client)
# 定时处理ROS事件 # 定时处理ROS事件
self.timer = QTimer() self.timer = QTimer()
self.timer.timeout.connect(self.spin_ros) self.timer.timeout.connect(self.spin_ros)
@@ -1716,6 +1720,15 @@ class ControlStationUI(QMainWindow):
receiver.stop() receiver.stop()
for receiver in self.monitor.ws_receivers: for receiver in self.monitor.ws_receivers:
receiver.stop() receiver.stop()
# Clean up serial receivers
for receiver in self.monitor.serial_receivers:
receiver.stop()
# Clean up CommandLongClient
if self.monitor.command_long_client:
try:
self.monitor.command_long_client.destroy_node()
except:
pass
self.monitor.destroy_node() self.monitor.destroy_node()
self.executor.shutdown() self.executor.shutdown()
except Exception as e: except Exception as e:
@@ -38,7 +38,7 @@ class SerialMode(Enum):
NOT_USE = auto() # 不使用 NOT_USE = auto() # 不使用
# ====================== Frame Processor Base and Implementation ===================== # ====================== Frame Processor 基類與實現 =====================
class FrameProcessor(ABC): class FrameProcessor(ABC):
"""協議處理器基類""" """協議處理器基類"""
@@ -47,10 +47,10 @@ class FrameProcessor(ABC):
self.buffer = bytearray() self.buffer = bytearray()
@abstractmethod @abstractmethod
def process_incoming(self, data: bytes) -> bytes: def process_incoming(self, data: bytes):
""" """
處理接收到的數據 處理接收到的數據
返回:已完整解析的 payload 列表 後續要丟到 UDP 去 返回:已完整解析的 payload 列表
""" """
pass pass
@@ -66,7 +66,7 @@ class FrameProcessor(ABC):
class RawFrameProcessor(FrameProcessor): class RawFrameProcessor(FrameProcessor):
"""原始數據直通處理器""" """原始數據直通處理器"""
def process_incoming(self, data: bytes) -> bytes: def process_incoming(self, data: bytes):
"""直接返回原始數據,不進行緩衝""" """直接返回原始數據,不進行緩衝"""
return [data] if data else [] return [data] if data else []
@@ -76,158 +76,69 @@ class RawFrameProcessor(FrameProcessor):
class XBeeFrameProcessor(FrameProcessor): class XBeeFrameProcessor(FrameProcessor):
""" """XBee API 協議處理器"""
XBee API 協議處理器
職責: def process_incoming(self, data: bytes):
- XBee API frame 的拆幀 / 組幀
- 0x90 (RX Packet) -> 解出 payload 回傳
- 0x88 (AT Response) -> 轉交 at_handler 處理(若有注入)
- 0x8B (TX Status) -> 目前忽略
- 其他 frame type -> 記 warning 忽略
若未來要做變化型 XBee(例如 API2 escape mode、不同 addressing),
繼承此類並覆寫 _encapsulate / _decapsulate / _try_extract_frame 即可。
"""
# XBee API frame type
FRAME_TYPE_TX_REQUEST = 0x10
FRAME_TYPE_AT_RESPONSE = 0x88
FRAME_TYPE_TX_STATUS = 0x8B
FRAME_TYPE_RX_PACKET = 0x90
def __init__(self, at_handler: "ATCommandHandler" = None):
super().__init__()
self.at_handler = at_handler
# ---- 對外契約 ----
def process_incoming(self, data: bytes) -> bytes:
"""處理 XBee API 幀並提取 payload""" """處理 XBee API 幀並提取 payload"""
self.buffer.extend(data) self.buffer.extend(data)
payloads = [] payloads = []
while True: while len(self.buffer) >= 3:
frame = self._try_extract_frame() # 尋找幀頭
if frame is None: if self.buffer[0] != 0x7E:
self.buffer.pop(0)
continue
# 讀取 payload 長度
length = (self.buffer[1] << 8) | self.buffer[2]
full_length = 3 + length + 1 # 起始符(1) + 長度(2) + payload + 校驗和(1)
# 等待完整幀
if len(self.buffer) < full_length:
break break
payload = self._dispatch_frame(frame)
# 提取完整 frame 並從緩衝區移除
frame = bytes(self.buffer[:full_length])
del self.buffer[:full_length]
# 判斷 frame 類型並處理
frame_type = frame[3]
if frame_type == 0x90: # RX Packet
payload = XBeeFrameHandler.decapsulate_data(frame)
if payload: if payload:
payloads.append(payload) payloads.append(payload)
elif frame_type == 0x88: # AT Response
# 可以在這裡處理 AT 指令回應
# response = XBeeFrameHandler.parse_at_command_response(frame)
# 目前忽略
pass
elif frame_type == 0x8B: # Transmit Status
# 傳輸狀態,目前忽略
pass
else:
logger.warning(f"Unknown XBee frame type: 0x{frame_type:02X}")
return payloads return payloads
def process_outgoing(self, data: bytes) -> bytes: def process_outgoing(self, data: bytes) -> bytes:
"""將數據封裝為 XBee API 傳輸幀""" """將數據封裝為 XBee API 傳輸幀"""
return self._encapsulate(data) return XBeeFrameProcessor.encapsulate_data(data)
# ---- 內部:拆幀與分派 ----
def _try_extract_frame(self) -> bytes:
"""
從 buffer 嘗試抓一個完整 frame
- 對齊幀頭 0x7E
- 長度不足時等待下次進 buffer
- 成功則從 buffer 切掉並回傳整個 frame bytes
"""
while len(self.buffer) >= 3:
if self.buffer[0] != 0x7E:
self.buffer.pop(0)
continue
length = (self.buffer[1] << 8) | self.buffer[2] # ====================== XBee Frame Handler =====================
full_length = 3 + length + 1 # 起始符(1) + 長度(2) + payload + 校驗和(1)
if len(self.buffer) < full_length: class XBeeFrameHandler:
return None """XBee API Frame 處理器"""
frame = bytes(self.buffer[:full_length])
del self.buffer[:full_length]
return frame
return None
def _dispatch_frame(self, frame: bytes) -> bytes:
"""根據 frame type 分派;若是 RX payload 回傳 bytes,其餘回傳 None"""
frame_type = frame[3]
if frame_type == self.FRAME_TYPE_RX_PACKET:
return self._decapsulate(frame)
if frame_type == self.FRAME_TYPE_AT_RESPONSE:
if self.at_handler is not None:
self.at_handler.handle_frame(frame)
return None
if frame_type == self.FRAME_TYPE_TX_STATUS:
return None
logger.warning(f"Unknown XBee frame type: 0x{frame_type:02X}")
return None
# ---- 內部:編解碼 ----
@staticmethod
def _encapsulate(
data: bytes,
dest_addr64: bytes = b'\x00\x00\x00\x00\x00\x00\xFF\xFF',
frame_id: int = 0x01,
) -> bytes:
"""
將 payload 包成 XBee TX Request (0x10)
- 使用廣播地址
- 添加適當的頭部和校驗和
"""
frame_type = XBeeFrameProcessor.FRAME_TYPE_TX_REQUEST
dest_addr16 = b'\xFF\xFE'
broadcast_radius = 0x00
options = 0x00
frame = struct.pack(">B", frame_type) + struct.pack(">B", frame_id)
frame += dest_addr64 + dest_addr16
frame += struct.pack(">BB", broadcast_radius, options) + data
checksum = 0xFF - (sum(frame) & 0xFF)
return b'\x7E' + struct.pack(">H", len(frame)) + frame + struct.pack("B", checksum)
@staticmethod @staticmethod
def _decapsulate(frame: bytes) -> bytes: def parse_at_command_response(frame: bytes) -> dict:
"""從 RX Packet (0x90) 取出 payload""" """解析 AT Command Response (0x88)"""
length = (frame[1] << 8) | frame[2]
rf_data_start = 3 + 12
return frame[rf_data_start:3 + length]
# ====================== Dongle Command Handler =====================
class ATCommandHandler:
"""
XBee AT 指令回應處理器
職責:
- 解析 AT Response frame (0x88)
- 依 AT 指令分派到對應的 handler
"""
def __init__(self, serial_port: str):
self.serial_port = serial_port
self.handlers = {
b'DB': self._handle_rssi,
b'SH': self._handle_serial_high,
b'SL': self._handle_serial_low,
# 可擴展其他 AT 指令
}
def handle_frame(self, frame: bytes) -> None:
"""接收一整個 AT Response frame,內部完成 parse + dispatch"""
parsed = self._parse(frame)
if parsed is None:
return
self._dispatch(parsed)
@staticmethod
def _parse(frame: bytes) -> dict:
"""解析 AT Command Response (0x88);不符格式回傳 None"""
if len(frame) < 8: if len(frame) < 8:
return None return None
if frame[3] != 0x88: frame_type = frame[3]
if frame_type != 0x88:
return None return None
frame_id = frame[4] frame_id = frame[4]
@@ -240,26 +151,81 @@ class ATCommandHandler:
'command': at_command, 'command': at_command,
'status': status, 'status': status,
'data': data, 'data': data,
'is_ok': status == 0x00, 'is_ok': status == 0x00
} }
def _dispatch(self, response: dict) -> None: @staticmethod
def parse_receive_packet(frame: bytes) -> dict:
"""解析 RX Packet (0x90) - 未來擴展用"""
# if len(frame) < 15 or frame[3] != 0x90:
# return None
# return {
# 'source_addr': frame[4:12],
# 'reserved': frame[12:14],
# 'options': frame[14],
# 'data': frame[15:-1]
# }
pass
return None
@staticmethod
def encapsulate_data(data: bytes, dest_addr64: bytes = b'\x00\x00\x00\x00\x00\x00\xFF\xFF', frame_id = 0x01) -> bytes:
"""
將數據封裝為 XBee API 傳輸幀
使用 XBee API 格式封裝數據:
- 傳輸請求幀 (0x10)
- 使用廣播地址
- 添加適當的頭部和校驗和
"""
frame_type = 0x10
dest_addr16 = b'\xFF\xFE'
broadcast_radius = 0x00
options = 0x00
frame = struct.pack(">B", frame_type) + struct.pack(">B", frame_id)
frame += dest_addr64 + dest_addr16
frame += struct.pack(">BB", broadcast_radius, options) + data
checksum = 0xFF - (sum(frame) & 0xFF)
return b'\x7E' + struct.pack(">H", len(frame)) + frame + struct.pack("B", checksum)
@staticmethod
def decapsulate_data(data: bytes):
# 獲取數據長度 (不包括校驗和)
length = (data[1] << 8) | data[2]
rf_data_start = 3 + 12
return data[rf_data_start:3 + length]
class ATCommandHandler:
"""AT 指令回應處理器"""
def __init__(self, serial_port: str):
self.serial_port = serial_port
self.handlers = {
b'DB': self._handle_rssi,
b'SH': self._handle_serial_high,
b'SL': self._handle_serial_low,
# 可擴展其他 AT 指令
}
def handle_response(self, response: dict):
"""根據 AT 指令類型分派處理""" """根據 AT 指令類型分派處理"""
if not response['is_ok']: if not response or not response['is_ok']:
logger.warning( if response:
f"[{self.serial_port}] AT {response['command'].decode()} " logger.warning(f"[{self.serial_port}] AT {response['command'].decode()} 失敗,狀態碼: {response['status']}")
f"失敗,狀態碼: {response['status']}"
)
return return
handler = self.handlers.get(response['command']) command = response['command']
handler = self.handlers.get(command)
if handler: if handler:
handler(response['data']) handler(response['data'])
else: else:
logger.debug( logger.debug(f"[{self.serial_port}] 未處理的 AT 指令: {command.decode()}")
f"[{self.serial_port}] 未處理的 AT 指令: "
f"{response['command'].decode()}"
)
def _handle_rssi(self, data: bytes): def _handle_rssi(self, data: bytes):
"""處理 DB (RSSI) 回應""" """處理 DB (RSSI) 回應"""
@@ -275,7 +241,8 @@ class ATCommandHandler:
pass pass
# ================ Serial UDP Socket Object ============== # ====================== Serial Handler =====================
class SerialHandler(asyncio.Protocol): class SerialHandler(asyncio.Protocol):
"""asyncio.Protocol 用於處理 Serial 收發""" """asyncio.Protocol 用於處理 Serial 收發"""
@@ -285,19 +252,22 @@ class SerialHandler(asyncio.Protocol):
self.serial_mode = serial_mode self.serial_mode = serial_mode
self.transport = None # Serial 自己的傳輸物件 self.transport = None # Serial 自己的傳輸物件
# 根據模式建立 processor(需要 AT handler 時一併在工廠內建好注入) # 根據模式創建對應的 processor
self.processor = self._create_processor(serial_mode, serial_port_str) self.processor = self._create_processor(serial_mode)
@staticmethod # AT 指令處理器(僅 XBee 模式使用)
def _create_processor(serial_mode: SerialMode, serial_port_str: str) -> FrameProcessor: if serial_mode == SerialMode.XBEEAPI2AT:
"""工廠方法:根據模式建立對應的 processor,必要時一併注入 AT handler""" self.at_handler = ATCommandHandler(serial_port_str)
else:
self.at_handler = None
def _create_processor(self, serial_mode: SerialMode) -> FrameProcessor:
"""工廠方法:根據模式創建處理器"""
if serial_mode == SerialMode.STRAIGHT: if serial_mode == SerialMode.STRAIGHT:
return RawFrameProcessor() return RawFrameProcessor()
elif serial_mode == SerialMode.XBEEAPI2AT:
if serial_mode == SerialMode.XBEEAPI2AT: return XBeeFrameProcessor()
at_handler = ATCommandHandler(serial_port_str) else:
return XBeeFrameProcessor(at_handler=at_handler)
logger.warning(f"Unknown serial mode: {serial_mode}, using Raw") logger.warning(f"Unknown serial mode: {serial_mode}, using Raw")
return RawFrameProcessor() return RawFrameProcessor()
@@ -321,6 +291,8 @@ class SerialHandler(asyncio.Protocol):
) )
# ====================== UDP Handler =====================
class UDPHandler(asyncio.DatagramProtocol): class UDPHandler(asyncio.DatagramProtocol):
"""asyncio.DatagramProtocol 用於處理 UDP 收發""" """asyncio.DatagramProtocol 用於處理 UDP 收發"""
@@ -628,7 +600,7 @@ if __name__ == '__main__':
SERIAL_PORT = '/dev/ttyUSB0' # 手動指定 SERIAL_PORT = '/dev/ttyUSB0' # 手動指定
SERIAL_BAUDRATE = 115200 SERIAL_BAUDRATE = 115200
UDP_REMOTE_PORT = 14561 UDP_REMOTE_PORT = 14571
sm.create_serial_link(SERIAL_PORT, SERIAL_BAUDRATE, UDP_REMOTE_PORT, SerialMode.XBEEAPI2AT) sm.create_serial_link(SERIAL_PORT, SERIAL_BAUDRATE, UDP_REMOTE_PORT, SerialMode.XBEEAPI2AT)
# SERIAL_PORT = '/dev/ttyACM0' # 手動指定 # SERIAL_PORT = '/dev/ttyACM0' # 手動指定
@@ -643,19 +615,3 @@ if __name__ == '__main__':
sm.remove_serial_link(1) sm.remove_serial_link(1)
time.sleep(3) time.sleep(3)
sm.shutdown() sm.shutdown()
'''
================= 改版記錄 ============================
2026.4.20
1. XBeeFrameHandler 結構移除
2. XBeeFrameProcessor 新增 _encapsulate, _decapsulate 編碼解碼 xbee 封包的功能 (原來在 XBeeFrameHandler 中)
3. XBeeFrameProcessor 新增 _try_extract_frame 處理被可能截斷的 UART 封包
4. XBeeFrameProcessor 新增 _dispatch_frame 分配封包到 UDP 或者 Dongle Command Handler
5. ATCommandHandler 新增 _parse 去拆解 0x88 AT Command Response
6. ATCommandHandler 新增 _dispatch 把拆解的結果 分配到 _handle_XXX
7. ATCommandHandler 新增各項 _handle_XXX (未實作)
'''
+116
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from typing import Optional from typing import Optional
import asyncio
import rclpy import rclpy
from rclpy.node import Node from rclpy.node import Node
@@ -185,3 +186,118 @@ class CommandLongClient(Node):
param7=0.0, param7=0.0,
timeout_sec=timeout_sec, timeout_sec=timeout_sec,
) )
# ============================================================================
# 【新增】非阻塞 async 包裝方法(用於 GUI 的非阻塞調用)
# 這些方法在 ThreadPoolExecutor 中運行同步版本,以避免阻塞事件循環
# ============================================================================
async def change_mode_async(
self,
*,
target_sysid: int,
custom_mode: float,
target_compid: int = 0,
base_mode: float = 1.0,
confirmation: int = 0,
timeout_sec: float = DEFAULT_TIMEOUT_SEC,
) -> CommandLongResult:
"""非阻塞 async 版本的 change_mode(在 ThreadPoolExecutor 中運行)"""
from concurrent.futures import ThreadPoolExecutor
loop = asyncio.get_event_loop()
executor = ThreadPoolExecutor(max_workers=1)
return await loop.run_in_executor(
executor,
lambda: self.change_mode(
target_sysid=target_sysid,
custom_mode=custom_mode,
target_compid=target_compid,
base_mode=base_mode,
confirmation=confirmation,
timeout_sec=timeout_sec,
)
)
async def arm_disarm_async(
self,
*,
target_sysid: int,
arm: bool,
target_compid: int = 0,
confirmation: int = 0,
param2: float = 0.0,
timeout_sec: float = DEFAULT_TIMEOUT_SEC,
) -> CommandLongResult:
"""非阻塞 async 版本的 arm_disarm(在 ThreadPoolExecutor 中運行)"""
from concurrent.futures import ThreadPoolExecutor
loop = asyncio.get_event_loop()
executor = ThreadPoolExecutor(max_workers=1)
return await loop.run_in_executor(
executor,
lambda: self.arm_disarm(
target_sysid=target_sysid,
arm=arm,
target_compid=target_compid,
confirmation=confirmation,
param2=param2,
timeout_sec=timeout_sec,
)
)
async def takeoff_async(
self,
*,
target_sysid: int,
altitude_m: float,
target_compid: int = 0,
min_pitch_deg: float = 0.0,
yaw_deg: float = 0.0,
latitude: Optional[float] = None,
longitude: Optional[float] = None,
timeout_sec: float = DEFAULT_TIMEOUT_SEC,
) -> CommandLongResult:
"""非阻塞 async 版本的 takeoff(在 ThreadPoolExecutor 中運行)"""
from concurrent.futures import ThreadPoolExecutor
loop = asyncio.get_event_loop()
executor = ThreadPoolExecutor(max_workers=1)
return await loop.run_in_executor(
executor,
lambda: self.takeoff(
target_sysid=target_sysid,
altitude_m=altitude_m,
target_compid=target_compid,
min_pitch_deg=min_pitch_deg,
yaw_deg=yaw_deg,
latitude=latitude,
longitude=longitude,
timeout_sec=timeout_sec,
)
)
async def land_async(
self,
*,
target_sysid: int,
target_compid: int = 0,
yaw_deg: float = 0.0,
latitude: Optional[float] = None,
longitude: Optional[float] = None,
altitude_m: float = 0.0,
timeout_sec: float = DEFAULT_TIMEOUT_SEC,
) -> CommandLongResult:
"""非阻塞 async 版本的 land(在 ThreadPoolExecutor 中運行)"""
from concurrent.futures import ThreadPoolExecutor
loop = asyncio.get_event_loop()
executor = ThreadPoolExecutor(max_workers=1)
return await loop.run_in_executor(
executor,
lambda: self.land(
target_sysid=target_sysid,
target_compid=target_compid,
yaw_deg=yaw_deg,
latitude=latitude,
longitude=longitude,
altitude_m=altitude_m,
timeout_sec=timeout_sec,
)
)
+45
View File
@@ -0,0 +1,45 @@
import rclpy
from rclpy.node import Node
from std_msgs.msg import String
import time
# import mavros_msgs.srv
class TalkerNode(Node):
def __init__(self):
start_time = time.time()
super().__init__('talker_node')
end_time = time.time()
print(f"Node initialization took {end_time - start_time:.2f} seconds")
self.publisher_ = self.create_publisher(String, 'hahatest/_1', 10)
self.timer = self.create_timer(1.0, self.timer_callback) # 每秒執行一次
self.get_logger().info('TalkerNode has been started.')
def timer_callback(self):
msg = String()
msg.data = 'Hello, ROS 2!'
self.publisher_.publish(msg)
self.get_logger().info(f'Published: "{msg.data}"')
def main(args=None):
rclpy.init(args=args)
node = TalkerNode()
print("Before sleep")
time.sleep(5) # 等待 5 秒鐘
print("After sleep")
try:
start_time = time.time()
while time.time() - start_time < 10: # 持續 10 秒鐘
rclpy.spin_once(node)
time.sleep(1) # 每秒執行一次
finally:
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
-188
View File
@@ -1,188 +0,0 @@
import time
from pymavlink import mavutil
# # 強制 source_system=255
# master = mavutil.mavlink_connection('udpout:localhost:14561', source_system=255)
# print("Starting Force-Ping Test...")
# while True:
# # 1. 發送 Heartbeat (證明連線活著)
# master.mav.heartbeat_send(mavutil.mavlink.MAV_TYPE_GCS,
# mavutil.mavlink.MAV_AUTOPILOT_INVALID, 0, 0, 0)
# # 2. 強制發送 PING (target 1, 1)
# # 我們加一個計數器,方便在 tcpdump 裡辨識
# ping_seq = int(time.time()) % 100
# master.mav.ping_send(int(time.time() * 1e6), ping_seq, 1, 1)
# print(f"Sent Heartbeat and PING (seq {ping_seq})")
# # 3. 非阻塞讀取,避免卡死
# msg = master.recv_match(blocking=False)
# if msg:
# print(f"--> Received something: {msg.get_type()}")
# time.sleep(1) # 每秒噴一次
# -------------------------------------------------------------------
# # 改成 udpin,這會讓你的程式守在 14561 等模擬器送資料進來
# # 這樣你就能收到模擬器的 Heartbeat,建立連線後再發 PING
# master = mavutil.mavlink_connection('udpin:localhost:14561', source_system=255)
# print("Waiting for simulator heartbeat...")
# # 這一行很重要:先收到 Heartbeat,pymavlink 才會設定好 target 地址
# master.wait_heartbeat()
# print(f"Heartbeat from system {master.target_system} component {master.target_component}")
# while True:
# # 既然收到了,現在發 PING 就會順著原路回去
# master.mav.ping_send(int(time.time() * 1e6), 1, master.target_system, master.target_component)
# print("PING sent!")
# msg = master.recv_match(type='PING', blocking=True, timeout=2)
# if msg:
# print(f"Got PING response: {msg}")
# time.sleep(1)
# -------------------------------------------------------------------
# import time
# from pymavlink import mavutil
# # 核心修改:因為 MAVProxy 正在往 14561 噴資料,我們必須「接住」它
# # 使用 udpin 會讓你的程式在 14561 監聽
# master = mavutil.mavlink_connection('udpin:localhost:14561', source_system=254)
# print("正在等待 MAVProxy 轉發的 Heartbeat...")
# # 只要這行跑過,代表你跟 MAVProxy 握手成功了
# master.wait_heartbeat()
# print(f"成功連線到系統 {master.target_system}!")
# while True:
# # 1. 務必發送 Heartbeat,讓 MAVProxy 知道要把回應送回這個 Port
# master.mav.heartbeat_send(
# mavutil.mavlink.MAV_TYPE_GCS,
# mavutil.mavlink.MAV_AUTOPILOT_INVALID,
# 0, 0, 0
# )
# # 2. 修改 target_component 為 1 (飛控核心)
# # 注意:這裡刻意將 seq 設為動態,方便你在 watch PING 裡觀察
# p_seq = int(time.time()) % 255
# master.mav.ping_send(
# int(time.time() * 1e6),
# p_seq,
# 1,
# 1 # <--- 改成 1 試試看!
# )
# print(f"Sent PING seq {p_seq} to 1/1")
# # 3. 讀取所有訊息,看看有沒有 PING 回來
# # 有時候回應的 target_system 可能是 255 (你的 ID)
# msg = master.recv_match(type='PING', blocking=True, timeout=1.0)
# if msg:
# print(f"🎉 成功!收到回應: {msg}")
# time.sleep(1)
# -------------------------------------------------------------------
# import time
# from pymavlink import mavutil
# # 建議一樣用 udpin 監聽 MAVProxy 的輸出
# master = mavutil.mavlink_connection('udpin:localhost:14561', source_system=255)
# def get_time_ns():
# return int(time.time() * 1e9) # 轉為奈秒
# print("開始測試 TIMESYNC 響應...")
# while True:
# now_ns = get_time_ns()
# # 發送 TIMESYNC 請求
# # tc1=0 代表這是請求,ts1=當前時間
# master.mav.timesync_send(0, now_ns)
# # 等待回應
# msg = master.recv_match(type='TIMESYNC', blocking=True, timeout=1.0)
# if msg and msg.tc1 != 0:
# # 收到回應,msg.tc1 應該等於我們剛才送出的 now_ns
# rtt_ms = (get_time_ns() - msg.tc1) / 1e6
# print(f"🕒 TIMESYNC RTT: {rtt_ms:.3f} ms")
# break # 測試一次就好
# else:
# print("等待 TIMESYNC 回應中...")
# time.sleep(1)
# -------------------------------------------------------------------
import time
import statistics
from pymavlink import mavutil
# 連接到 MAVProxy 轉發的端口
master = mavutil.mavlink_connection('udpin:localhost:14561', source_system=255)
# 測試設定
TARGET_SYS = 2 # 你想測試的載具 ID
SAMPLES = 20 # 取 20 次樣本來算平均和抖動
rtt_history = []
print(f"開始測試對 System {TARGET_SYS} 的通道品質...")
def get_time_ns():
return int(time.time() * 1e9)
try:
while len(rtt_history) < SAMPLES:
ts1_send = get_time_ns()
# 發送 TIMESYNC (V1/V2 格式相容)
# 針對特定系統發送
master.mav.timesync_send(0, ts1_send)
# 等待回應 (過濾目標系統的訊息)
msg = master.recv_match(type='TIMESYNC', blocking=True, timeout=1.0)
if msg and msg.get_srcSystem() == TARGET_SYS and msg.tc1 != 0:
ts1_recv = get_time_ns()
rtt_ns = ts1_recv - msg.tc1
rtt_ms = rtt_ns / 1e6
rtt_history.append(rtt_ms)
print(f"Sample {len(rtt_history)}: RTT = {rtt_ms:.3f} ms")
else:
print("Request timeout or mismatched ID...")
time.sleep(0.1) # 快速採樣
# 4. 計算統計數據
avg_latency = statistics.mean(rtt_history)
std_dev = statistics.stdev(rtt_history) # 這就是 Jitter 的一種表現
max_rtt = max(rtt_history)
min_rtt = min(rtt_history)
print("\n--- 物理通道品質分析報告 ---")
print(f"目標載具 : System {TARGET_SYS}")
print(f"平均延遲 (Avg) : {avg_latency:.3f} ms")
print(f"最小延遲 (Min) : {min_rtt:.3f} ms")
print(f"最大延遲 (Max) : {max_rtt:.3f} ms")
print(f"抖動幅度 (Jitter): {std_dev:.3f} ms (標準差)")
if std_dev > 10:
print("警告:網路抖動過高,可能影響控制穩定性!")
else:
print("通道品質:良好")
except KeyboardInterrupt:
print("測試終止。")
@@ -1,46 +0,0 @@
from pymavlink.dialects.v20 import common as mavlink
class NullBuffer:
def write(self, data): pass
def seek(self, pos): pass
def tell(self): return 0
class CapturingBuffer:
def __init__(self):
self.last_data = b''
def write(self, data):
self.last_data = bytes(data)
def seek(self, pos): pass
def tell(self): return 0
# 1. 初始化(僅作為編碼器)
# file 參數可以是任何擁有 write() 方法的物件,這裡用 BytesIO 模擬
# 初始化方法一:使用 BytesIO
# import io
# out_buf = io.BytesIO()
# mav = mavlink.MAVLink(out_buf, srcSystem=1, srcComponent=191)
# 初始化方法二:使用自定義的 NullBuffer
mav = mavlink.MAVLink(NullBuffer(), srcSystem=1, srcComponent=191)
mav.seq = 254
# 2. 建立心跳包並取得二進制數據
# 連續產出「不同」的訊息物件
for i in range(3):
msg = mav.heartbeat_encode(mavlink.MAV_TYPE_GCS, 0, 0, 0, 0)
data = msg.pack(mav)
mav.seq = (mav.seq + 1) & 0xFF
# MAVLink 2 的序列號在第 3 個 Byte (Index 2)
print(f"第 {i} 次發送, Seq: {data[4]}, Raw: {data.hex()}")
print("分隔線")
buf = CapturingBuffer()
mav_buf2 = mavlink.MAVLink(buf, srcSystem=1, srcComponent=191)
for i in range(3):
mav_buf2.heartbeat_send(mavlink.MAV_TYPE_GCS, 0, 0, 0, 0)
data = buf.last_data
print(f"第 {i} 次發送, Seq: {data[4]}, Raw: {data.hex()}")
File diff suppressed because it is too large Load Diff
-71
View File
@@ -1,71 +0,0 @@
import asyncio
import serial_asyncio
SERIAL_PORT = '/dev/ttyUSB0' # 修改為你的 serial port
SERIAL_BAUDRATE = 115200 # 修改為你的 baudrate
UDP_REMOTE_IP = '192.168.1.100' # 修改為目標 IP
UDP_REMOTE_PORT = 5005 # 修改為目標 port
UDP_LOCAL_PORT = 5006 # 本地 UDP 監聽 port
class SerialToUDP(asyncio.Protocol):
def __init__(self, udp_transport):
self.udp_transport = udp_transport
def connection_made(self, transport):
self.transport = transport
def data_received(self, data):
# Serial 收到資料,轉發到 UDP
self.udp_transport.sendto(data, (UDP_REMOTE_IP, UDP_REMOTE_PORT))
def write_to_serial(self, data):
self.transport.write(data)
class UDPToSerial(asyncio.DatagramProtocol):
def __init__(self, serial_proto):
self.serial_proto = serial_proto
def datagram_received(self, data, addr):
# UDP 收到資料,轉發到 Serial
self.serial_proto.write_to_serial(data)
async def main():
loop = asyncio.get_running_loop()
# 定義協議工廠函數
def create_empty_protocol():
return asyncio.DatagramProtocol()
# 建立 UDP1 傳輸
udp_transport, _ = await loop.create_datagram_endpoint(
create_empty_protocol,
local_addr=('0.0.0.0', UDP_LOCAL_PORT)
)
# 建立 Serial 傳輸
serial_proto = SerialToUDP(udp_transport)
def get_serial_protocol():
return serial_proto
_, serial_transport = await serial_asyncio.create_serial_connection(
loop, get_serial_protocol, SERIAL_PORT, baudrate=SERIAL_BAUDRATE
)
# 建立 UDP2 監聽
udp_proto = UDPToSerial(serial_proto)
def get_udp_protocol():
return udp_proto
udp_listen_transport, _ = await loop.create_datagram_endpoint(
get_udp_protocol,
local_addr=('0.0.0.0', UDP_LOCAL_PORT)
)
# 保持運行
try:
await asyncio.Future()
except asyncio.CancelledError:
pass
if __name__ == '__main__':
asyncio.run(main())
+27
View File
@@ -0,0 +1,27 @@
#!/bin/bash
# 網站清單
DOMAINS=("google.com" "smarter.nchu.edu.tw")
echo "網站 SSL 憑證剩餘天數:"
echo "---------------------------"
for domain in "${DOMAINS[@]}"; do
end_date=$(echo | openssl s_client -servername "$domain" -connect "$domain:443" 2>/dev/null |
openssl x509 -noout -enddate | cut -d= -f2)
end_timestamp=$(date -d "$end_date" +%s)
now_timestamp=$(date +%s)
remaining_days=$(( (end_timestamp - now_timestamp) / 86400 ))
if [ $remaining_days -lt 0 ]; then
status="已過期 ❌"
elif [ $remaining_days -lt 15 ]; then
status="即將到期 ⚠️"
else
status="正常 ✅"
fi
printf "%-20s 到期日:%-25s 剩餘天數:%3d 天 %s\n" "$domain" "$end_date" "$remaining_days" "$status"
done
-40
View File
@@ -1,40 +0,0 @@
import socket
import sys
import select
# 設定來源 IP 與 Port
SRC_IP = '127.0.0.1' # 監聽所有介面
SRC_PORT = 16661 # 請自行修改
# 建立 UDP 監聽 socket
src_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
src_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
src_sock.bind((SRC_IP, SRC_PORT))
print(f"Listening for UDP on {SRC_IP}:{SRC_PORT}")
# 設定目標 Unix socket 路徑
UNIX_SOCKET_PATH = '/tmp/unix_socket_mavlink.sock' # 請自行修改
# 建立 Unix socket 連線
unix_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
try:
unix_sock.connect(UNIX_SOCKET_PATH)
except Exception as e:
print(f"Failed to connect to unix socket: {e}")
sys.exit(1)
while True:
# 使用 select 監聽兩個 socket
readable, _, _ = select.select([src_sock, unix_sock], [], [])
for sock in readable:
if sock is src_sock:
data, addr = src_sock.recvfrom(4096)
if data:
unix_sock.sendall(data)
# print(f"Received UDP data from {addr}: {data}") # debug
# break # debug
elif sock is unix_sock:
data = unix_sock.recv(4096)
if data:
# 回送到最近收到資料的 UDP client
src_sock.sendto(data, addr)