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

smart camera fixture connect and remove proper management

Browse Source
This commit is contained in:
gg 2025-01-27 14:37:48 +01:00
parent 57e5a7797e
commit a88d7f5cd8
8 changed files with 75 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/components/archive_synchronizer.py
View File

@ -91,7 +91,8 @@ class ArchiveSynchronizer(Component):
else:
pass
# self.log.info(f"({self.name}) id {record.id}: failed to archive remotely")
self.main_window.run_request.emit(record.save, [], {})
if self.main_window is not None:
self.main_window.run_request.emit(record.save, [], {})
if self.hold_time > 0:
QThread.msleep(self.hold_time)

8
src/components/component.py
View File

@ -174,9 +174,9 @@ class Component(QObject):
self._lock.release()
return running
def wait_completion(self, timeout=50):
def wait_completion(self, timeout=5):
"""
waits untill the requested action has been completed by the component
waits until the requested action has been completed by the component
this will return immediately if threaded=False was passed at component initialization
"""
if self._threaded:
@ -197,7 +197,7 @@ class Component(QObject):
else:
self._do_reconfigure()
def pause(self):
def pause(self,timeout=5):
"""will pause periodic calls to _get and sources or requestors triggers"""
if self._threaded:
self._lock.acquire(max(self._lock.available(), 1))
@ -207,7 +207,7 @@ class Component(QObject):
return
if self._threaded:
self._pause.emit()
self.wait_completion()
self.wait_completion(timeout=timeout)
else:
self._do_pause()

2
src/components/hikrobot_sc/hikrobot_dll.py
View File

@ -226,7 +226,7 @@ def MV_VS_GetFrame(handle, name=""):
stFrameData = MV_VS_DATA()
nRet = mv_lib.MV_VS_GetResultData(handle, byref(stFrameData), 2000)
nRet = mv_lib.MV_VS_GetResultData(handle, byref(stFrameData), 500)
if nRet == MV_VS_OK:
# Print frame information
# print(f"Frame Width: [{stFrameData.nImageWidth}], Height: [{stFrameData.nImageHeight}], ImageLen: [{stFrameData.nImageLen}]")

80
src/components/hikrobot_sc/hikrobot_sc.py
View File

