This commit is contained in:
neonb 2025-01-21 10:51:06 +01:00
parent 6ed0cd9f99
commit 855084f7e7
3 changed files with 33 additions and 62 deletions

View File

@ -51,7 +51,7 @@ class ModbusComponent(Component):
strict=False,
)
if not self.client.connect():
QMessageBox.critical(None, "ERRORE", f"ERRORE TECNA - VERIFICARE CONNESSIONE USB")
QMessageBox.critical(None, "ERRORE", f"ERRORE MODBUS - VERIFICARE CONNESSIONE USB")
exit(-1)
#raise ConnectionError("device not reachable (could not connect): {} ({})".format(self.name, self.port))

View File

@ -1,11 +1,9 @@
import logging
import time
from .modbus_component import ModbusComponent
import logging
class PipeCutterComponent(ModbusComponent):
def __init__(self, config=None, name="pipe_cutter", period=1, lazy=True, paused=False, threaded=True):
def __init__(self, config=None, name=None, period=1, lazy=True, paused=False, threaded=True):
"""Initialize the Pipe Cutter Component."""
super().__init__(config=config, name=name, period=period, lazy=lazy, paused=paused, threaded=threaded)
self.log = logging.getLogger(self.name)
@ -13,74 +11,49 @@ class PipeCutterComponent(ModbusComponent):
self.current_od_of_pipe = None
self.machine_status = None
def config_changed(self):
super().config_changed()
def read_total_length(self):
"""Read the total length (Register 4X30)."""
"""Read the total length from Register 4X30."""
try:
self.current_total_length = self.read(register_address=30, data_type="16bit_uint")
self.log.info(f"Total length (Register 4X30): {self.current_total_length}")
self.current_total_length = self._read(register=30)
self.log.info(f"Total Length Read (Register 4X30): {self.current_total_length}")
return self.current_total_length
except Exception as e:
self.log.error(f"Failed to read total length (Register 4X30): {e}")
self.log.error(f"Error reading total length: {e}")
raise
def read_od_of_pipe(self):
"""Read the OD of the pipe (Register 4X60)."""
"""Read the OD of the pipe from Register 4X60."""
try:
self.current_od_of_pipe = self.read(register_address=60, data_type="16bit_uint")
self.log.info(f"OD of the pipe (Register 4X60): {self.current_od_of_pipe}")
self.current_od_of_pipe = self._read(register=60)
self.log.info(f"OD of Pipe Read (Register 4X60): {self.current_od_of_pipe}")
return self.current_od_of_pipe
except Exception as e:
self.log.error(f"Failed to read OD of the pipe (Register 4X60): {e}")
self.log.error(f"Error reading OD of pipe: {e}")
raise
def read_machine_status(self):
"""Read the machine status from Register 4X766."""
try:
self.machine_status = self.read(register_address=1,register=766 ,data_type="16bit_uint")
self.log.info(f"Machine status (Register 4X766): {self.machine_status}")
if not hasattr(self, 'client') or self.client is None:
self.log.error(
"Attempting to read machine status but 'client' is not initialized.")
raise AttributeError(
"'PipeCutterComponent' object has no attribute 'client'.")
self.machine_status = self._read(register=766)
self.log.info(f"Machine Status Read (Register 4X766): {self.machine_status}")
return self.machine_status
except Exception as e:
self.log.error(f"Failed to read machine status (Register 4X766): {e}")
except AttributeError as ae:
# Log more detailed instance state for debugging
self.log.error(
f"AttributeError while reading machine status: {ae}. Component State: "
f"name={self.name}, config={self.config}, client={getattr(self, 'client', None)}"
)
raise
def write_total_length(self, value):
try:
self.write(register_address=1,register=30, value=value, data_type="16bit_uint")
self.log.info(f"Wrote total length (Register 4X30): {value}")
except Exception as e:
self.log.error(f"Failed to write total length (Register 4X30): {e}")
self.log.error(f"Error reading machine status: {e}")
raise
def write_od_of_pipe(self, value):
try:
self.write(register_address=1,register=60, value=value, data_type="16bit_uint")
self.log.info(f"Wrote OD of pipe (Register 4X60): {value}")
except Exception as e:
self.log.error(f"Failed to write OD of pipe (Register 4X60): {e}")
raise
def write_bit_with_delay(self, register_address, bit_position, delay_ms):
"""
Set a specific bit in a Modbus register, wait for a specified time, and reset it.
:param register_address: The register address to write to.
:param bit_position: The bit position to modify (0-15).
:param delay_ms: The delay in milliseconds before resetting the bit to 0 .
"""
try:
# Step 1: Set the bit to 1
current_value = self.read(register_address, data_type="16bit_uint")
new_value = current_value | (1 << bit_position) # Set the bit
self.write(register_address, new_value, data_type="16bit_uint")
self.log.info(f"Set bit {bit_position} in register {register_address} to 1")
# Step 2: Wait for the specified delay
time.sleep(delay_ms / 1000) # Convert milliseconds to seconds
# Step 3: Reset the bit to 0
current_value = self.read(register_address, data_type="16bit_uint")
new_value = current_value & ~(1 << bit_position) # Clear the bit
self.write(register_address, new_value, data_type="16bit_uint")
self.log.info(f"Reset bit {bit_position} in register {register_address} to 0")
except Exception as e:
self.log.error(f"Failed to set and reset bit {bit_position} in register {register_address}: {e}")
raise

View File

@ -11,10 +11,8 @@ class Test_Pipe_Cutter(Test_Test):
A class to manage and operate a pipe cutter machine in a test environment.
"""
def __init__(self, config, components, recipe=None, run_once=False,
reset_on_start=False, enable_override=False, bench_name="generic", step=None,):
super().__init__(components=components, recipe=recipe, step=None, run_once=run_once,
reset_on_start=reset_on_start, enable_override=enable_override,)
def __init__(self, config, components, recipe=None, run_once=False,reset_on_start=False, enable_override=False, bench_name="generic", step=None,):
super().__init__(components=components, recipe=recipe, step=None, run_once=run_once,reset_on_start=reset_on_start, enable_override=enable_override,)
self.bench_name = bench_name
self.cutter = PipeCutterComponent()
self.current_cut_length = 0.0