diff --git a/src/test/modbus_read_data.py b/src/test/modbus_read_data.py index 3f4dc8e..33629c3 100644 --- a/src/test/modbus_read_data.py +++ b/src/test/modbus_read_data.py @@ -4,33 +4,36 @@ import serial def read_register(client, register_address, count=1): - """ - Reads data from the Modbus server. - - :param client: An instance of ModbusClient - :param register_address: The address of the register to read - :param count: Number of registers to read - :return: List of register values or None if there was an error - """ + """Reads data from the Modbus server.""" try: result = client.read_holding_registers(register_address, count=count, slave=1) if result.isError(): print(f"Error reading register {register_address}") return None return result.registers - except ModbusIOException as modbus_error: - print(f"Modbus IO Exception while reading register {register_address}: {modbus_error}") - return None except Exception as e: print(f"Exception while reading register {register_address}: {e}") return None +def write_register(client, register_address, value): + """Writes data to the Modbus server.""" + try: + result = client.write_register(register_address, value, slave=1) + if result.isError(): + print(f"Error writing to register {register_address}") + return False + return True + except Exception as e: + print(f"Exception while writing to register {register_address}: {e}") + return False + + def main(): # Modbus client configuration client = ModbusClient( method="rtu", - port="COM4", # Change this to the appropriate port + port="COM4", stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, @@ -38,7 +41,6 @@ def main(): timeout=1, # Timeout in seconds ) - # Attempt to connect to the Modbus server if client.connect(): print("Modbus client connected successfully.") else: @@ -54,6 +56,32 @@ def main(): print(f"Register {register_address} value: {value[0]}") else: print(f"Failed to read register {register_address}.") + + # Read and write register 20 + print(f"Reading register 20...") + value = read_register(client, 20) + if value is not None: + print(f"Register 20 value: {value[0]}") + + print(f"Writing 1 to register 20...") + if write_register(client, 20, 1): + print("Successfully wrote to register 20.") + + # Verify the write by reading again + print(f"Reading register 20 again...") + value = read_register(client, 20) + if value is not None: + print(f"Register 20 new value: {value[0]}") + + + else: + print("Failed to write to register 20.") + + + + else: + print("Failed to read register 20 before writing.") + except Exception as e: print(f"Unexpected error occurred: {e}") finally: