dev

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 register_address: The address of the register to write.
-    :param value: The value to write to the register
+    :param bit_position: The specific bit to write (0-15).
-    :return: True if the write was successful, False otherwise
+    :param delay_ms: The delay in milliseconds before writing the value.
     """
     try:
-        result = client.write_register(register_address, value, slave=1)
+        # Step 1: Wait for the specified delay
-        if result.isError():
+        print(
-            print(f"Error writing value {value} to register {register_address}: {result}")
+            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
+        # Define the registers and criteria
-        register_address = 60  # Address to read/write
+        monitor_register_address = 766  # Register to monitor for the value `105`
-        write_value = 900  # Value to write to the register
+        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
+        # Continuously monitor register
-        print(f"Reading register {register_address}...")
+        while True:
-        read_data = read_register(client, register_address)
+            #print(f"Reading register {monitor_register_address}...")
-        if read_data is not None:
+            read_data = read_register(client, monitor_register_address)
-            print(f"Register {register_address} value before write: {read_data[0]}")
+            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
+            # Add a delay between readings
-        print(f"Writing value {write_value} to register {register_address}...")
+            time.sleep(1)
-        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]}")
 
     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.")