cam solution ok

This commit is contained in:
edo 2025-08-22 10:44:32 +02:00
parent 9193b2cc6d
commit b2e62fac3a
4 changed files with 94 additions and 263 deletions

View File

@ -73,77 +73,11 @@ class HikrobotSmartCamera(Component):
return -2 return -2
time.sleep(0.2) time.sleep(0.2)
# --- Start of Added Code: Switch Scheme --- if hasattr(self.config, "vision_config"):
# As per SDK PDF Chapter 5.11, this section switches the camera's active scheme. solution_name = self.config.vision_config.get("solution_name", None)
# It reads a comma-separated list of scheme names from the config file. else:
try: solution_name = None
solution_name=self.config.vision_config.get("solution_name", None) self.switch_scheme(solution_name,cam_idx)
if solution_name is not None:
scheme_name = solution_name
self.log.info(f"CAM{cam_idx} Attempting to switch to scheme: {scheme_name.decode()}")
# Step 1: Set the scheme name to be switched using MV_VS_SetStringValue.
# The node name is "SolutionName".
nRet = mv_lib.MV_VS_SetStringValue(pHandle, b"SolutionName", c_char_p(scheme_name))
if nRet != MV_VS_OK:
self.log.error(f"CAM{cam_idx} Set SolutionName failed! nRet [{nRet & 0xFFFFFFFF:#x}]")
# Clean up resources before returning
mv_lib.MV_VS_Logout(pHandle)
mv_lib.MV_VS_DestroyHandle(pHandle)
return -2
# Step 2: Call the scheme switch command using MV_VS_SetCommandValue.
# The command is "CommandSolutionSwitch".
nRet = mv_lib.MV_VS_SetCommandValue(pHandle, b"CommandSolutionSwitch")
if nRet != MV_VS_OK:
self.log.error(f"CAM{cam_idx} CommandSolutionSwitch failed! nRet [{nRet & 0xFFFFFFFF:#x}]")
# Clean up resources before returning
mv_lib.MV_VS_Logout(pHandle)
mv_lib.MV_VS_DestroyHandle(pHandle)
return -2
self.log.info(f"CAM{cam_idx} Switched to scheme '{scheme_name.decode()}' successfully.")
# Step 3: Wait for the scheme switch to complete.
# As per SDK PDF Chapter 3.3.2, we poll "SolutionSwitchStatus".
# A value of 0 indicates the switch is complete.
self.log.info(f"CAM{cam_idx} Waiting for scheme switch to complete...")
switch_timeout = 30 # 30-second timeout for the switch
start_time = time.time()
while True:
if time.time() - start_time > switch_timeout:
self.log.error(f"CAM{cam_idx} Timed out waiting for scheme switch.")
mv_lib.MV_VS_Logout(pHandle)
mv_lib.MV_VS_DestroyHandle(pHandle)
return -2
# Create a variable to store the status
status = ctypes.c_int64()
nRet = mv_lib.MV_VS_GetIntValue(pHandle, b"SolutionSwitchStatus", byref(status))
if nRet != MV_VS_OK:
self.log.error(f"CAM{cam_idx} Get SolutionSwitchStatus failed! nRet [{nRet & 0xFFFFFFFF:#x}]")
mv_lib.MV_VS_Logout(pHandle)
mv_lib.MV_VS_DestroyHandle(pHandle)
return -2
if status.value == 0:
self.log.info(f"CAM{cam_idx} Scheme switch completed successfully.")
break
else:
self.log.info(f"CAM{cam_idx} Scheme switch not yet complete. Waiting...")
time.sleep(1)
else:
self.log.warning(f"CAM{cam_idx} No scheme name defined in config for this camera index. Using default scheme.")
except KeyError:
self.log.warning("'schemes' key not found in vision_config. Using camera's default/current scheme.")
except Exception as e:
self.log.error(f"CAM{cam_idx} An unexpected error occurred during scheme switching: {e}")
# Clean up resources before returning
mv_lib.MV_VS_Logout(pHandle)
mv_lib.MV_VS_DestroyHandle(pHandle)
return -2
# --- End of Added Code ---
nRet = mv_lib.MV_VS_SetBoolValue(pHandle, b"CommandImageMode", 0) nRet = mv_lib.MV_VS_SetBoolValue(pHandle, b"CommandImageMode", 0)
if nRet != MV_VS_OK: if nRet != MV_VS_OK:
@ -160,6 +94,84 @@ class HikrobotSmartCamera(Component):
self.log.error(f"CAM{cam_idx} Start run failed! [{nRet & 0xFFFFFFFF:#x}]") self.log.error(f"CAM{cam_idx} Start run failed! [{nRet & 0xFFFFFFFF:#x}]")
self.connected = True self.connected = True
def switch_scheme(self,solution_name,cam_idx):
# --- Start of Added Code: Switch Scheme ---
# As per SDK PDF Chapter 5.11, this section switches the camera's active scheme.
# It reads a comma-separated list of scheme names from the config file.
pHandle= self.cam_list[cam_idx]["handle"]
try:
if solution_name is not None:
scheme_name = solution_name
self.log.info(f"CAM{cam_idx} Attempting to switch to scheme: {scheme_name}")
scheme_name_b = bytes(scheme_name,"utf-8")
# Step 1: Set the scheme name to be switched using MV_VS_SetStringValue.
# The node name is "SolutionName".
nRet = mv_lib.MV_VS_SetStringValue(pHandle, b"SrcOperateSolutionName", c_char_p(scheme_name_b))
if nRet != MV_VS_OK:
self.log.error(f"CAM{cam_idx} Set SolutionName failed! nRet [{nRet & 0xFFFFFFFF:#x}]")
# Clean up resources before returning
mv_lib.MV_VS_Logout(pHandle)
mv_lib.MV_VS_DestroyHandle(pHandle)
return -2
# Step 2: Call the scheme switch command using MV_VS_SetCommandValue.
# The command is "CommandSolutionSwitch".
nRet = mv_lib.MV_VS_SetCommandValue(pHandle, b"CommandProjectLoad")
if nRet != MV_VS_OK:
self.log.error(f"CAM{cam_idx} CommandSolutionSwitch failed! nRet [{nRet & 0xFFFFFFFF:#x}]")
# Clean up resources before returning
mv_lib.MV_VS_Logout(pHandle)
mv_lib.MV_VS_DestroyHandle(pHandle)
return -2
self.log.info(f"CAM{cam_idx} Switched to scheme '{scheme_name}' successfully.")
# Step 3: Wait for the scheme switch to complete.
# As per SDK PDF Chapter 3.3.2, we poll "SolutionSwitchStatus".
# A value of 0 indicates the switch is complete.
self.log.info(f"CAM{cam_idx} Waiting for scheme switch to complete...")
switch_timeout = 30 # 30-second timeout for the switch
start_time = time.time()
while True:
if time.time() - start_time > switch_timeout:
self.log.error(f"CAM{cam_idx} Timed out waiting for scheme switch.")
mv_lib.MV_VS_Logout(pHandle)
mv_lib.MV_VS_DestroyHandle(pHandle)
return -2
# Create a variable to store the status
status = ctypes.c_uint8()
nRet = mv_lib.MV_VS_GetEnumValue(pHandle, b"ScPushState", byref(status))
if nRet != MV_VS_OK:
self.log.error(f"CAM{cam_idx} Get SolutionSwitchStatus failed! nRet [{nRet & 0xFFFFFFFF:#x}]")
mv_lib.MV_VS_Logout(pHandle)
mv_lib.MV_VS_DestroyHandle(pHandle)
return -2
if status.value == 0:
self.log.info(f"CAM{cam_idx} Scheme switch completed successfully.")
break
else:
self.log.info(f"CAM{cam_idx} Scheme switch not yet complete. Waiting...")
time.sleep(1)
else:
self.log.warning(
f"CAM{cam_idx} No scheme name defined in config for this camera index. Using default scheme.")
except KeyError:
self.log.warning("'schemes' key not found in vision_config. Using camera's default/current scheme.")
except Exception as e:
self.log.error(f"CAM{cam_idx} An unexpected error occurred during scheme switching: {e}")
# Clean up resources before returning
mv_lib.MV_VS_Logout(pHandle)
mv_lib.MV_VS_DestroyHandle(pHandle)
return -2
# --- End of Added Code ---
@Component.reconfig_on_error @Component.reconfig_on_error
def _get(self): def _get(self):
if not self.connected: if not self.connected:
@ -222,6 +234,9 @@ class HikrobotSmartCamera(Component):
if len(self.cam_list) == self.num_cameras: if len(self.cam_list) == self.num_cameras:
# Dummy acquisition for each camera to avoid reading past images # Dummy acquisition for each camera to avoid reading past images
for cam_idx in range(self.num_cameras): for cam_idx in range(self.num_cameras):
solution_name = self.config.vision_config.get("solution_name", None)
self.log.info(f"Setting Solution {solution_name} to camera {cam_idx}")
self.switch_scheme(solution_name,cam_idx)
cam = self.cam_list[cam_idx] cam = self.cam_list[cam_idx]
nRet = mv_lib.MV_VS_SetCommandValue(cam["handle"], b"AcquisitionStart") nRet = mv_lib.MV_VS_SetCommandValue(cam["handle"], b"AcquisitionStart")
frame_res = MV_VS_GetFrame(cam["handle"]) frame_res = MV_VS_GetFrame(cam["handle"])

