LOGO! device control over a Modbus-TCP - Holding register control crashes via PyModbus calls

Question

I'm trying to control a lot of holding registers and show them in a window using Pymodbus and Tkinker

here is the code:

import tkinter as tk
from pymodbus.client.sync import ModbusTcpClient
from pymodbus.exceptions import ConnectionException, ModbusIOException
import time

# Dirección IP del LOGO!
logo_ip = '192.168.0.2'
# Puerto Modbus TCP (por defecto es 502)
logo_port = 502

client = ModbusTcpClient(logo_ip, port=logo_port)
Control = [528, 529, 530, 531, 532, 533, 534, 535]
TempDir = [536, 537, 538, 539, 540, 541, 542, 543]
SalidasDir = [8193, 8194, 8195, 8196, 8197, 8198, 8199, 8200]

FunCajonEst = [0] * 8
TempFuncDir = list(range(33))

try:
    connection = client.connect()
    if connection:
        print("Conexión exitosa al LOGO!")
    
    else:
        print("Error al conectar al LOGO!")

finally:
    print("Proceda")


# Definir las listas Temperatura y SalidasC2
Temperatura = [0] * 8



# Leer los valores iniciales desde el dispositivo Modbus
def leer_valores():
    global Temperatura, FunCajonEst
    x = 0
    for i in range(33):
        try:
            result = client.read_holding_registers(TempFuncDir[i], 1)
            if not isinstance(result, ModbusIOException) and result.registers[0] == 1:
                if x < len(FunCajonEst):  # Asegurar que x no exceda el tamaño de FunCajonEst
                    FunCajonEst[x] = i
                x += 1
                if x == 4:
                    x = 0
        except Exception as e:
            print(f"Error leyendo el registro {TempFuncDir[i]}: {e}")

    for i in range(8):
        try:
            result = client.read_holding_registers(TempDir[i], 1)
            if not isinstance(result, ModbusIOException):
                Temperatura[i] = result.registers[0]
            result = client.read_coils(SalidasDir[i], 1)
            if not isinstance(result, ModbusIOException):
                SalidasC2[i] = result.bits[0]
        except Exception as e:
            print(f"Error leyendo el registro o la bobina {i}: {e}")

leer_valores()

def create_window():
    window = tk.Tk()
    window.title("Interfaz con Tkinter")

    # Crear el marco principal
    main_frame = tk.Frame(window)
    main_frame.pack(padx=10, pady=10)

    # Lista para almacenar los valores de las cajas de texto
    Temp = [0] * 8
    # Lista para almacenar las referencias a las cajas de texto
    entries = []
    # Lista para almacenar las referencias a las etiquetas
    labels = []

    def on_button_click(index):
        try:
            if index < len(entries):  # Verificar que el índice esté dentro del rango
                value = int(entries[index].get())
                if index < len(Temp):  # Verificar que el índice esté dentro del rango
                    Temp[index] = value
                    print(f"Temperatura de control[{index}] = {value}")

                    write_result = client.write_register(Control[index], value)
                    if write_result.isError():
                        print(f"Error al escribir en holding register: {write_result}")
                    else:
                        print(f"Escritura en holding register exitosa: {write_result}")
                else:
                    print(f"Índice {index} fuera de rango para Temp")
            else:
                print(f"Índice {index} fuera de rango para entries")
        except ValueError:
            print(f"El valor en la caja de texto {index+1} no es un entero válido")
        except ConnectionException:
            print(f"Error de conexión durante la escritura en el registro Modbus. Reintentando...")
            time.sleep(1)
            on_button_click(index)
        except Exception as e:
            print(f"Error durante la escritura en el registro Modbus: {e}")

    # Función para actualizar etiquetas
    def actualizar_textos():
        leer_valores()
        for i in range(8):
            if i * 2 < len(labels):  # Verificar que el índice esté dentro del rango
                labels[i * 2].config(text=f"Temperatura {i+1}: {Temperatura[i]}")
            if i * 2 + 1 < len(labels):  # Verificar que el índice esté dentro del rango
                labels[i * 2 + 1].config(text=f"Modo de funcionamiento: {FunCajonEst[i]}")

    # Crear las filas con botones, cajas de texto y etiquetas
    for i in range(8):
        # Crear y mostrar la etiqueta con el valor de Temperatura
        message_label = tk.Label(main_frame, text=f"Temperatura {i+1}: {Temperatura[i]}")
        message_label.pack(pady=3)
        labels.append(message_label)

        message_label = tk.Label(main_frame, text=f"Modo de funcionamiento: {FunCajonEst[i]}")
        message_label.pack(pady=3)
        labels.append(message_label)

        frame = tk.Frame(main_frame)
        frame.pack(fill=tk.X, pady=3)

        button = tk.Button(frame, text=f"Control de temperatura {i+1}", command=lambda i=i: on_button_click(i))
        button.pack(side=tk.LEFT, padx=3)

        entry = tk.Entry(frame)
        entry.pack(side=tk.RIGHT, fill=tk.X, expand=True, padx=5)
        entries.append(entry)

    # Crear el botón "Recargar"
    reload_button = tk.Button(window, text="Recargar", command=actualizar_textos)
    reload_button.pack(pady=10)

    window.mainloop()

create_window()

The problem seems to be in the call to write_result = client.write_register(Control[index], value) because the code does not return an error and stops working, instead it returns:

