wenchun
/
AirTrapMine
forked from chiyu1468/AirTrapMine
1
0
Fork 0
Code Pull Requests Activity

(Tested) fix mainOrchestrator.py

Browse Source 
完善 serial 支援
Chiyu Chen 5 months ago
parent 5134fa8466
commit 62356cc056
2 changed files with 171 additions and 16 deletions
Show Stats Download Patch File Download Diff File

177
src/fc_network_adapter/fc_network_adapter/mainOrchestrator.py
Unescape Escape View File

@ -38,15 +38,21 @@ class PanelState:
self.object_manager_state = "Stopped"
self.serial_manager_state = "Stopped"
self.socket_object_list = [] # 已有的 mavlink object
self.linked_serial_dict = {} # 已連線的 serial 端口
self.linked_serial_dict = {} # 已連線的 serial 端口 serial id num : serial_port string
self.panel_info_msg_list = [] # 顯示在面板上的資訊訊息
# 這邊是儲存關於 socket object 的資料
# 關於創建通道時的暫存資訊
self.udp_info_temp = {"IP": "127.0.0.1", "Port": "", "Direction": ""} # 暫存 UDP 設定資訊
self.serial_info_temp = {"Port": "", "Baud": 115200, "CommunicationType": "", "Go2Middleware": True} # 暫存 Serial 設定資訊
self.socket_info_single = {"socket_type": "", "socket_state": "", "bridge_msg_types": "", "return_msg_types": "",
self.serial_info_temp = {"Port": "", "Baud": 115200, "CommunicationType": "", "Go2Middleware": False} # 暫存 Serial 設定資訊
# 關於顯示通道資訊
self.socket_info_single = {
"socket_type": "", "socket_state": "", "bridge_msg_types": "", "return_msg_types": "",
"target_sockets": "", "primary_socket_id": "", "socket_connection_string": "",
"InfoReady": False} # 暫存單一 socket 的資訊
self.serial_info_single = {
"serial_port": "", "baudrate": "", "receiver_type": "", "target_port": "",
"InfoReady": False} # 暫存單一 serial 連結的資訊
def intoSTART(self):
self.panel_status = "Running"
@ -178,6 +184,15 @@ class ControlPanel:
def show_object_info(self, stdscr, socket_id, state: PanelState):
"""顯示物件詳細資訊的對話框"""
start = time.time()
while not state.socket_info_single.get('InfoReady', False):
# 太久沒有回應
if time.time() - start > 2:
state.panel_info_msg_list.append(("Fail! Socket Info NOT Aquire!", time.time()))
return
time.sleep(0.05) # 等待資訊準備好
height, width = stdscr.getmaxyx()
dialog_height = 15
dialog_width = min(70, width - 4)
@ -188,12 +203,9 @@ class ControlPanel:
dialog_win.border()
dialog_win.addstr(0, 2, f" Socket #{socket_id} 詳細資訊 ", curses.A_BOLD)
while not state.socket_info_single.get('InfoReady', False):
time.sleep(0.05) # 等待資訊準備好
# 這裡顯示基本資訊
dialog_win.addstr(2, 2, f"Socket ID : {socket_id}")
# dialog_win.addstr(3, 2, f"Socket status : 運行中")
dialog_win.addstr(3, 2, f"Socket status : {state.socket_info_single.get('socket_state', 'N/A')}")
# show_str = ", ".join(map(str, state.socket_info_single.get('socket_type', '')))
dialog_win.addstr(4, 2, f"Socket Type : {state.socket_info_single.get('socket_type', '')}")
dialog_win.addstr(4, 30, f"{state.socket_info_single.get('socket_connection_string', '')}")
@ -300,6 +312,10 @@ class ControlPanel:
# MenuNode("Telemetry", "數傳模式", "SET_SERIAL_COMM_TELEMETRY"),
]),
MenuNode("Set Baud", "設定 Baud", "TEXT_BAUD_SERIAL"),
MenuNode("Link to Middleware", "方便功能 可以直接建立 UDP object", "LINK_SERIAL_TO_MIDDLEWARE_UDP", children=[
MenuNode("Yes", action = "LINK_SERIAL_TO_MIDDLEWARE_UDP_YES"),
MenuNode("No", action = "LINK_SERIAL_TO_MIDDLEWARE_UDP_NO"),
]),
MenuNode("Create", "建立此串口", "CREATE_SERIAL_PORT"),
MenuNode("返回", "回到列表", "BACK"),
])
@ -369,8 +385,55 @@ class ControlPanel:
menu.current_page = page
return menu
def show_linked_serial_info(self):
pass
def show_linked_serial_info(self, stdscr, serial_id, state: PanelState):
"""顯示 Serial 連結詳細資訊的對話框"""
start = time.time()
while not state.serial_info_single.get('InfoReady', False):
# 太久沒有回應
if time.time() - start > 2:
state.panel_info_msg_list.append(("Fail! Serial Info NOT Aquire!", time.time()))
return
time.sleep(0.05) # 等待資訊準備好
height, width = stdscr.getmaxyx()
dialog_height = 15
dialog_width = min(70, width - 4)
start_y = (height - dialog_height) // 2
start_x = (width - dialog_width) // 2
dialog_win = curses.newwin(dialog_height, dialog_width, start_y, start_x)
dialog_win.border()
dialog_win.addstr(0, 2, f" Serial Link #{serial_id} 詳細資訊 ", curses.A_BOLD)
# 從 linked_serial_dict 獲取資訊
serial_info = state.linked_serial_dict.get(serial_id, {})
if not serial_info:
dialog_win.addstr(2, 2, f"無法取得 Serial #{serial_id} 的資訊")
else:
# 顯示基本資訊
dialog_win.addstr(2, 2, f"Serial ID : {serial_id}")
dialog_win.addstr(3, 2, f"Serial Port : {state.serial_info_single.get('serial_port', 'N/A')}")
dialog_win.addstr(4, 2, f"Baudrate : {state.serial_info_single.get('baudrate', 'N/A')}")
dialog_win.addstr(5, 2, f"Communication : {state.serial_info_single.get('receiver_type', 'N/A')}")
dialog_win.addstr(6, 2, f"Target UDP Port : {state.serial_info_single.get('target_port', 'N/A')}")
dialog_win.addstr(7, 2, f"Status : {state.serial_info_single.get('status', 'Running')}")
# 如果有額外資訊可以繼續添加
if 'thread_alive' in serial_info:
thread_status = "Alive" if serial_info['thread_alive'] else "Stopped"
dialog_win.addstr(8, 2, f"Thread Status : {thread_status}")
state.serial_info_single['InfoReady'] = False # 重置狀態以便下次使用
dialog_win.addstr(dialog_height - 2, 2, "按任意鍵返回...")
dialog_win.refresh()
dialog_win.getch()
del dialog_win
stdscr.clear()
stdscr.refresh()
# ================ 關於 主要選單 的部份 ===================
@ -481,6 +544,9 @@ class ControlPanel:
desc = f"{child.desc} [{state.serial_info_temp['CommunicationType']}]"
elif child.action == "TEXT_BAUD_SERIAL" and state.serial_info_temp["Baud"]:
desc = f"{child.desc} [{state.serial_info_temp['Baud']}]"
elif child.action == "LINK_SERIAL_TO_MIDDLEWARE_UDP":
link_status = "Yes" if state.serial_info_temp["Go2Middleware"] else "No"
desc = f"{child.desc} [{link_status}]"
line = f"{marker}{child.name:15s} – {desc}"
attr = curses.A_REVERSE if i == current_idx else curses.A_NORMAL
@ -643,6 +709,16 @@ class ControlPanel:
menu_stack.pop()
idx_stack.pop()
elif selected.action == "LINK_SERIAL_TO_MIDDLEWARE_UDP_YES":
logger.info("mark A")
state.serial_info_temp["Go2Middleware"] = True
menu_stack.pop()
idx_stack.pop()
elif selected.action == "LINK_SERIAL_TO_MIDDLEWARE_UDP_NO":
state.serial_info_temp["Go2Middleware"] = False
menu_stack.pop()
idx_stack.pop()
elif selected.action == "CREATE_SERIAL_PORT":
state.serial_info_temp["Port"] = menu_stack[-1].name # 從選單取得 Port 名稱
cmd_q.put("CREATE_SERIAL_PORT")
@ -663,6 +739,24 @@ class ControlPanel:
menu_stack.append(created_list_menu)
idx_stack.append(0)
elif selected.action == "INSPECT_LINKED_SERIAL":
# 顯示 Serial 連結詳細資訊
if hasattr(selected, 'serial_id'):
cmd_q.put(("INSPECT_LINKED_SERIAL", selected.serial_id))
self.show_linked_serial_info(stdscr, selected.serial_id, state)
elif selected.action == "REMOVE_LINKED_SERIAL":
# 移除 Serial 連結
if hasattr(selected, 'serial_id'):
cmd_q.put(("REMOVE_LINKED_SERIAL", selected.serial_id))
# 返回上層（回到列表）
if len(menu_stack) > 1:
menu_stack.pop()
idx_stack.pop()
# 一樣退兩層
menu_stack.pop()
idx_stack.pop()
elif selected.action == "LIST_MAV_OBJECT":
# 動態生成 mavlink_object 列表選單
created_list_menu = self.create_object_list_menu(state, page=0)
@ -714,7 +808,6 @@ class ControlPanel:
if hasattr(selected, 'socket_id'):
target_id = self.select_target_socket(stdscr, selected.socket_id, state)
if target_id is not None:
# cmd_q.put(("MAVOBJ_MAKE_LINK", selected.socket_id, target_id))
cmd_q.put(("MAVOBJ_ADD_TARGET", selected.socket_id, target_id))
cmd_q.put(("MAVOBJ_ADD_TARGET", target_id, selected.socket_id)) # 雙向連結
@ -770,6 +863,7 @@ class ControlPanel:
class Orchestrator:
def __init__(self, stop_sig):
self.stop_evt = stop_sig # 外部操作去中斷 "面板" 執行緒的訊號 (內部自己停止的話不需要用這個)
self.occupied_ip_ports = {} # 紀錄已被佔用的 ip:port 組合 "ip str" : [port int, port int, ...]
# === 1) 面板部分的準備 ===
self.cmd_q = queue.Queue()
@ -855,7 +949,7 @@ class Orchestrator:
mav_obj = self.manager.managed_objects.get(socket_id, None)
if mav_obj:
self.panelState.socket_info_single["socket_type"] = mav_obj.socket_type
# self.panelState.socket_info_single["socket_state"] = "Running" if mav_obj.running
self.panelState.socket_info_single["socket_state"] = mav_obj.state.name
self.panelState.socket_info_single["bridge_msg_types"] = mav_obj.bridge_msg_types
self.panelState.socket_info_single["return_msg_types"] = mav_obj.return_msg_types
self.panelState.socket_info_single["primary_socket_id"] = mav_obj.primary_socket_id
@ -864,6 +958,18 @@ class Orchestrator:
self.panelState.socket_info_single["socket_connection_string"] = f"{ip_info[0]}:{ip_info[1]}"
# getattr(mav_obj.mavlink_socket, "connection_string", "")
self.panelState.socket_info_single["InfoReady"] = True # 標記資訊已準備好
elif action == "INSPECT_LINKED_SERIAL":
serial_id = cmd[1]
serial_obj = self.plumber.serial_objects.get(serial_id, None)
if serial_obj:
self.panelState.serial_info_single["serial_port"] = serial_obj.serial_port
self.panelState.serial_info_single["baudrate"] = serial_obj.baudrate
self.panelState.serial_info_single["receiver_type"] = serial_obj.receiver_type.name
self.panelState.serial_info_single["target_port"] = serial_obj.target_port
self.panelState.serial_info_single["InfoReady"] = True # 標記資訊已準備好
elif action == "REMOVE_LINKED_SERIAL":
serial_id = cmd[1]
self.plumber.remove_serial_link(serial_id)
elif cmd == "CREATE_UDP_INBOUND":
self.panelState.udp_info_temp["direction"] = "inbound"
@ -919,18 +1025,35 @@ class Orchestrator:
# =============== 面板動作 - Mavlink Object ===============
def create_udp_object(self):
# 監聽部分
if self.panelState.udp_info_temp["direction"] == "inbound":
connection_string = f"udp:{self.panelState.udp_info_temp['IP']}:{self.panelState.udp_info_temp['Port']}"
# 監聽的 port 要先檢查是否被佔用
ip = self.panelState.udp_info_temp['IP']
port = int(self.panelState.udp_info_temp['Port'])
port_check_list = self.occupied_ip_ports.get(ip, []) + self.occupied_ip_ports.get("0.0.0.0", [])
if port in port_check_list:
self.panelState.panel_info_msg_list.append((f"Failed! Port {port} on IP {ip} occupied.", time.time()-1))
return
# 再記錄被佔用的 port
if ip in self.occupied_ip_ports:
self.occupied_ip_ports[ip].append(port)
else:
self.occupied_ip_ports[ip] = [port]
# 外放資訊部分
elif self.panelState.udp_info_temp["direction"] == "outbound":
connection_string = f"udpout:{self.panelState.udp_info_temp['IP']}:{self.panelState.udp_info_temp['Port']}"
try:
# 檢測這個 connection_string 是否能成功建立 mavlink 連結
mavlink_socket = mavutil.mavlink_connection(connection_string)
except Exception as e:
self.panelState.panel_info_msg_list.append((f"Failed to create UDP {self.panelState.udp_info_temp['direction']} object: {e}", time.time()-1))
return
# mavlink 連結建立成功 把他丟到 mavlink_object
mavlink_object = mo.mavlink_object(mavlink_socket)
mavlink_object.socket_type = "UDP " + self.panelState.udp_info_temp['direction'].capitalize()
# 再把 mavlink_object 丟到 manager 的 event loop 裡面去管理與執行
self.manager.add_mavlink_object(mavlink_object)
self.panelState.panel_info_msg_list.append((f"Created UDP {self.panelState.udp_info_temp['direction']} object: {connection_string}", time.time()))
@ -967,6 +1090,14 @@ class Orchestrator:
# =============== 面板動作 - Serial Manager ===============
def create_serial_port_object(self):
# 先檢查是否已有相同的 Serial Port 被建立
serial_port_strs = self.panelState.linked_serial_dict.values() # linked_serial_dict 會在上面的 mainLoop 被不斷更新
if self.panelState.serial_info_temp['Port'] in serial_port_strs:
self.panelState.panel_info_msg_list.append(
(f"Fail! Serial Port {self.panelState.serial_info_temp['Port']} already linked.", time.time())
)
return
# 獲取可用的 udp port
udp_port_tmp = find_available_port(19000, 20000)
@ -1004,6 +1135,16 @@ class Orchestrator:
self.panelState.panel_info_msg_list.append((f"Failed to create Serial Port object at {self.panelState.serial_info_temp['Port']}.", time.time()))
return
self.panelState.panel_info_msg_list.append((f"Created Serial Port object at {self.panelState.serial_info_temp['Port']}.", time.time()))
# 自動建立對應的 UDP 監聽端口
if self.panelState.serial_info_temp['Go2Middleware']:
self.panelState.udp_info_temp['IP'] = "127.0.0.1"
self.panelState.udp_info_temp['Port'] = str(udp_port_tmp)
self.panelState.udp_info_temp['direction'] = "inbound"
self.create_udp_object()
def main():
stop_evt = threading.Event()
@ -1024,3 +1165,15 @@ if __name__ == "__main__":
main()
'''
================= 改版記錄 ============================
2025.12.23
1. 新增 serial 通道的資訊顯示完整化
2. 新增 serial 通道刪除功能
3. 新增 serial 直接順便開 ip object
4. 修改 避免 serial 與 ip port 重複建立相同的通道
5. 修改 show_object_info 與 show_linked_serial_info 改變檢核 Ready 方式 避免卡死
'''

6
src/fc_network_adapter/fc_network_adapter/serialManager.py
Unescape Escape View File

@ -359,7 +359,7 @@ class serial_manager:
self.loop = None
self.running = False
self.serial_count = 0
self.serial_objects = {} # serial id num : serial object
self.serial_objects = {} # serial id num : serial_object
def __del__(self):
self.loop = None
@ -626,6 +626,8 @@ if __name__ == '__main__':
linked_serial = sm.get_serial_link()
print(linked_serial)
time.sleep(10)
sm.remove_serial_link(1)
time.sleep(3)
sm.shutdown()
Write Preview
Loading…
Cancel
Save