v0.1.7.0(2024/06/25)-未发布
1. [aio.py] 取消了在本文件中开启openapi线程的做法，并修改如下：
- 通过包的方式导入其他模块
- 使用current_path来规避文件路径问题
- 声名了 self.hr 变量，用来接收openapi的实例化
- 修改了对于segment button的错误调用
- 设定progress bar的长度是10
- 完善了segmented_button_callback函数
- 在detect_network函数中增加heartbeat初始化
- tabview_click函数中新增textbox清屏功能，以及实例化openapi，并做检测
2. [openapi.py] 取消了初始化中无限循环检测，因为阻塞了aio主界面进程！！！socket也无法多次连接！！！浪费了好多时间！！！很生气！！！！
- 通过tabview切换来实现重新连接，并保留了异常处理部分
- 将所有的 __xxxx 函数都替换成 xxxx 函数，去掉了 __
- 使用current_path来规避文件路径问题
3. [do_brake.py] 初步完成了机器状态收集的功能，还需要完善
- 使用current_path来规避文件路径问题
- 新增validate_resp函数，校验数据
- 完善了调用接口
This commit is contained in:
@ -1,3 +1,4 @@
|
||||
import sys
|
||||
import tkinter
|
||||
from os.path import exists, dirname
|
||||
from os import getcwd
|
||||
@ -7,12 +8,10 @@ import customtkinter
|
||||
from time import time, strftime, localtime, sleep
|
||||
from urllib.request import urlopen
|
||||
from socket import setdefaulttimeout
|
||||
import data_process.brake as brake
|
||||
import data_process.current as current
|
||||
import data_process.iso as iso
|
||||
import data_process.wavelogger as wavelogger
|
||||
import automatic_test.openapi as openapi
|
||||
from data_process import *
|
||||
from automatic_test import *
|
||||
|
||||
current_path = dirname(__file__)
|
||||
customtkinter.set_appearance_mode("System") # Modes: "System" (standard), "Dark", "Light"
|
||||
customtkinter.set_default_color_theme("blue") # Themes: "blue" (standard), "green", "dark-blue"
|
||||
customtkinter.set_widget_scaling(1.1) # widget dimensions and text size
|
||||
@ -73,6 +72,7 @@ class App(customtkinter.CTk):
|
||||
super().__init__()
|
||||
self.my_font = customtkinter.CTkFont(family="Consolas", size=16, weight="bold")
|
||||
self.w_param = 84
|
||||
self.hr = None
|
||||
# =====================================================================
|
||||
# configure window
|
||||
self.title("AIO - All in one automatic toolbox")
|
||||
@ -142,7 +142,7 @@ class App(customtkinter.CTk):
|
||||
# For data process tab END =====================================================================
|
||||
# For automatic test tab START =====================================================================
|
||||
# create buttons
|
||||
self.seg_button = customtkinter.CTkSegmentedButton(self.tabview.tab('Automatic Test'), font=self.my_font, command=lambda value='机器状态': self.thread_it(self.segmented_button_callback()))
|
||||
self.seg_button = customtkinter.CTkSegmentedButton(self.tabview.tab('Automatic Test'), font=self.my_font, command=lambda value='机器状态': self.thread_it(self.segmented_button_callback))
|
||||
self.seg_button.grid(row=1, column=2, columnspan=12, padx=(20, 10), pady=(10, 10), sticky="ew")
|
||||
self.seg_button.configure(values=["无效功能", "触发急停", "停止运动", "继续运动", "零点位姿", "机器状态", "告警信息"])
|
||||
self.seg_button.set("无效功能")
|
||||
@ -150,7 +150,7 @@ class App(customtkinter.CTk):
|
||||
# create progress bar
|
||||
self.progressbar = customtkinter.CTkProgressBar(self.tabview.tab('Automatic Test'))
|
||||
self.progressbar.grid(row=5, column=1, padx=5, pady=5, sticky="ew")
|
||||
self.progressbar.configure(mode="determinnate")
|
||||
self.progressbar.configure(mode="determinnate", width=10)
|
||||
self.progressbar.start()
|
||||
# create widgits_at
|
||||
for widgit in widgits_at:
|
||||
@ -189,44 +189,70 @@ class App(customtkinter.CTk):
|
||||
tkinter.messagebox.showwarning(title="版本更新", message="连接服务器失败，无法确认当前是否是最新版本......")
|
||||
# functions below ↓ ----------------------------------------------------------------------------------------
|
||||
|
||||
def segmented_button_callback(self):
|
||||
value = self.seg_button.get()
|
||||
print(f"segment button is triggered: {value}")
|
||||
match value:
|
||||
case '触发急停':
|
||||
pass
|
||||
case '停止运动':
|
||||
pass
|
||||
case '继续运动':
|
||||
pass
|
||||
case '零点位姿':
|
||||
pass
|
||||
case '机器状态':
|
||||
pass
|
||||
case '告警信息':
|
||||
pass
|
||||
|
||||
def detect_network(self):
|
||||
current_path = dirname(__file__)
|
||||
while True:
|
||||
with open(f'{current_path}/../assets/templates/heartbeat', 'r', encoding='utf-8') as f_h:
|
||||
pb_color = 'green' if f_h.read().strip() == '1' else 'red'
|
||||
self.progressbar.configure(progress_color=pb_color)
|
||||
sleep(3)
|
||||
|
||||
def thread_it(self, func, *args):
|
||||
""" 将函数打包进线程 """
|
||||
self.myThread = Thread(target=func, args=args)
|
||||
self.myThread.daemon = True # 主线程退出就直接让子线程跟随退出,不论是否运行完成。
|
||||
self.myThread.start()
|
||||
|
||||
def segmented_button_callback(self):
|
||||
value = self.seg_button.get()
|
||||
self.textbox.configure(state='normal')
|
||||
|
||||
match value:
|
||||
case '触发急停':
|
||||
self.textbox.delete(index1='1.0', index2='end')
|
||||
self.write2textbox(f"segment button is triggered: {value}")
|
||||
case '停止运动':
|
||||
self.textbox.delete(index1='1.0', index2='end')
|
||||
self.write2textbox(f"segment button is triggered: {value}")
|
||||
case '继续运动':
|
||||
self.textbox.delete(index1='1.0', index2='end')
|
||||
self.write2textbox(f"segment button is triggered: {value}")
|
||||
case '零点位姿':
|
||||
self.textbox.delete(index1='1.0', index2='end')
|
||||
self.write2textbox(f"segment button is triggered: {value}")
|
||||
case '机器状态':
|
||||
self.textbox.configure(state='normal')
|
||||
self.textbox.delete(index1='1.0', index2='end')
|
||||
with open(f'{current_path}/../assets/templates/heartbeat', 'r', encoding='utf-8') as f_h:
|
||||
connection_state = f_h.read().strip()
|
||||
|
||||
if connection_state == '0':
|
||||
self.write2textbox("无法连接机器人，检查是否已经使用Robot Assist软件连接机器，重试中...")
|
||||
else:
|
||||
do_brake.main(self.hr, 'get_state', self.write2textbox)
|
||||
|
||||
case '告警信息':
|
||||
self.textbox.delete(index1='1.0', index2='end')
|
||||
self.write2textbox(f"segment button is triggered: {value}")
|
||||
|
||||
self.textbox.configure(state='disable')
|
||||
|
||||
def detect_network(self):
|
||||
with open(f"{current_path}/../assets/templates/heartbeat", "w", encoding='utf-8') as f_h:
|
||||
f_h.write('0')
|
||||
while True:
|
||||
with open(f'{current_path}/../assets/templates/heartbeat', 'r', encoding='utf-8') as f_h:
|
||||
pb_color = 'green' if f_h.read().strip() == '1' else 'red'
|
||||
self.progressbar.configure(progress_color=pb_color)
|
||||
sleep(3)
|
||||
|
||||
def tabview_click(self):
|
||||
self.initialization()
|
||||
self.textbox.configure(state='normal')
|
||||
self.textbox.delete(index1='1.0', index2='end')
|
||||
self.textbox.configure(state='disabled')
|
||||
|
||||
tab_name = self.tabview.get()
|
||||
if tab_name == 'Data Process':
|
||||
self.menu_main_dp.set("Start Here!")
|
||||
elif tab_name == 'Automatic Test':
|
||||
self.menu_main_at.set("Start Here!")
|
||||
with open(f"{current_path}/../assets/templates/heartbeat", "r", encoding='utf-8') as f_h:
|
||||
connection_state = f_h.read().strip()
|
||||
if connection_state == '0' or self.hr is None:
|
||||
self.hr = openapi.HmiRequest()
|
||||
|
||||
def initialization(self):
|
||||
tab_name = self.tabview.get()
|
||||
@ -492,10 +518,6 @@ class App(customtkinter.CTk):
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
t_hr = Thread(target=openapi.HmiRequest)
|
||||
t_hr.daemon = True
|
||||
t_hr.start()
|
||||
|
||||
aio = App()
|
||||
aio.net_detect = Thread(target=aio.detect_network)
|
||||
aio.net_detect.daemon = True
|
||||
|
||||
Reference in New Issue
Block a user