diff --git a/src/test/modbus_read_data.py b/src/test/modbus_read_data.py index 157f0ba..15be258 100644 --- a/src/test/modbus_read_data.py +++ b/src/test/modbus_read_data.py @@ -5,8 +5,6 @@ import logging import time - - def read_register(client, register_address, count=1): """ Reads data from the Modbus server. @@ -17,41 +15,74 @@ def read_register(client, register_address, count=1): :return: List of register values or None if there was an error """ try: + # Read holding register result = client.read_holding_registers(register_address, count=count, slave=1) if result.isError(): print(f"Error reading register {register_address}: {result}") 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"Error reading register {register_address}: {e}") + print(f"Exception while reading register {register_address}: {e}") return None -def write_register(client, register_address, value): +def write_bit_in_register_with_delay(client, register_address, bit_position, delay_ms): """ - Writes data to a Modbus register. + Writes a specific bit in a Modbus register after a delay. :param client: An instance of ModbusClient - :param register_address: The address of the register to write - :param value: The value to write to the register - :return: True if the write was successful, False otherwise + :param register_address: The address of the register to write. + :param bit_position: The specific bit to write (0-15). + :param delay_ms: The delay in milliseconds before writing the value. """ try: - result = client.write_register(register_address, value, slave=1) - if result.isError(): - print(f"Error writing value {value} to register {register_address}: {result}") + # Step 1: Wait for the specified delay + print( + f"Waiting for {delay_ms} milliseconds before modifying bit {bit_position} in register {register_address}...") + time.sleep(delay_ms / 1000) # Convert milliseconds to seconds + + # Step 2: Read the current value of the register + read_result = client.read_holding_registers(register_address, count=1, slave=1) + if read_result.isError(): + print(f"Error reading register {register_address} before modifying.") return False + current_value = read_result.registers[0] + + # Step 3: Set the specific bit to 1 + new_value = current_value | (1 << bit_position) # Set the specified bit + write_result = client.write_register(register_address, new_value, slave=1) + + current_value = read_result.registers[0] + new_value = current_value & ~(1 << bit_position) + # Clear the bit + write_result = client.write_register(register_address, new_value, slave=1) + if write_result.isError(): + print(f"Error writing to register {register_address}.") + return False + print(f"Successfully set bit {bit_position} in register {register_address} to 1") + return True + except ModbusIOException as modbus_error: + print(f"Modbus IO Exception while writing bit {bit_position} to register {register_address}: {modbus_error}") + return False except Exception as e: - print(f"Error writing value {value} to register {register_address}: {e}") + print(f"Failed to write bit {bit_position} in register {register_address}: {e}") return False def main(): + # Configure logging (optional, useful for debugging) + #logging.basicConfig(level=logging.INFO) + #logger = logging.getLogger("pymodbus") + #logger.setLevel(logging.DEBUG) + # Modbus client configuration client = ModbusClient( method="rtu", - port="COM4", # Ensure the correct port for your setup + port="COM4", stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, @@ -68,31 +99,33 @@ def main(): return try: - # Define the register address and value - register_address = 60 # Address to read/write - write_value = 900 # Value to write to the register + # Define the registers and criteria + monitor_register_address = 766 # Register to monitor for the value `105` + target_register_address = 600 # Register to write the bit + bit_position = 1 # Bit position to write (e.g., `600.01`) + delay_ms = 2000 # 2-second delay before writing the bit - # Step 1: Read the register value - print(f"Reading register {register_address}...") - read_data = read_register(client, register_address) - if read_data is not None: - print(f"Register {register_address} value before write: {read_data[0]}") + # Continuously monitor register + while True: + #print(f"Reading register {monitor_register_address}...") + read_data = read_register(client, monitor_register_address) + if read_data is not None: + print(f"Register {monitor_register_address} value: {read_data[0]}") + # Check if the value is 105, then trigger the write to the bit + if read_data[0] == 105: + print( + f"Register {monitor_register_address} value is 105. Writing to register {target_register_address}.{bit_position}...") + write_bit_in_register_with_delay(client, target_register_address, bit_position, delay_ms) + else: + print(f"Failed to read data from register {monitor_register_address}") - # Step 2: Write a value to the register - print(f"Writing value {write_value} to register {register_address}...") - if write_register(client, register_address, write_value): - print(f"Successfully wrote value {write_value} to register {register_address}.") - else: - print(f"Failed to write value {write_value} to register {register_address}.") - - # Step 3: Read the register again to confirm the write - print(f"Reading register {register_address} after write...") - read_data = read_register(client, register_address) - if read_data is not None: - print(f"Register {register_address} value after write: {read_data[0]}") + # Add a delay between readings + time.sleep(1) except KeyboardInterrupt: - print("\nStopping operation...") + print("\nOperation interrupted by user.") + except Exception as e: + print(f"Unexpected error occurred: {e}") finally: client.close() print("Modbus client connection closed.")