View File

@ -110,9 +110,7 @@ class Vision(Component):
self.set_recipe(vision_recipe) self.set_recipe(vision_recipe)
self.vision_recipe=vision_recipe self.vision_recipe=vision_recipe
# Store vision_config in the config dictionary instead of as an attribute # Store vision_config in the config dictionary instead of as an attribute
if "general" not in self.config: self.config.vision_config = self.vision_config
self.config["general"] = {}
self.config["general"]["vision_config"] = self.vision_config
self.recipe_watcher = QFileSystemWatcher([]) self.recipe_watcher = QFileSystemWatcher([])
self.recipe_watcher.fileChanged.connect(self._set_recipe) self.recipe_watcher.fileChanged.connect(self._set_recipe)
# LOAD MODEL # LOAD MODEL

View File

@ -178,10 +178,8 @@ try:
if "vision" in self.components and "hikrobot_sc" in self.components: if "vision" in self.components and "hikrobot_sc" in self.components:
self.components["vision"].set_sources({"hikrobot_sc": self.components["hikrobot_sc"].out}) self.components["vision"].set_sources({"hikrobot_sc": self.components["hikrobot_sc"].out})
# Connect the recipe_changed_signal from Vision to HikrobotSmartCamera
self.components["vision"].recipe_changed_signal.connect(
self.components["hikrobot_sc"].update_solution_name
)
# connect tecna to screwdriver # connect tecna to screwdriver
if "screwdriver" in self.components and "tecna_t3" in self.components: if "screwdriver" in self.components and "tecna_t3" in self.components:

