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

dev TODO test for tecna compatibility

Browse Source
This commit is contained in:
neonb 2025-01-22 12:27:15 +01:00
parent 9653812a13
commit fe460f030a
4 changed files with 77 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/components/modbus_component.py
View File

@ -73,10 +73,10 @@ class ModbusComponent(Component):
break
return read
def _write(self, register, value):
def _write(self, register, value,**kwargs):
while True:
self.lock.lock()
wrote = self.client.write_registers(register, value, skip_encode=True)
wrote = self.client.write_registers(register, value, skip_encode=True,**kwargs)
self.lock.unlock()
if wrote.isError():
self.log.exception(traceback.format_exception(type(wrote), wrote, wrote.__traceback__))
@ -118,7 +118,7 @@ class ModbusComponent(Component):
return self._decode(self._read(register,**kwargs), *args, data_type=data_type, gain=gain, offset=offset)
def write(self, register, data, *args, data_type="16bit_uint", gain=1, offset=0, **kwargs):
self._write(register, self._encode(data, *args, data_type=data_type, gain=gain, offset=offset, **kwargs))
self._write(register, self._encode(data, *args, data_type=data_type, gain=gain, offset=offset),**kwargs)
# def _get(self, data):
# # print("MODBUS", str(int(QThread.currentThreadId())), flush=True)

45
src/components/pipe_cutter_component.py
View File

@ -46,10 +46,10 @@ class PipeCutterComponent(ModbusComponent):
"""
try:
# Read holding register at 4x767 (register 767) for a single register (count=1)
response = self.read(register=766, count=1,)
print(response) #DEBUG
response = self.read(register=766, count=1, )
status = response
super()._get([status])
# super()._get([status])
print(f"response is {response}") # DEBUG
return status
except KeyError as ke:
@ -145,7 +145,7 @@ class PipeCutterComponent(ModbusComponent):
return super().write(
register,
self._convert_to_format(
data,
int(data),
formatting=formatting,
encoding_map=encoding_map,
),
@ -153,8 +153,10 @@ class PipeCutterComponent(ModbusComponent):
data_type=data_type,
gain=gain,
offset=offset,
slave=1,
**kwargs,
)
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.
@ -164,16 +166,18 @@ class PipeCutterComponent(ModbusComponent):
:param delay_ms: The delay in milliseconds before resetting the bit to 0 .
"""
try:
# Step 1: Set the bit to 1
# Step 1: Get the current value of the register
current_value = self.read(register_address, data_type="16bit_uint")
# Step 2: Set the bit to 1
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
# Step 3: Wait for the specified delay
time.sleep(delay_ms / 1000) # Convert milliseconds to seconds
# Step 3: Reset the bit to 0
# Step 4: 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")
@ -181,3 +185,30 @@ class PipeCutterComponent(ModbusComponent):
except Exception as e:
self.log.error(f"Failed to set and reset bit {bit_position} in register {register_address}: {e}")
raise
def write_total_length(self, value):
"""
Write a specified value to register 30 (4X30).
:param value: The value to be written to the register.
"""
try:
# Write the value to register 30
self.write(register=30, data=value, data_type="16bit_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}")
raise
def write_od_of_pipe(self, value):
"""
Write a specified value to register 30 (4X30).
:param value: The value to be written to the register.
"""
try:
# Write the value to register 30
self.write(register=60, data=value, data_type="16bit_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}")
raise

4
src/test/modbus_read_data.py
View File

@ -69,8 +69,8 @@ def main():
try:
# Define the register address and value
register_address = 767 # Address to read/write
write_value = 102 # Value to write to the register
register_address = 60 # Address to read/write
write_value = 900 # Value to write to the register
# Step 1: Read the register value
print(f"Reading register {register_address}...")

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

@ -1,7 +1,7 @@
from PyQt5.QtGui import QMovie
import logging
from ui.test_test import Test_Test
from components import PipeCutterComponent
import time
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@ -12,33 +12,48 @@ class Test_Pipe_Cutter(Test_Test):
"""
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,)
super().__init__(components=components, recipe=recipe, step=step, run_once=run_once,reset_on_start=reset_on_start, enable_override=enable_override,)
self.config = config
self.bench_name = bench_name
self.step=step
self.current_cut_length = 0
self.get_connection = None
self.current_cut_length = 0.0
self.max_cut_length = config.get("max_cut_length", 10.0)
self.movie = QMovie(config.get("config/instruction_images/st-ten-10/pipe.gif ")) # Animation Path
self.current_status = None
gif_path = config.get("config/instruction_images/st-ten-10/pipe.gif",) # GIF file path
self.movie = QMovie(gif_path)
self.pipe_gif.setMovie(self.movie)
self.movie.start()
self.pipe_gif.setMovie(self.movie)
def start(self, recipe=None, step=None, pieces=None):
print(self.config)
self.get_connection = self.components["pipe_cutter"].out.connect(self.get)
self.components["pipe_cutter"].resume()
show = super().start(recipe=recipe, step=step, pieces=pieces)
if show is False:
return show
if step is not None :
self.start_cutting()
self.display_cutting_animation(True)
self.stop_cutting()
return show
def start_cutting(self,recipe=None,step=None):
"""
Start the pipe cutting process and activate the pipe cutter component.
"""
self.current_status = self.components["pipe_cutter"]._get()
#print(f"current status is :{str(self.current_status)}")
#print(self.step.spec["lenght"])
if self.current_status == 102 : # ready for operation
try:
self.reset_machine()
self.components["pipe_cutter"].write_total_length(self.recipe.spec.get("length", None))
self.components["pipe_cutter"].write_od_of_pipe(self.recipe.spec.get("diameter", None))
#self.reset_machine()
self.components["pipe_cutter"].write_total_length(str(self.step.spec["lenght"]))
self.components["pipe_cutter"].write_od_of_pipe(str(self.step.spec["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}")
@ -53,14 +68,17 @@ class Test_Pipe_Cutter(Test_Test):
Stop the pipe cutting process and deactivate the pipe cutter component.
"""
try:
self.current_status = self.components["pipe_cutter"]._get()
if self.current_status in(106,107):
self.components["pipe_cutter"].write_bit_with_delay(600, 2, 100)
self.reset_machine()
#self.reset_machine()
except Exception as e:
logger.error(f"Failed to stop the pipe cutting process: {e}")
def stop(self):
self.components["pipe_cutter"].pause()
self.disconnect(self.get_connection)
super().stop()
def reset_machine(self):
"""
Reset the pipe cutter machine to its initial state.
@ -70,18 +88,4 @@ class Test_Pipe_Cutter(Test_Test):
except Exception as e:
logger.error(f"Failed to reset the pipe cutter machine: {e}")
def display_cutting_animation(self, show: bool):
"""
Show or hide the cutting animation GIF.
Args:
show (bool): Whether to display or stop the animation.
"""
if show:
self.pipe_gif.setVisible(True)
self.movie.start()
logger.info("Cutting animation started.")
else:
self.movie.stop()
self.pipe_gif.setVisible(False)
logger.info("Cutting animation stopped.")