Repository navigation
Expand file tree
/
Copy pathcamera.py
More file actions
498 lines (399 loc) · 19.3 KB
/
Copy pathcamera.py
File metadata and controls
498 lines (399 loc) · 19.3 KB
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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
# \file camera.py
# \author IDS Imaging Development Systems GmbH
# \date 2024-02-20
#
# \brief This sample showcases the usage of the ids_peak API
# in setting camera parameters, starting/stopping the image acquisition
# and how to record a video using the ids_peak_ipl API.
#
# \version 1.0
#
# Copyright (C) 2024, IDS Imaging Development Systems GmbH.
#
# The information in this document is subject to change without notice
# and should not be construed as a commitment by IDS Imaging Development Systems GmbH.
# IDS Imaging Development Systems GmbH does not assume any responsibility for any errors
# that may appear in this document.
#
# This document, or source code, is provided solely as an example of how to utilize
# IDS Imaging Development Systems GmbH software libraries in a sample application.
# IDS Imaging Development Systems GmbH does not assume any responsibility
# for the use or reliability of any portion of this document.
#
# General permission to copy or modify is hereby granted.
import os
import time
import numpy as np
import cv2
import threading
#import sys
from os.path import exists
from video_recorder import VideoRecorder
from collections import deque
from ids_peak import ids_peak
from ids_peak_ipl import ids_peak_ipl
from ids_peak import ids_peak_ipl_extension
from calibration import find_homography
from PyQt5.QtCore import QTimer
TARGET_PIXEL_FORMAT = ids_peak_ipl.PixelFormatName_BGRa8
os.environ["LD_PRELOAD"] = os.environ.get("LD_PRELOAD", "") + ":/lib/aarch64-linux-gnu/libGLdispatch.so.0"
os.environ["QT_XCB_GL_INTEGRATION"] = "none"
class Camera:
def __init__(self, device_manager, interface):
if interface is None:
raise ValueError("Interface is None")
self.device_manager = device_manager
self._interface = interface
self._device = None
self._datastream = None
self.acquisition_mode = 0 #0: Real Time, #1: HW Trigger, #2: SW Trigger
self.acquisition_running = False
self.node_map = None
self._buffer_list = []
self.target_gain = 1
self.max_gain = 1
self.target_dgain = 1
self.killed = False
self.save_image = False
self.calibrate = 0
self.frame_times = deque(maxlen=120) # ✅ Store timestamps of the last 120 frames
self.translation_matrix = np.eye(3)
self.asset_dir = "./Assets"
self.save_dir = "./Saved_Media"
os.makedirs(self.asset_dir, exist_ok=True)
os.makedirs(self.save_dir, exist_ok=True)
# Default the hardware trigger line
self.hardware_trigger_line = "Line0"
self._get_device()
self._setup_device_and_datastream()
self._interface.set_camera(self)
self._image_converter = ids_peak_ipl.ImageConverter()
self.video_recorder = VideoRecorder(interface) # ✅ Initialize video recorder
def __del__(self):
self.close()
def _get_device(self):
# Update device manager to refresh the camera list
self.device_manager.Update()
if self.device_manager.Devices().empty():
print("No device found. Exiting Program.")
sys.exit(1)
selected_device = None
# Initialize first device found if only one is available
if len(self.device_manager.Devices()) == 1:
selected_device = 0
else:
# List all available devices
for i, device in enumerate(self.device_manager.Devices()):
print(
f"{str(i)}: {device.ModelName()} ("
f"{device.ParentInterface().DisplayName()} ; "
f"{device.ParentInterface().ParentSystem().DisplayName()} v."
f"{device.ParentInterface().ParentSystem().version()})")
while True:
try:
# Let the user decide which device to open
selected_device = int(input("Select device to open: "))
if selected_device < len(self.device_manager.Devices()):
break
else:
print("Invalid ID.")
except ValueError:
print("Please enter a correct id.")
continue
# Opens the selected device in control mode
self._device = self.device_manager.Devices()[selected_device].OpenDevice(
ids_peak.DeviceAccessType_Control)
self.node_map = self._device.RemoteDevice().NodeMaps()[0]
self.node_map.FindNode("GainSelector").SetCurrentEntry("AnalogAll")
self.max_gain = self.node_map.FindNode("Gain").Maximum()
# Load the default settings
self.node_map.FindNode("UserSetSelector").SetCurrentEntry("Default")
self.node_map.FindNode("UserSetLoad").Execute()
self.node_map.FindNode("UserSetLoad").WaitUntilDone()
def get_actual_fps(self):
"""Calculate FPS as the average over the last 2 seconds."""
current_time = time.time()
self.frame_times.append(current_time) # ✅ Store current timestamp
# ✅ Remove timestamps older than 2 seconds
while self.frame_times and self.frame_times[0] < current_time - 2:
self.frame_times.popleft()
# ✅ Calculate FPS based on the number of frames in the last 2 seconds
if len(self.frame_times) > 1:
self.GUIfps = round(len(self.frame_times) / 2) # Average over 2 seconds
else:
self.GUIfps = 0 # No frames captured in the last 2 seconds
return self.GUIfps
def _init_data_stream(self):
# Open device's datastream
self._datastream = self._device.DataStreams()[0].OpenDataStream()
# Allocate image buffer for image acquisition
self.revoke_and_allocate_buffer()
def conversion_supported(self, source_pixel_format: int) -> bool:
"""
Check if the image_converter supports the conversion of the
`source_pixel_format` to our `TARGET_PIXEL_FORMAT`
"""
return any(
TARGET_PIXEL_FORMAT == supported_pixel_format
for supported_pixel_format in
self._image_converter.SupportedOutputPixelFormatNames(
source_pixel_format))
def _setup_device_and_datastream(self):
self._datastream = self._device.DataStreams()[0].OpenDataStream()
# Disable auto gain and auto exposure to enable custom gain in program
self._find_and_set_remote_device_enumeration("GainAuto", "Off")
self._find_and_set_remote_device_enumeration("ExposureAuto", "Off")
# Set camera frame rate to 60 FPS
try:
self.node_map.FindNode("AcquisitionFrameRate").SetValue(60)
print("Acquisition frame rate set to 60 FPS")
except Exception as e:
print(f"Failed to set AcquisitionFrameRate: {e}")
# Allocate image buffer for image acquisition
payload_size = self.node_map.FindNode("PayloadSize").Value()
# Use more buffers
max_buffer = self._datastream.NumBuffersAnnouncedMinRequired() * 5
for idx in range(max_buffer):
buffer = self._datastream.AllocAndAnnounceBuffer(payload_size)
self._datastream.QueueBuffer(buffer)
print("Allocated buffers, finished opening device")
def close(self):
self.stop_recording()
self.stop_realtime_acquisition()
self.stop_hardware_acquisition()
# If datastream has been opened, revoke and deallocate all buffers
if self._datastream is not None:
try:
for buffer in self._datastream.AnnouncedBuffers():
self._datastream.RevokeBuffer(buffer)
except Exception as e:
print(f"Exception (close): {str(e)}")
def _find_and_set_remote_device_enumeration(self, name: str, value: str):
all_entries = self.node_map.FindNode(name).Entries()
available_entries = []
for entry in all_entries:
if (entry.AccessStatus() != ids_peak.NodeAccessStatus_NotAvailable
and entry.AccessStatus() != ids_peak.NodeAccessStatus_NotImplemented):
available_entries.append(entry.SymbolicValue())
if value in available_entries:
self.node_map.FindNode(name).SetCurrentEntry(value)
def set_remote_device_value(self, name: str, value: any):
try:
self.node_map.FindNode(name).SetValue(value)
except ids_peak.Exception:
self.interface.warning(f"Could not set value for {name}!")
#RealTime Acquisition
def start_realtime_acquisition(self):
self.acquisition_mode = 0 #0: Real Time, #1: HW Trigger, #2: SW Trigger
if self._device is None or self.acquisition_running:
return False
if self._datastream is None:
self._init_data_stream()
for buffer in self._buffer_list:
self._datastream.QueueBuffer(buffer)
# Constant Acquisition Test
try:
allEntries = self.node_map.FindNode("TriggerSelector").Entries()
availableEntries = []
for entry in allEntries:
if (entry.AccessStatus() != ids_peak.NodeAccessStatus_NotAvailable
and entry.AccessStatus() != ids_peak.NodeAccessStatus_NotImplemented):
availableEntries.append(entry.SymbolicValue())
if len(availableEntries) == 0:
raise Exception("RT Acquisition not supported")
elif "ExposureStart" not in availableEntries:
self.node_map.FindNode("TriggerSelector").SetCurrentEntry(
availableEntries[0])
else:
self.node_map.FindNode(
"TriggerSelector").SetCurrentEntry("ExposureStart")
# Set trigger mode to 'Off' for continuous acquisition
self.node_map.FindNode("TriggerMode").SetCurrentEntry("Off")
# Start the data stream and acquisition
self._datastream.StartAcquisition()
self.node_map.FindNode("AcquisitionStart").Execute()
self.acquisition_running = True
except Exception as e:
print(f"Exception during start_realtime_acquisition: {e}")
return False
return True
def stop_realtime_acquisition(self):
if self._device is None or self.acquisition_running is False:
return
try:
self.node_map.FindNode("AcquisitionStop").Execute()
# Kill the datastream to exit out of pending `WaitForFinishedBuffer`
# calls
self._datastream.KillWait()
self._datastream.StopAcquisition(ids_peak.AcquisitionStopMode_Default)
# Discard all buffers from the acquisition engine
# They remain in the announced buffer pool
self._datastream.Flush(ids_peak.DataStreamFlushMode_DiscardAll)
self.acquisition_running = False
# Unlock parameters
self.node_map.FindNode("TLParamsLocked").SetValue(0)
self.revoke_and_allocate_buffer()
print("Closed RT Acq")
except Exception as e:
print(f"Exception (stop acquisition): {str(e)}")
# Hardware Acquisition
def start_hardware_acquisition(self):
self.acquisition_mode = 1 #0: Real Time, #1: HW Trigger, #2: SW Trigger
if self._device is None or self.acquisition_running:
return False
if self._datastream is None:
self._init_data_stream()
for buffer in self._buffer_list:
self._datastream.QueueBuffer(buffer)
# HW Acquisition Test
try:
allEntries = self.node_map.FindNode("TriggerSelector").Entries()
availableEntries = []
for entry in allEntries:
if (entry.AccessStatus() != ids_peak.NodeAccessStatus_NotAvailable
and entry.AccessStatus() != ids_peak.NodeAccessStatus_NotImplemented):
availableEntries.append(entry.SymbolicValue())
if len(availableEntries) == 0:
raise Exception("Hardware Trigger not supported")
elif "ExposureStart" not in availableEntries:
self.node_map.FindNode("TriggerSelector").SetCurrentEntry(
availableEntries[0])
else:
self.node_map.FindNode(
"TriggerSelector").SetCurrentEntry("ExposureStart")
self.node_map.FindNode("TriggerMode").SetCurrentEntry("On")
# Use the hardware_trigger_line variable
self.node_map.FindNode("TriggerSource").SetCurrentEntry(self.hardware_trigger_line)
# Start the data stream and acquisition
self._datastream.StartAcquisition()
self.node_map.FindNode("AcquisitionStart").Execute()
self.acquisition_running = True
# Log after selecting new trigger line
print("Hardware Acquisition started!")
print("Trigger Mode:", self.node_map.FindNode("TriggerMode").CurrentEntry().SymbolicValue())
print("Trigger Source:", self.node_map.FindNode("TriggerSource").CurrentEntry().SymbolicValue())
except Exception as e:
print(f"Exception during start_hardware_acquisition: {e}")
return False
return True
def stop_hardware_acquisition(self):
if self._device is None or self.acquisition_running is False:
return
try:
self.node_map.FindNode("AcquisitionStop").Execute()
# Kill the datastream to exit out of pending `WaitForFinishedBuffer`
# calls
self._datastream.KillWait()
self._datastream.StopAcquisition(ids_peak.AcquisitionStopMode_Default)
# Discard all buffers from the acquisition engine
# They remain in the announced buffer pool
self._datastream.Flush(ids_peak.DataStreamFlushMode_DiscardAll)
self.acquisition_running = False
# Unlock parameters
self.node_map.FindNode("TLParamsLocked").SetValue(0)
self.revoke_and_allocate_buffer()
print("Closed HW Acq")
except Exception as e:
print(f"Exception (stop hardware acquisition): {str(e)}")
def start_recording(self):
if self._datastream is None:
self._init_data_stream()
fps = int(self.node_map.FindNode("AcquisitionFrameRate").Value()) if self.acquisition_mode == 0 else self.GUIfps
self.video_recorder.start_recording(fps)
def stop_recording(self):
self.video_recorder.stop_recording()
def _valid_name(self, path: str, ext: str):
num = 0
def build_string():
return f"{path}_{num}{ext}"
while exists(build_string()):
num += 1
return build_string()
def start_calibration(self):
def delayed_capture():
# ✅ Step 4: Capture Image After Projection
save_path = os.path.join(self.asset_dir, "calibration_capture_image.png")
# We need to wait for a new image to be captured
print("capturing img")
latest_image = None
while latest_image is None:
latest_image = self.get_data_stream_image() # ✅ Get a new frame
# Save the captured image
ids_peak_ipl.ImageWriter.WriteAsPNG(save_path, latest_image)
#Compute Homography
threading.Thread(target=compute_homography, daemon=True).start()
def compute_homography():
try:
homography_matrix = find_homography()
self.translation_matrix = homography_matrix
print("✅ Homography Computed Successfully!")
self._interface.on_projection_received(np.array(cv2.imread("./Assets/custom_registration_image.png")), self.translation_matrix)
except Exception as e:
print(f"❌ Error calculating homography: {e}")
"""Handles the entire calibration process separately from image acquisition."""
print("Starting Calibration...")
# ✅ Step 1: Display Calibration Pattern
self._interface.on_projection_received(np.array(cv2.imread("./Assets/custom_registration_image.png")))
# ✅ Step 2: Wait for the Projection to Fully Appear
QTimer.singleShot(80, delayed_capture)
def revoke_and_allocate_buffer(self):
if self._datastream is None:
return
try:
# Check if buffers are already allocated
if self._datastream is not None:
# Remove buffers from the announced pool
for buffer in self._datastream.AnnouncedBuffers():
self._datastream.RevokeBuffer(buffer)
self._buffer_list = []
payload_size = self.node_map.FindNode("PayloadSize").Value()
buffer_amount = self._datastream.NumBuffersAnnouncedMinRequired()
for _ in range(buffer_amount):
buffer = self._datastream.AllocAndAnnounceBuffer(payload_size)
self._buffer_list.append(buffer)
except Exception as e:
self._interface.warning(str(e))
def change_pixel_format(self, pixel_format: str):
try:
self.node_map.FindNode("PixelFormat").SetCurrentEntry(pixel_format)
self.revoke_and_allocate_buffer()
except Exception as e:
self._interface.warning(f"Cannot change pixelformat: {str(e)}")
def get_data_stream_image(self):
try:
cwd = os.getcwd()
converted_ipl_image = None
buffer = self._datastream.WaitForFinishedBuffer(500) # Short timeout
if buffer is None:
return None # No buffer available, skip processing
ipl_image = ids_peak_ipl_extension.BufferToImage(buffer)
converted_ipl_image = self._image_converter.Convert(ipl_image, TARGET_PIXEL_FORMAT)
self._datastream.QueueBuffer(buffer)
self._interface.on_image_received(converted_ipl_image)
self.video_recorder.add_frame(converted_ipl_image) # ✅ Use new VideoRecorder
if self.save_image:
save_path = self._valid_name(os.path.join(self.save_dir, "image"), ".png")
ids_peak_ipl.ImageWriter.WriteAsPNG(save_path, converted_ipl_image)
print(f"Image Saved at {save_path}")
self.save_image = False
return converted_ipl_image
except ids_peak.Exception as e:
print(f"No buffer available: {e}")
return None
def acquisition_thread(self):
while not self.killed:
try:
self.get_data_stream_image()
except Exception as e:
self._interface.warning(f"Acquisition error: {str(e)}")
self.save_image = False
def change_hardware_trigger_line(self, new_line: str):
# Change the hardware trigger line
self.hardware_trigger_line = new_line
print(f"Hardware trigger line set to: {new_line}")
# Reinitialize the hardware acquisition
if self.acquisition_running and self.acquisition_mode == 1:
self.stop_hardware_acquisition()
QTimer.singleShot(500, self.start_hardware_acquisition)
# self.start_hardware_acquisition()
return new_line