@ -1,6 +1,7 @@
import copy
import ctypes
import struct
import time
from ctypes import *
import sys
@ -26,17 +27,24 @@ class HikrobotSmartCamera(Component):
nRet = mv_lib.MV_VS_EnumDevices(ctypes.byref(device_info_list))
if nRet != MV_VS_OK:
print(f"Enum devices failed! nRet [{nRet:#x}]")
QMessageBox.critical(None, "Errore", f"ERRORE CONNESSIONE TELECAMERE\n")
return -2
else:
print("Enum devices OK")
enum_ok = False
for n in range(3):
if nRet != MV_VS_OK:
print(f"Enum devices failed! nRet [{nRet:#x}]")
return -2
else:
print("Enum devices OK")
if device_info_list.nDeviceNum > 0:
print_device_info(device_info_list)
else:
print("Find No Devices!")
if device_info_list.nDeviceNum > 0:
print_device_info(device_info_list)
if device_info_list.nDeviceNum >= self.num_cameras:
enum_ok = True
break
else:
print("wrong number of devices")
else:
print("Find No Devices!")
if not enum_ok:
return -2
self.num_cameras = device_info_list.nDeviceNum
@ -47,28 +55,36 @@ class HikrobotSmartCamera(Component):
pHandle = c_void_p()
nRet = mv_lib.MV_VS_CreateHandle(byref(pHandle), MV_VS_DEVICE_INFO)
if nRet != MV_VS_OK:
self.log.error(f"Create Handle failed! [{nRet&0xFFFFFFFF:#x}]")
self.log.error(f"CAM{cam_idx} Create Handle failed! [{nRet&0xFFFFFFFF:#x}]")
return -2
self.cam_list[cam_idx] = {"handle": pHandle}
time.sleep(0.1)
nRet = mv_lib.MV_VS_Login(pHandle, b"Admin", b"Abc1234")
if nRet != MV_VS_OK:
self.log.error(f"Login failed! [{nRet&0xFFFFFFFF:#x}]")
login_ok = False
for n in range(3):
nRet = mv_lib.MV_VS_Login(pHandle, b"Admin", b"Abc1234")
if nRet != MV_VS_OK:
self.log.error(f"CAM{cam_idx} Login failed! [{nRet&0xFFFFFFFF:#x}]")
time.sleep(1)
else:
login_ok = True
self.log.info(f"CAM{cam_idx} Login OK")
break
if not login_ok:
return -2
time.sleep(0.2)
nRet = mv_lib.MV_VS_SetBoolValue(pHandle, b"CommandImageMode", 0)
if nRet != MV_VS_OK:
self.log.error(f"CommandImageMode failed! [{nRet&0xFFFFFFFF:#x}]")
self.log.error(f"CAM{cam_idx} CommandImageMode failed! [{nRet&0xFFFFFFFF:#x}]")
return -2
nRet = mv_lib.MV_VS_SetEnumValue(pHandle, b"AcquisitionMode", 0)
if nRet != MV_VS_OK:
self.log.error(f"AcquisitionMode failed! [{nRet&0xFFFFFFFF:#x}]")
self.log.error(f"CAM{cam_idx} AcquisitionMode failed! [{nRet&0xFFFFFFFF:#x}]")
return -2
# Start the camera test
nRet = mv_lib.MV_VS_StartRun(pHandle)
if nRet != MV_VS_OK:
self.log.error(f"Start run failed! [{nRet & 0xFFFFFFFF:#x}]")
self.log.error(f"CAM{cam_idx} Start run failed! [{nRet & 0xFFFFFFFF:#x}]")
self.connected = True
@Component.reconfig_on_error
@ -87,7 +103,7 @@ class HikrobotSmartCamera(Component):
# Trigger acquisition & test execution
nRet = mv_lib.MV_VS_SetCommandValue(cam["handle"], b"AcquisitionStart")
if nRet != MV_VS_OK:
self.log.error(f"AcquisitionStart failed! [{nRet&0xFFFFFFFF:#x}]")
self.log.error(f"CAM{cam_idx} AcquisitionStart failed! [{nRet&0xFFFFFFFF:#x}]")
self.connected = False
# get frame
frame_res = MV_VS_GetFrame(cam["handle"])
@ -112,22 +128,16 @@ class HikrobotSmartCamera(Component):
super()._get([concat_frame, concat_results])
def resume(self):
self.log.info(f"RESUMING")
if not self.connected:
self.config_changed()
if len(self.cam_list) == self.num_cameras:
# Dummy acquisition for each camera to avoid reading past images
for cam_idx in range(self.num_cameras):
cam = self.cam_list[cam_idx]
nRet = mv_lib.MV_VS_SetCommandValue(cam["handle"], b"AcquisitionStart")
frame_res = MV_VS_GetFrame(cam["handle"])
super().resume()
def pause(self):
if self.connected:
for cam_idx in range(self.num_cameras):
cam = self.cam_list[cam_idx]
nRet = mv_lib.MV_VS_SetCommandValue(cam["handle"], b"AcquisitionStop")
frame_res = MV_VS_GetFrame(cam["handle"])
self.log.info(f"PAUSED")
super().pause()
if len(self.cam_list) == self.num_cameras:
# Dummy acquisition for each camera to avoid reading past images
for cam_idx in range(self.num_cameras):
cam = self.cam_list[cam_idx]
nRet = mv_lib.MV_VS_SetCommandValue(cam["handle"], b"AcquisitionStart")
frame_res = MV_VS_GetFrame(cam["handle"])
else:
QMessageBox.critical(None, "Errore", f"ERRORE CONNESSIONE TELECAMERE\n")
super().resume()

