fix merge while merging from main

This commit is contained in:
gitea 2024-08-16 15:57:00 +08:00
commit 8f342832b2
53 changed files with 6134 additions and 1212 deletions

8
.gitignore vendored
View File

@ -7,4 +7,10 @@ aio/venv
aio/__pycache__/
aio/code/automatic_test/__pycache__/
aio/code/data_process/__pycache__/
aio/assets/templates/c_msg.log
aio/assets/templates/c_msg.log*
aio/code/durable_action/__pycache__/
aio/assets/templates/durable/
aio/assets/templates/.__c_msg.lock
aio/code/commons/__pycache__/
aio/assets/templates/debug.log
dial_gauge/results.xlsx

View File

@ -8,6 +8,7 @@
4. wavelogger 波形处理,几乎不花费时间
5. 制动自动化测试
6. 电机电流自动化测试
7. 耐久工程曲线指标采集(仅适用于六轴)
---
@ -30,9 +31,10 @@
### 四、打包方法
打包时,只需要修改 clibs.py 中的 PREFIX 即可,调试时再修改回来
```
pyinstaller.exe -F --version-file file_version_info.txt -i .\icon.ico .\aio.py -p .\brake.py -p .\current.py
pyinstaller --noconfirm --onedir --windowed --add-data "C:/Users/Administrator/AppData/Local/Programs/Python/Python312/Lib/site-packages/customtkinter;customtkinter/" --version-file ..\assets\file_version_info.txt -i ..\assets\icon.ico ..\code\aio.py -p ..\code\data_process\brake.py -p ..\code\data_process\iso.py -p ..\code\data_process\current.py -p ..\code\data_process\wavelogger.py
pyinstaller --noconfirm --onedir --windowed --optimize 2 --contents-directory . --upx-dir "D:/Syncthing/common/A_Program/upx-4.2.4-win64/" --add-data "C:/Users/Administrator/AppData/Local/Programs/Python/Python312/Lib/site-packages/customtkinter;customtkinter/" --add-data "D:/Syncthing/company/D-测试工作/X-自动化测试/01-AIO/rokae/aio/assets/templates:templates" --version-file ../assets/file_version_info.txt -i ../assets/icon.ico ../code/aio.py -p ../code/data_process/brake.py -p ../code/data_process/iso.py -p ../code/data_process/current.py -p ../code/data_process/wavelogger.py -p ../code/commons/openapi.py -p ../code/commons/clibs.py -p ../code/automatic_test/btn_functions.py -p ../code/automatic_test/do_current.py -p ../code/automatic_test/do_brake.py -p ../code/durable_action/factory_test.py
```
---
@ -132,14 +134,18 @@ pyinstaller --noconfirm --onedir --windowed --add-data "C:/Users/Administrator/A
> **需要注意的点**
1. 使用之前需要手动修改!!负载信息!!点位信息!!,确保所有点位不会发生撞击之后,再进行自动化测试
2. 工程文件不能手动重命名需要重命名存档可以导入HMI然后另存为
3. 务必正确填写configs.xlsx中的Target页面A1单元格可以选择正负方向急停但不完全保证100%大概有95%左右的准确度
4. 由于xCore系统问题运行过程中可能会出现机器人宕机问题如果遇到可以手动重启控制柜重新运行
5. 运行过程中,如果是因为机器问题无法达到额定百分比速度,会在日志输出框提示,注意观察
6. 运行自动化程序之前,确保机器处于正常状态,无故障,未触发急停
7. 需要额外硬件接线详细参考configs.xlsx中急停接线图sheet页
8. 注意观察二轴100%臂展时,是否可以获取到正确的数据
1. 修改该文件点位信息
2. 修改“编程”→“工具列表”中的工具信息制动只允许使用tool33/tool66/tool100/inertia这四个工具不可重命名
3. 除了调试行,其他行请勿修改,包括增加空行等,尤其是有特别注释的地方
4. stop0_related半静态任务需要注意DO0_0的命名不同控制柜可能不一样
5. 需要导入 autotest.xml 寄存器文件以及新建modbus总线设备命名为autotest其他默认即可
6. 其他无需修改如有其他需求可联系fanmingfu@rokae.com沟通
7. 需要额外硬件接线详细参考configs.xlsx中急停接线图sheet页其中DO0_0视硬件情况可能为其他名称
8. 运行自动化程序之前,确保机器处于正常状态,无故障,未触发急停
9. 运行过程中,如果是因为机器问题无法达到额定百分比速度,会在日志输出框提示,注意观察
10. 由于xCore系统问题运行过程中可能会出现机器人宕机问题如果遇到可以手动重启控制柜重新运行
11. 务必正确填写configs.xlsx中的Target页面A1单元格可以选择正负方向急停
12. 工程文件可以手动重命名,按照机型存档,或者导出用于自动化测试
#### 6) 电机电流自动化测试
@ -150,7 +156,16 @@ pyinstaller --noconfirm --onedir --windowed --add-data "C:/Users/Administrator/A
> **需要注意的点**
基本同第五点
基本同第五点,针对五轴以及以下的机型,缺失轴数据可以填写其他有效轴任意一轴的点位信息
#### 7) 耐久工程曲线指标采集
可实现固定周期指定曲线指标的采集,用于长时间观察指标的变化情况,输入文件:
- configs.xlsx执行之前需要手动修改好configs.xlsx中的参数以及间隔时间
- target.zip需要确认工程点位和动作无问题后保存导出
> 重新运行时,必须突出软件,重新运行
#### 其他
customtkinter的tabview组件不支持修改字体大小可以参考 [Changing Font of a Tabview](https://github.com/TomSchimansky/CustomTkinter/issues/2296) 进行手动修改源码实现:
@ -464,4 +479,125 @@ v0.1.9.1(2024/07/12)
- socket类种修改tab_name的值为'openapi'
v0.1.9.2(2024/07/13)
1. [current: current.py]: 在find_point函数种当无法找到正确点位时继续执行而不是直接终止执行
1. [APIs: do_brake.py]
- 修改ready_to_go信号的接收逻辑适配大负载机型
2. [APIs: do_current.py]
- 修改ready_to_go信号的接收逻辑适配大负载机型
- 调整单轴测试时间为35s适配大负载机型调整堵转电流持续时间15s适当减少测试时间
- 将act信号置为False的动作放在初始化增加程序健壮性
- 修改所有输出文件的命名,在扩展名之前加入时间戳
- 删除多余的时序矫正语句——item['value'].reverse(),使输出的曲线为平滑的自然顺序
3. [current: current.py]
- 在find_point函数种当无法找到正确点位时继续执行而不是直接终止执行
- max功能计算逻辑矫正应该是取绝对值的最大值
- 整体梳理了trq/trqh的传递路径现已修正完毕
- 减速比rr数据源修改为configs.xlsx
4. 在current工程main函数增加 VelSet 100语句
v0.1.9.3(2024/07/15)
1. [APIs: openapi.py]
- 修改modbus连接失败报错输出形式使之只在automatic test页面显示
- 将该文件移动至toplevel为后面扩展做准备
- 修改heartbeat文件路径使后续打包的时候更方便
2. [APIs: aio.py]
- 修改heartbeat文件路径使后续打包的时候更方便
- 修改write2textbox函数的打印逻辑先判断网络相关
v0.1.9.4(2024/07/15)
1. [profile: aio.py]完善durable text相关逻辑
2. [profile: do_brake/do_current/btn_functions.py]删除validate_resp函数修改execution函数
3. [profile: factory_test.py]
- 新增耐久/老化测试程序
- 实现六轴折线图显示
4. [profile: openapi.py]:多次合并遗留问题处理
5. templates文件夹组织架构调整
v0.2.0.0(2024/07/17)
1. [profile: aio.py]
- 增加velocity相关逻辑
- 修改负载信息为曲线信息
2. [profile: factory_test.py]
- 增加velocity相关逻辑
3. [profile: current.py]
- 修正减速比获取的规则
4. [profile: openapi.py]
- HmiRequest模块日志取消记录move.monitor相关
- HmiRequest模块增加了durable_lock变量控制文件读写互斥
v0.2.0.1(2024/07/19)
1. [main: aio.py]
- 修改了x轴显示使之为时间刻度
- 修改pre_warning函数增加了durable test的初始化
2. [main: factory_test.py]
- 增加了数据计算错误的判断逻辑
- 增加了历史数据保存的逻辑
- 增加了文件读写互斥的逻辑
- 修改功能为输出有效电流和最大电流,并将数据结构简化
v0.2.0.2(2024/07/26)
1. [main: current.py]
- 修正堵转电流无法正确写入结果文件的问题
2. [main: do_brake.py]
- 初始速度采集等待时间设置为可通过configs.xlsx配置文件调整的
- 初次速度采集停止逻辑修改为tasks.stop指令未验证
- 急停信号触发前pending时间设置为固定值10s
- 实现正负方向速度采集逻辑
- 工程名变更逻辑实现修改为通配符,方便后续根据机型保存文件
- 增加超差后写诊断的逻辑并可以通过configs.xlsx配置文件调整
- 程序输出中增加时间戳,方便调试定位日志时间
3. [main: do_current.py]
- 工程名变更逻辑实现修改为通配符,方便后续根据机型保存文件
4. 为工程文件添加更详细的注释
5. 补充了do_current/do_brake的流程图
6. [main: openapi.py]
- 将modbus motor_on/off的实现方法改为高电平脉冲触发
7. configs.xlsx配置表新增write_diagnosis/get_init_speed两个参数
v0.2.0.3(2024/07/27)
1. [APIs: do_brake.py]: 精简程序,解决 OOM 问题
2. [APIs: do_current.py]: 精简程序,解决 OOM 问题
3. [APIs: factory_test.py]: 精简程序,解决 OOM 问题
4. [APIsL openapi.py]
- 心跳修改为 1 s因为 OOM 问题的解决依赖于长久的打开曲线开关,此时对于 hr.c_msg 的定时清理是个挑战,将心跳缩短,有利于清理日志后,避免丢失心跳
- 新增 diagnosis.save 命令,但是执行时,有问题,待解决
v0.2.0.4(2024/07/30)
1. [APIs: do_brake.py]: 修复制动数据处理过程中,只取曲线的最后 240 个数据
2. [APIs: aio.py]: 判定版本处,删除 self.destroy(),因为该语句会导致异常发生
- 心跳修改为 1s因为 OOM 问题的解决依赖于长久的打开曲线开关,此时对于 hr.c_msg 的定时清理是个挑战,将心跳缩短,有利于清理日志后,避免丢失心跳
- 新增 diagnosis.save 命令,但是执行时,有问题,待解决
v0.2.0.5(2024/07/31)
此版本改动较大,公共部分做了规整,放置到新建文件夹 commons 当中,并所有自定义模块引入 logging 模块,记录重要信息
1. [t_change_ui: clibs.py]
- 调整代码组织结构,新增模块,将公共函数以及类合并入此
- 将一些常量放入该模块
- 引入logging/concurrent_log_handler模块并作初始化操作供其他模块使用按50M切割最多保留10份
- prj_to_xcore函数设置工程名部分重写修复了多个prj工程可能不能执行的问题并优化输入密码的部分
2. [t_change_ui: openapi.py]
- 完全重写了 get_from_id 函数,使更精准
- 在 msg_storage 函数中,增加 logger保留所有响应消息
- 删除 heartbeat 函数中的日志保存功能部分
- 心跳再次修改为 2s...
3. [t_change_ui: aio.py]
- 增加了日志初始化部分
- detect_network 函数中修改重新实例化HR间隔为 4s对应心跳
- create_plot 函数中增加 close('all'),解决循环画图不销毁占用内存的问题
4. [t_change_ui: do_brake.py]
- 使用一直打开曲线的方法规避解决了 OOM 的问题,同时修改数据处理方式,只取最后 12s
- 优化 ssh 输入密码的部分
5. [t_change_ui: do_current.py]
- 保持电流,只取最后 15s
- 优化 ssh 输入密码的部分
6. [t_change_ui: all the part]: 引入 commons 包,并定制了 logging 输出,后续持续优化
7. [APIs: btn_functions.py]: 重写了告警输出函数,从日志中拿数据
8. [APIs: aio.py]: 将日志框输出的内容,也保存至日志文件
9. [APIs: do_brake.py]
- 修改获取初始速度的逻辑只获取configs文件中配置的时间内的速度
- 新增 configs 参数 single_brake可针对特定条件做测试
10. [APIs: all]: 添加了 logger.setLevel(INFO),只有添加这个,单个模块内才生效
v0.2.0.6(2024/08/09)
1. [t_change_ui: all files]
- 修改了 logger 的实现
- 尤其是 clibs.py使用日志字典重写了日志记录的功能

Binary file not shown.

View File

@ -6,8 +6,8 @@ VSVersionInfo(
ffi=FixedFileInfo(
# filevers and prodvers should be always a tuple with four items: (1, 2, 3, 4)
# Set not needed items to zero 0.
filevers=(0, 1, 9, 2),
prodvers=(0, 1, 9, 2),
filevers=(0, 2, 0, 5),
prodvers=(0, 2, 0, 5),
# Contains a bitmask that specifies the valid bits 'flags'r
mask=0x3f,
# Contains a bitmask that specifies the Boolean attributes of the file.
@ -31,12 +31,12 @@ VSVersionInfo(
'040904b0',
[StringStruct('CompanyName', 'Rokae - https://www.rokae.com/'),
StringStruct('FileDescription', 'All in one automatic toolbox'),
StringStruct('FileVersion', '0.1.9.2 (2024-07-13)'),
StringStruct('FileVersion', '0.2.0.5 (2024-08-02)'),
StringStruct('InternalName', 'AIO.exe'),
StringStruct('LegalCopyright', '© 2024-2024 Manford Fan'),
StringStruct('OriginalFilename', 'AIO.exe'),
StringStruct('ProductName', 'AIO'),
StringStruct('ProductVersion', '0.1.9.2 (2024-07-13)')])
StringStruct('ProductVersion', '0.2.0.5 (2024-08-02)')])
]),
VarFileInfo([VarStruct('Translation', [1033, 1200])])
]

Binary file not shown.

View File

@ -1,9 +1,9 @@
openpyxl==3.1.2
pdfplumber==0.11.0
concurrent_log_handler==0.9.25
customtkinter==5.2.2
Jinja2==3.1.4
lxml==5.2.2
numpy==1.26.4
matplotlib==3.9.1
numpy==2.0.1
openpyxl==3.1.2
pandas==2.2.2
pillow==10.3.0
pyinstaller==6.7.0
paramiko==3.4.0
pdfplumber==0.11.0
pymodbus==3.6.9

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

@ -5,8 +5,8 @@
"data": {
"open": false,
"display_open": false,
"overrun": true,
"turn_area": true,
"overrun": false,
"turn_area": false,
"delay_motion": false
}
}

View File

@ -1 +1 @@
0.1.9.2 @ 07/13/2024
0.2.0.5 @ 08/02/2024

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@ -1 +0,0 @@
__all__ = ['automatic_test', 'data_process']

View File

