Eduardo_Scripniciuc/st-ten-1
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dev PIPE

Browse Source
This commit is contained in:
germano laptop 2025-01-27 12:13:56 +01:00
parent b62549148d
commit 4ff84cf342
4 changed files with 79 additions and 119 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/components/pipe_cutter_component.py
View File

@ -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}")

90
src/test/modbus_read_data.py
View File

@ -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:

3
src/ui/test/test.py
View File

@ -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"):

103
src/ui/test_pipe_cutter/test_pipe_cutter.py
View File

@ -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()