2
src/ui/barcode_recipe_selection/barcode_recipe_selection.py
View File

@ -91,7 +91,6 @@ class Barcode_Recipe_Selection(Test_Test):
self.barcode_input_l.setPalette(self.status_palettes[False])
self.focus_timer.start(3000)
def set_recipe(self):
try:
# LOOKUP RECIPE
@ -102,7 +101,6 @@ class Barcode_Recipe_Selection(Test_Test):
self.focus_timer.start(3000)
self.parent_assembly_widget().set_text("RICETTA NON TROVATA",bg_color="red")
def set_focus(self):
self.barcode_input_l.setFocus()
self.barcode_input_l.setPlainText("")

12
src/ui/main_window/main_window.py
View File

@ -1,7 +1,12 @@
import argparse
import sip
from PyQt5.QtCore import pyqtSignal
from ui.window import Window
parser = argparse.ArgumentParser(prog='STTEN', description='Leak test system',add_help=False)
parser.add_argument('-w', '--width')
parser.add_argument('-h', '--height')
args,unspec = parser.parse_known_args()
class Main_Window(Window):
do = pyqtSignal(dict)
@ -17,6 +22,13 @@ class Main_Window(Window):
self.run_request.connect(self.run_request_handler)
# print("MAIN_WINDOW ", str(int(QThread.currentThreadId())), flush=True)
if "width" in args:
if args.width is not None:
self.setFixedWidth(int(args.width))
if "height" in args:
if args.height is not None:
self.setFixedHeight(int(args.height))
@staticmethod
def _do(config):
return config["f"](*config.get("a", []), **config.get("k", {}))

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

@ -4,7 +4,7 @@ import os
import sys
import weakref
from datetime import datetime, timedelta
from PyQt5.QtCore import QTimer, pyqtSlot
from PyQt5.QtCore import QTimer, pyqtSlot, pyqtSignal
from PyQt5.QtWidgets import QMessageBox
from lib.db import Archive, Recipes, Users
from lib.helpers import get_shift
@ -32,6 +32,7 @@ from src.components.rfid_pn532 import RFID_PN532
class Test(Widget):
def __init__(self, config, components=None, main_window=None):
super().__init__()
self.autotest_timer = None
@ -269,7 +270,9 @@ class Test(Widget):
self.log.info(f"cycle step: {self.step!r} action: {action!r} current index:{self.cycle_index}")
current_w = self.centralWidget
if hasattr(current_w, "stop"):
self.log.info(f"stopping widget {self.step.step_type}")
current_w.stop()
if action == "change_recipe":
self.log.info(f"cycle next: action: {action!r}")
@ -819,7 +822,7 @@ class Test(Widget):
@pyqtSlot(bool)
def handle_rfid_error(self, connected):
if connected:
self.error_label.setText("") # Update the label from Designer
self.error_label.setText("") # Update the GUI label
else:
self.error_label.setText("Errore RFID.") # Update the label from Designer
self.error_label.setStyleSheet("color: red;")
self.error_label.setText("Errore RFID.") # Update the GUI label
self.error_label.setStyleSheet("color: red;")

7
src/ui/test_vision/test_vision.py
View File

@ -43,7 +43,7 @@ class Test_Vision(Test_Test):
if cam_type in ("uvc_camera","galaxy_camera","hikrobot_sc"):
if cam_type == "hikrobot_sc":
if not self.components[cam_type].connected:
wait_time = 50
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.exec_()
self.components[cam_type].resume()
@ -59,11 +59,12 @@ class Test_Vision(Test_Test):
def stop(self):
# disable camera-vision loop
cam_type = self.components["vision"].vision_config["camera_type"]
self.components["vision"].pause()
if cam_type in ("uvc_camera", "galaxy_camera", "hikrobot_sc"):
self.components[cam_type].pause()
self.components[cam_type].pause(timeout=60)
else:
return
self.components["vision"].pause()
self.disconnect(self.get_connection)
self.disconnect(self.request_frame_connection)
super().stop()