Error al escribir en holding register: Modbus Error: [Input/Output] [WinError 10054] Se ha forzado la interrupción de una conexión existente por el host remoto

This code is just a more complex version of this one, which also does not work:

from pymodbus.client.sync import ModbusTcpClient
from pymodbus.exceptions import ConnectionException

# Dirección IP del LOGO!
logo_ip = '192.168.0.2'
# Puerto Modbus TCP (por defecto es 502)
logo_port = 502

client = ModbusTcpClient(logo_ip, port=logo_port)

try:
    connection = client.connect()
    if connection:
        print("Conexión exitosa al LOGO!")

        write_result = client.write_register(528, 10)
        if write_result.isError():
            print(f"Error al escribir en holding register: {write_result}")
        else:
            print(f"Escritura en holding register exitosa: {write_result}")
        # Dirección del holding register en el LOGO!
        
    
        read_result = client.read_holding_registers(529, 1)
        read_value = read_result
        print(f"Valor leído del holding register 529: {read_value}")
        
        client.close()
    else:
        print("Error al conectar al LOGO!")
except ConnectionException as e:
    print(f"Error de conexión: {e}")
finally:
    client.close()

I read some posts about this issue, but I could not figure out what I´m doing wrong, in fact, this worked this morning.

In resume, my problem is that I can't write a holding register of my LOGO! device, and I don´t know why.

Answer 1

(...) what I´m doing wrong (...) ?"
_{/ in fact, this worked this morning /}

Welcome to the zoo of distributed-systems. This zoo is full of strange animals and cages are left open, so courage & keep walking!

Good news - the TCP port# is set correct, as documented to be reserved = 502

Bad news - the rest is on us to design, so as to survive in this wild zoo, even if indeed wild things started to happen and blocked our intents, / ... that worked this morning /.

A good Design Strategy for such a wild zoo ?

# ASSUME NOTHING            ;;; the oldest assembly language MACRO I keep using until today

We simply do not get much guaranteed in distributed-systems, the less in heterogeneous distributed-systems and even less in COTS devices, used in industrial environments with blind, optimistic assumptions lead designs with zero self-healing and no redundancy to cope with isolated failure(s) that simply happen. Failures do happen. Aerospace people know that, MIL people know that, automotive people know that, life-saving people know that, lawyers know that, insurance people live on that. We know that too, only we all do not know when these happen. So using good defensive designs is our responsibility, so as to survive some, better all identifiable local, isolated levels of failures.

We can still design our code to survive. On the level of generic Python the code syntax allows us to use "fusing"-isolations using with <constructor> as <name> :-context-handlers and classical "fusing"-scopes in try: ... except <exception(s)>: ... finally: ...-imperative code structures, that can help you build your code "fused"-enough by these syntax-compositions to run indeed "assumption-free", so as not to expect anything to happen as expected, when expected or even at all.

Looks like an Ol' Paranoid Eden?

Well, the ZOO is cruel and animals are wild - your code has to survive that - from cut cable(s), frozen controller, panicked L3-network, crashed Windows, ... ZOO is indeed full of surprises in the realms of real world distributed-systems.

The best next step ?

You already use Tkinter for GUI. The superpower of Tkinter is in its internal capability to deliver to you a well tuned soft-realtime code-execution system for free. So you can enjoy this facette of Tkinter to equip a simply serial Python code execution to actually work with a "cooperative-concurrency" -- right, built using Tkinter timed ( and time-outed, like watchdogs ), multiplexed, fail-safe ( best as distributed-cooperating fail-safe ) ModBus-register-monitors, that can help you survive many of your ZOO beasts, be it so designed. So the goal is achievable without any rocket science.

For this the Tkinter-toolbox has many additional beauties than just the fancy-looking GUI-layer on top of these - each "button" can be set to use - timed-operations ( .after()- + .after_cancel()-methods and many others for event-driven, absolute-time and relative-time activations or cancelling thereof, like in security count-downs, like the train engines' control room "dead-man button" to take automated self-defensive actions if & once something did not happen in due time ... ), all that neatly coordinated and controlled by the ( hidden ) .mainloop()-scheduler, i.e. you can benefit from using per-register operated ( both timed and isolated to be independent from all others ) update-attempts ( assuming in their calls these may brutally fail to operate, and let Python control exception incidents accordingly, using defensive handling like in :

    # ASSUME NOTHING :o)
    # ... RECORD ENTRY IN JOURNAL-section
    #            FOR A (still possible) POST-MORTEM REVIEW AS NECROLOG
    try:
          write_result = client.write_register( Control[index],
                                                value
                                                )
    except:
          # client.write_register() call crashed - analyze why :
          # ... LOG
          # ... 
          # ... SET
          write_result = a_preDEFed_crash_proxy_value
          # ... HANDLE
          # ...
    finally:
          if write_result.isError():
             # ... LOG
             # ...
             # ... GUI, if indeed deemed useful - ref. NASA Skylab Station Control
             print( f"Error al escribir en holding register: {write_result}" )
             # ... HANDLE
             # ...
          else:
             # ... GUI, if indeed deemed useful - ref. NASA Apollo Guidance System SYNDROME
             print( f"Escritura en holding register exitosa: {write_result}" )
             # ... PROCESS
             # ...
    # ... ROLL-BACK JOURNAL-section FROM NECROLOG ( as we did not die here - hooray, more work to do )
    # ... Keep Walking!