@ -1,21 +1,35 @@
import tkinter
from os.path import exists, dirname
from os import getcwd
from os.path import exists
from os import getcwd, remove
from threading import Thread
import tkinter.messagebox
import customtkinter
from time import time, strftime, localtime, sleep
from urllib.request import urlopen
from socket import setdefaulttimeout
from data_process import *
from automatic_test import *
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
from data_process import brake, current, iso, wavelogger
from automatic_test import do_current, do_brake, btn_functions
from durable_action import factory_test
from commons import openapi, clibs
from matplotlib.pyplot import rcParams, figure, subplots_adjust, close
from matplotlib import use
from pandas import DataFrame, read_excel
current_path = dirname(__file__)
with open(clibs.log_data_hmi, 'w') as _hmi, open(clibs.log_data_debug, 'w', encoding='utf-8') as _debug:
for i in range(1, 11):
try:
remove(f'{clibs.log_data_hmi}.{i}')
except FileNotFoundError:
pass
logger = clibs.log_prod
logger.info("日志文件初始化完成...")
use('Agg')
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
customtkinter.set_window_scaling(1.1) # window geometry dimensions
setdefaulttimeout(3)
# global vars
btns_func = {
'start': {'btn': '', 'row': 1, 'text': '开始运行'},
@ -35,6 +49,10 @@ widgits_at = {
'path': {'label': '', 'entry': '', 'row': 2, 'col': 2, 'text': '数据文件夹路径'},
'loadsel': {'label': '', 'optionmenu': '', 'row': 2, 'col': 1, 'text': '负载信息'},
}
widgits_da = {
'path': {'label': '', 'entry': '', 'row': 1, 'col': 2, 'text': '数据文件夹路径'},
'curvesel': {'label': '', 'optionmenu': '', 'row': 1, 'col': 1, 'text': '指标选择'},
}
class App(customtkinter.CTk):
@ -44,6 +62,11 @@ class App(customtkinter.CTk):
self.w_param = 84
self.hr = None
self.md = None
self.canvas = None
self.flg = 0
self.df_copy = None
self.old_curve = None
self.myThread = None
# =====================================================================
# configure window
self.title("AIO - All in one automatic toolbox")
@ -70,7 +93,7 @@ class App(customtkinter.CTk):
btns_func['log']['btn'].configure(command=lambda: self.thread_it(self.func_log_callback))
btns_func['end']['btn'].configure(command=lambda: self.thread_it(self.func_end_callback))
# create version info
self.label_version = customtkinter.CTkLabel(self.frame_func, justify='left', text="Vers: 0.1.9.1\nDate: 07/12/2024", font=self.my_font, text_color="#4F4F4F")
self.label_version = customtkinter.CTkLabel(self.frame_func, justify='left', text="Vers: 0.2.0.5\nDate: 08/02/2024", font=self.my_font, text_color="#4F4F4F")
self.frame_func.rowconfigure(6, weight=1)
self.label_version.grid(row=6, column=0, padx=20, pady=20, sticky='s')
# =====================================================================
@ -79,6 +102,7 @@ class App(customtkinter.CTk):
self.tabview.grid(row=0, column=1, padx=10, pady=5, sticky="nsew")
self.tabview.add("Data Process")
self.tabview.add("Automatic Test")
self.tabview.add("Durable Action")
# create main menu for data process
self.menu_main_dp = customtkinter.CTkOptionMenu(self.tabview.tab('Data Process'), values=["init", "brake", "current", "iso", "wavelogger"], font=self.my_font, text_color='yellow', button_color='red', fg_color='green', command=self.func_main_callback)
self.menu_main_dp.grid(row=1, column=1, sticky='we', padx=5, pady=10)
@ -114,7 +138,7 @@ class App(customtkinter.CTk):
# For automatic test tab START =====================================================================
# create buttons
self.seg_button = customtkinter.CTkSegmentedButton(self.tabview.tab('Automatic Test'), dynamic_resizing=False, font=customtkinter.CTkFont(size=16, weight='bold'), 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.grid(row=1, column=2, columnspan=12, padx=(65, 10), pady=(10, 10), sticky="ew")
self.seg_button.configure(dynamic_resizing=False, values=["功能切换", "触发急停", "恢复急停", "待定功能", "功能待定", "机器状态", "告警信息"])
self.seg_button.set("功能切换")
# create progress bar
@ -126,7 +150,7 @@ class App(customtkinter.CTk):
for widgit in widgits_at:
if widgit == 'path':
widgits_at[widgit]['label'] = customtkinter.CTkLabel(self.tabview.tab('Automatic Test'), text=f'{widgit.upper()}', font=self.my_font)
widgits_at[widgit]['label'].grid(row=widgits_at[widgit]['row'], column=widgits_at[widgit]['col'], sticky='e', pady=5)
widgits_at[widgit]['label'].grid(row=widgits_at[widgit]['row'], column=widgits_at[widgit]['col'], sticky='e', padx=(20, 5), pady=5)
widgits_at[widgit]['entry'] = customtkinter.CTkEntry(self.tabview.tab('Automatic Test'), width=670, placeholder_text=widgits_at[widgit]['text'], font=self.my_font)
widgits_at[widgit]['entry'].grid(row=widgits_at[widgit]['row'], column=widgits_at[widgit]['col']+1, columnspan=11, padx=(5, 10), pady=5, sticky='we')
widgits_at[widgit]['entry'].configure(state='disabled')
@ -136,6 +160,23 @@ class App(customtkinter.CTk):
widgits_at[widgit]['optionmenu'].set(widgits_at[widgit]['text'])
widgits_at[widgit]['optionmenu'].configure(state='disabled')
# For automatic test tab END =====================================================================
# For durable_action tab START =====================================================================
# create progress bar
self.progressbar_da = customtkinter.CTkProgressBar(self.tabview.tab('Durable Action'))
self.progressbar_da.grid(row=2, column=1, padx=5, pady=5, sticky="ew")
self.progressbar_da.configure(mode="determinnate", width=self.w_param)
self.progressbar_da.start()
for widgit in widgits_da:
if widgit == 'path':
widgits_da[widgit]['label'] = customtkinter.CTkLabel(self.tabview.tab('Durable Action'), text=f'{widgit.upper()}', font=self.my_font)
widgits_da[widgit]['label'].grid(row=widgits_da[widgit]['row'], column=widgits_da[widgit]['col'], sticky='e', padx=(20, 5), pady=10)
widgits_da[widgit]['entry'] = customtkinter.CTkEntry(self.tabview.tab('Durable Action'), width=670, placeholder_text=widgits_da[widgit]['text'], font=self.my_font)
widgits_da[widgit]['entry'].grid(row=widgits_da[widgit]['row'], column=widgits_da[widgit]['col']+1, columnspan=11, padx=(5, 10), pady=10, sticky='we')
elif widgit in ['curvesel']:
widgits_da[widgit]['optionmenu'] = customtkinter.CTkOptionMenu(self.tabview.tab('Durable Action'), dynamic_resizing=False, button_color='#708090', fg_color='#778899', values=['device_servo_trq_feedback', '[max] device_servo_trq_feedback'], font=self.my_font)
widgits_da[widgit]['optionmenu'].grid(row=widgits_da[widgit]['row'], column=widgits_da[widgit]['col'], padx=5, pady=10, sticky='we')
widgits_da[widgit]['optionmenu'].set(widgits_da[widgit]['text'])
# For durable_action tab END =====================================================================
# create textbox
self.textbox = customtkinter.CTkTextbox(self, wrap='none', font=customtkinter.CTkFont(family="consolas", size=14), text_color="blue", fg_color='#E9E9E9', border_width=2, border_color='#CDCDCD', border_spacing=5)
self.textbox.grid(row=6, column=1, columnspan=13, ipadx=10, ipady=10, padx=10, pady=(5, 10), sticky='nsew')
@ -153,6 +194,60 @@ class App(customtkinter.CTk):
tkinter.messagebox.showwarning(title="版本更新", message="连接服务器失败,无法确认当前是否是最新版本......")
# functions below ↓ ----------------------------------------------------------------------------------------
def create_canvas(self, _figure):
self.canvas = FigureCanvasTkAgg(_figure, self.tabview.tab('Durable Action'))
self.canvas.draw()
self.canvas.get_tk_widget().configure(height=600)
self.canvas.get_tk_widget().grid(row=3, column=1, rowspan=3, columnspan=13, padx=20, pady=10, sticky="nsew")
def create_plot(self):
rcParams['font.sans-serif'] = ['SimHei']
rcParams['axes.unicode_minus'] = False
rcParams['figure.dpi'] = 100
rcParams['font.size'] = 14
rcParams['lines.marker'] = 'o'
curvesel = widgits_da['curvesel']['optionmenu'].get()
while True:
if not self.hr.durable_lock:
self.hr.durable_lock = 1
if curvesel == 'device_servo_trq_feedback':
df = read_excel(clibs.durable_data_current_xlsx)
_title = 'device_servo_trq_feedback'
elif curvesel == '[max] device_servo_trq_feedback':
_title = '[max] device_servo_trq_feedback'
df = read_excel(clibs.durable_data_current_max_xlsx)
else:
_title = 'device_servo_trq_feedback'
df = read_excel(clibs.durable_data_current_xlsx)
self.hr.durable_lock = 0
break
else:
sleep(1)
if not df.equals(self.df_copy) or self.flg == 0 or curvesel != self.old_curve:
self.flg = 1
self.df_copy = df.copy()
self.old_curve = widgits_da['curvesel']['optionmenu'].get()
close('all')
_figure = figure(frameon=True, facecolor='#E9E9E9')
subplots_adjust(left=0.04, right=0.98, bottom=0.1, top=0.95)
_ = df['time'].to_list()
_xticks = [str(_i) for _i in _]
ax = _figure.add_subplot(1, 1, 1)
ax.set_xticks(range(len(_xticks)))
ax.set_xticklabels(_xticks)
df.plot(grid=True, x='time', y='axis1', ax=ax)
df.plot(grid=True, x='time', y='axis2', ax=ax)
df.plot(grid=True, x='time', y='axis3', ax=ax)
df.plot(grid=True, x='time', y='axis4', ax=ax)
df.plot(grid=True, x='time', y='axis5', ax=ax)
df.plot(grid=True, x='time', y='axis6', ax=ax, title=_title, legend='upper left', rot=30)
self.create_canvas(_figure)
def thread_it(self, func, *args):
""" 将函数打包进线程 """
self.myThread = Thread(target=func, args=args)
@ -166,7 +261,7 @@ class App(customtkinter.CTk):
self.seg_button.configure(state='disabled')
# self.tabview.configure(state='disabled')
self.textbox.delete(index1='1.0', index2='end')
with open(f'{current_path}/../assets/templates/heartbeat', 'r', encoding='utf-8') as f_h:
with open(clibs.heartbeat, 'r', encoding='utf-8') as f_h:
c_state = f_h.read().strip()
if c_state == '0' and value != '功能切换':
@ -181,32 +276,44 @@ class App(customtkinter.CTk):
# self.tabview.configure(state='normal')
def detect_network(self):
with open(f"{current_path}/../assets/templates/heartbeat", "w", encoding='utf-8') as f_hb:
df = DataFrame(clibs.durable_data_current)
df.to_excel(clibs.durable_data_current_xlsx, index=False)
df = DataFrame(clibs.durable_data_current_max)
df.to_excel(clibs.durable_data_current_max_xlsx, index=False)
with open(clibs.heartbeat, "w", encoding='utf-8') as f_hb:
f_hb.write('0')
self.hr = openapi.HmiRequest(self.write2textbox)
self.md = openapi.ModbusRequest(self.write2textbox)
while True:
with open(f'{current_path}/../assets/templates/heartbeat', 'r', encoding='utf-8') as f_hb:
if self.tabview.get() == 'Durable Action':
self.create_plot()
with open(clibs.heartbeat, 'r', encoding='utf-8') as f_hb:
c_state = f_hb.read().strip()
pb_color = 'green' if c_state == '1' else 'red'
self.progressbar.configure(progress_color=pb_color)
self.progressbar_da.configure(progress_color=pb_color)
if c_state == '0':
self.hr.t_bool = False
sleep(3)
sleep(4)
del self.hr
self.hr = openapi.HmiRequest(self.write2textbox)
sleep(3)
def tabview_click(self):
self.initialization()
tab_name = self.tabview.get()
if tab_name == 'Data Process':
self.flg = 0
self.menu_main_dp.set("Start Here!")
elif tab_name == 'Automatic Test':
self.flg = 0
self.menu_main_at.set("Start Here!")
self.seg_button.configure(state='normal')
elif tab_name == 'Durable Action':
pass
def initialization(self):
tab_name = self.tabview.get()
@ -227,7 +334,7 @@ class App(customtkinter.CTk):
self.menu_sub_dp.grid_forget()
elif tab_name == 'Automatic Test':
for widgit in widgits_at:
if widgit in ['path', 'av1', 'av2', 'av3', 'av4', 'av5', 'av6', 'rc1', 'rc2', 'rc3', 'rc4', 'rc5', 'rc6', 'rr1', 'rr2', 'rr3', 'rr4', 'rr5', 'rr6']:
if widgit in ['path', ]:
widgits_at[widgit]['label'].configure(text=f'{widgit.upper()}', text_color='black')
widgits_at[widgit]['entry'].delete(0, tkinter.END)
widgits_at[widgit]['entry'].configure(placeholder_text=widgits_at[widgit]['text'], state='normal')
@ -237,6 +344,15 @@ class App(customtkinter.CTk):
widgits_at[widgit]['optionmenu'].set(widgits_at[widgit]['text'])
widgits_at[widgit]['optionmenu'].configure(state='disabled')
self.seg_button.set("功能切换")
elif tab_name == 'Durable Action':
for widgit in widgits_da:
if widgit in ['path', ]:
widgits_da[widgit]['label'].configure(text=f'{widgit.upper()}', text_color='black')
widgits_da[widgit]['entry'].delete(0, tkinter.END)
widgits_da[widgit]['entry'].configure(placeholder_text=widgits_at[widgit]['text'], state='normal')
elif widgit in ['curvesel']:
widgits_da[widgit]['optionmenu'].configure(state='normal')
widgits_da[widgit]['optionmenu'].set(widgits_da[widgit]['text'])
def func_main_callback(self, func_name):
self.initialization()
@ -332,6 +448,7 @@ class App(customtkinter.CTk):
self.textbox.tag_add(color, 'insert', 'end')
self.textbox.tag_config(tagName=color, foreground=color)
tab_name_cur = self.tabview.get()
logger.info(text)
if tab_name == tab_name_cur:
if wait != 0:
@ -347,7 +464,7 @@ class App(customtkinter.CTk):
self.textbox.insert(index='end', text=text + '\n', tags=color)
self.textbox.update()
self.textbox.see('end')
elif tab_name == 'openapi' and tab_name_cur == 'Automatic Test':
elif tab_name == 'openapi' and tab_name_cur == 'Automatic Test' or tab_name_cur == 'Durable Action':
if wait != 0:
self.textbox.insert(index='end', text=text, tags=color)
self.textbox.update()
@ -453,12 +570,24 @@ class App(customtkinter.CTk):
return 0, 0
else:
return 0, 0
elif tab_name == 'Durable Action':
path = widgits_da['path']['entry'].get().strip()
curvesel = widgits_da['curvesel']['optionmenu'].get()
c1 = exists(path)
c2 = curvesel in ['device_servo_trq_feedback', '[max] device_servo_trq_feedback']
if c1 and c2:
return 7, path, curvesel
else:
return 0, 0
def func_start_callback(self):
self.textbox.delete(index1='1.0', index2='end')
flag, *args = self.check_param()
func_dict = {1: brake.main, 2: current.main, 3: iso.main, 4: wavelogger.main, 5: do_brake.main, 6: do_current.main}
func_dict = {
1: brake.main, 2: current.main, 3: iso.main, 4: wavelogger.main, 5: do_brake.main, 6: do_current.main,
7: factory_test.main
}
if flag == 1:
func_dict[flag](path=args[0], vel=args[1], trq=args[2], estop=args[3], w2t=self.write2textbox)
elif flag == 2:
@ -473,10 +602,19 @@ class App(customtkinter.CTk):
elif flag == 6:
self.pre_warning()
func_dict[flag](path=args[0], hr=self.hr, md=self.md, loadsel=args[1], w2t=self.write2textbox)
elif flag == 7:
self.pre_warning()
func_dict[flag](path=args[0], hr=self.hr, md=self.md, w2t=self.write2textbox)
else:
tkinter.messagebox.showerror(title="参数错误", message="请检查对应参数是否填写正确!", )
def pre_warning(self):
if self.tabview.get() == 'Durable Action':
df = DataFrame(clibs.durable_data_current)
df.to_excel(clibs.durable_data_current_xlsx, index=False)
df = DataFrame(clibs.durable_data_current_max)
df.to_excel(clibs.durable_data_current_max_xlsx, index=False)
if tkinter.messagebox.askyesno(title="开始运行", message="确认机器已按照测试规范更新固件,并提按照测试机型前修改好工程?"):
pass
else:

View File

@ -1 +0,0 @@
__all__ = ['openapi', 'btn_functions', 'do_brake', 'do_current']

View File

@ -1,72 +1,55 @@
from json import loads
from sys import argv
from commons import clibs
def validate_resp(_id, response, w2t):
match _id:
case 'DATA ERR':
w2t(f"数据处理错误,需要确认", 0, 4, 'red', tab_name='Automatic Test')
case 'DATA READ ERR':
w2t(f"无法读取数据,需要确认", 0, 3, 'red', tab_name='Automatic Test')
case 'NOT SUPPORT':
w2t(f"不支持的功能,需要确认", 0, 2, 'red', tab_name='Automatic Test')
if not response:
w2t(f"无法获取{id}请求的响应信息", 0, 1, 'red', tab_name='Automatic Test')
def execution(cmd, hr, w2t, **kwargs):
_id = hr.execution(cmd, **kwargs)
_msg = hr.get_from_id(_id)
if not _msg:
w2t(f"无法获取{_id}请求的响应信息", 0, 6, 'red', tab_name='Automatic Test')
else:
_response = loads(_msg)
validate_resp(_id, _response, w2t)
return _response
tab_name = clibs.tab_names['at']
logger = clibs.log_prod
def trigger_estop(md, w2t):
md.trigger_estop()
w2t("触发急停成功,可点击机器状态验证。", 0, 0, 'green', 'Automatic Test')
w2t("触发急停成功,可点击机器状态验证。", 0, 0, 'green', tab_name)
def reset_estop(md, w2t):
md.reset_estop()
w2t("恢复急停成功,可点击机器状态验证。", 0, 0, 'green', 'Automatic Test')
w2t("恢复急停成功,可点击机器状态验证。", 0, 0, 'green', tab_name)
def get_state(hr, w2t):
# 获取机器状态
_response = execution('state.get_state', hr, w2t)
_response = clibs.execution('state.get_state', hr, w2t, tab_name)
stat_desc = {'engine': '上电状态', 'operate': '操作模式', 'rc_state': '控制器状态', 'robot_action': '机器人动作', 'safety_mode': '安全模式', 'servo_mode': '伺服工作模式', 'task_space': '工作任务空间'}
for component, state in _response['data'].items():
w2t(f"{stat_desc[component]}: {state}", tab_name='Automatic Test')
w2t(f"{stat_desc[component]}: {state}", tab_name=tab_name)
# 获取设备伺服信息
_response = execution('device.get_params', hr, w2t)
_response = clibs.execution('device.get_params', hr, w2t, tab_name)
dev_desc = {0: '伺服版本', 1: '伺服参数', 2: '安全板固件', 3: '控制器', 4: '通讯总线', 5: '解释器', 6: '运动控制', 8: '力控版本', 9: '末端固件', 10: '机型文件', 11: '环境包'}
dev_vers = {}
for device in _response['data']['devices']:
dev_vers[device['type']] = device['version']
for i in sorted(dev_desc.keys()):
w2t(f"{dev_desc[i]}: {dev_vers[i]}", tab_name='Automatic Test')
w2t(f"{dev_desc[i]}: {dev_vers[i]}", tab_name=tab_name)
# 设置示教器模式
_response = execution('state.set_tp_mode', hr, w2t, tp_mode='without')
_response = clibs.execution('state.set_tp_mode', hr, w2t, tab_name, tp_mode='without')
def warning_info(hr, w2t):
for msg in hr.c_msg:
if 'alarm' in msg.lower():
w2t(str(loads(msg)), tab_name='Automatic Test')
for msg in hr.c_msg_xs:
if 'alarm' in msg.lower():
w2t(str(loads(msg)), tab_name='Automatic Test')
for postfix in ['', '.2', '.3', '.4', '.5', '.6', '.7', '.8', '.9', '.10']:
log_name = clibs.log_data_hmi + postfix
try:
with open(log_name, 'r', encoding='utf-8') as f_log:
for line in f_log:
if 'alarm' in line:
w2t(line.strip(), tab_name=tab_name)
except FileNotFoundError:
pass
def main(hr, md, func, w2t):
if hr is None:
w2t("无法连接机器人检查是否已经使用Robot Assist软件连接机器重试中...", 0, 49, 'red', tab_name='Automatic Test')
w2t("无法连接机器人检查是否已经使用Robot Assist软件连接机器重试中...", 0, 49, 'red', tab_name)
# func: get_state/
match func:
case 'trigger_estop':

View File

@ -1,36 +1,20 @@
from random import randint
from time import sleep, time
from time import sleep, time, strftime, localtime
from sys import argv
from os import scandir, mkdir
from os.path import exists
from os import mkdir
from paramiko import SSHClient, AutoAddPolicy
from json import loads
from openpyxl import load_workbook
import pandas
from pandas import DataFrame, concat
from commons import clibs
RADIAN = 57.3 # 180 / 3.1415926
def traversal_files(path, w2t):
if not exists(path):
msg = f'数据文件夹{path}不存在,请确认后重试......'
w2t(msg, 0, 1, 'red', tab_name='Automatic Test')
else:
dirs = []
files = []
for item in scandir(path):
if item.is_dir():
dirs.append(item.path)
elif item.is_file():
files.append(item.path)
return dirs, files
tab_name = clibs.tab_names['at']
logger = clibs.log_prod
def check_files(path, loadsel, data_dirs, data_files, w2t):
if len(data_dirs) != 0 or len(data_files) != 5:
w2t('初始路径下不允许有文件夹,且初始路径下只能存在如下五个文件,确认后重新运行!', 0, 0, 'red', tab_name='Automatic Test')
w2t(' 1. configs.xlsx\n 2. reach33/reach66/reach100_xxxx.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name='Automatic Test')
w2t('初始路径下不允许有文件夹,且初始路径下只能存在如下五个文件,确认后重新运行!', 0, 0, 'red', tab_name)
w2t(' 1. configs.xlsx\n 2. reach33/reach66/reach100_xxxx.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name)
config_file = reach33 = reach66 = reach100 = prj_file = None
for data_file in data_files:
@ -46,8 +30,8 @@ def check_files(path, loadsel, data_dirs, data_files, w2t):
elif filename.endswith('.zip'):
prj_file = data_file
else:
w2t('初始路径下不允许有文件夹,且初始路径下只能存在如下五个文件,确认后重新运行!', 0, 0, 'red', tab_name='Automatic Test')
w2t(' 1. configs.xlsx\n 2. reach33/reach66/reach100_xxxx.xlsx\n 3. xxxx.zip', 0, 2, 'red', tab_name='Automatic Test')
w2t('初始路径下不允许有文件夹,且初始路径下只能存在如下五个文件,确认后重新运行!', 0, 0, 'red', tab_name)
w2t(' 1. configs.xlsx\n 2. reach33/reach66/reach100_xxxx.xlsx\n 3. xxxx.zip', 0, 2, 'red', tab_name)
if config_file and reach33 and reach66 and reach100 and prj_file:
result_dirs = []
@ -65,74 +49,24 @@ def check_files(path, loadsel, data_dirs, data_files, w2t):
if _reach == 'reach100':
mkdir(f"{path}\\j3\\{dir_name}")
w2t("数据目录合规性检查结束,未发现问题......", tab_name='Automatic Test')
w2t("数据目录合规性检查结束,未发现问题......", 0, 0, 'blue', tab_name)
return config_file, reach33, reach66, reach100, prj_file, result_dirs
else:
w2t('初始路径下不允许有文件夹,且初始路径下只能存在如下五个文件,确认后重新运行!', 0, 0, 'red', tab_name='Automatic Test')
w2t(' 1. configs.xlsx\n 2. reach33/reach66/reach100_xxxx.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name='Automatic Test')
def prj_to_xcore(prj_file):
ssh = SSHClient()
ssh.set_missing_host_key_policy(AutoAddPolicy())
ssh.connect('192.168.0.160', 22, username='luoshi', password='luoshi2019')
sftp = ssh.open_sftp()
sftp.put(prj_file, '/tmp/target.zip')
cmd = 'cd /tmp; rm -rf target/; mkdir target; unzip -d target/ -q target.zip; '
cmd += 'chmod 777 -R target/; rm target.zip'
ssh.exec_command(cmd)
cmd = 'sudo rm -rf /home/luoshi/bin/controller/projects/target; '
cmd += 'sudo mv /tmp/target/ /home/luoshi/bin/controller/projects/'
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
stdin.write('luoshi2019' + '\n')
stdin.flush()
print(stdout.read().decode()) # 必须得输出一下stdout才能正确执行sudo
print(stderr.read().decode()) # 顺便也执行以下stderr
_prj_name = prj_file.split('\\')[-1].removesuffix('.zip')
cmd = 'cd /home/luoshi/bin/controller/; '
cmd += f'sudo mv projects/target/_build/{_prj_name}.prj projects/target/_build/target.prj'
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
stdin.write('luoshi2019' + '\n')
stdin.flush()
print(stdout.read().decode()) # 必须得输出一下stdout才能正确执行sudo
print(stderr.read().decode()) # 顺便也执行以下stderr
ssh.close()
def validate_resp(_id, response, w2t):
match _id:
case 'DATA ERR':
w2t(f"数据处理错误,需要确认", 0, 4, 'red', tab_name='Automatic Test')
case 'DATA READ ERR':
w2t(f"无法读取数据,需要确认", 0, 3, 'red', tab_name='Automatic Test')
case 'NOT SUPPORT':
w2t(f"不支持的功能,需要确认", 0, 2, 'red', tab_name='Automatic Test')
if not response:
w2t(f"无法获取{id}请求的响应信息", 0, 1, 'red', tab_name='Automatic Test')
def execution(cmd, hr, w2t, **kwargs):
_id = hr.execution(cmd, **kwargs)
_msg = hr.get_from_id(_id)
if not _msg:
w2t(f"无法获取{_id}请求的响应信息", 0, 6, 'red', tab_name='Automatic Test')
else:
_response = loads(_msg)
validate_resp(_id, _response, w2t)
return _response
w2t('初始路径下不允许有文件夹,且初始路径下只能存在如下五个文件,确认后重新运行!', 0, 0, 'red', tab_name)
w2t(' 1. configs.xlsx\n 2. reach33/reach66/reach100_xxxx.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name)
def gen_result_file(path, curve_data, axis, _reach, _load, _speed, count):
_d2d_vel = {'hw_joint_vel_feedback': []}
_d2d_trq = {'device_servo_trq_feedback': []}
_d2d_stop = {'device_safety_estop': []}
for data in curve_data:
for data in curve_data[-240:]: # 保留最后12s的数据
dict_results = data['data']
# dict_results.reverse()
for item in dict_results:
try:
item['value'].reverse()
except KeyError:
continue
if item.get('channel', None) == axis-1 and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == axis-1 and item.get('name', None) == 'device_servo_trq_feedback':
@ -140,17 +74,17 @@ def gen_result_file(path, curve_data, axis, _reach, _load, _speed, count):
elif item.get('channel', None) == 0 and item.get('name', None) == 'device_safety_estop':
_d2d_stop['device_safety_estop'].extend(item['value'])
df1 = pandas.DataFrame.from_dict(_d2d_vel)
df2 = pandas.DataFrame.from_dict(_d2d_trq)
df3 = pandas.DataFrame.from_dict(_d2d_stop)
df = pandas.concat([df1, df2, df3], axis=1)
df1 = DataFrame.from_dict(_d2d_vel)
df2 = DataFrame.from_dict(_d2d_trq)
df3 = DataFrame.from_dict(_d2d_stop)
df = concat([df1, df2, df3], axis=1)
_filename = f"{path}\\j{axis}\\reach{_reach}_load{_load}_speed{_speed}\\reach{_reach}_load{_load}_speed{_speed}_{count}.data"
df.to_csv(_filename, sep='\t', index=False)
def run_rl(path, loadsel, hr, md, config_file, prj_file, result_dirs, w2t):
def run_rl(path, loadsel, hr, md, config_file, result_dirs, w2t):
_count = 0
speed_max = 0
_total = 63
display_pdo_params = [
{"name": "hw_joint_vel_feedback", "channel": 0},
{"name": "hw_joint_vel_feedback", "channel": 1},
@ -168,36 +102,59 @@ def run_rl(path, loadsel, hr, md, config_file, prj_file, result_dirs, w2t):
]
wb = load_workbook(config_file, read_only=True)
ws = wb['Target']
if ws.cell(row=1, column=1).value == 'positive':
md.write_pon(True)
elif ws.cell(row=1, column=1).value == 'negative':
md.write_pon(False)
else:
w2t("configs.xlsx中Target页面A1单元格填写不正确检查后重新运行...", 0, 111, 'red', 'Automatic Test')
write_diagnosis = float(ws.cell(row=3, column=10).value)
get_init_speed = float(ws.cell(row=4, column=10).value)
single_brake = str(ws.cell(row=5, column=10).value)
logger.info(f"write_diagnosis = {write_diagnosis}, get_init_speed = {get_init_speed}, single_brake = {single_brake}")
if ws.cell(row=1, column=1).value == 'positive':
md.write_pon(1)
elif ws.cell(row=1, column=1).value == 'negative':
md.write_pon(0)
else:
w2t("configs.xlsx中Target页面A1单元格填写不正确检查后重新运行...", 0, 111, 'red', tab_name)
clibs.execution('diagnosis.open', hr, w2t, tab_name, open=True, display_open=True)
clibs.execution('diagnosis.set_params', hr, w2t, tab_name, display_pdo_params=display_pdo_params)
for condition in result_dirs:
_reach = condition.split('_')[0].removeprefix('reach')
_load = condition.split('_')[1].removeprefix('load')
_speed = condition.split('_')[2].removeprefix('speed')
# for single condition test
_single_axis = 0
if single_brake != '0':
_total = 3
_single_axis = int(single_brake.split('-')[0])
if _reach != single_brake.split('-')[1] or _load != single_brake.split('-')[2] or _speed != single_brake.split('-')[3]:
continue
for axis in range(1, 4):
# for single condition test
if _single_axis != 0 and _single_axis != axis:
continue
md.write_axis(axis)
speed_max = 0
if axis == 3 and _reach != '100':
continue
w2t(f"-"*90, 0, 0, 'purple', tab_name)
for count in range(1, 4):
_count += 1
w2t(f"[{_count}/63-{count}] 正在执行{axis}{condition}的制动测试......", 0, 0, 'purple', 'Automatic Test')
this_time = strftime("%Y-%m-%d %H:%M:%S", localtime(time()))
prj_path = 'target/_build/target.prj'
w2t(f"[{this_time} | {_count}/{_total}] 正在执行 {axis}{condition} 的第 {count} 次制动测试...", 0, 0, 'purple', tab_name)
# 1. 关闭诊断曲线,触发软急停,并解除,目的是让可能正在运行着的机器停下来,切手动模式并下电
md.trigger_estop()
md.reset_estop()
md.write_act(False)
_response = execution('diagnosis.open', hr, w2t, open=False, display_open=False)
sleep(1) # 让曲线彻底关闭
_response = execution('state.switch_manual', hr, w2t)
_response = execution('state.switch_motor_off', hr, w2t)
md.clear_alarm()
md.write_act(0)
sleep(write_diagnosis) # 软急停超差后等待写诊断时间可通过configs.xlsx配置
while count == 1:
# 2. 修改未要执行的场景
ssh = SSHClient()
ssh.set_missing_host_key_policy(AutoAddPolicy())
@ -207,129 +164,133 @@ def run_rl(path, loadsel, hr, md, config_file, prj_file, result_dirs, w2t):
elif ws.cell(row=1, column=1).value == 'negative':
_rl_cmd = f"brake_E(j{axis}_{_reach}_n, j{axis}_{_reach}_p, p_speed, p_tool)"
else:
w2t("configs.xlsx中Target页面A1单元格填写不正确检查后重新运行...", 0, 111, 'red', 'Automatic Test')
w2t("configs.xlsx中Target页面A1单元格填写不正确检查后重新运行...", 0, 111, 'red', tab_name)
_rl_speed = f"VelSet {_speed}"
_rl_tool = f"tool p_tool = tool{loadsel.removeprefix('tool')}"
cmd = 'cd /home/luoshi/bin/controller/; '
cmd += 'sudo sed -i "/brake_E/d" projects/target/_build/brake/main.mod; '
cmd += f'sudo sed -i "/DONOTDELETE/i {_rl_cmd}" projects/target/_build/brake/main.mod; '
cmd += 'sudo sed -i "/VelSet/d" projects/target/_build/brake/main.mod; '
cmd += f'sudo sed -i "/MoveAbsJ/i {_rl_speed}" projects/target/_build/brake/main.mod'
cmd += f'sudo sed -i "/MoveAbsJ/i {_rl_speed}" projects/target/_build/brake/main.mod; '
cmd += 'sudo sed -i "/tool p_tool/d" projects/target/_build/brake/main.mod; '
cmd += f'sudo sed -i "/VelSet/i {_rl_tool}" projects/target/_build/brake/main.mod; '
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
stdin.write('luoshi2019' + '\n')
stdin.flush()
print(stdout.read().decode()) # 必须得输出一下stdout才能正确执行sudo
print(stderr.read().decode()) # 顺便也执行以下stderr
stdout.read().decode() # 需要read一下才能正常执行
stderr.read().decode()
# 3. reload工程后pp2main并且自动模式和上电最后运行程序
prj_path = 'target/_build/target.prj'
_response = execution('overview.reload', hr, w2t, prj_path=prj_path, tasks=['brake', 'stop0_related'])
_response = execution('rl_task.pp_to_main', hr, w2t, tasks=['brake', 'stop0_related'])
_response = execution('state.switch_auto', hr, w2t)
_response = execution('state.switch_motor_on', hr, w2t)
_response = execution('rl_task.run', hr, w2t, tasks=['brake', 'stop0_related'])
for i in range(3):
clibs.execution('overview.reload', hr, w2t, tab_name, prj_path=prj_path, tasks=['brake', 'stop0_related'])
clibs.execution('rl_task.pp_to_main', hr, w2t, tab_name, tasks=['brake', 'stop0_related'])
clibs.execution('state.switch_auto', hr, w2t, tab_name)
clibs.execution('state.switch_motor_on', hr, w2t, tab_name)
clibs.execution('rl_task.run', hr, w2t, tab_name, tasks=['brake', 'stop0_related'])
_t_start = time()
while True:
if md.read_ready_to_go() == 1:
md.write_act(True)
break
else:
sleep(1)
if (time() - _t_start) // 20 > 1:
w2t("20s内未收到机器人的运行信号需要确认RL程序编写正确并正常执行...", 0, 111, 'red', tab_name)
else:
w2t("未收到机器人的运行信号需要确认RL程序编写正确并正常执行...", 0, 111, 'red', 'Automatic Test')
# 4. 第一次打开诊断曲线并执行采集8s之后触发软急停关闭曲线采集找出最大速度传递给RL程序最后清除相关记录
if count == 1:
_response = execution('diagnosis.open', hr, w2t, open=True, display_open=True)
_response = execution('diagnosis.set_params', hr, w2t, display_pdo_params=display_pdo_params)
sleep(10) # 前10秒获取实际最大速度
md.trigger_estop()
_response = execution('diagnosis.open', hr, w2t, open=False, display_open=False)
sleep(1)
# 4. 找出最大速度传递给RL程序最后清除相关记录
sleep(get_init_speed+5) # 冗余5s指定时间后获取实际【正|负】方向的最大速度可通过configs.xlsx配置
clibs.execution('rl_task.stop', hr, w2t, tab_name, tasks=['brake'])
# 找出最大速度
for _msg in hr.c_msg:
_c_msg = hr.c_msg.copy()
_number = 0
for _msg in _c_msg:
if _number > get_init_speed*20: # 最开始回零点的时候,二轴速度可以达到最大值,实际摆动时,某一方向可能达不到
break
if 'diagnosis.result' in _msg:
_number += 1
dict_results = loads(_msg)['data']
for item in dict_results:
if item.get('channel', None) == axis-1 and item.get('name', None) == 'hw_joint_vel_feedback':
_ = abs(RADIAN*sum(item['value'])/len(item['value']))
_ = clibs.RADIAN * sum(item['value']) / len(item['value'])
if ws.cell(row=1, column=1).value == 'positive':
speed_max = max(_, speed_max)
elif ws.cell(row=1, column=1).value == 'negative':
speed_max = min(_, speed_max)
logger.info(f"speed max = {speed_max}")
speed_max = abs(speed_max)
speed_target = float(ws.cell(row=3, column=axis+1).value) * float(_speed) / 100
if speed_max < speed_target*0.95 or speed_max > speed_target*1.05:
w2t(f"Axis: {axis}-{count} | Speed: {speed_max} | Shouldbe: {speed_target}", 0, 0, 'indigo', 'Automatic Test')
w2t(f"Axis: {axis}-{count} | Speed: {speed_max} | Shouldbe: {speed_target}", 0, 0, 'indigo', tab_name)
md.write_speed_max(speed_max)
sleep(1)
for _msg in hr.c_msg:
if 'diagnosis.result' in _msg:
_index = hr.c_msg.index(_msg)
del hr.c_msg[_index:]
hr.c_msg_xs.clear()
if len(hr.c_msg) > 270:
del hr.c_msg[270:]
if speed_max < 10:
md.clear_alarm()
w2t("未获取到正确的速度,即将重新获取...", 0, 0, 'red', tab_name)
continue
else:
break
# 5. 清除软急停重新运行程序并打开曲线发送继续运动信号当速度达到最大值时通过DO触发急停
md.reset_estop()
_response = execution('overview.reload', hr, w2t, prj_path=prj_path, tasks=['brake', 'stop0_related'])
_response = execution('rl_task.pp_to_main', hr, w2t, tasks=['brake', 'stop0_related'])
_response = execution('state.switch_auto', hr, w2t)
_response = execution('state.switch_motor_on', hr, w2t)
_response = execution('rl_task.run', hr, w2t, tasks=['brake', 'stop0_related'])
md.reset_estop() # 其实没必要
md.clear_alarm()
clibs.execution('overview.reload', hr, w2t, tab_name, prj_path=prj_path, tasks=['brake', 'stop0_related'])
clibs.execution('rl_task.pp_to_main', hr, w2t, tab_name, tasks=['brake', 'stop0_related'])
clibs.execution('state.switch_auto', hr, w2t, tab_name)
clibs.execution('state.switch_motor_on', hr, w2t, tab_name)
clibs.execution('rl_task.run', hr, w2t, tab_name, tasks=['brake', 'stop0_related'])
for i in range(3):
if md.read_ready_to_go() == 1:
md.write_act(True)
md.write_act(1)
break
else:
sleep(1)
else:
w2t("未收到机器人的运行信号需要确认RL程序编写正确并正常执行...", 0, 111, 'red', 'Automatic Test')
w2t("未收到机器人的运行信号需要确认RL程序编写正确并正常执行...", 0, 111, 'red', tab_name)
_response = execution('diagnosis.open', hr, w2t, open=True, display_open=True)
_response = execution('diagnosis.set_params', hr, w2t, display_pdo_params=display_pdo_params)
sleep(randint(3, 6))
md.write_probe(True)
sleep(10) # 排除从其他位姿到零点位姿,再到轴极限位姿的时间
md.write_probe(1)
_t_start = time()
while True:
if md.read_brake_done() == 1:
sleep(1) # 保证所有数据均已返回
md.write_probe(False)
_response = execution('diagnosis.open', hr, w2t, open=False, display_open=False)
sleep(1) # 保证所有数据均已返回
sleep(1) # 保证速度归零
md.write_probe(0)
break
else:
if (time() - _t_start) > 30:
w2t(f"30s内未触发急停该条数据无效需要确认RL/Python程序编写正确并正常执行或者判别是否是机器本体问题...", 0, 0, 'red', 'Automatic Test')
md.write_probe(False)
_response = execution('diagnosis.open', hr, w2t, open=False, display_open=False)
sleep(1) # 保证所有数据均已返回
w2t(f"30s内未触发急停该条数据无效需要确认RL/Python程序编写正确并正常执行或者判别是否是机器本体问题比如正负方向速度是否一致...", 0, 0, 'red', tab_name)
md.write_probe(0)
break
else:
sleep(1)
# 6. 保留数据并处理输出
curve_data = []
for _msg in hr.c_msg:
_c_msg = hr.c_msg.copy()
for _msg in _c_msg:
if 'diagnosis.result' in _msg:
curve_data.insert(0, loads(_msg))
else:
for _msg in hr.c_msg:
if 'diagnosis.result' in _msg:
_index = hr.c_msg.index(_msg)
del hr.c_msg[_index:]
hr.c_msg_xs.clear()
break
if len(hr.c_msg) > 270:
del hr.c_msg[270:]
gen_result_file(path, curve_data, axis, _reach, _load, _speed, count)
else:
w2t(f"\n{loadsel.removeprefix('tool')}%负载的制动性能测试执行完毕,如需采集其他负载,须切换负载类型,并更换其他负载,重新执行。", 0, 0, 'green', 'Automatic Test')
w2t(f"\n{loadsel.removeprefix('tool')}%负载的制动性能测试执行完毕,如需采集其他负载,须切换负载类型,并更换其他负载,重新执行。", 0, 0, 'green', tab_name)
def main(path, hr, md, loadsel, w2t):
_s_time = time()
data_dirs, data_files = traversal_files(path, w2t)
data_dirs, data_files = clibs.traversal_files(path, w2t)
config_file, reach33, reach66, reach100, prj_file, result_dirs = check_files(path, loadsel, data_dirs, data_files, w2t)
prj_to_xcore(prj_file)
run_rl(path, loadsel, hr, md, config_file, prj_file, result_dirs, w2t)
clibs.prj_to_xcore(prj_file)
run_rl(path, loadsel, hr, md, config_file, result_dirs, w2t)
_e_time = time()
time_total = _e_time - _s_time
w2t(f"处理总时长:{time_total // 3600:02.0f} h {time_total % 3600 // 60:02.0f} m {time_total % 60:02.0f} s", 0, 0, 'green', 'Automatic Test')
w2t(f"处理总时长:{time_total // 3600:02.0f} h {time_total % 3600 // 60:02.0f} m {time_total % 60:02.0f} s", 0, 0, 'green', tab_name)
if __name__ == '__main__':

View File

@ -1,33 +1,33 @@
import os
from os import mkdir
from time import sleep, time
from sys import argv
from os import scandir
from os.path import exists
from paramiko import SSHClient, AutoAddPolicy
from json import loads
import pandas
from pandas import DataFrame, concat
from commons import clibs
def traversal_files(path, w2t):
if not exists(path):
msg = f'数据文件夹{path}不存在,请确认后重试......'
w2t(msg, 0, 1, 'red', tab_name='Automatic Test')
else:
dirs = []
files = []
for item in scandir(path):
if item.is_dir():
dirs.append(item.path)
elif item.is_file():
files.append(item.path)
return dirs, files
logger = clibs.log_prod
tab_name = clibs.tab_names['at']
display_pdo_params = [
{"name": "hw_joint_vel_feedback", "channel": 0},
{"name": "hw_joint_vel_feedback", "channel": 1},
{"name": "hw_joint_vel_feedback", "channel": 2},
{"name": "hw_joint_vel_feedback", "channel": 3},
{"name": "hw_joint_vel_feedback", "channel": 4},
{"name": "hw_joint_vel_feedback", "channel": 5},
{"name": "device_servo_trq_feedback", "channel": 0},
{"name": "device_servo_trq_feedback", "channel": 1},
{"name": "device_servo_trq_feedback", "channel": 2},
{"name": "device_servo_trq_feedback", "channel": 3},
{"name": "device_servo_trq_feedback", "channel": 4},
{"name": "device_servo_trq_feedback", "channel": 5},
]
def check_files(path, loadsel, data_dirs, data_files, w2t):
if len(data_dirs) != 0 or len(data_files) != 3:
w2t('初始路径下不允许有文件夹,且初始路径下只能存在如下三个文件,确认后重新运行!', 0, 0, 'red', tab_name='Automatic Test')
w2t(' 1. configs.xlsx\n 2. T_电机电流.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name='Automatic Test')
w2t('初始路径下不允许有文件夹,且初始路径下只能存在如下三个文件,确认后重新运行!', 0, 0, 'red', tab_name)
w2t(' 1. configs.xlsx\n 2. T_电机电流.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name)
config_file = current_file = prj_file = None
for data_file in data_files:
@ -39,74 +39,22 @@ def check_files(path, loadsel, data_dirs, data_files, w2t):
elif filename.endswith('.zip'):
prj_file = data_file
else:
w2t('初始路径下不允许有文件夹,且初始路径下只能存在如下三个文件,确认后重新运行!', 0, 0, 'red', tab_name='Automatic Test')
w2t(' 1. configs.xlsx\n 2. T_电机电流.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name='Automatic Test')
w2t('初始路径下不允许有文件夹,且初始路径下只能存在如下三个文件,确认后重新运行!', 0, 0, 'red', tab_name)
w2t(' 1. configs.xlsx\n 2. T_电机电流.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name)
if config_file and current_file and prj_file:
w2t("数据目录合规性检查结束,未发现问题......", tab_name='Automatic Test')
w2t("数据目录合规性检查结束,未发现问题......", tab_name=tab_name)
if loadsel == 'tool100':
os.mkdir(f"{path}\\single")
os.mkdir(f"{path}\\s_1")
os.mkdir(f"{path}\\s_2")
os.mkdir(f"{path}\\s_3")
mkdir(f"{path}\\single")
mkdir(f"{path}\\s_1")
mkdir(f"{path}\\s_2")
mkdir(f"{path}\\s_3")
elif loadsel == 'inertia':
os.mkdir(f"{path}\\inertia")
mkdir(f"{path}\\inertia")
return config_file, current_file, prj_file
else:
w2t('初始路径下不允许有文件夹,且初始路径下只能存在如下三个文件,确认后重新运行!', 0, 0, 'red', tab_name='Automatic Test')
w2t(' 1. configs.xlsx\n 2. T_电机电流.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name='Automatic Test')
def prj_to_xcore(prj_file):
ssh = SSHClient()
ssh.set_missing_host_key_policy(AutoAddPolicy())
ssh.connect('192.168.0.160', 22, username='luoshi', password='luoshi2019')
sftp = ssh.open_sftp()
sftp.put(prj_file, '/tmp/target.zip')
cmd = 'cd /tmp; rm -rf target/; mkdir target; unzip -d target/ -q target.zip; '
cmd += 'chmod 777 -R target/; rm target.zip'
ssh.exec_command(cmd)
cmd = 'sudo rm -rf /home/luoshi/bin/controller/projects/target; '
cmd += 'sudo mv /tmp/target/ /home/luoshi/bin/controller/projects/'
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
stdin.write('luoshi2019' + '\n')
stdin.flush()
print(stdout.read().decode()) # 必须得输出一下stdout才能正确执行sudo
print(stderr.read().decode()) # 顺便也执行以下stderr
_prj_name = prj_file.split('\\')[-1].removesuffix('.zip')
cmd = 'cd /home/luoshi/bin/controller/; '
cmd += f'sudo mv projects/target/_build/{_prj_name}.prj projects/target/_build/target.prj'
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
stdin.write('luoshi2019' + '\n')
stdin.flush()
print(stdout.read().decode()) # 必须得输出一下stdout才能正确执行sudo
print(stderr.read().decode()) # 顺便也执行以下stderr
ssh.close()
def validate_resp(_id, response, w2t):
match _id:
case 'DATA ERR':
w2t(f"数据处理错误,需要确认", 0, 4, 'red', tab_name='Automatic Test')
case 'DATA READ ERR':
w2t(f"无法读取数据,需要确认", 0, 3, 'red', tab_name='Automatic Test')
case 'NOT SUPPORT':
w2t(f"不支持的功能,需要确认", 0, 2, 'red', tab_name='Automatic Test')
if not response:
w2t(f"无法获取{id}请求的响应信息", 0, 1, 'red', tab_name='Automatic Test')
def execution(cmd, hr, w2t, **kwargs):
_id = hr.execution(cmd, **kwargs)
_msg = hr.get_from_id(_id)
if not _msg:
w2t(f"无法获取{_id}请求的响应信息", 0, 7, 'red', tab_name='Automatic Test')
else:
_response = loads(_msg)
validate_resp(_id, _response, w2t)
return _response
w2t('初始路径下不允许有文件夹,且初始路径下只能存在如下三个文件,确认后重新运行!', 0, 0, 'red', tab_name)
w2t(' 1. configs.xlsx\n 2. T_电机电流.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name)
def data_proc_regular(path, filename, channel, scenario_time):
@ -115,21 +63,22 @@ def data_proc_regular(path, filename, channel, scenario_time):
lines = f_obj.readlines()
_d2d_vel = {'hw_joint_vel_feedback': []}
_d2d_trq = {'device_servo_trq_feedback': []}
for line in lines:
for line in lines[-500:]: # 保留最后25s的数据
data = eval(line.strip())['data']
for item in data:
try:
item['value'].reverse()
except KeyError:
continue
if item.get('channel', None) == channel and item.get('name', None) == 'hw_joint_vel_feedback':
item['value'].reverse()
_d2d_vel['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == channel and item.get('name', None) == 'device_servo_trq_feedback':
item['value'].reverse()
_d2d_trq['device_servo_trq_feedback'].extend(item['value'])
df1 = pandas.DataFrame.from_dict(_d2d_vel)
df2 = pandas.DataFrame.from_dict(_d2d_trq)
df = pandas.concat([df1, df2], axis=1)
_filename = f'{path}\\single\\j{channel+1}_single.data'
df1 = DataFrame.from_dict(_d2d_vel)
df2 = DataFrame.from_dict(_d2d_trq)
df = concat([df1, df2], axis=1)
_filename = f'{path}\\single\\j{channel+1}_single_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
elif channel in list(range(6, 9)):
with open(filename, 'r', encoding='utf-8') as f_obj:
@ -149,7 +98,10 @@ def data_proc_regular(path, filename, channel, scenario_time):
for line in lines:
data = eval(line.strip())['data']
for item in data:
try:
item['value'].reverse()
except KeyError:
continue
if item.get('channel', None) == 0 and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel_0['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == 0 and item.get('name', None) == 'device_servo_trq_feedback':
@ -175,59 +127,62 @@ def data_proc_regular(path, filename, channel, scenario_time):
elif item.get('channel', None) == 5 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq_5['device_servo_trq_feedback'].extend(item['value'])
df_01 = pandas.DataFrame.from_dict(_d2d_vel_0)
df_02 = pandas.DataFrame.from_dict(_d2d_trq_0)
df = pandas.concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j1_s_{channel-5}_{scenario_time}.data'
df_01 = DataFrame.from_dict(_d2d_vel_0)
df_02 = DataFrame.from_dict(_d2d_trq_0)
df = concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j1_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = pandas.DataFrame.from_dict(_d2d_vel_1)
df_02 = pandas.DataFrame.from_dict(_d2d_trq_1)
df = pandas.concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j2_s_{channel-5}_{scenario_time}.data'
df_01 = DataFrame.from_dict(_d2d_vel_1)
df_02 = DataFrame.from_dict(_d2d_trq_1)
df = concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j2_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = pandas.DataFrame.from_dict(_d2d_vel_2)
df_02 = pandas.DataFrame.from_dict(_d2d_trq_2)
df = pandas.concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j3_s_{channel-5}_{scenario_time}.data'
df_01 = DataFrame.from_dict(_d2d_vel_2)
df_02 = DataFrame.from_dict(_d2d_trq_2)
df = concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j3_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = pandas.DataFrame.from_dict(_d2d_vel_3)
df_02 = pandas.DataFrame.from_dict(_d2d_trq_3)
df = pandas.concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j4_s_{channel-5}_{scenario_time}.data'
df_01 = DataFrame.from_dict(_d2d_vel_3)
df_02 = DataFrame.from_dict(_d2d_trq_3)
df = concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j4_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = pandas.DataFrame.from_dict(_d2d_vel_4)
df_02 = pandas.DataFrame.from_dict(_d2d_trq_4)
df = pandas.concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j5_s_{channel-5}_{scenario_time}.data'
df_01 = DataFrame.from_dict(_d2d_vel_4)
df_02 = DataFrame.from_dict(_d2d_trq_4)
df = concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j5_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = pandas.DataFrame.from_dict(_d2d_vel_5)
df_02 = pandas.DataFrame.from_dict(_d2d_trq_5)
df = pandas.concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j6_s_{channel-5}_{scenario_time}.data'
df_01 = DataFrame.from_dict(_d2d_vel_5)
df_02 = DataFrame.from_dict(_d2d_trq_5)
df = concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j6_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
elif channel in list(range(9, 15)):
with open(filename, 'r', encoding='utf-8') as f_obj:
lines = f_obj.readlines()
_d2d_vel = {'hw_joint_vel_feedback': []}
_d2d_trq = {'device_servo_trq_feedback': []}
for line in lines:
for line in lines[-300:]: # 保留最后15s的数据
data = eval(line.strip())['data']
for item in data:
try:
item['value'].reverse()
except KeyError:
continue
if item.get('channel', None) == channel-9 and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == channel-9 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq['device_servo_trq_feedback'].extend(item['value'])
df1 = pandas.DataFrame.from_dict(_d2d_vel)
df2 = pandas.DataFrame.from_dict(_d2d_trq)
df = pandas.concat([df1, df2], axis=1)
_filename = f'{path}\\single\\j{channel-8}_hold.data'
df1 = DataFrame.from_dict(_d2d_vel)
df2 = DataFrame.from_dict(_d2d_trq)
df = concat([df1, df2], axis=1)
_filename = f'{path}\\single\\j{channel-8}_hold_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
@ -239,16 +194,19 @@ def data_proc_inertia(path, filename, channel):
for line in lines:
data = eval(line.strip())['data']
for item in data:
try:
item['value'].reverse()
except KeyError:
continue
if item.get('channel', None) == channel+3 and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == channel+3 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq['device_servo_trq_feedback'].extend(item['value'])
df1 = pandas.DataFrame.from_dict(_d2d_vel)
df2 = pandas.DataFrame.from_dict(_d2d_trq)
df = pandas.concat([df1, df2], axis=1)
_filename = f'{path}\\inertia\\j{channel+4}_inertia.data'
df1 = DataFrame.from_dict(_d2d_vel)
df2 = DataFrame.from_dict(_d2d_trq)
df = concat([df1, df2], axis=1)
_filename = f'{path}\\inertia\\j{channel+4}_inertia_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
@ -300,119 +258,95 @@ def run_rl(path, hr, md, loadsel, w2t):
conditions = c_inertia
disc = disc_inertia
for condition in conditions:
number = conditions.index(condition)
w2t(f"正在执行{disc[number][0]}测试......", 0, 0, 'purple', 'Automatic Test')
# 1. 关闭诊断曲线,触发软急停,并解除,目的是让可能正在运行着的机器停下来,切手动模式并下电
_response = execution('diagnosis.open', hr, w2t, open=False, display_open=False)
# preparation 触发软急停,并解除,目的是让可能正在运行着的机器停下来
clibs.execution('diagnosis.open', hr, w2t, tab_name, open=True, display_open=True)
clibs.execution('diagnosis.set_params', hr, w2t, tab_name, display_pdo_params=display_pdo_params)
# clibs.execution('diagnosis.save', hr, w2t, tab_name, save=True) # 这条命令有问题
md.trigger_estop()
md.reset_estop()
_response = execution('state.switch_manual', hr, w2t)
_response = execution('state.switch_motor_off', hr, w2t)
# 2. 修改未要执行的场景
for condition in conditions:
number = conditions.index(condition)
w2t(f"正在执行{disc[number][0]}测试......", 0, 0, 'purple', tab_name)
# 1. 将act重置为False并修改未要执行的场景
md.write_act(False)
ssh = SSHClient()
ssh.set_missing_host_key_policy(AutoAddPolicy())
ssh.connect('192.168.0.160', 22, username='luoshi', password='luoshi2019')
ssh.connect(clibs.ip_addr, 22, username='luoshi', password='luoshi2019')
cmd = 'cd /home/luoshi/bin/controller/; '
cmd += 'sudo sed -i "/scenario/d" projects/target/_build/current/main.mod; '
cmd += f'sudo sed -i "/DONOTDELETE/i {condition}" projects/target/_build/current/main.mod'
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
stdin.write('luoshi2019' + '\n')
stdin.flush()
print(stdout.read().decode()) # 必须得输出一下stdout才能正确执行sudo
print(stderr.read().decode()) # 顺便也执行以下stderr
stdout.read().decode() # 需要read一下才能正常执行
stderr.read().decode()
# 3. reload工程后pp2main并且自动模式和上电
# 2. reload工程后pp2main并且自动模式和上电
prj_path = 'target/_build/target.prj'
_response = execution('overview.reload', hr, w2t, prj_path=prj_path, tasks=['current'])
_response = execution('overview.get_cur_prj', hr, w2t)
_response = execution('rl_task.pp_to_main', hr, w2t, tasks=['current'])
_response = execution('state.switch_auto', hr, w2t)
_response = execution('state.switch_motor_on', hr, w2t)
clibs.execution('overview.reload', hr, w2t, tab_name, prj_path=prj_path, tasks=['current'])
clibs.execution('rl_task.pp_to_main', hr, w2t, tab_name, tasks=['current'])
clibs.execution('state.switch_auto', hr, w2t, tab_name)
clibs.execution('state.switch_motor_on', hr, w2t, tab_name)
# 4. 开始运行程序单轴运行15s
_response = execution('rl_task.run', hr, w2t, tasks=['current'])
for i in range(3):
# 3. 开始运行程序单轴运行35s
clibs.execution('rl_task.run', hr, w2t, tab_name, tasks=['current'])
_t_start = time()
while True:
if md.read_ready_to_go() == 1:
md.write_act(True)
sleep(1)
md.write_act(False)
break
else:
sleep(1)
if (time() - _t_start) // 20 > 1:
w2t("20s内未收到机器人的运行信号需要确认RL程序编写正确并正常执行...", 0, 111, 'red', tab_name)
else:
w2t("未收到机器人的运行信号需要确认RL程序编写正确并正常执行...", 0, 111, 'red', 'Automatic Test')
sleep(1)
# 5. 打开诊断曲线,并执行采集
sleep(7) # 保证程序已经运行起来,其实主要是为了保持电流的采集而设定
_response = execution('diagnosis.open', hr, w2t, open=True, display_open=True)
display_pdo_params = [
{"name": "hw_joint_vel_feedback", "channel": 0},
{"name": "hw_joint_vel_feedback", "channel": 1},
{"name": "hw_joint_vel_feedback", "channel": 2},
{"name": "hw_joint_vel_feedback", "channel": 3},
{"name": "hw_joint_vel_feedback", "channel": 4},
{"name": "hw_joint_vel_feedback", "channel": 5},
{"name": "device_servo_trq_feedback", "channel": 0},
{"name": "device_servo_trq_feedback", "channel": 1},
{"name": "device_servo_trq_feedback", "channel": 2},
{"name": "device_servo_trq_feedback", "channel": 3},
{"name": "device_servo_trq_feedback", "channel": 4},
{"name": "device_servo_trq_feedback", "channel": 5},
{"name": "device_safety_estop", "channel": 0},
]
_response = execution('diagnosis.set_params', hr, w2t, display_pdo_params=display_pdo_params)
# 4. 打开诊断曲线,并执行采集
sleep(10) # 保证程序已经运行起来,其实主要是为了保持电流的采集而设定
scenario_time = 0
if number < 6 or number > 8:
if number < 6:
sleep(35)
elif number > 8:
sleep(15)
else:
_t_start = time()
while True:
scenario_time = md.read_scenario_time()
if float(scenario_time) > 1:
w2t(f"场景{number-5}的周期时间:{scenario_time}", 0, 0, 'green', 'Automatic Test')
w2t(f"场景{number-5}的周期时间:{scenario_time}", 0, 0, 'green', tab_name)
break
else:
if (time()-_t_start)//60 > 3:
w2t(f"未收到场景{number-5}的周期时间需要确认RL程序编写正确并正常执行...", 0, 111, 'red', 'Automatic Test')
w2t(f"未收到场景{number-5}的周期时间需要确认RL程序编写正确并正常执行...", 0, 111, 'red', tab_name)
else:
sleep(5)
sleep(1) # 一定要延迟一秒再读一次scenario time寄存器因为一开始读取的数值不准确
scenario_time = md.read_scenario_time()
sleep(float(scenario_time)*0.2) # 再运行周期的20%即可
# 6. 关闭诊断曲线,停止程序运行,下电并且换成手动模式
_response = execution('diagnosis.open', hr, w2t, open=False, display_open=False)
_response = execution('rl_task.stop', hr, w2t, tasks=['current'])
_response = execution('state.switch_motor_off', hr, w2t)
_response = execution('state.switch_manual', hr, w2t)
sleep(1) # 保证所有数据均已返回
# 7. 保留数据并处理输出
for _msg in hr.c_msg:
# 5.停止程序运行,保留数据并处理输出
clibs.execution('rl_task.stop', hr, w2t, tab_name, tasks=['current'])
_c_msg = hr.c_msg.copy()
for _msg in _c_msg:
if 'diagnosis.result' in _msg:
disc[number][1].insert(0, loads(_msg))
else:
_index = 210
for _msg in hr.c_msg:
if 'diagnosis.result' in _msg:
_index = hr.c_msg.index(_msg)
break
del hr.c_msg[_index:]
hr.c_msg_xs.clear()
if len(hr.c_msg) > 270:
del hr.c_msg[270:]
gen_result_file(path, loadsel, disc, number, scenario_time)
else:
if loadsel == 'tool100':
w2t("单轴和场景电机电流采集完毕,如需采集惯量负载,须切换负载类型,并更换惯量负载,重新执行。", 0, 0, 'green', 'Automatic Test')
w2t("单轴和场景电机电流采集完毕,如需采集惯量负载,须切换负载类型,并更换惯量负载,重新执行。", 0, 0, 'green', tab_name)
elif loadsel == 'inertia':
w2t("惯量负载电机电流采集完毕,如需采集单轴/场景/保持电机电流,须切换负载类型,并更换偏置负载,重新执行。", 0, 0, 'green', 'Automatic Test')
w2t("惯量负载电机电流采集完毕,如需采集单轴/场景/保持电机电流,须切换负载类型,并更换偏置负载,重新执行。", 0, 0, 'green', tab_name)
def main(path, hr, md, loadsel, w2t):
data_dirs, data_files = traversal_files(path, w2t)
data_dirs, data_files = clibs.traversal_files(path, w2t)
config_file, current_file, prj_file = check_files(path, loadsel, data_dirs, data_files, w2t)
prj_to_xcore(prj_file)
clibs.prj_to_xcore(prj_file)
run_rl(path, hr, md, loadsel, w2t)

191
aio/code/commons/clibs.py Normal file
View File

@ -0,0 +1,191 @@
from os import scandir
from threading import Thread
from time import sleep
from os.path import exists
from paramiko import SSHClient, AutoAddPolicy
from socket import setdefaulttimeout
from logging import getLogger
from logging.config import dictConfig
import concurrent_log_handler
ip_addr = '192.168.0.160' # for product
# ip_addr = '192.168.84.129' # for test
RADIAN = 57.3 # 180 / 3.1415926
MAX_FRAME_SIZE = 1024
TIMEOUT = 5
setdefaulttimeout(TIMEOUT)
tab_names = {'dp': 'Data Process', 'at': 'Automatic Test', 'da': 'Duration Action', 'op': 'openapi'}
# PREFIX = '' # for pyinstaller packaging
PREFIX = '../assets/' # for source code debug
log_data_hmi = f'{PREFIX}templates/c_msg.log'
log_data_debug = f'{PREFIX}templates/debug.log'
heartbeat = f'{PREFIX}templates/heartbeat'
durable_data_current_xlsx = f'{PREFIX}templates/durable/durable_data_current.xlsx'
durable_data_current_max_xlsx = f'{PREFIX}templates/durable/durable_data_current_max.xlsx'
durable_data_current = {
'time': list(range(1, 19)),
'axis1': [0 for _ in range(18)],
'axis2': [0 for _ in range(18)],
'axis3': [0 for _ in range(18)],
'axis4': [0 for _ in range(18)],
'axis5': [0 for _ in range(18)],
'axis6': [0 for _ in range(18)],
}
durable_data_current_max = {
'time': list(range(1, 19)),
'axis1': [0 for _ in range(18)],
'axis2': [0 for _ in range(18)],
'axis3': [0 for _ in range(18)],
'axis4': [0 for _ in range(18)],
'axis5': [0 for _ in range(18)],
'axis6': [0 for _ in range(18)],
}
# version表示版本该键值为从1开始的整数。该key必选除此之外其它key都是可选。
# formatters日志格式化器其value值为一个字典该字典的每个键值对都代表一个Formatter键值对中key代表Formatter ID(自定义ID)value为字典描述如何配置相应的Formatter实例。默认格式为 %(message)s
# filters日志过滤器其value值为一个字典该字典的每个键值对都代表一个Filter键值对中key代表Filter ID(自定义ID)value为字典描述如何配置相应的Filter实例。
# handlers日志处理器其value值为一个字典该字典的每个键值对都代表一个Handler键值对中key代表Handler ID(自定义ID)value为字典描述如何配置相应的Handler实例包含以下配置key
# class (必选):日志处理器类全称
# level (可选)指定该日志处理器需要处理哪些级别的日志低于该级别的日志将不被该handler处理。level可以为代表日志级别的整数或者表大写字符串字符串日志级别和数字日志级别对应关系如下
# CRITICAL 50
# ERROR 40
# WARNING 30
# INFO 20
# DEBUG 10
# NOTSET 0
f_complex = '%(asctime)s # %(name)s-%(levelname)s-%(module)s-%(funcName)s-%(lineno)d # %(message)s'
f_simple = '%(levelname)s-%(module)s-%(funcName)s-%(lineno)d: %(message)s'
log_dicts = {
'version': 1,
'disable_existing_loggers': True,
'formatters': {
'standard': {
'format': f_complex,
'style': '%',
'datefmt': '%Y-%m-%dT%H:%M:%S',
},
'test': {
'format': f_simple,
'style': '%',
'datefmt': '%Y-%m-%dT%H:%M:%S',
},
},
'filters': {},
'handlers': {
'console': {
'level': 'DEBUG',
'class': 'logging.StreamHandler',
'formatter': 'test',
},
'hmi.log': {
'level': 'DEBUG',
'class': 'concurrent_log_handler.ConcurrentRotatingFileHandler',
'filename': log_data_hmi,
'maxBytes': 1024*1024*50,
'backupCount': 10,
'encoding': 'utf-8',
'formatter': 'standard',
},
'debug.log': {
'level': 'INFO',
'class': 'logging.FileHandler',
'filename': log_data_debug,
'encoding': 'utf-8',
'formatter': 'standard',
},
},
'loggers': {
'normal': {
'handlers': ['hmi.log', 'debug.log'],
'level': 'DEBUG',
'propagate': False
},
'debug': {
'handlers': ['console'],
'level': 'DEBUG',
'propagate': False
},
'': {
'handlers': ['hmi.log', 'debug.log'],
'level': 'DEBUG',
'propagate': False
},
}
}
dictConfig(log_dicts)
log_prod = getLogger('normal')
log_debug = getLogger('debug')
class GetThreadResult(Thread):
def __init__(self, func, args=()):
super(GetThreadResult, self).__init__()
self.func = func
self.args = args
self.result = 0
def run(self):
sleep(1)
self.result = self.func(*self.args)
def get_result(self):
Thread.join(self) # 等待线程执行完毕
try:
return self.result
except Exception as Err:
return None
def traversal_files(path, w2t):
# 功能:以列表的形式分别返回指定路径下的文件和文件夹,不包含子目录
# 参数:路径
# 返回值:路径下的文件夹列表 路径下的文件列表
if not exists(path):
msg = f'数据文件夹{path}不存在,请确认后重试......'
w2t(msg, 0, 1, 'red')
else:
dirs = []
files = []
for item in scandir(path):
if item.is_dir():
dirs.append(item.path)
elif item.is_file():
files.append(item.path)
return dirs, files
def prj_to_xcore(prj_file):
ssh = SSHClient()
ssh.set_missing_host_key_policy(AutoAddPolicy())
ssh.connect(ip_addr, 22, username='luoshi', password='luoshi2019')
sftp = ssh.open_sftp()
sftp.put(prj_file, '/tmp/target.zip')
cmd = 'cd /tmp; rm -rf target/; mkdir target; unzip -d target/ -q target.zip; '
cmd += 'chmod 777 -R target/; rm target.zip'
ssh.exec_command(cmd)
cmd = 'sudo rm -rf /home/luoshi/bin/controller/projects/target; '
cmd += 'sudo mv /tmp/target/ /home/luoshi/bin/controller/projects/'
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
stdin.write('luoshi2019' + '\n')
stdout.read().decode() # 需要read一下才能正常执行
stderr.read().decode()
cmd = 'cd /home/luoshi/bin/controller/; '
cmd += 'sudo chmod -R 755 projects; rm /tmp/*.prj; sudo mv projects/target/_build/*.prj /tmp; cd /tmp; '
cmd += 'prj=($(ls *.prj)); sudo mv ${prj[0]} /home/luoshi/bin/controller/projects/target/_build/target.prj; '
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
stdin.write('luoshi2019' + '\n')
stdout.read().decode() # 需要read一下才能正常执行
stderr.read().decode()
ssh.close()
def execution(cmd, hr, w2t, tab_name, **kwargs):
_id = hr.execution(cmd, **kwargs)
_msg = hr.get_from_id(_id)
if not _msg:
w2t(f"无法获取{_id}请求的响应信息", 0, 6, 'red', tab_name)
else:
return eval(_msg.split('#')[2])

View File

@ -1,16 +1,14 @@
from json import load, dumps
from socket import socket, setdefaulttimeout, AF_INET, SOCK_STREAM
from json import load, dumps, loads
from socket import socket, AF_INET, SOCK_STREAM
from threading import Thread
import selectors
from time import time, sleep
from os.path import dirname
from pymodbus.client.tcp import ModbusTcpClient
from pymodbus.payload import BinaryPayloadDecoder, BinaryPayloadBuilder
from pymodbus.constants import Endian
from commons import clibs
MAX_FRAME_SIZE = 1024
setdefaulttimeout(2)
current_path = dirname(__file__)
logger = clibs.log_prod
class ModbusRequest(object):
@ -18,102 +16,92 @@ class ModbusRequest(object):
super().__init__()
self.w2t = w2t
self.tab_name = 'openapi'
self.host = '192.168.0.160'
self.host = clibs.ip_addr
self.port = 502
self.interval = 0.3
self.c = ModbusTcpClient(self.host, self.port)
self.c.connect()
def motor_off(self):
try:
self.c.write_register(40002, 0)
sleep(self.interval)
self.c.write_register(40002, 1)
sleep(self.interval)
self.c.write_register(40002, 0)
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法正常下电连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法正常下电连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def motor_on(self):
try:
self.c.write_register(40003, 0)
sleep(self.interval)
self.c.write_register(40003, 1)
sleep(self.interval)
self.c.write_register(40003, 0)
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法正常上电连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法正常上电连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def trigger_estop(self):
try:
self.c.write_register(40012, 0)
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法触发软急停连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法触发软急停连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def reset_estop(self):
try:
self.c.write_register(40012, 1)
sleep(0.2)
sleep(self.interval)
self.c.write_register(40001, 0)
sleep(0.2)
sleep(self.interval)
self.c.write_register(40001, 1)
sleep(0.2)
sleep(self.interval)
self.c.write_register(40001, 0)
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法重置软急停连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法重置软急停连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def clear_alarm(self):
try:
self.c.write_register(40000, 1)
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法清除告警连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法清除告警连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def get_cart_vel(self):
try:
results = self.c.read_holding_registers(40537, 7)
print(f"cart vel: {results.registers}")
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法读取笛卡尔速度连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法读取笛卡尔速度连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def get_jnt_vel(self):
try:
results = self.c.read_holding_registers(40579, 7)
print(f"joint vel: {results.registers}")
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法读取关节速度连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法读取关节速度连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def get_tcp_vel(self):
try:
results = self.c.read_holding_registers(40607, 7)
print(f"tcp vel: {results.registers}")
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法读取TCP速度连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法读取TCP速度连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def get_tcp_mag_vel(self):
try:
results = self.c.read_holding_registers(40621, 1)
print(f"tcp mag: {results.registers}")
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法读取TCP合成速度连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法读取TCP合成速度连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def write_act(self, number):
try:
self.c.write_register(41000, number)
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法发送执行信号连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法发送执行信号连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def read_ready_to_go(self):
try:
results = self.c.read_holding_registers(41001, 1)
return results.registers[0]
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法读取准备信号连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法读取准备信号连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def read_scenario_time(self):
try:
@ -122,15 +110,13 @@ class ModbusRequest(object):
result = f"{result.decode_32bit_float():.3f}"
return result
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法读取准备信号连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法读取准备信号连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def write_stop0(self, number):
try:
self.c.write_register(41004, number)
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法通过IO操作stop0急停连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法通过IO操作stop0急停连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def write_speed_max(self, speed):
try:
@ -139,16 +125,14 @@ class ModbusRequest(object):
payload = builder.build()
self.c.write_registers(41005, payload, skip_encode=True)
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法写入速度值连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法写入速度值连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def read_brake_done(self):
try:
results = self.c.read_holding_registers(41007, 1)
return results.registers[0]
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法读取制动已执行信号连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法读取制动已执行信号连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def write_axis(self, axis):
try:
@ -157,22 +141,19 @@ class ModbusRequest(object):
payload = builder.to_registers()
self.c.write_registers(41008, payload)
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法写入速度值连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法写入速度值连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def write_probe(self, probe):
try:
self.c.write_register(41010, probe)
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法写入速度探测信号连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法写入速度探测信号连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
def write_pon(self, pon): # positive or negative
try:
self.c.write_register(41011, pon)
except Exception as Err:
self.w2t(f"{Err}")
self.w2t("无法写入正负方向信号连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
self.w2t(f"{Err}\n无法写入正负方向信号连接Modbus失败需要确认网络是否通畅或是未正确导入寄存器文件...", 0, 100, 'red', self.tab_name)
class HmiRequest(object):
@ -196,6 +177,7 @@ class HmiRequest(object):
self.half_length = 0
self.index = 0
self.reset_index = 0
self.durable_lock = 0
self.sock_conn()
self.t_heartbeat = Thread(target=self.heartbeat)
@ -210,21 +192,19 @@ class HmiRequest(object):
def sock_conn(self):
# while True:
with open(f"{current_path}/../../assets/templates/heartbeat", "r", encoding='utf-8') as f_hb:
with open(clibs.heartbeat, "r", encoding='utf-8') as f_hb:
c_state = f_hb.read().strip()
if c_state == '0':
try:
self.c = socket(AF_INET, SOCK_STREAM)
self.c.connect(('192.168.0.160', 5050))
# self.c.connect(('192.168.84.129', 5050))
self.c.connect((clibs.ip_addr, 5050))
self.c.setblocking(False)
self.c_xs = socket(AF_INET, SOCK_STREAM)
self.c_xs.connect(('192.168.0.160', 6666))
# self.c_xs.connect(('192.168.84.129', 6666))
self.c_xs.connect((clibs.ip_addr, 6666))
self.c_xs.setblocking(False)
self.w2t("Connection success", 0, 0, 'green', tab_name=self.tab_name)
with open(f"{current_path}/../../assets/templates/heartbeat", "w", encoding='utf-8') as f_hb:
with open(clibs.heartbeat, "w", encoding='utf-8') as f_hb:
f_hb.write('1')
md = ModbusRequest(self.w2t)
md.reset_estop()
@ -234,7 +214,7 @@ class HmiRequest(object):
md.write_axis(1)
except Exception as Err:
self.w2t("Connection failed...", 0, 0, 'red', tab_name=self.tab_name)
with open(f"{current_path}/../../assets/templates/heartbeat", "w", encoding='utf-8') as f_hb:
with open(clibs.heartbeat, "w", encoding='utf-8') as f_hb:
f_hb.write('0')
def header_check(self, index, data):
@ -262,28 +242,29 @@ class HmiRequest(object):
# print(f"in head check data: {data}")
self.broke = 100
index += MAX_FRAME_SIZE
index += clibs.MAX_FRAME_SIZE
return index, 0, 0
def heartbeat(self):
while self.t_bool:
_id = self.execution('controller.heart')
_flag = '0' if self.get_from_id(_id) is None else '1'
print(f"hb = {_flag}", end=' ')
print(f"len(c_msg) = {len(self.c_msg)}", end=' ')
print(f"len(c_msg_xs) = {len(self.c_msg_xs)}", end='\n')
with open(f"{current_path}/../../assets/templates/heartbeat", "w", encoding='utf-8') as f_hb:
# print(f"hb = {_flag}", end=' ')
# print(f"len(c_msg) = {len(self.c_msg)}", end=' ')
# print(f"len(c_msg_xs) = {len(self.c_msg_xs)}", end='\n')
with open(clibs.heartbeat, "w", encoding='utf-8') as f_hb:
f_hb.write(_flag)
if _flag == '0':
self.w2t(f"{_id} 心跳丢失,连接失败,重新连接中...", 0, 7, 'red', tab_name=self.tab_name)
sleep(1.5)
# with open(f"{current_path}/../../assets/templates/c_msg.log", "w", encoding='utf-8') as f:
# for msg in self.c_msg:
# f.write(str(loads(msg)) + '\n')
sleep(2)
def msg_storage(self, response, flag=0):
# response是解码后的字符串
messages = self.c_msg if flag == 0 else self.c_msg_xs
if len(messages) < 20000:
logger.debug(f"{loads(response)}")
if 'move.monitor' in response:
pass
elif len(messages) < 10000:
messages.insert(0, response)
else:
messages.insert(0, response)
@ -327,7 +308,7 @@ class HmiRequest(object):
# print(f"broke == 0 index = {self.index-8}")
# print(f"broke == 0 INIT pkg size = {self.pkg_size}")
# print(f"broke == 0 data = {data}")
if self.index > MAX_FRAME_SIZE:
if self.index > clibs.MAX_FRAME_SIZE:
break
# 详见解包原理数据.txtself.pkg_size 永远是除了当前data之外剩余未处理的数据大小
if self.pkg_size <= len(data) - self.index:
@ -371,15 +352,15 @@ class HmiRequest(object):
# print('flag = 0 encounter broke == 3')
self.broke = 3
self.index += MAX_FRAME_SIZE
self.index += clibs.MAX_FRAME_SIZE
self.reset_index = 1
break # 因为 index + 2 的大小超过 MAX_FRAME_SIZE
break # 因为 index + 2 的大小超过 clibs.MAX_FRAME_SIZE
elif self.index+_frame_size-6 > len(data):
self.response = data[self.index:].decode()
self.pkg_size -= (len(data) - self.index) # 详见解包原理数据.txtself.pkg_size
self.leftover = (_frame_size-6-(len(data)-self.index))
self.index += MAX_FRAME_SIZE
self.index += clibs.MAX_FRAME_SIZE
self.reset_index = 1
# print(f"in flag=0 else data = {data}")
@ -448,9 +429,9 @@ class HmiRequest(object):
# print('flag = 1 encounter broke == 3')
self.broke = 3
self.index += MAX_FRAME_SIZE
self.index += clibs.MAX_FRAME_SIZE
self.reset_index = 1
break # 因为 index + 2 的大小超过 MAX_FRAME_SIZE
break # 因为 index + 2 的大小超过 clibs.MAX_FRAME_SIZE
# print(f"in pkg size > 0 loop after if data = {data}")
# print(f"in pkg size > 0 loop after if index = {self.index}")
# print(f"in pkg size > 0 loop after if pkg size = {self.pkg_size}")
@ -467,7 +448,7 @@ class HmiRequest(object):
self.response += data[self.index:].decode()
self.leftover -= (len(data) - self.index)
self.pkg_size -= (len(data) - self.index)
self.index += MAX_FRAME_SIZE
self.index += clibs.MAX_FRAME_SIZE
self.reset_index = 1
# print(f"in pkg size > 0 loop after else data = {data}")
# print(f"in pkg size > 0 loop after else index = {self.index}")
@ -521,14 +502,20 @@ class HmiRequest(object):
def get_from_id(self, msg_id, flag=0):
for i in range(3):
messages = self.c_msg if flag == 0 else self.c_msg_xs
for msg in messages:
if msg_id is None:
return None
elif msg_id in msg:
return msg
with open(clibs.log_data_hmi, mode='r', encoding='utf-8') as f_log:
for line in f_log:
if msg_id in line.strip():
return line
sleep(1)
else:
else: # 尝试在上一次分割的日志中查找,只做一次
sleep(1)
try:
with open(clibs.log_data_hmi+'.1', mode='r', encoding='utf-8') as f_log:
for line in f_log:
if msg_id in line.strip():
return line
except FileNotFoundError:
pass
return None
def package(self, cmd):
@ -543,7 +530,7 @@ class HmiRequest(object):
def unpackage(self, sock):
def to_read(conn, mask):
data = conn.recv(MAX_FRAME_SIZE)
data = conn.recv(clibs.MAX_FRAME_SIZE)
if data:
# print(data)
self.get_response(data)
@ -590,7 +577,7 @@ class HmiRequest(object):
if flg == 0: # for old protocols
req = None
try:
with open(f'{current_path}/../../assets/templates/{command}.json', encoding='utf-8',
with open(f'{clibs.PREFIX}templates/json/{command}.json', encoding='utf-8',
mode='r') as f_json:
req = load(f_json)
except:
@ -616,6 +603,8 @@ class HmiRequest(object):
req['data']['type'] = kwargs['type']
req['data']['bias'] = kwargs['bias']
req['data']['value'] = kwargs['value']
case 'diagnosis.save':
req['data']['save'] = kwargs['save']
case _:
pass
@ -626,7 +615,7 @@ class HmiRequest(object):
self.c.send(self.package(cmd))
sleep(0.5)
except Exception as Err:
self.w2t(f"{cmd}\n请求发送失败...{Err}", 0, 0, 'red', tab_name=self.tab_name)
self.w2t(f"{cmd}: 请求发送失败...{Err}", 0, 0, 'red', tab_name=self.tab_name)
return req['id']

View File

@ -1 +0,0 @@
__all__ = ['brake', 'current', 'iso', 'wavelogger']

View File

@ -1,49 +1,12 @@
# coding: utf-8
from os import scandir
from os.path import isfile, exists
from os.path import isfile
from sys import argv
from openpyxl import load_workbook
from time import time, sleep, strftime, localtime
from threading import Thread
from pandas import read_csv
from commons import clibs
class GetThreadResult(Thread):
def __init__(self, func, args=()):
super(GetThreadResult, self).__init__()
self.func = func
self.args = args
self.result = 0
def run(self):
sleep(1)
self.result = self.func(*self.args)
def get_result(self):
Thread.join(self) # 等待线程执行完毕
try:
return self.result
except Exception as Err:
return None
def traversal_files(path, w2t):
# 功能:以列表的形式分别返回指定路径下的文件和文件夹,不包含子目录
# 参数:路径
# 返回值:路径下的文件夹列表 路径下的文件列表
if not exists(path):
msg = f'数据文件夹{path}不存在,请确认后重试......'
w2t(msg, 0, 1, 'red')
else:
dirs = []
files = []
for item in scandir(path):
if item.is_dir():
dirs.append(item.path)
elif item.is_file():
files.append(item.path)
return dirs, files
logger = clibs.log_prod
def check_files(path, raw_data_dirs, result_files, w2t):
@ -83,7 +46,7 @@ def check_files(path, raw_data_dirs, result_files, w2t):
规则解释AA/BB/CC 指的是臂展/负载/速度的比例例如reach66_load100_speed3366%臂展100%负载以及33%速度情况下的测试结果文件夹"""
w2t(msg, 0, 4, 'red')
_, raw_data_files = traversal_files(raw_data_dir, w2t)
_, raw_data_files = clibs.traversal_files(raw_data_dir, w2t)
if len(raw_data_files) != 3:
msg = f"数据目录 {raw_data_dir} 下数据文件个数错误,每个数据目录下有且只能有三个以 .data 为后缀的数据文件"
w2t(msg, 0, 5, 'red')
@ -109,6 +72,7 @@ def get_configs(configfile, w2t):
return av, rr
def now_doing_msg(docs, flag, w2t):
# 功能:输出正在处理的文件或目录
# 参数文件或目录start 或 done 标识
@ -228,7 +192,7 @@ def data_process(result_file, raw_data_dirs, av, rr, vel, trq, estop, w2t):
global stop
stop = 0
t_excel = GetThreadResult(load_workbook, args=(result_file, ))
t_excel = clibs.GetThreadResult(load_workbook, args=(result_file, ))
t_wait = Thread(target=w2t_local, args=('.', 1, w2t))
t_excel.start()
t_wait.start()
@ -242,7 +206,7 @@ def data_process(result_file, raw_data_dirs, av, rr, vel, trq, estop, w2t):
for raw_data_dir in raw_data_dirs:
if raw_data_dir.split('\\')[-1].split('_')[0] == prefix:
now_doing_msg(raw_data_dir, 'start', w2t)
_, data_files = traversal_files(raw_data_dir, w2t)
_, data_files = clibs.traversal_files(raw_data_dir, w2t)
# 数据文件串行处理模式---------------------------------
# count = 1
# for data_file in data_files:
@ -280,7 +244,7 @@ def main(path, vel, trq, estop, w2t):
# 参数initialization函数的返回值
# 返回值:-
time_start = time()
raw_data_dirs, result_files = traversal_files(path, w2t)
raw_data_dirs, result_files = clibs.traversal_files(path, w2t)
try:
# threads = []

View File

@ -1,31 +1,13 @@
from openpyxl import load_workbook
from os import scandir
from os.path import exists
from sys import argv
from pandas import read_csv, concat, set_option
from re import match
from threading import Thread
from time import sleep
from csv import reader, writer
from commons import clibs
class GetThreadResult(Thread):
def __init__(self, func, args=()):
super(GetThreadResult, self).__init__()
self.func = func
self.args = args
self.result = 0
def run(self):
sleep(1)
self.result = self.func(*self.args)
def get_result(self):
Thread.join(self) # 等待线程执行完毕
try:
return self.result
except Exception as Err:
return None
logger = clibs.log_prod
def w2t_local(msg, wait, w2t):
@ -38,27 +20,8 @@ def w2t_local(msg, wait, w2t):
break
def traversal_files(path, w2t):
# 功能:以列表的形式分别返回指定路径下的文件和文件夹,不包含子目录
# 参数:路径
# 返回值:路径下的文件夹列表 路径下的文件列表
if not exists(path):
msg = f'数据文件夹{path}不存在,请确认后重试......'
w2t(msg, 0, 8, 'red')
else:
dirs = []
files = []
for item in scandir(path):
if item.is_dir():
dirs.append(item.path)
elif item.is_file():
files.append(item.path)
return dirs, files
def initialization(path, sub, w2t):
_, data_files = traversal_files(path, w2t)
_, data_files = clibs.traversal_files(path, w2t)
count = 0
for data_file in data_files:
@ -69,8 +32,8 @@ def initialization(path, sub, w2t):
count += 1
else:
if not (match('j[1-7].*\\.data', filename) or match('j[1-7].*\\.csv', filename)):
print(f"不合规 {data_file}")
msg = f"所有文件必须以 jx_ 开头,以 .data/csv 结尾x取值1-7请检查后重新运行。"
msg = f"不合规 {data_file}\n"
msg += f"所有文件必须以 jx_ 开头,以 .data/csv 结尾x取值1-7请检查后重新运行。"
w2t(msg, 0, 6, 'red')
if not ((sub == 'cycle' and count == 2) or (sub != 'cycle' and count == 1)):
@ -79,7 +42,7 @@ def initialization(path, sub, w2t):
return data_files
def current_max(data_files, rcs, trqh, w2t):
def current_max(data_files, rcs, trq, w2t):
current = {1: [], 2: [], 3: [], 4: [], 5: [], 6: [], 7: []}
for data_file in data_files:
if data_file.endswith('.data'):
@ -93,8 +56,8 @@ def current_max(data_files, rcs, trqh, w2t):
axis = int(data_file.split('\\')[-1].split('_')[0].removeprefix('j'))
rca = rcs[axis-1]
col = df.columns.values[trqh-1]
c_max = df[col].max()
col = df.columns.values[trq-1]
c_max = df[col].abs().max()
scale = 1 if data_file.endswith('.csv') else 1000
_ = abs(c_max/scale*rca)
@ -118,7 +81,7 @@ def current_max(data_files, rcs, trqh, w2t):
return current
def current_avg(data_files, rcs, trqh, w2t):
def current_avg(data_files, rcs, trq, w2t):
current = {1: [], 2: [], 3: [], 4: [], 5: [], 6: [], 7: []}
for data_file in data_files:
if data_file.endswith('.data'):
@ -132,7 +95,7 @@ def current_avg(data_files, rcs, trqh, w2t):
axis = int(data_file.split('\\')[-1].split('_')[0].removeprefix('j'))
rca = rcs[axis-1]
col = df.columns.values[trqh - 1]
col = df.columns.values[trq-1]
c_std = df[col].std()
c_avg = df[col].mean()
@ -158,7 +121,7 @@ def current_avg(data_files, rcs, trqh, w2t):
return current
def current_cycle(dur, data_files, rcs, vel, trq, trqh, rpms, w2t):
def current_cycle(dur, data_files, rcs, rrs, vel, trq, trqh, rpms, w2t):
result = None
hold = []
single = []
@ -174,7 +137,7 @@ def current_cycle(dur, data_files, rcs, vel, trq, trqh, rpms, w2t):
w2t(f"正在打开文件 {result},需要 10s 左右", 1, 0, 'orange')
global stop
stop = 0
t_excel = GetThreadResult(load_workbook, args=(result, ))
t_excel = clibs.GetThreadResult(load_workbook, args=(result, ))
t_wait = Thread(target=w2t_local, args=('.', 1, w2t))
t_excel.start()
t_wait.start()
@ -189,14 +152,14 @@ def current_cycle(dur, data_files, rcs, vel, trq, trqh, rpms, w2t):
for axis, cur_value in avg.items():
try:
shtname = f"J{axis}"
wb[shtname]["J4"].value = float(cur_value)
wb[shtname]["J4"].value = float(cur_value[0])
except:
pass
if dur == 0:
p_single(wb, single, vel, trq, rpms, w2t)
p_single(wb, single, vel, trq, rpms, rrs, w2t)
else:
p_scenario(wb, single, vel, trq, rpms, dur, w2t)
p_scenario(wb, single, vel, trq, rpms, rrs, dur, w2t)
w2t(f"正在保存文件 {result},需要 10s 左右", 1, 0, 'orange')
stop = 0
@ -239,7 +202,7 @@ def find_point(data_file, pos, flag, df, _row_s, _row_e, w2t, exitcode, threshol
w2t(f"[{pos}] {data_file}数据有误,需要检查,无法找到有效起始点或结束点...", 0, 0, 'red')
def p_single(wb, single, vel, trq, rpms, w2t):
def p_single(wb, single, vel, trq, rpms, rrs, w2t):
# 1. 先找到第一个速度为零的点,数据从后往前找,一开始就是零的情况不予考虑
# 2. 记录第一个点的位置,继续向前查找第二个速度为零的点,同理,一开始为零的点不予考虑
# 3. 记录第二个点的位置,并将其中的数据拷贝至对应位置
@ -253,7 +216,7 @@ def p_single(wb, single, vel, trq, rpms, w2t):
set_option("display.precision", 2)
if data_file.endswith('.data'):
df = read_csv(data_file, sep='\t')
rr = float(wb['统计'].cell(row=2, column=axis+1).value)
rr = rrs[axis-1]
addition = 180 / 3.1415926 * 60 / 360 * rr
elif data_file.endswith('.csv'):
df = read_csv(data_file, sep=',', encoding='gbk', header=8)
@ -270,6 +233,7 @@ def p_single(wb, single, vel, trq, rpms, w2t):
col_names = list(df.columns)
df_1 = df[col_names[vel-1]].multiply(rpm*addition)
df_2 = df[col_names[trq-1]].multiply(scale)
# print(df_1.abs().max())
df = concat([df_1, df_2], axis=1)
_step = 5 if data_file.endswith('.csv') else 50
@ -320,7 +284,7 @@ def p_single(wb, single, vel, trq, rpms, w2t):
cell.value = None
def p_scenario(wb, single, vel, trq, rpms, dur, w2t):
def p_scenario(wb, single, vel, trq, rpms, rrs, dur, w2t):
for data_file in single:
cycle = 0.001
axis = int(data_file.split('\\')[-1].split('_')[0].removeprefix('j'))
@ -332,7 +296,7 @@ def p_scenario(wb, single, vel, trq, rpms, dur, w2t):
set_option("display.precision", 2)
if data_file.endswith('.data'):
df = read_csv(data_file, sep='\t')
rr = float(wb['统计'].cell(row=2, column=axis+1).value)
rr = rrs[axis-1]
addition = 180 / 3.1415926 * 60 / 360 * rr
elif data_file.endswith('.csv'):
df = read_csv(data_file, sep=',', encoding='gbk', header=8)
@ -376,6 +340,7 @@ def get_configs(configfile, w2t):
_wb = load_workbook(configfile, read_only=True)
_ws = _wb['Target']
rcs = []
rrs = []
rpms = []
for i in range(2, 9):
try:
@ -388,18 +353,23 @@ def get_configs(configfile, w2t):
except:
rcs.append(0.0)
return rpms, rcs
try:
rrs.append(float(_ws.cell(row=2, column=i).value))
except:
rrs.append(0.0)
return rpms, rcs, rrs
def main(path, sub, dur, vel, trq, trqh, w2t):
data_files = initialization(path, sub, w2t)
rpms, rcs = get_configs(path + '\\configs.xlsx', w2t)
rpms, rcs, rrs = get_configs(path + '\\configs.xlsx', w2t)
if sub == 'max':
current_max(data_files, rcs, trqh, w2t)
current_max(data_files, rcs, trq, w2t)
elif sub == 'avg':
current_avg(data_files, rcs, trqh, w2t)
current_avg(data_files, rcs, trq, w2t)
elif sub == 'cycle':
current_cycle(dur, data_files, rcs, vel, trq, trqh, rpms, w2t)
current_cycle(dur, data_files, rcs, rrs, vel, trq, trqh, rpms, w2t)
else:
pass

View File

@ -1,27 +1,10 @@
# _*_ encodingutf-8 _*_
import pdfplumber
from openpyxl import load_workbook
from os import scandir, remove
from os.path import exists
from os import remove
from sys import argv
from commons import clibs
def traversal_files(path, w2t):
# 功能:以列表的形式分别返回指定路径下的文件和文件夹,不包含子目录
# 参数:路径
# 返回值:路径下的文件夹列表 路径下的文件列表
if not exists(path):
msg = f'数据文件夹{path}不存在,请确认后重试......'
w2t(msg, 0, 1, 'red')
else:
dirs = files = []
for item in scandir(path):
if item.is_dir():
dirs.append(item.path)
elif item.is_file():
files.append(item.path)
return dirs, files
logger = clibs.log_prod
def p_iso(file, p_files, ws, tmpfile):
@ -153,7 +136,7 @@ def p_iso_1000(file, p_files, ws, tmpfile):
def main(path, w2t):
dirs, files = traversal_files(path, 1)
dirs, files = clibs.traversal_files(path, 1)
try:
wb = load_workbook(path + "/iso-results.xlsx")

View File

@ -1,31 +1,10 @@
import os
import random
from pandas import read_csv
from csv import reader
from sys import argv
from os.path import exists
from os import scandir, remove
from openpyxl import Workbook
from random import randint
from commons import clibs
def traversal_files(path, w2t):
# 功能:以列表的形式分别返回指定路径下的文件和文件夹,不包含子目录
# 参数:路径
# 返回值:路径下的文件夹列表 路径下的文件列表
if not exists(path):
msg = f'数据文件夹{path}不存在,请确认后重试......'
w2t(msg, 0, 1, 'red')
else:
dirs = []
files = []
for item in scandir(path):
if item.is_dir():
dirs.append(item.path)
elif item.is_file():
files.append(item.path)
return dirs, files
logger = clibs.log_prod
def find_point(bof, step, pos, data_file, flag, df, row, w2t):
@ -95,7 +74,7 @@ def get_cycle_info(data_file, df, row, step, w2t):
def initialization(path, w2t):
_, data_files = traversal_files(path, w2t)
_, data_files = clibs.traversal_files(path, w2t)
for data_file in data_files:
if not data_file.lower().endswith('.csv'):
@ -126,7 +105,7 @@ def single_file_proc(ws, data_file, df, low, high, cycle, w2t):
_step = 5
_data = {}
row_max = df.index[-1]-100
print(data_file)
# print(data_file)
while _row < row_max:
if count not in _data.keys():
_data[count] = []
@ -149,7 +128,7 @@ def single_file_proc(ws, data_file, df, low, high, cycle, w2t):
ws.cell(row=1, column=i).value = f"{i-1}次测试"
ws.cell(row=i, column=1).value = f"{i-1}次精度变化"
print(_data)
# print(_data)
for i in sorted(_data.keys()):
_row = 2
_column = i + 1
@ -162,9 +141,9 @@ def execution(data_files, w2t):
wb = Workbook()
for data_file in data_files:
ws, df, low, high, cycle = preparation(data_file, wb, w2t)
print(f"low = {low}")
print(f"high = {high}")
print(f"cycle = {cycle}")
# print(f"low = {low}")
# print(f"high = {high}")
# print(f"cycle = {cycle}")
single_file_proc(ws, data_file, df, low, high, cycle, w2t)
wd = data_files[0].split('\\')

View File

@ -0,0 +1,224 @@
from sys import argv
from json import loads
from time import sleep, time, strftime, localtime
from pandas import DataFrame
from openpyxl import load_workbook
from math import sqrt
from numpy import power
from csv import writer
from commons import clibs
logger = clibs.log_prod
tab_name = clibs.tab_names['da']
count = 0
display_pdo_params = [
# {"name": "hw_joint_vel_feedback", "channel": 0},
# {"name": "hw_joint_vel_feedback", "channel": 1},
# {"name": "hw_joint_vel_feedback", "channel": 2},
# {"name": "hw_joint_vel_feedback", "channel": 3},
# {"name": "hw_joint_vel_feedback", "channel": 4},
# {"name": "hw_joint_vel_feedback", "channel": 5},
{"name": "device_servo_trq_feedback", "channel": 0},
{"name": "device_servo_trq_feedback", "channel": 1},
{"name": "device_servo_trq_feedback", "channel": 2},
{"name": "device_servo_trq_feedback", "channel": 3},
{"name": "device_servo_trq_feedback", "channel": 4},
{"name": "device_servo_trq_feedback", "channel": 5},
]
title = [
'time', 'trq-1', 'trq-2', 'trq-3', 'trq-4', 'trq-5', 'trq-6', 'trq-max-1', 'trq-max-2', 'trq-max-3', 'trq-max-4',
'trq-max-5', 'trq-max-6'
]
def check_files(data_dirs, data_files, w2t):
if len(data_dirs) != 0 or len(data_files) != 2:
w2t('初始路径下不允许有文件夹,且初始路径下只能存在如下文件,确认后重新运行!\n1. target.zip\n2. configs.xlsx', 0, 10, 'red', tab_name)
_files = [data_files[0].split('\\')[-1], data_files[1].split('\\')[-1]]
_files.sort()
if _files != ['configs.xlsx', 'target.zip']:
w2t('初始路径下只能存在如下文件,确认后重新运行!\n1. target.zip\n2. configs.xlsx', 0, 10, 'red', tab_name)
data_files.sort()
return data_files
def run_rl(path, config_file, data_all, hr, md, w2t):
# 1. 关闭诊断曲线,触发软急停,并解除,目的是让可能正在运行着的机器停下来,切手动模式并下电
clibs.execution('diagnosis.open', hr, w2t, tab_name, open=True, display_open=True)
clibs.execution('diagnosis.set_params', hr, w2t, tab_name, display_pdo_params=display_pdo_params)
md.trigger_estop()
md.reset_estop()
md.write_act(False)
sleep(1) # 让曲线彻底关闭
# 2. reload工程后pp2main并且自动模式和上电
prj_path = 'target/_build/target.prj'
clibs.execution('overview.reload', hr, w2t, tab_name, prj_path=prj_path, tasks=['current'])
clibs.execution('rl_task.pp_to_main', hr, w2t, tab_name, tasks=['current'])
clibs.execution('state.switch_auto', hr, w2t, tab_name)
clibs.execution('state.switch_motor_on', hr, w2t, tab_name)
# 3. 开始运行程序
clibs.execution('rl_task.run', hr, w2t, tab_name, tasks=['current'])
_t_start = time()
while True:
if md.read_ready_to_go() == 1:
md.write_act(True)
break
else:
if (time() - _t_start) // 20 > 1:
w2t("20s内未收到机器人的运行信号需要确认RL程序编写正确并正常执行...", 0, 111, 'red', tab_name)
else:
sleep(1)
# 4. 获取初始数据,周期时间,首次的各轴平均电流值,打开诊断曲线,并执行采集
sleep(20) # 初始化 scenario time 为 0
_t_start = time()
while True:
scenario_time = md.read_scenario_time()
if float(scenario_time) > 1:
w2t(f"场景的周期时间:{scenario_time}s", 0, 0, 'green', tab_name)
break
else:
if (time() - _t_start) // 60 > 3:
w2t(f"未收到场景的周期时间需要确认RL程序编写正确并正常执行...", 0, 111, 'red', tab_name)
else:
sleep(5)
sleep(1) # 一定要延迟一秒再读一次scenario time寄存器因为一开始读取的数值不准确
scenario_time = float(md.read_scenario_time())
sleep(scenario_time*0.2)
# 6. 准备初始数据,关闭诊断曲线,保留数据并处理输出
with open(f'{path}\\results.csv', mode='a+', newline='') as f_csv:
csv_writer = writer(f_csv)
csv_writer.writerow(title)
_wb = load_workbook(config_file, read_only=True)
_ws = _wb['Target']
wait_time = float(_ws.cell(row=2, column=10).value)
rcs = []
for i in range(6):
rcs.append(float(_ws.cell(row=6, column=i + 2).value))
get_durable_data(path, data_all, scenario_time, wait_time, rcs, hr, md, w2t)
# 7. 继续运行
while True:
# 固定间隔,更新一次数据,打开曲线,获取周期内电流,关闭曲线
sleep(wait_time+scenario_time+7)
# 保留数据并处理输出
get_durable_data(path, data_all, scenario_time, wait_time, rcs, hr, md, w2t)
def get_durable_data(path, data, scenario_time, wait_time, rcs, hr, md, w2t):
_data_list = []
_c_msg = hr.c_msg.copy()
for _msg in _c_msg:
if 'diagnosis.result' in _msg:
_data_list.insert(0, loads(_msg))
else:
hr.c_msg_xs.clear()
if len(hr.c_msg) > 270:
del hr.c_msg[270:]
# with open(f'{path}\\log.txt', 'w', encoding='utf-8') as f_obj:
# for _ in _data_list:
# f_obj.write(f"{_}\n")
_d2d_trq = {0: [], 1: [], 2: [], 3: [], 4: [], 5: []}
for line in _data_list:
for item in line['data']:
for i in range(6):
item['value'].reverse()
if item.get('channel', None) == i and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq[i].extend(item['value'])
if len(_d2d_trq[0]) / 1000 > scenario_time + 1:
this_time = strftime("%Y-%m-%d %H:%M:%S", localtime(time()))
next_time = strftime("%Y-%m-%d %H:%M:%S", localtime(time()+wait_time+10+scenario_time)).split()[-1]
_df = DataFrame(_d2d_trq)
_flg = 0
_res = []
for i in range(6):
def overmax_data(df, index, number, flag):
if number > 100:
md.trigger_estop()
hr.durable_quit = 1
df.to_excel(f'{path}\\{this_time}.xlsx')
w2t(f"[{this_time}] {flag}-axis-{index} 数据过大错误,需要检查确定。", 0, 10, 'red', tab_name)
try:
_ = sqrt(_df[i].apply(lambda x: power((rcs[i]*x/1000), 2)).sum()/len(_df[i]))
except:
md.trigger_estop()
_df.to_excel(path+"\\err_data.xlsx")
w2t(f"{i}calculate error", 0, 11, 'red', tab_name)
if not _flg:
del data[0]['time'][0]
data[0]['time'].append(this_time.split()[-1])
del data[1]['time'][0]
data[1]['time'].append(this_time.split()[-1])
_res.append(this_time)
_flg = 1
del data[0][f"axis{i + 1}"][0]
overmax_data(_df, i, _, 'trq')
data[0][f"axis{i + 1}"].append(_)
_res.append(_)
_ = rcs[i] * _df[i].abs().max() / 1000
overmax_data(_df, i, _, 'trq-max')
del data[1][f"axis{i + 1}"][0]
data[1][f"axis{i + 1}"].append(_)
_res.append(_)
_df_1 = DataFrame(data[0])
_df_2 = DataFrame(data[1])
with open(f'{path}\\results.csv', mode='a+', newline='') as f_csv:
def change_order(res):
_time = res[0:1]
_trq = []
_trq_max = []
for _item in res[1::2]:
_trq.append(_item)
for _item in res[2::2]:
_trq_max.append(_item)
return _time + _trq + _trq_max
csv_writer = writer(f_csv)
csv_writer.writerow(change_order(_res))
while True:
if not hr.durable_lock:
hr.durable_lock = 1
_df_1.to_excel(clibs.durable_data_current_xlsx, index=False)
_df_2.to_excel(clibs.durable_data_current_max_xlsx, index=False)
hr.durable_lock = 0
break
else:
sleep(1)
global count
count += 1
w2t(f"[{this_time}] 当前次数:{count:09d} | 预计下次数据更新时间:{next_time}", 0, 0, '#008B8B', tab_name)
break
else:
md.trigger_estop()
with open(f'{path}\\device_servo_trq_feedback_0.txt', 'w', encoding='utf-8') as f_obj:
for _ in _d2d_trq[0]:
f_obj.write(f"{_}\n")
w2t("采集的数据时间长度不够,需要确认。", 0, 10, 'red', tab_name)
def main(path, hr, md, w2t):
data_all = [clibs.durable_data_current, clibs.durable_data_current_max]
data_dirs, data_files = clibs.traversal_files(path, w2t)
config_file, prj_file = check_files(data_dirs, data_files, w2t)
clibs.prj_to_xcore(prj_file)
run_rl(path, config_file, data_all, hr, md, w2t)
if __name__ == '__main__':
main(*argv[1:])

2
dial_gauge/README.md Normal file
View File

@ -0,0 +1,2 @@
## 千分表数据自动采集

6
dial_gauge/conf.ini Normal file
View File

@ -0,0 +1,6 @@
[md_rtu]
port = COM10
[md_tcp]
addr = 192.168.0.160
port = 502

99
dial_gauge/gauge.py Normal file
View File

@ -0,0 +1,99 @@
from configparser import ConfigParser
from pymodbus.client import ModbusSerialClient, ModbusTcpClient
from time import sleep, strftime, localtime, time
from openpyxl import load_workbook, Workbook
from os import rename
# import pymodbus
# pymodbus.pymodbus_apply_logging_config("DEBUG")
def initializations():
try:
now = strftime("%Y%m%d%H%M%S", localtime(time()))
rename('results.xlsx', f'results_{now}.xlsx')
except FileNotFoundError:
pass
wb = Workbook()
ws = wb.active
ws.title = 'results'
ws['A1'].value = '时间'
ws['B1'].value = '次数'
ws['C1'].value = '结果'
wb.save('results.xlsx')
wb.close()
def do_connections():
configs = ConfigParser()
configs.read('conf.ini')
# ================ 配置Modbus TCP客户端 ================
tcp_host = configs.get('md_tcp', 'addr')
tcp_port = configs.getint('md_tcp', 'port')
client_tcp = ModbusTcpClient(tcp_host, tcp_port)
if client_tcp.connect():
print(f"Modbus TCP已连接到{tcp_host}:{tcp_port}...")
else:
raise Exception(f"Modbus TCP无法连接到{tcp_host}:{tcp_port}...")
# ================ 配置Modbus RTU客户端 ================
rtu_port = configs.get('md_rtu', 'port')
client_rtu = ModbusSerialClient(
method='rtu',
port=rtu_port, # 根据实际情况调整端口Windows上可能是 'COM3'
baudrate=38400, # 根据协议设定波特率为38400
timeout=1,
parity='N', # 无奇偶校验
stopbits=2, # 2个停止位
bytesize=8 # 8个数据位
)
if client_rtu.connect():
print(f"Modbus RTU已连接到系统{rtu_port}端口...")
else:
raise Exception(f"Modbus RTU无法连接到系统{rtu_port}端口...")
return client_tcp, client_rtu
def get_gauge_data(client_tcp, client_rtu):
count = 2
client_tcp.write_register(41000, 1) # 将 act 信号置为 True
# ================ 功能实现 ================
while True:
while True:
res_tcp = client_tcp.read_holding_registers(41001, 1) # 获取 time_to_go 的值
sleep(0.5)
if not res_tcp.registers[0]:
continue
else:
break
try:
res_rtu = client_rtu.read_holding_registers(0x0000, 2, slave=1)
plus_or_minus = 1 if res_rtu.registers[0] == 0 else -1
result = res_rtu.registers[1] * plus_or_minus / 10000
now = strftime("%Y-%m-%d %H:%M:%S", localtime(time()))
wb = load_workbook('results.xlsx')
ws = wb['results']
ws[f'A{count}'].value = now
ws[f'B{count}'].value = count-1
ws[f'C{count}'].value = float(result)
wb.save('results.xlsx')
wb.close()
count += 1
except:
client_tcp.write_register(41000, 0) # 将 act 信号置为 False
sleep(2) #
def main():
initializations()
client_tcp, client_rtu = do_connections()
get_gauge_data(client_tcp, client_rtu)
if __name__ == '__main__':
main()

43
dial_gauge/gauge.xml Normal file
View File

@ -0,0 +1,43 @@
<?xml version="1.0" encoding="UTF-8"?>
<m>
<l>
<c name="addr" type="2" value="41000"/>
<c name="addr_1st" type="2" value="0"/>
<c name="addr_2nd" type="2" value="0"/>
<c name="bit_bias" type="2" value="0"/>
<c name="byte_bias" type="4" value="0"/>
<c name="description" type="10" value="pc to robot"/>
<c name="dev_name" type="10" value="autotest"/>
<c name="dev_type" type="10" value="MODBUS"/>
<c name="end_addr" type="2" value="41000"/>
<c name="function" type="10" value=""/>
<c name="len" type="2" value="1"/>
<c name="name" type="10" value="act"/>
<c name="retain" type="1" value="false"/>
<c name="rw" type="10" value="rd"/>
<c name="type" type="10" value="bool"/>
<c name="value"/>
<c name="value_single" type="10" value=""/>
<c name="bias" type="2" value="0"/>
</l>
<l>
<c name="addr" type="2" value="41001"/>
<c name="addr_1st" type="2" value="0"/>
<c name="addr_2nd" type="2" value="0"/>
<c name="bit_bias" type="2" value="0"/>
<c name="byte_bias" type="4" value="0"/>
<c name="description" type="10" value="robot to pc"/>
<c name="dev_name" type="10" value="autotest"/>
<c name="dev_type" type="10" value="MODBUS"/>
<c name="end_addr" type="2" value="41001"/>
<c name="function" type="10" value=""/>
<c name="len" type="2" value="1"/>
<c name="name" type="10" value="time_to_go"/>
<c name="retain" type="1" value="false"/>
<c name="rw" type="10" value="rdwr"/>
<c name="type" type="10" value="bool"/>
<c name="value"/>
<c name="value_single" type="10" value=""/>
<c name="bias" type="2" value="0"/>
</l>
</m>

Binary file not shown.