diff --git a/src/components/pipe_cutter_component.py b/src/components/pipe_cutter_component.py index 9d45e74..bfc7b8c 100644 --- a/src/components/pipe_cutter_component.py +++ b/src/components/pipe_cutter_component.py @@ -191,7 +191,7 @@ class PipeCutterComponent(ModbusComponent): """ try: # Write the value to register 30 - self.write(register=30, data=value, data_type="16bit_uint") + self.write(register=30, data=value, data_type="32bit_uint") self.log.info(f"Successfully wrote value {value} to register 30 (4X30).") except Exception as e: self.log.error(f"Failed to write value {value} to register 30: {e}") diff --git a/src/test/modbus_read_data.py b/src/test/modbus_read_data.py index 15be258..3f4dc8e 100644 --- a/src/test/modbus_read_data.py +++ b/src/test/modbus_read_data.py @@ -1,8 +1,6 @@ -import serial from pymodbus.client import ModbusSerialClient as ModbusClient from pymodbus.exceptions import ModbusIOException -import logging -import time +import serial def read_register(client, register_address, count=1): @@ -15,10 +13,9 @@ 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}") + print(f"Error reading register {register_address}") return None return result.registers except ModbusIOException as modbus_error: @@ -29,66 +26,16 @@ def read_register(client, register_address, count=1): return None -def write_bit_in_register_with_delay(client, register_address, bit_position, delay_ms): - """ - 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 bit_position: The specific bit to write (0-15). - :param delay_ms: The delay in milliseconds before writing the value. - """ - try: - # 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"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", + port="COM4", # Change this to the appropriate port stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, baudrate=38400, timeout=1, # Timeout in seconds - strict=False, ) # Attempt to connect to the Modbus server @@ -99,31 +46,14 @@ def main(): return try: - # 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 - - # 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) + # Read registers 30 and 60 + for register_address in [30, 60]: + print(f"Reading register {register_address}...") + value = read_register(client, register_address) + if value is not None: + print(f"Register {register_address} value: {value[0]}") else: - print(f"Failed to read data from register {monitor_register_address}") - - # Add a delay between readings - time.sleep(1) - - except KeyboardInterrupt: - print("\nOperation interrupted by user.") + print(f"Failed to read register {register_address}.") except Exception as e: print(f"Unexpected error occurred: {e}") finally: diff --git a/src/ui/test/test.py b/src/ui/test/test.py index 2afd73f..fe9d8f7 100755 --- a/src/ui/test/test.py +++ b/src/ui/test/test.py @@ -362,6 +362,9 @@ class Test(Widget): self.data["recipe"] = model_to_dict(self.recipe) w = self.cycle_available_steps[self.step.step_type] show = None + if self.step.step_type == "pipe_cutter": + if current_w != w : + self.setCentralWidget(w) if self.step.step_type == "leak_2": self.setCentralWidget(w) # NEED TO PRESHOW UI if hasattr(w, "start"): diff --git a/src/ui/test_pipe_cutter/test_pipe_cutter.py b/src/ui/test_pipe_cutter/test_pipe_cutter.py index 97979ec..661482a 100644 --- a/src/ui/test_pipe_cutter/test_pipe_cutter.py +++ b/src/ui/test_pipe_cutter/test_pipe_cutter.py @@ -28,22 +28,18 @@ class Test_Pipe_Cutter(Test_Test): self.movie.start() - self.pipe_gif.setMovie(self.movie) - def start(self, recipe=None, step=None, pieces=None): """ Iterate through pipe cutter statuses, handling each status appropriately. + Stop cutting only when the status transitions from 102 -> 103 -> 102. """ - self.get_connection = self.components["pipe_cutter"].out.connect(self.get) - self.components["pipe_cutter"].resume() - + self.pipe_gif.setMovie(self.movie) # Call the parent start process show = super().start(recipe=recipe, step=step, pieces=pieces) if show is False: return show - # Read the current status of the pipe cutter - self.current_status = self.components["pipe_cutter"].read(register=766) + # Define status messages status_actions = { 102: "ready for operation", 103: "running", @@ -51,42 +47,66 @@ class Test_Pipe_Cutter(Test_Test): 100: "need to reset" } - for idx, status in enumerate([self.current_status]): # Iterate over the current status - print(f"Processing status {status_actions.get(status, 'unknown')} at index {idx}") + # Initialize variables to track current and previous status + previous_status = None + self.current_status = None - if status == 102: # Ready for operation - print(f"Status {status}: Starting the cutting process.") + self.get_connection = self.components["pipe_cutter"].out.connect(self.get) + self.components["pipe_cutter"].resume() + + + # Continuously monitor and handle pipe cutter statuses + while True: + self.current_status = self.components["pipe_cutter"].read(register=766) + print(self.current_status) + + # Status: 102 (ready for operation) + if self.current_status == 102: + if previous_status == 103: + #print("Transition detected: 103 -> 102. Stopping the cutting process.") + self.stop_cutting() + + # Start cutting if machine is ready self.start_cutting(recipe=recipe, step=step) - elif status == 103: # Running - print(f"Status {status}: The pipe cutter is running. Monitoring the system.") - self.display_text(text="TAGLIO CORRUGATO IN CORSO", bg_color="yellow") + # Status: 103 (running) + elif self.current_status == 103: + #print(f"Status {self.current_status}: The pipe cutter is running. Monitoring the system.") + self.display_text(text="TAGLIO CORRUGATO IN CORSO", bg_color="green") - elif status == 105: # Emergency - print(f"Status {status}: Emergency detected! Resetting the system.") + # Status: 105 (emergency) + elif self.current_status == 105: + #print(f"Status {self.current_status}: Emergency detected! Resetting the system.") try: - self.display_text(text="EMERGENZA INTERNEVUTA", bg_color="red") + self.display_text(text="EMERGENZA INTERVENUTA RIPETERE IL TEST", bg_color="red") + self.start_cutting() except Exception as e: logger.error(f"Failed to handle the emergency state: {e}") - elif status == 100: # Need to reset - print(f"Status {status}: The system requires resetting. Performing reset.") + # Status: 100 (need to reset) + elif self.current_status == 100: + #print(f"Status {self.current_status}: The system requires resetting. Performing reset.") try: - self.reset() - self.display_text(text="RESET IN CORSO", bg_color="yellow") + self.display_text(text="RESET IN CORSO", bg_color="yellow") + self.reset_machine() + self.display_text(text="TAGLIO CORRUGATO IN CORSO", bg_color="green") + self.start_cutting(recipe=recipe, step=step) + + except Exception as e: logger.error(f"Failed to reset the system: {e}") + # Unknown status else: - # Handle unknown statuses - print(f"Unknown status: {status}. Skipping operation.") - logger.warning(f"Unknown pipe cutter status {status} detected.") + # print(f"Unknown status {self.current_status}. Skipping operation.") + logger.warning(f"Unknown pipe cutter status {self.current_status} detected.") - # Final steps after all statuses are processed - if step is not None: - self.stop_cutting() - self.stop() + previous_status = self.current_status + if step is not None and self.end_cut: + print("Cutting process complete. Exiting monitoring loop.") + break + self.stop() return show def start_cutting(self, recipe=None, step=None): @@ -95,21 +115,29 @@ class Test_Pipe_Cutter(Test_Test): """ self.lenght = self.step.spec["lenght"] self.diameter = self.step.spec["diameter"] - try: - self.components["pipe_cutter"].write_total_length(self.lenght) - self.components["pipe_cutter"].write_od_of_pipe(self.diameter) - time.sleep(1) - self.components["pipe_cutter"].write_bit_with_delay(600, 0, 100) - print(f"Status 102: Pipe cutting process started successfully.") - except Exception as e: - logger.error(f"Failed to start the pipe cutting process: {e}") + if self.current_status == 102: + + try: + self.lenght = int(self.lenght)*100 + self.diameter = int(self.diameter)*100 + print(self.lenght) + self.components["pipe_cutter"].write_od_of_pipe(self.diameter) + time.sleep(1) + self.components["pipe_cutter"].write_od_of_pipe(self.diameter) + time.sleep(1) + self.components["pipe_cutter"].write_bit_with_delay(600, 0, 100) + except Exception as e: + logger.error(f"Failed to start the pipe cutting process: {e}") def stop_cutting(self): """ Stop the pipe cutting process and deactivate the pipe cutter component. """ try: - self.components["pipe_cutter"].write_bit_with_delay(600, 2, 100) + self.components["pipe_cutter"].write_bit_with_delay(600, 2, 100) + if self.current_status is 105: + self.end_cut = False + else: self.end_cut = True except Exception as e: logger.error(f"Failed to stop the pipe cutting process: {e}") @@ -117,7 +145,6 @@ class Test_Pipe_Cutter(Test_Test): def stop(self): self.components["pipe_cutter"].pause() self.disconnect(self.get_connection) - self.display_text(text="TAGLIO INTERROTTO", bg_color="yellow") super().stop()