from pymodbus.client import ModbusSerialClient import struct import time import signal import sys ## Clase de configuración de las bombas con referencias del PLC # class Bomba_class(): ## Declaración de variables de dispositivos fisicos y diccionario de errores def __init__(self): ## Id para la bomba que llenará. Para el caso de Colbun, solo una que llenará y vaciará. self.id_llenado = 29 self.id_vaciado = 29 self.entrada_usb = '/dev/ttyUSB0' self.estados = { 'conexion': False, 'errores_config': [], 'respuesta':0, } self.Preparar_bomba(self.id_llenado, 0x0001) #self.Imprimir_configuracion(29) try: signal.signal(signal.SIGINT, self.signal_handler)# Captura ctrl + c signal.signal(signal.SIGTERM, self.signal_handler)# Captura ctrl + z except: sys.exit('Configuracion de parada de emergencia fallida. Intente nuevamente') ## Función que contiene las configuraciones iniciales de las bombas. # @int id de la bomba. # @int direccion_giro 1 para sentido horario y 0 para anti horario. def Preparar_bomba(self, id, direccion_giro=0x0001): client = ModbusSerialClient(framer='rtu', port=self.entrada_usb, baudrate=9600, timeout=1, parity='N', stopbits=1, bytesize=8) # Conectar al dispositivo connection = client.connect() if connection: """ Este ejemplo y toda configuración que le sigue esta pensado para la bomba: -----------------------------------WT600--------------------------------- Consideracion importante para escribir valores flotantes: result = client.write_registers(16, self.float_to_registers(550.0), slave=id) self.float_to_registers(550.0) -> ([17417],[32768]) En este ejemplo tenemos que se utiliza la funcion 'write_registers'. La cual escribirá n cantidad de registros(2) ([17417],[32768]) desde la memoria indicada(16). De modo que en la memoria 16, será 17417 y en la 17 será 32768. Que representan en cuestion 550.0 La consideracion final es que las configuraciones que requieren flotantes necesitan dos memorias (Ej. RPM = 16 y 17) y es requerido escribirlas utilizando el ejemplo primero. """ self.estados['conexion'] = True ## Configurar modo de comunicacion result = client.write_register(0, 0x0001, slave=id) if result.isError(): self.estados['errores_config'].append(f"Error al configurar la comunicacion: {result}") #print(f"Error al configurar la comunicacion: {result}") #Configurar Modo de trabajo # 0 = Transfer Mode(Contiuo) | 1 = Filling Mode(Intervalos) "La configuracion de los intervalos no esta considerada en este listado. Ver el manual de RS485 y memorias 9-11-13-15" result = client.write_register(1, 0, slave=id) if result.isError(): self.estados['errores_config'].append(f"Error al configurar el modo de trabajo: {result}") #print(f"Error al configurar el modo de trabajo: {result}") #Configurar cabezal instalado (Linea inferior para indicar el modelo del cabezal correcto) # 0-YZ1515x,1-313D,2-BZ15,3-YZ2515x,4-BZ25,5-DG(6),6-DG(10),7-KZ25,8-YZ35,9-KZ35,10-WP110(3),11-WP110(4),12-WP110(6) result = client.write_register(2, 7, slave=id) # Int (0-12) if result.isError(): self.estados['errores_config'].append(f"Error al configurar el cabezal instalado: {result}") #print(f"Error al configurar el cabezal instalado: {result}") #Configurar manguera en el cabezal (Consultar linea inferior para indicar manguera correcta) #0-13#,1-14#,2-19#,3-16#,4-25#,5-17#,6-18#,7-15#,8-24#,9-35#,10-36#,11-0.5*0.8,12-1*1,13-2*1,14-3*1,15-2.4*0.8,16-4.8*1,17-73#,18-82#,19-8#,20-90#,21-88#,22-92# """ ID de mangueras compatibles respecto a los cabezales. Si la manguera no coincide con el cabezal, el registro devolverá un error. 0 "YZ1515x",{0,1,2,3,4,5,6}, 1 "313D ",{0,1,2,3,4,5,6}, 2 "BZ15 ",{0,1,2,3,4,5,6}, 3 "YZ2515x",{7,8,9 }, 4 "BZ25 ",{7,8 }, 5 "DG(6) ",{11,12,13,14,15,16 }, 6 "DG(10) ",{11,12,13,14,15,16 }, 7 "KZ25 ",{7,8,9,10 }, 8 "YZ35 ",{17,18 }, 9 "KZ35 ",{19,20,21,22}, """ result = client.write_register(3, 10, slave=id) if result.isError(): self.estados['errores_config'].append(f"Error al configurar manguera instalada: {result}") #print(f"Error al configurar manguera instalada: {result}") # Configurar la dirección de giro (1 para horario, 0 para antihorario) if not direccion_giro > 0x0001: result = client.write_register(4, direccion_giro, slave=id) if result.isError(): self.estados['errores_config'].append(f"Error al configurar la direccion de giro: {result}") #print(f"Error al configurar la direccion de giro: {result}") else: print('Valor de direccion de giro incompatible! Giro horario configurado por defecto') #Flowrate no se configura manualmente por memoria, si no, al cambiar uno de los valores de la formula de abajo. """Flowrate(0.001 ml/min - 9999 l/min) = RPM * Flow rate coeficient """ #Configurar angulo de succion (Angulo que se forma en la mangera al entrar en el cabezal) # Velocidad < 100rpm el angulo es 10°-720°. Para velocidad = 100rpm es 36°-720°. Por defecto esta a 0° result = client.write_register(7, 120, slave=id) # Int if result.isError(): self.estados['errores_config'].append(f"Error al configurar angulo de succion: {result}") #print(f"Error al configurar angulo de succion: {result}") """#Configurar velocidad de succion # Rango de velocidad 10-300rpm. Por defecto son 10rpm result = client.write_register(8, 10, slave=id) # Int if result.isError(): print(f"Error al configurar velocidad de succión: {result}") """ #Configurar veces de llenado (Veces que la bomba realizara llenados respecto del tiempo configurado en la memoria 11) # 0 a 9999 veces. 0 es loop infinito result = client.write_register(15, 1, slave=id) # Int if result.isError(): self.estados['errores_config'].append(f"Error al configurar veces de llenado: {result}") #print(f"Error al configurar veces de llenado: {result}") #Configurar RPM # 0.0rpm - 600.0rpm. result = client.write_registers(16, self.float_to_registers(600), slave=id) # Float if result.isError(): self.estados['errores_config'].append(f"Error al configurar RPMs: {result}") #print(f"Error al configurar RPMs: {result}") client.close() else: self.estados['conexion'] = False print("Error al conectar") ## Función para activar el motor de la bomba (Ejecución indefinida). # @int id de la bomba. # @int direccion 1 para sentido horario y 0 para anti horario. def Activar_bomba(self, id, direccion = 0): client = ModbusSerialClient(framer='rtu', port=self.entrada_usb, baudrate=9600, timeout=1, parity='N', stopbits=1, bytesize=8) connection = client.connect() if connection: result = client.write_register(4, direccion, slave=id) result = client.write_coil(0x0001, 0x0001, slave=id) #El segundo 0x0001 es ON para la bomba self.estados['respuesta'] = result.isError() if result.isError(): print(f"Error al iniciar la bomba: {result}") client.close() else: print(f"Error al conectar para iniciar la bomba {id}") ## Función para detener el motor del la bomba. # @int id de la bomba. def Detener_bomba(self, id): client = ModbusSerialClient(framer='rtu', port=self.entrada_usb, baudrate=9600, timeout=1, parity='N', stopbits=1, bytesize=8) connection = client.connect() if connection: result = client.write_coil(0x0001, 0x0000, slave=id) #El 0x0000 es OFF para la bomba self.estados['respuesta'] = result.isError() if result.isError(): print(f"Error al detener la bomba: {result}") client.close() else: print(f"Error al conectar para detener la bomba {id}") ## Función que imprime las memorias. # @int id de la bomba. def Imprimir_configuracion(self, id): # Principalmente ideada para imprimir valores flotantes client = ModbusSerialClient(framer='rtu', port=self.entrada_usb, baudrate=9600, timeout=1, parity='N', stopbits=1, bytesize=8) connection = client.connect() if connection: from pymodbus.constants import Endian from pymodbus.payload import BinaryPayloadDecoder result = client.read_holding_registers(address=9, count=2, slave=id) #print(result.registers) if not result.isError(): print(BinaryPayloadDecoder.fromRegisters(result.registers, byteorder=Endian.BIG, wordorder=Endian.BIG).decode_32bit_float()) result = client.read_holding_registers(address=11, count=2, slave=id) #print(result.registers) if not result.isError(): print(BinaryPayloadDecoder.fromRegisters(result.registers, byteorder=Endian.BIG, wordorder=Endian.BIG).decode_32bit_float()) client.close() ## Función que captura la señal kill de forma global y detiene las bombas. def signal_handler(self, sig, frame): self.Detener_bomba(29) sys.exit('Bomba Detenida por CTRL + C') ## Función para ingresar correctamente valores flotantes a la memoria de la bomba. def float_to_registers(self, value): packed = struct.pack('>f', value) return struct.unpack('>HH', packed) ## Función que retorna el estado de las bombas y el listado de problemas asociados. def Estado(self): if not self.estados['conexion']: return 2,self.estados['errores_config'].append('No hay conexión a la bomba') if self.estados['conexion'] and (len(self.estados['errores_config']) > 0 or not self.estados['respuesta']): return 1,self.estados['errores_config'] return 0,None