View File

@ -1,109 +1,12 @@
import sys import sys
import logging
import time
from PyQt5.QtCore import pyqtSignal, QTimer from PyQt5.QtCore import pyqtSignal
from PyQt5.QtGui import QColor, QImage, QPalette, QPixmap from PyQt5.QtGui import QColor, QImage, QPalette, QPixmap
from PyQt5.QtWidgets import QHeaderView, QProgressBar, QTableWidgetItem, QDialog, QVBoxLayout, QLabel from PyQt5.QtWidgets import QHeaderView, QProgressBar, QTableWidgetItem
from src.lib.helpers.blocking_dialog import BlockingDialog from lib.helpers.blocking_dialog import BlockingDialog
from src.ui.helpers import calc_foreground_color from ui.helpers import calc_foreground_color
from src.ui.test_test import Test_Test from ui.test_test import Test_Test
class SchemeProgressDialog(QDialog):
"""
Dialog to show scheme switching progress and block the UI until complete.
"""
def __init__(self, parent=None, target_scheme=""):
super().__init__(parent)
self.setWindowTitle("Switching Camera Scheme")
self.setModal(True)
self.setMinimumWidth(400)
# Add a logger
self.log = logging.getLogger("SchemeProgressDialog")
layout = QVBoxLayout()
self.setLayout(layout)
self.label = QLabel(f"Switching camera to scheme: {target_scheme}\nPlease wait...")
layout.addWidget(self.label)
self.progress_bar = QProgressBar()
self.progress_bar.setMinimum(0)
self.progress_bar.setMaximum(100)
layout.addWidget(self.progress_bar)
self.status_label = QLabel("Status: Initializing...")
layout.addWidget(self.status_label)
self.debug_label = QLabel("Debug info: Waiting for camera response...")
layout.addWidget(self.debug_label)
# Add a timeout mechanism (30 seconds)
self.timeout_timer = QTimer(self)
self.timeout_timer.timeout.connect(self.check_timeout)
self.timeout_timer.start(30000) # 30 seconds timeout
# Track the last progress update time
self.last_update_time = time.time()
# Create a timer to update the waiting time display
self.update_timer = QTimer(self)
self.update_timer.timeout.connect(self.update_waiting_time)
self.update_timer.start(1000) # Update every second
def update_waiting_time(self):
"""
Update the display of how long we've been waiting
"""
elapsed = int(time.time() - self.last_update_time)
self.debug_label.setText(f"Debug info: Waiting for {elapsed} seconds since last update")
def check_timeout(self):
"""
Check if we've been waiting too long and close the dialog
"""
self.log.warning("Scheme switching timeout after 30 seconds")
self.debug_label.setText("Debug info: Timeout after 30 seconds. Continuing anyway.")
self.reject()
def update_progress(self, progress, status):
"""
Update the progress bar and status label based on the progress signal.
"""
self.log.info(f"Progress update: {progress}%, Status: {status}")
# Reset the last update time
self.last_update_time = time.time()
# Update the progress bar
self.progress_bar.setValue(progress)
# Update the status label with more detailed information
if "Initializing" in status:
self.status_label.setText(f"Status: {status}")
if "Camera busy" in status:
self.debug_label.setText(f"Debug info: Camera is busy. Waiting for it to become available...")
else:
self.debug_label.setText(f"Debug info: Camera is initializing. This may take some time.")
else:
self.status_label.setText(f"Status: {status}")
self.debug_label.setText(f"Debug info: Progress update received at {time.strftime('%H:%M:%S')}")
# Check for completion or failure
if progress >= 100 and status == "Success":
self.log.info("Scheme switching completed successfully")
self.timeout_timer.stop()
self.update_timer.stop()
self.accept()
elif status == "Failed":
self.log.error("Scheme switching failed")
self.timeout_timer.stop()
self.update_timer.stop()
self.reject()
class Test_Vision(Test_Test): class Test_Vision(Test_Test):
@ -115,8 +18,6 @@ class Test_Vision(Test_Test):
self.ok_counter_limit = 3 self.ok_counter_limit = 3
else: else:
self.ok_counter_limit = 1 self.ok_counter_limit = 1
# Get logger
self.log = logging.getLogger("Test_Vision")
# DETECTIONS TABLE # DETECTIONS TABLE
self.results_table_headers = { self.results_table_headers = {
"class": "RILEVATO", "class": "RILEVATO",
@ -145,48 +46,6 @@ class Test_Vision(Test_Test):
wait_time = 60 wait_time = 60
dialog = BlockingDialog(self.ui,wait_time=wait_time, message=f"COLLEGARE CAVO ETHERNET E ATTENDERE {wait_time} SECONDI PER IL CORRETTO AVVIO DELLE TELECAMERE") dialog = BlockingDialog(self.ui,wait_time=wait_time, message=f"COLLEGARE CAVO ETHERNET E ATTENDERE {wait_time} SECONDI PER IL CORRETTO AVVIO DELLE TELECAMERE")
dialog.exec_() dialog.exec_()
# Check if we need to switch the camera scheme
self.log.info("Checking if camera scheme needs to be switched...")
# Get the target scheme from the recipe name
target_scheme = self.step.spec.get("recipe", "").replace(".ini", "")
# Get the current scheme name from the camera
current_scheme = self.components[cam_type].get_current_scheme()
self.log.info(f"Current camera scheme: {current_scheme}, Target scheme: {target_scheme}")
# If schemes don't match or a scheme switch is already in progress, show the progress dialog
if current_scheme != target_scheme or self.components[cam_type].current_operation == "switch_scheme":
self.log.info(f"Waiting for camera to switch to scheme: {target_scheme}")
# Create and show the progress dialog
progress_dialog = SchemeProgressDialog(self.ui, target_scheme=target_scheme)
# Connect to the progress signal
progress_connection = self.components[cam_type].progress_signal.connect(
progress_dialog.update_progress
)
# If a scheme switch is not already in progress, start one
if self.components[cam_type].current_operation != "switch_scheme":
self.log.info(f"Initiating scheme switch to: {target_scheme}")
self.components[cam_type].switch_scheme(target_scheme)
# Show the dialog and wait for it to complete
result = progress_dialog.exec_()
# Disconnect from the progress signal
self.components[cam_type].progress_signal.disconnect(progress_connection)
if result == QDialog.Rejected:
# Scheme switching failed, but we'll continue anyway
self.log.warning(f"Failed to switch to scheme: {target_scheme}, continuing anyway")
else:
self.log.info(f"Successfully switched to scheme: {target_scheme}")
else:
self.log.info(f"Camera already using correct scheme: {current_scheme}")
self.components[cam_type].resume() self.components[cam_type].resume()
else: else:
return return
@ -376,42 +235,3 @@ class Test_Vision(Test_Test):
super().emit_ko() super().emit_ko()
if "neo_pixels" in self.components and self.components["neo_pixels"].ready: if "neo_pixels" in self.components and self.components["neo_pixels"].ready:
self.components["neo_pixels"].set_all_pixel_color("#000000") self.components["neo_pixels"].set_all_pixel_color("#000000")
def test_scheme_switching(self):
"""
Test method to verify that the scheme switching mechanism works correctly.
This can be called manually for testing purposes.
"""
if "vision" not in self.components or "hikrobot_sc" not in self.components:
self.log.error("Cannot test scheme switching: vision or hikrobot_sc component not available")
return False
# Get the current scheme
current_scheme = self.components["hikrobot_sc"].get_current_scheme()
self.log.info(f"Current scheme: {current_scheme}")
# Choose a test recipe different from the current scheme
test_recipes = ["test_absence", "test_presence"]
test_recipe = test_recipes[1] if current_scheme == test_recipes[0] else test_recipes[0]
self.log.info(f"Testing scheme switch to: {test_recipe}")
# Use the synchronous scheme switching method directly (no timeout)
success = self.components["hikrobot_sc"].switch_scheme_sync(test_recipe)
if success:
self.log.info(f"Successfully switched to scheme: {test_recipe}")
# Verify the scheme was switched
new_scheme = self.components["hikrobot_sc"].get_current_scheme()
self.log.info(f"Verified current scheme: {new_scheme}")
if new_scheme == test_recipe:
self.log.info("Scheme switching test passed!")
return True
else:
self.log.error(f"Scheme verification failed. Expected: {test_recipe}, Got: {new_scheme}")
return False
else:
self.log.error(f"Failed to switch to scheme: {test_recipe}")
return False