161 lines
6.2 KiB
Python
161 lines
6.2 KiB
Python
import copy
|
|
import ctypes
|
|
import struct
|
|
import time
|
|
from ctypes import *
|
|
import sys
|
|
|
|
from PyQt5.QtCore import pyqtSignal, QMutex
|
|
from PyQt5.QtWidgets import QMessageBox
|
|
|
|
from components.component import Component
|
|
|
|
from .hikrobot_dll import *
|
|
|
|
|
|
class HikrobotSmartCamera(Component):
|
|
def __init__(self, config=None, name=None, period=0.5, lazy=True, paused=False, threaded=True):
|
|
super().__init__(config=config, name=name, period=period, lazy=lazy, paused=paused, threaded=threaded)
|
|
self.simulate = "--sim-camera" in sys.argv
|
|
self.num_cameras=1
|
|
self.connected = False
|
|
self.ok_memory = True
|
|
self.ok_frames = []
|
|
self.ok_results =[]
|
|
|
|
def config_changed(self):
|
|
self.connected = False
|
|
global MV_VS_OK
|
|
device_info_list = MV_VS_DEVICE_INFO_LIST()
|
|
|
|
nRet = mv_lib.MV_VS_EnumDevices(ctypes.byref(device_info_list))
|
|
|
|
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)
|
|
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
|
|
self.cam_list = {}
|
|
for cam_idx in range(self.num_cameras):
|
|
self.log.info(f"INIT CAMERA # {cam_idx}")
|
|
MV_VS_DEVICE_INFO = device_info_list.pDeviceInfo[cam_idx]
|
|
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"CAM{cam_idx} Create Handle failed! [{nRet&0xFFFFFFFF:#x}]")
|
|
return -2
|
|
self.cam_list[cam_idx] = {"handle": pHandle}
|
|
time.sleep(0.1)
|
|
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"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"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"CAM{cam_idx} Start run failed! [{nRet & 0xFFFFFFFF:#x}]")
|
|
self.connected = True
|
|
|
|
@Component.reconfig_on_error
|
|
def _get(self):
|
|
if not self.connected:
|
|
self.config_changed()
|
|
if not self.connected:
|
|
return
|
|
concat_frame = None
|
|
concat_results = []
|
|
rot = self.config.vision_config["rotations"].split(",")
|
|
for cam_idx in range(self.num_cameras):
|
|
cam = self.cam_list[cam_idx]
|
|
self.log.info(f"GET FRAME CAMERA # {cam_idx}")
|
|
if self.ok_results[cam_idx]:
|
|
# keep last good result
|
|
frame = self.ok_frames[cam_idx]
|
|
res = False
|
|
else:
|
|
# Trigger acquisition & test execution
|
|
nRet = mv_lib.MV_VS_SetCommandValue(cam["handle"], b"AcquisitionStart")
|
|
if nRet != MV_VS_OK:
|
|
self.log.error(f"CAM{cam_idx} AcquisitionStart failed! [{nRet&0xFFFFFFFF:#x}]")
|
|
self.connected = False
|
|
frame_res = None
|
|
else:
|
|
# get frame
|
|
frame_res = MV_VS_GetFrame(cam["handle"])
|
|
|
|
if frame_res is not None:
|
|
res = frame_res["res"]["ScDeviceSolutionRunningResult"]
|
|
frame = frame_res["frame"]
|
|
# rotate frame
|
|
if rot[cam_idx] == "1":
|
|
frame = cv2.rotate(frame, cv2.ROTATE_90_CLOCKWISE)
|
|
if rot[cam_idx] == "2":
|
|
frame = cv2.rotate(frame, cv2.ROTATE_90_COUNTERCLOCKWISE)
|
|
if rot[cam_idx] == "3":
|
|
frame = cv2.rotate(frame, cv2.ROTATE_180)
|
|
else:
|
|
self.connected = False
|
|
if concat_frame is None:
|
|
concat_frame = copy.deepcopy(frame)
|
|
else:
|
|
concat_frame = cv2.hconcat([concat_frame, frame])
|
|
concat_results.append(not res)
|
|
if not res:
|
|
self.ok_results[cam_idx] = True
|
|
self.ok_frames[cam_idx] = copy.deepcopy(frame)
|
|
|
|
super()._get([concat_frame, self.ok_results])
|
|
|
|
def resume(self):
|
|
self.log.info(f"RESUMING")
|
|
if not self.connected:
|
|
self.config_changed()
|
|
if self.connected:
|
|
# init OK result memory
|
|
self.ok_frames = [None for n in range(self.num_cameras)]
|
|
self.ok_results = [False for n in range(self.num_cameras)]
|
|
|
|
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()
|