import asyncio import json import re import sys import time import pymodbus import serial import threading import binascii import datetime import requests from pymodbus.client import ModbusTcpClient from pymodbus import ModbusException from pymodbus.exceptions import ModbusIOException import utils from utils import PLCDropParamsData, PLCTestReqResultData, print_with_timestamp PLC_CONNECT_REGISTER = 'D10' PLC_READ_INDEX_REGISTER = 'D15' # 代表1个通道(工位)的夹具最多可放多少个样品，可选数 [1 2 4] # D11 代表第1通道 的运行状态 1--试验中 0--停止中 # D12 代表第1通道 的运行状态 1--试验中 0--停止中 # D13 代表第1通道 的运行状态 1--试验中 0--停止中 # D14 代表第1通道 的运行状态 1--试验中 0--停止中 def get_visible_string(s): """ 清理字符串，移除非法字符和空字符 :param s: 原始字符串 :return: 清理后的字符串 """ if not s: return "" # 移除空字符和不可打印字符 # 只保留ASCII可打印字符 (32-126) visible_str = ''.join( char for char in s if 32 <= ord(char) <= 126 or char in '\n\r\t' ) # 移除首尾空白字符 visible_str = visible_str.strip() return visible_str class ModbusDtMachine: def __init__(self, plc_type, comm_port, comm_config={}, drop_register={}, plc_address=0x01, tcp_modbus_unit=0x01, machine_SN='DTM_STA_00001'): self.connect_thread = None self.drop_register = drop_register self.plc_address = plc_address self.tcp_modbus_unit = tcp_modbus_unit self.DTM_SN = machine_SN self.plc_address = tcp_modbus_unit self.plc_comm_lock = threading.Lock() self.client = None self.connection = None self.preConnection = None self.plc_type = plc_type self.comm = comm_port self.comm_config = comm_config self.exit_connect_thread = False self.nextTestResultBufferFlag = 0 # 从PLC读取测试结果，采用A B 双buffer制，上位机读A时，PLC写B，反过来成立 # print_with_timestamp("dtmmodbus client init", comm_port, comm_config)#cyx # 将三菱PLC使用的MXXX Dxxxx的寄存器地址转换为modbus 的数值寄存器地址 # D0160 --> 160 def conv_register_address(self, s): address_type = None # 使用正则表达式检查格式，确保是一个或两个字母后跟三或四位数字 pattern = re.compile(r'^([MD]|HD)\d{1,6}$') if not pattern.match(s): raise ValueError( "Invalid format. The string must start with 'M', 'D', or 'HD' followed by three or four digits.") address_type = s[0] if s[1].isdigit() else s[:2] # 获取前缀，可以是'M', 'D', 或者 'HD' address_value = int(s[len(address_type):]) # 获取后面的数字部分 converted_address = address_value # 根据PLC的前缀来决定如何映射到Modbus地址 if address_type == 'M': converted_address = address_value # 假设'M'类型的地址映射到Modbus的Coil modbus_type = "Coil" elif address_type == 'HD': # 假设'D'类型的地址映射到Modbus的Holding Register if self.plc_type == 'xinjie': converted_address = address_value + 41088 else: # mhi converted_address = address_value modbus_type = "Holding_Register" elif address_type == 'D': # 'HD'可能指的是特殊功能或更高位数据的地址 if self.plc_type == 'xinjie': converted_address = address_value else: # mhi converted_address = address_value modbus_type = "Data_Register" return modbus_type, converted_address def conv_drop_height(self, height): if self.plc_type == 'xinjie': height_register_value = int(100 * height) else: # mhi height_register_value = int(100 * height) return height_register_value def is_connected_to_tcp_server(self): if self.connection: return True else: return False def connect_to_modbus_server(self, host='127.0.0.1', port=5020): if self.client: # 已经连接到modbus Server ，则立即返回 print_with_timestamp(f"Modbus client--plc {self.comm} {self.plc_address} connectted state is OK", color='blue') return # 创建 Modbus 客户端实例 self.client = ModbusTcpClient(host, port=port) # 连接到 Modbus 服务器 self.connection = self.client.connect() if self.connection: print_with_timestamp( f"Modbus tcp client-- {self.tcp_modbus_unit}-{self.plc_address}-{self.comm} connected modbus tcp server successfully" , color='green') else: print_with_timestamp( f"Modbus tcp client-- {self.tcp_modbus_unit}-{self.plc_address}--{self.comm} connect to modbus tcp server error", color='red') def connect_monitor(self): while not self.exit_connect_thread: if not self.is_connected_to_tcp_server(): self.connect_to_modbus_server() # 等待 5 秒后继续下一次连接尝试 time.sleep(5) def start_connect_thread(self): self.connect_thread = threading.Thread(target=self.connect_monitor) self.connect_thread.start() def stop_connect_thread(self): self.exit_connect_thread = True self.connect_thread.join() # 等待线程结束 def set_station_dropheight(self, station_no, height): # 数据库中的height 是float 并且mm为单位 value_return = {'status': 'error', 'message': f"set station{station_no} dropheight error"} height_register_value = self.conv_drop_height(height) # 将mm 单位 的跌落高度转换 寄存器的值 try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['height']) print_with_timestamp(f'ModbusDtMachine set {self.DTM_SN} station {station_no} dropheight {height} mm', address_type, address_value, height_register_value) # cyx resp = self.client.write_registers(address_value, [height_register_value], self.plc_address) if resp.isError() is False: value_return = {'status': 'success', 'message': f"set station {station_no} dropheight success"} else: print_with_timestamp(f"ModbusDtMachine set station {station_no} dropheight success error") except ModbusException as error: value_return['error'] = error pass return value_return def set_station_cycles(self, station_no, cycles): value_return = {'status': 'error', 'message': f"set station{station_no} cycles error"} try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['cycles']) print_with_timestamp(f"ModbusDtMachine set {self.DTM_SN} station {station_no} cycles {cycles} ", address_type, address_value, cycles) # cyx resp = self.client.write_register(address_value, cycles, self.plc_address) if resp.isError() is False: value_return = {'status': 'success', 'message': f"set station{station_no} cycles success"} except ModbusException as error: value_return['error'] = error pass return value_return def set_station_finished(self, station_no, finished): value_return = {'status': 'error', 'message': f"set station{station_no} finished error"} try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['cyclesFinished']) resp = self.client.write_register(address_value, finished, self.plc_address) if resp.isError() is False: value_return = {'status': 'success', 'message': f"set station{station_no} finished success"} except ModbusException as error: value_return['error'] = error pass return value_return def set_station_dropdirection(self, station_no, direction): value_return = {'status': 'error', 'message': f"set station{station_no} finished error"} direction_register = self.drop_register[station_no].get('direction') print_with_timestamp(f"ModbusDtMachine set_station_dropdirection direction register {direction_register} ") if direction_register is None or direction is None or direction_register == "": value_return = {'status': 'success', 'message': f"drop direction set is not demand"} return value_return try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['direction']) resp = self.client.write_register(address_value, direction, self.plc_address) if resp.isError() is False: value_return = {'status': 'success', 'message': f"set station{station_no} finished success"} except ModbusException as error: value_return['error'] = error pass return value_return def set_station_dutInfo(self, station_no, dutInfo): value_return = {'status': 'error', 'message': f"set station{station_no} dutInfo error"} dutBasic_register = self.drop_register[station_no].get('dutBasic') # print_with_timestamp(f"ModbusDtMachine set_station_dutInfo --0 {dutInfo} {dutBasic_register} {type(dutInfo)} ") if dutBasic_register is None or dutInfo is None or dutBasic_register == "": value_return = {'status': 'success', 'message': f"dut info set is not demand"} return value_return try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['dutBasic']) # 20250827 增加判断充UI传下来的是单个dut的SN，还是多个dut的SN数组 dut_info_type = 0 if dutInfo is None else 2 if isinstance(dutInfo, list) else 1 if isinstance(dutInfo, dict) else 0 if dut_info_type == 1: dut0 = dutInfo elif dut_info_type == 2: dut0 = dutInfo[0] else: dut0 = {} # print_with_timestamp(f"ModbusDtMachine set_station_dutInfo --1 {dut0} {dut_info_type} ") dutProject = dut0.get("project", "") # 项目代码 dutPhase = dut0.get("phase", "") # 项目阶段 dutProjectType = dut0.get("projectType", "") # 项目方案 dutWeeks = str(dut0.get("weeks", 0)) # 测试周次 dutWorkOrder = dut0.get("workOrder", "") # 工单 registers_value = [0x0001] # 为产品信息有效标识 registers_value.extend(utils.string_to_registers(dutProject, 36)) registers_value.extend(utils.string_to_registers(dutPhase, 36)) registers_value.extend(utils.string_to_registers(dutProjectType, 36)) registers_value.extend(utils.string_to_registers(dutWeeks, 36)) registers_value.extend(utils.string_to_registers(dutWorkOrder, 36)) registers_value_bottom = [] if dut_info_type == 1: # 20250830 为了兼容1个工位(通道) 可能同时放置4个被测样品 registers_value_bottom.extend(utils.string_to_registers(dut0.get('SN', " "), 36)) # 产品序列号 elif dut_info_type == 2: pass # registers_value.extend(utils.string_to_registers(dut0.get('SN', " "), 36)) # 产品序列号 for dut in dutInfo: # 20250830 为了兼容1个工位(通道) 可能同时放置4个被测样品 registers_value_bottom.extend(utils.string_to_registers(dut.get('SN', " "), 36)) else: # dufInfo 参数非法 return value_return # self.client_write 不能写太多数量的寄存器 resp1 = self.client.write_registers(address_value, registers_value, self.plc_address) resp2 = self.client.write_registers(address_value + len(registers_value), registers_value_bottom, self.plc_address) # print_with_timestamp(f"ModbusDtMachine set_station_dutInfo dutInfo register--6 {dutBasic_register} {registers_value} " # f"{resp1} {resp2}") if resp1.isError() is False and resp2.isError() is False: value_return = {'status': 'success', 'message': f"set station{station_no} dutInfo success"} except ModbusException as error: value_return['error'] = error pass return value_return def read_station_dutInfo(self, station_no, sn_count=1): """ 从 Modbus 寄存器读取试验样品信息并转换为字符串 :param station_no: 工位号 :param sn_count: SN 数量，默认为1（单个产品） :return: 包含产品信息的字典 """ value_return = {'status': 'error', 'message': f"get station{station_no} dutInfo error"} try: # 获取寄存器地址 dutBasic_register = self.drop_register[station_no].get('dutBasic') if dutBasic_register is None or dutBasic_register == "": value_return = {'status': 'success', 'message': f"dut info get is not demand"} return value_return address_type, address_value = self.conv_register_address(dutBasic_register) # 计算需要读取的寄存器数量 # 固定部分: 1个标识符 + 5个字段 × 18个寄存器/字段 = 91个寄存器 fixed_registers_count = 1 + 5 * 18 # 标识符 + 5个字段(每个36字节=18寄存器) # 第一次读取：固定部分 response_fixed = self.client.read_holding_registers( address_value, fixed_registers_count, self.plc_address ) if response_fixed.isError(): value_return['message'] = f"Failed to read fixed registers: {response_fixed}" return value_return fixed_registers = response_fixed.registers # 解析标识符 identifier = fixed_registers[0] if identifier != 0x0001: value_return['message'] = f"Invalid identifier: {identifier}" return value_return # 解析固定字段 dut_info = {} index = 1 # 跳过标识符 # 项目代码 (36字节 = 18寄存器) project_registers = fixed_registers[index:index + 18] dut_info['project'] = utils.registers_to_string(project_registers).strip() index += 18 # 项目阶段 (36字节 = 18寄存器) phase_registers = fixed_registers[index:index + 18] dut_info['phase'] = utils.registers_to_string(phase_registers).strip() index += 18 # 项目方案 (36字节 = 18寄存器) projectType_registers = fixed_registers[index:index + 18] dut_info['projectType'] = utils.registers_to_string(projectType_registers).strip() index += 18 # 测试周次 (36字节 = 18寄存器) weeks_registers = fixed_registers[index:index + 18] weeks_str = utils.registers_to_string(weeks_registers).strip() dut_info['weeks'] = int(weeks_str) if weeks_str.isdigit() else 0 index += 18 # 工单 (36字节 = 18寄存器) workOrder_registers = fixed_registers[index:index + 18] dut_info['workOrder'] = utils.registers_to_string(workOrder_registers).strip() # 第二次读取：SN部分 sn_list = [] if sn_count > 0: # 计算SN部分的起始地址和数量 sn_start_address = address_value + fixed_registers_count sn_registers_count = sn_count * 18 # 每个SN占36字节=18寄存器 response_sn = self.client.read_holding_registers( sn_start_address, sn_registers_count, self.plc_address ) if response_sn.isError(): value_return['message'] = f"Failed to read SN registers: {response_sn}" return value_return sn_registers = response_sn.registers # 解析SN列表 for i in range(sn_count): start_idx = i * 18 end_idx = start_idx + 18 if end_idx <= len(sn_registers): sn_part = sn_registers[start_idx:end_idx] # 检查是否全为0 if all(reg == 0 for reg in sn_part): continue # 全0则放弃 sn_raw = utils.registers_to_string(sn_part).strip() sn = get_visible_string(sn_raw) if not sn: continue else: sn_list.append(sn) # 根据SN数量组织返回数据 if sn_count == 1: dut_info['SN'] = sn_list[0] if sn_list else "" value_return = { 'status': 'success', 'message': f"read station{station_no} dutInfo success", 'data': dut_info } else: # 为每个SN创建一个完整的信息副本 duts_info = [] for sn in sn_list: dut_copy = dut_info.copy() dut_copy['SN'] = sn duts_info.append(dut_copy) value_return = { 'status': 'success', 'message': f"read station{station_no} dutInfo success", 'data': duts_info } except Exception as e: value_return['error'] = str(e) print_with_timestamp(f"Error in read_station_dutInfo: {e}", color='red') return value_return def read_station_dropheight(self, station_no): value_return = {'status': 'error', 'message': f"read station{station_no} dropheight error"} try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['height']) resp = self.client.read_holding_registers(address_value, 0x1, self.plc_address) if resp.isError() is False and len(resp.registers) >= 1: value_return = {'status': 'success', 'message': f"read station{station_no} height success", 'value': resp.registers[0]} except ModbusException as error: value_return['error'] = error pass return value_return def read_station_result_counter(self, station_no): value_return = {'status': 'error', 'message': f"set station{station_no} result counter error"} try: if self.drop_register[station_no].get('stationResultCounter'): address_type, address_value = self.conv_register_address( self.drop_register[station_no]['stationResultCounter']) resp = self.client.read_holding_registers(address_value, 0x2, self.plc_address) # print_with_timestamp(f"dtMachine modbus read_station_result_counter " # f"{self.drop_register[station_no].get('stationResultCounter')} {resp}") if resp.isError() is False and len(resp.registers) >= 2: value_return = {'status': 'success', 'message': f"read station{station_no} result counter success", 'value': (resp.registers[0] | resp.registers[1] * 0x10000)} except ModbusException as error: value_return['error'] = error pass return value_return def read_station_cyclesFinished(self, station_no): value_return = {'status': 'error', 'message': f"set station{station_no} cyclesFinished error"} try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['cyclesFinished']) resp = self.client.read_holding_registers(address_value, 0x1, self.plc_address) if resp.isError() is False and len(resp.registers) >= 1: value_return = {'status': 'success', 'message': f"read station{station_no} cyclesFinished success", 'value': resp.registers[0]} except ModbusException as error: value_return['error'] = error pass return value_return def read_station_cyclesFinishedBulk(self, station_no, result_index=0, length=33): value_return = {'status': 'error', 'message': f"set station{station_no} cyclesFinishedBulk error"} try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['cyclesFinished']) address_value = address_value + PLCTestReqResultData.required_registers_size * result_index resp = self.client.read_holding_registers(address_value, length, self.plc_address) if resp.isError() is False and len(resp.registers) >= length: value_return = {'status': 'success', 'message': f"read station{station_no} cyclesFinishedBulk success", 'value': resp.registers, 'address': address_value} """ nextTestResultBufferFlag = 0 if result_index == 6: nextTestResultBufferFlag = 1 if self.drop_register[station_no].get('resultSwitch') and result_index in [2, 6]: try: address_type, flag_address_value = self.conv_register_address(self.drop_register[station_no]['resultSwitch']) resp = self.client.write_register(flag_address_value, nextTestResultBufferFlag, self.plc_address) if resp.isError() is True: print_with_timestamp(f"set station {station_no} nextTestResultBufferFlag error!!!!!") except Exception as error: print_with_timestamp(f"write switch register error {error}", color='red') # cyx """ except ModbusException as error: value_return['error'] = error print_with_timestamp(f"read_station_cyclesFinishedBulk error {error}", color='red') pass return value_return def read_station_dropParamsBulk(self, station_no, length=33): value_return = {'status': 'error', 'message': f"set station{station_no} dropParamsBulk error"} direction_register = self.drop_register[station_no].get('direction') if direction_register is None or direction_register == "": value_return = {'status': 'success', 'message': f"drop params read is not demand"} return value_return try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['direction']) resp = self.client.read_holding_registers(address_value, length, self.plc_address) if resp.isError() is False and len(resp.registers) >= length: value_return = {'status': 'success', 'message': f"read station{station_no} dropParamsBulk success", 'value': resp.registers} except ModbusException as error: value_return['error'] = error pass return value_return def start_station(self, station_no): value_return = {'status': 'error', 'message': f"start station{station_no} error"} try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['stop']) resp1 = self.client.write_coils(address_value, [0x00], self.plc_address) address_type, address_value = self.conv_register_address(self.drop_register[station_no]['start']) resp2 = self.client.write_coils(address_value, [0x01], self.plc_address) if resp1.isError() is False and resp2.isError() is False: value_return = {'status': 'success', 'message': f"start station{station_no} success"} except ModbusException as error: value_return['error'] = error pass return value_return def resume_station(self, station_no): value_return = {'status': 'error', 'message': f"resume station{station_no} error"} try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['stop']) resp1 = self.client.write_coils(address_value, [0x00], self.plc_address) address_type, address_value = self.conv_register_address(self.drop_register[station_no]['start']) resp2 = self.client.write_coils(address_value, [0x01], self.plc_address) if resp1.isError() is False and resp2.isError() is False: value_return = {'status': 'success', 'message': f"resume station{station_no} success"} except ModbusException as error: value_return['error'] = error pass return value_return def stop_station(self, station_no): value_return = {'status': 'error', 'message': f"stop station{station_no} error"} try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['start']) resp1 = self.client.write_coils(address_value, [0x00], self.plc_address) address_type, address_value = self.conv_register_address(self.drop_register[station_no]['stop']) resp2 = self.client.write_coils(address_value, [0x01], self.plc_address) if resp1.isError() is False and resp2.isError() is False: value_return = {'status': 'success', 'message': f"stop station{station_no} success"} except ModbusException as error: value_return['error'] = error pass return value_return def station_start_status(self, station_no): value_return = {'status': 'error'} try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['start']) resp1 = self.client.read_coils(address_value, 0x01, self.plc_address) if resp1.isError() is False: value_return = {'status': 'success', 'message': f" station{station_no} status success", 'value': resp1.bits[0]} except ModbusException as error: value_return['error'] = error pass return value_return def station_pause_status(self, station_no): # 不管是上位机还是HMI，如果暂停了机器，需要去设置这个值 value_return = {'status': 'error'} if self.drop_register[station_no].get('pause') is None: value_return = {'status': 'error', 'message': "register is not exist"} return value_return try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['pause']) resp1 = self.client.read_coils(address_value, 0x01, self.plc_address) if resp1.isError() is False: value_return = {'status': 'success', 'message': f" station{station_no} status success", 'value': resp1.bits[0]} except ModbusException as error: value_return['error'] = error pass return value_return def station_stop_status(self, station_no): # 读PLC 停止试验的寄存器 value_return = {'status': 'error'} try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['stop']) resp1 = self.client.read_coils(address_value, 0x01, self.plc_address) if resp1.isError() is False: value_return = {'status': 'success', 'message': f" station{station_no} status success", 'value': resp1.bits[0]} except ModbusException as error: value_return['error'] = error pass return value_return def station_run_status(self, station_no): # 读PLC 停止试验的寄存器 value_return = {'status': 'error', 'value': 0x00} if 'runStatus' not in self.drop_register[station_no]: return value_return try: address_type, address_value = self.conv_register_address(self.drop_register[station_no]['runStatus']) resp1 = self.client.read_coils(address_value, 0x01, self.plc_address) if resp1.isError() is False: value_return = {'status': 'success', 'message': f" station{station_no} status success", 'value': resp1.bits[0]} except ModbusException as error: value_return['error'] = error pass return value_return def read_plc_connect_state(self): value_return = {'status': 'error', 'message': f"get plc connect state error--0", 'value': 0x00} """ if self.is_connected_to_tcp_server() is False and self.preConnection: print_with_timestamp(f"Plc {self.comm}-{self.tcp_modbus_unit}-{self.plc_address} disconnect", color='red') if self.is_connected_to_tcp_server() and self.preConnection is None: print_with_timestamp(f"Plc {self.comm}-{self.tcp_modbus_unit}-{self.plc_address} connectted", color='green') self.preConnection = self.connection """ if self.is_connected_to_tcp_server() is False: value_return = {'status': 'error', 'message': f"get plc connect state error modbus tcp client is not connectted"} return value_return try: type, address_value = self.conv_register_address(PLC_CONNECT_REGISTER) resp = self.client.read_holding_registers(address_value, 0x9, self.plc_address) # 10 定义为PLC 连接状态寄存器 # print_with_timestamp(f"dtMachine_modbus read_plc_connect_state {self.comm} {resp.registers}") if resp.isError() is False and len(resp.registers) >= 9: # 20250905 读取5个 叠加运行状态 value_return = {'status': 'success', 'message': f"get plc connect state success", 'value': resp.registers} except ModbusException as error: value_return['error'] = error pass except Exception as error: value_return['error1'] = error return value_return def read_plc_read_index(self): value_return = {'status': 'error', 'message': f"get plc connect state error--0", 'value': 0x00} """ if self.is_connected_to_tcp_server() is False and self.preConnection: print_with_timestamp(f"Plc {self.comm}-{self.tcp_modbus_unit}-{self.plc_address} disconnect", color='red') if self.is_connected_to_tcp_server() and self.preConnection is None: print_with_timestamp(f"Plc {self.comm}-{self.tcp_modbus_unit}-{self.plc_address} connectted", color='green') self.preConnection = self.connection """ if self.is_connected_to_tcp_server() is False: value_return = {'status': 'error', 'message': f"get plc connect state error modbus tcp client is not connectted"} return value_return try: type, address_value = self.conv_register_address(PLC_READ_INDEX_REGISTER) resp = self.client.read_holding_registers(address_value, 0x4, self.plc_address) # 15 定义为PLC # print_with_timestamp(f"dtMachine_modbus read_plc_connect_state {self.comm} {resp.registers}") if resp.isError() is False and len(resp.registers) >= 4: value_return = {'status': 'success', 'message': f"get plc read index success", 'value': resp.registers} except ModbusException as error: value_return['error'] = error pass except Exception as error: value_return['error1'] = error return value_return def register_test(self, action, address, value): type, address_value = self.conv_register_address(address) if (type == 'Data_Register' or type == 'Holding_Register') and action.lower() == 'read': value_return = {'status': 'error', 'message': f"register test {action}:{address} error"} try: resp = self.client.read_holding_registers(address_value, 0x1, self.plc_address) if resp.isError() is False and len(resp.registers) >= 1: value_return = {'status': 'success', 'message': f"register test {action}:{address} success", 'value': resp.registers[0]} except pymodbus.exceptions.ModbusIOExceptio as error: print_with_timestamp('ModbusDtMachine register_test', action, address, value, self.client.get_last_error()) # cyx value_return['error'] = error except Exception as error: print_with_timestamp('ModbusDtMachine register_test', action, address, value, self.client.get_last_error()) # cyx value_return['error'] = error pass return value_return if (type == 'Data_Register' or type == 'Holding_Register') and action.lower() == 'write': try: print_with_timestamp('ModbusDtMachine register_test', action, address, value, address_value) # cyx resp = self.client.write_register(address_value, int(value), self.plc_address) if resp.isError() is False: value_return = {'status': 'success', 'message': f"register test {action}:{address} success"} except ModbusIOException as error: print_with_timestamp('ModbusDtMachine register_test', action, address, value, self.client.get_last_error()) # cyx value_return['error'] = error except Exception as error: print_with_timestamp('ModbusDtMachine register_test', action, address, value, self.client.get_last_error()) # cyx value_return['error'] = error pass return value_return if type == 'Coil' and action.lower() == 'read': try: resp = self.client.read_coils(address_value, 0x01, self.plc_address) if resp.isError() is False: value_return = {'status': 'success', 'message': f" register test {action}:{address} success", 'value': resp.bits[0]} except ModbusIOException as error: print_with_timestamp('ModbusDtMachine register_test', action, address, value, self.client.get_last_error()) # cyx value_return['error'] = error except Exception as error: print_with_timestamp('ModbusDtMachine register_test', action, address, value, self.client.get_last_error()) # cyx value_return['error'] = error return value_return if type == 'Coil' and action.lower() == 'write': try: resp = self.client.write_coils(address_value, [value], self.plc_address) if resp.isError() is False: value_return = {'status': 'success', 'message': f"register test {action}:{address} success"} except ModbusIOException as error: print_with_timestamp('ModbusDtMachine register_test', action, address, value, self.client.get_last_error()) # cyx value_return['error'] = error except Exception as error: print_with_timestamp('ModbusDtMachine register_test', action, address, value, self.client.get_last_error()) # cyx value_return['error'] = error return value_return pass