PupilCalib/IDSManager.py at main · AYESDIE/PupilCalib · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
import cv2
import warnings
from CameraManager import CameraManager

from ids_peak import ids_peak
from ids_peak_ipl import ids_peak_ipl
from ids_peak import ids_peak_ipl_extension

import numpy


class IDSManager(CameraManager):
    def __init__(self):
        super().__init__()
        self.s_manager_name = "IDSManager"
        ids_peak.Library.Initialize()

        # Initialize and find device
        self.device_manager = ids_peak.DeviceManager.Instance()

        self.device_manager.Update()

        if self.device_manager.Devices().empty():
            warnings.warn("No devices found, the program will not run", ResourceWarning)

        self.device = self.device_manager.Devices()[0].OpenDevice(ids_peak.DeviceAccessType_Control)
        self.node_map_remote_device = self.device.RemoteDevice().NodeMaps()[0]

        # Start data stream
        self.data_stream = self.device.DataStreams()
        if self.device.DataStreams().empty():
            warnings.warn("Data stream is not available", ResourceWarning)

        self.data_stream = self.device.DataStreams()[0].OpenDataStream()

        # TODO: Set ROI

        # Clear buffers
        self.data_stream.Flush(ids_peak.DataStreamFlushMode_DiscardAll)
        for buffer in self.data_stream.AnnouncedBuffers():
            self.data_stream.RevokeBuffer(buffer)

        payload_size = self.node_map_remote_device.FindNode("PayloadSize").Value()
        num_buffers_min_required = self.data_stream.NumBuffersAnnouncedMinRequired()

        for count in range(num_buffers_min_required):
            buffer = self.data_stream.AllocAndAnnounceBuffer(payload_size)
            self.data_stream.QueueBuffer(buffer)

        # Start Acquisition

        self.data_stream.StartAcquisition(ids_peak.AcquisitionStartMode_Default, ids_peak.DataStream.INFINITE_NUMBER)
        self.node_map_remote_device.FindNode("TLParamsLocked").SetValue(1)
        self.node_map_remote_device.FindNode("AcquisitionStart").Execute()

        if self.loadCameraCalibration():
            self.setCalibration(True)

    def captureCurrentFrame(self):
        buffer = self.data_stream.WaitForFinishedBuffer(5000)

        ipl_image = ids_peak_ipl_extension.BufferToImage(buffer)
        converted_ipl_image = ipl_image.ConvertTo(ids_peak_ipl.PixelFormatName_BGRa8)

        # Queue buffer so that it can be used again
        self.data_stream.QueueBuffer(buffer)

        # Get raw image data from converted image and construct a QImage from it
        self.m_current_frame = converted_ipl_image.get_numpy_1D()
        self.m_current_frame = cv2.cvtColor(numpy.reshape(self.m_current_frame, [converted_ipl_image.Height(), converted_ipl_image.Width(), -1]), cv2.COLOR_BGR2GRAY)
        self.m_current_frame = cv2.resize(self.m_current_frame, (900, 600), cv2.INTER_AREA)

        self.detectAndShowAprilTag()
        self.applyCalibrationMatrix()