-
-
Notifications
You must be signed in to change notification settings - Fork 32
Expand file tree
/
Copy pathDualCamera_separate_transform.py
More file actions
1766 lines (1454 loc) · 72.8 KB
/
Copy pathDualCamera_separate_transform.py
File metadata and controls
1766 lines (1454 loc) · 72.8 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
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import tkinter as tk
from tkinter import ttk, messagebox, filedialog
import threading
import time
import queue
import os
import sys
import platform
from datetime import datetime
# Add the camera SDK paths (modify these paths according to your setup)
BASE = os.path.dirname(os.path.abspath(__file__))
sys.path.append(os.path.join(BASE, "haikang_sdk", "Python"))
sys.path.append(os.path.join(BASE, "haikang_sdk", "Python", "MvImport"))
# Import frame camera modules
try:
from ctypes import *
import numpy as np
import cv2
from MvCameraControl_class import (
MvCamera, MV_CC_DEVICE_INFO_LIST, MV_CC_DEVICE_INFO,
MV_GIGE_DEVICE, MV_USB_DEVICE, MV_GENTL_CAMERALINK_DEVICE,
MV_GENTL_CXP_DEVICE, MV_GENTL_XOF_DEVICE, MV_FRAME_OUT,
MV_CC_RECORD_PARAM, MV_CC_INPUT_FRAME_INFO, MV_CC_PIXEL_CONVERT_PARAM,
MVCC_INTVALUE, MVCC_ENUMVALUE, MVCC_FLOATVALUE, MvGvspPixelType
)
from CameraParams_const import MV_ACCESS_Exclusive
from CameraParams_header import MV_TRIGGER_MODE_OFF, MV_FormatType_AVI
from PixelType_header import PixelType_Gvsp_Mono8, PixelType_Gvsp_RGB8_Packed
FRAME_CAMERA_AVAILABLE = True
except ImportError as e:
print(f"Frame camera SDK not available: {e}")
FRAME_CAMERA_AVAILABLE = False
# Import event camera modules
try:
from metavision_core.event_io.raw_reader import initiate_device
from metavision_core.event_io import EventsIterator
from metavision_sdk_core import PeriodicFrameGenerationAlgorithm, ColorPalette
from metavision_sdk_ui import EventLoop, BaseWindow, MTWindow, UIAction, UIKeyEvent
EVENT_CAMERA_AVAILABLE = True
except ImportError as e:
print(f"Event camera SDK not available: {e}")
EVENT_CAMERA_AVAILABLE = False
# Platform-specific imports for always on top functionality
if platform.system() == "Windows":
try:
import win32gui
import win32con
WINDOWS_AVAILABLE = True
except ImportError:
WINDOWS_AVAILABLE = False
print("win32gui not available. Install pywin32 for full always-on-top support")
else:
WINDOWS_AVAILABLE = False
def set_window_always_on_top(window_title, always_on_top=True):
"""Set a window to always on top (Windows only)"""
if not WINDOWS_AVAILABLE:
return False
try:
hwnd = win32gui.FindWindow(None, window_title)
if hwnd:
if always_on_top:
win32gui.SetWindowPos(hwnd, win32con.HWND_TOPMOST, 0, 0, 0, 0,
win32con.SWP_NOMOVE | win32con.SWP_NOSIZE)
else:
win32gui.SetWindowPos(hwnd, win32con.HWND_NOTOPMOST, 0, 0, 0, 0,
win32con.SWP_NOMOVE | win32con.SWP_NOSIZE)
return True
except Exception as e:
print(f"Error setting window always on top: {e}")
return False
class FrameCameraController:
"""Modified version of CameraRecorder for GUI integration"""
def __init__(self, status_callback=None, screen_info=None):
self.cam = None
self.device_info = None
self.recording_thread = None
self.preview_thread = None
self.command_queue = queue.Queue()
self.frame_queue = queue.Queue(maxsize=5)
self.is_recording = False
self.is_grabbing = False
self.show_preview = False
self.exit_flag = False
self.current_filename = None
self.record_params = None
self.stats_lock = threading.Lock()
self.status_callback = status_callback
self.screen_info = screen_info or {}
# Always on top state - DEFAULT TO TRUE
self.preview_always_on_top = True
# Frame transformation settings - DEFAULT VERTICAL FLIP ON
self.vertical_flip = True
self.horizontal_flip = False
self.rotation_angle = 0 # 0, 90, 180, 270
self.use_opencv_recording = False # Use OpenCV when transformations are applied
self.opencv_writer = None
self.stats = {
'frames_captured': 0,
'frames_displayed': 0,
'recording_start_time': None,
'last_fps': 0.0
}
# For pixel format conversion
self.convert_param = MV_CC_PIXEL_CONVERT_PARAM()
self.rgb_buffer = None
self.rgb_buffer_size = 0
def find_cameras(self):
"""Find available frame cameras"""
if not FRAME_CAMERA_AVAILABLE:
return []
try:
# Initialize SDK
MvCamera.MV_CC_Initialize()
except:
pass
dev_list = MV_CC_DEVICE_INFO_LIST()
layers = (MV_GIGE_DEVICE | MV_USB_DEVICE | MV_GENTL_CAMERALINK_DEVICE |
MV_GENTL_CXP_DEVICE | MV_GENTL_XOF_DEVICE)
ret = MvCamera.MV_CC_EnumDevices(layers, dev_list)
if ret == 0:
cameras = []
for i in range(dev_list.nDeviceNum):
device_info = cast(dev_list.pDeviceInfo[i], POINTER(MV_CC_DEVICE_INFO)).contents
name = self._get_device_name(device_info)
cameras.append((i, name, device_info))
return cameras
return []
def _get_device_name(self, device_info):
"""Extract device name from device info"""
if device_info.nTLayerType == MV_GIGE_DEVICE:
raw = device_info.SpecialInfo.stGigEInfo.chModelName
elif device_info.nTLayerType == MV_USB_DEVICE:
raw = device_info.SpecialInfo.stUsb3VInfo.chModelName
elif device_info.nTLayerType == MV_GENTL_CAMERALINK_DEVICE:
raw = device_info.SpecialInfo.stCMLInfo.chModelName
elif device_info.nTLayerType == MV_GENTL_CXP_DEVICE:
raw = device_info.SpecialInfo.stCXPInfo.chModelName
elif device_info.nTLayerType == MV_GENTL_XOF_DEVICE:
raw = device_info.SpecialInfo.stXoFInfo.chModelName
else:
return "Unknown_Device"
return bytes(raw).split(b'\x00', 1)[0].decode(errors='ignore')
def connect_camera(self, device_info):
"""Connect to a specific camera"""
if not FRAME_CAMERA_AVAILABLE:
return False
try:
# Force disconnect any existing camera first
self.force_disconnect()
# Initialize SDK if not already done
try:
MvCamera.MV_CC_Initialize()
except:
pass
self.cam = MvCamera()
self.device_info = device_info
# Create handle
if self.cam.MV_CC_CreateHandle(device_info) != 0:
return False
# Open device
if self.cam.MV_CC_OpenDevice(MV_ACCESS_Exclusive, 0) != 0:
self.cam.MV_CC_DestroyHandle()
return False
# For GigE, set optimal packet size
if device_info.nTLayerType == MV_GIGE_DEVICE:
pkt = self.cam.MV_CC_GetOptimalPacketSize()
if pkt > 0:
self.cam.MV_CC_SetIntValue("GevSCPSPacketSize", pkt)
# Turn off trigger mode
self.cam.MV_CC_SetEnumValue("TriggerMode", MV_TRIGGER_MODE_OFF)
# Setup recording parameters
self._setup_recording_params()
self._notify_status("Frame camera connected")
return True
except Exception as e:
self._notify_status(f"Frame camera connection failed: {e}")
return False
def disconnect_camera(self):
"""Disconnect camera gracefully"""
self.force_disconnect()
self._notify_status("Frame camera disconnected")
def force_disconnect(self):
"""Force disconnect with brutal cleanup"""
try:
# Stop all operations first
self.stop_grabbing()
# Give threads time to finish
time.sleep(0.1)
# Force close camera
if self.cam:
try:
self.cam.MV_CC_CloseDevice()
except:
pass
try:
self.cam.MV_CC_DestroyHandle()
except:
pass
self.cam = None
# Close OpenCV writer if open
if self.opencv_writer:
try:
self.opencv_writer.release()
except:
pass
self.opencv_writer = None
# Reset all flags
self.device_info = None
self.is_recording = False
self.is_grabbing = False
self.show_preview = False
self.exit_flag = True
self.use_opencv_recording = False
# Clear queues
while not self.command_queue.empty():
try:
self.command_queue.get_nowait()
except:
break
while not self.frame_queue.empty():
try:
self.frame_queue.get_nowait()
except:
break
except Exception as e:
print(f"Error in force disconnect: {e}")
def _setup_recording_params(self):
"""Setup recording parameters"""
if not self.cam:
return
record_param = MV_CC_RECORD_PARAM()
memset(byref(record_param), 0, sizeof(MV_CC_RECORD_PARAM))
# Get camera parameters
st_param = MVCC_INTVALUE()
memset(byref(st_param), 0, sizeof(MVCC_INTVALUE))
# Width
ret = self.cam.MV_CC_GetIntValue("Width", st_param)
if ret == 0:
record_param.nWidth = st_param.nCurValue
# Height
ret = self.cam.MV_CC_GetIntValue("Height", st_param)
if ret == 0:
record_param.nHeight = st_param.nCurValue
# Pixel format
st_enum_value = MVCC_ENUMVALUE()
memset(byref(st_enum_value), 0, sizeof(MVCC_ENUMVALUE))
ret = self.cam.MV_CC_GetEnumValue("PixelFormat", st_enum_value)
if ret == 0:
record_param.enPixelType = MvGvspPixelType(st_enum_value.nCurValue)
# Frame rate
st_float_value = MVCC_FLOATVALUE()
memset(byref(st_float_value), 0, sizeof(MVCC_FLOATVALUE))
ret = self.cam.MV_CC_GetFloatValue("ResultingFrameRate", st_float_value)
if ret != 0:
ret = self.cam.MV_CC_GetFloatValue("AcquisitionFrameRate", st_float_value)
if ret != 0:
st_float_value.fCurValue = 30.0
record_param.fFrameRate = st_float_value.fCurValue
record_param.nBitRate = 5000
record_param.enRecordFmtType = MV_FormatType_AVI
# Setup RGB buffer
self.rgb_buffer_size = record_param.nWidth * record_param.nHeight * 3
self.rgb_buffer = (c_ubyte * self.rgb_buffer_size)()
self.record_params = record_param
def start_grabbing(self):
"""Start grabbing frames"""
if not self.cam or self.is_grabbing:
return self.is_grabbing
ret = self.cam.MV_CC_StartGrabbing()
if ret == 0:
self.is_grabbing = True
self.exit_flag = False
self.recording_thread = threading.Thread(target=self._recording_worker, daemon=True)
self.recording_thread.start()
self._notify_status("Frame grabbing started")
return True
else:
self._notify_status(f"Failed to start grabbing: 0x{ret:08x}")
return False
def stop_grabbing(self):
"""Stop grabbing frames"""
if self.is_grabbing:
if self.is_recording:
self.stop_recording()
self.stop_preview()
self.exit_flag = True
if self.recording_thread:
self.recording_thread.join(timeout=2.0)
try:
self.cam.MV_CC_StopGrabbing()
except:
pass
self.is_grabbing = False
self._notify_status("Frame grabbing stopped")
def start_recording(self, filename=None):
"""Start recording"""
# Auto-start grabbing if needed
if not self.is_grabbing:
if not self.start_grabbing():
return False
if not filename:
device_name = self._get_device_name(self.device_info).replace(" ", "_")
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{device_name}_{timestamp}.avi"
self.current_filename = filename
# Determine recording method based on transformations
has_transformations = (self.vertical_flip or self.horizontal_flip or self.rotation_angle != 0)
if has_transformations:
# Use OpenCV recording for transformed frames
self.use_opencv_recording = True
self.command_queue.put(('START_OPENCV_RECORDING', filename))
else:
# Use SDK recording for non-transformed frames
self.use_opencv_recording = False
self.command_queue.put(('START_RECORDING', filename))
return True
def stop_recording(self):
"""Stop recording"""
if self.use_opencv_recording:
self.command_queue.put(('STOP_OPENCV_RECORDING',))
else:
self.command_queue.put(('STOP_RECORDING',))
def start_preview(self):
"""Start preview with auto-start grabbing"""
# Auto-start grabbing if needed
if not self.is_grabbing:
if not self.start_grabbing():
return False
if not self.show_preview:
self.show_preview = True
self.preview_thread = threading.Thread(target=self._preview_worker, daemon=True)
self.preview_thread.start()
return True
return False
def stop_preview(self):
"""Stop preview"""
if self.show_preview:
self.show_preview = False
if self.preview_thread:
self.preview_thread.join(timeout=2.0)
def toggle_preview_always_on_top(self):
"""Toggle always on top for preview window"""
self.preview_always_on_top = not self.preview_always_on_top
if WINDOWS_AVAILABLE:
set_window_always_on_top("Frame Camera Preview", self.preview_always_on_top)
self._notify_status(f"Frame preview always on top: {'ON' if self.preview_always_on_top else 'OFF'}")
def toggle_vertical_flip(self):
"""Toggle vertical flip"""
self.vertical_flip = not self.vertical_flip
self._notify_status(f"Frame vertical flip: {'ON' if self.vertical_flip else 'OFF'}")
return self.vertical_flip
def toggle_horizontal_flip(self):
"""Toggle horizontal flip"""
self.horizontal_flip = not self.horizontal_flip
self._notify_status(f"Frame horizontal flip: {'ON' if self.horizontal_flip else 'OFF'}")
return self.horizontal_flip
def rotate_frame(self):
"""Rotate frame by 90 degrees clockwise"""
self.rotation_angle = (self.rotation_angle + 90) % 360
self._notify_status(f"Frame rotation: {self.rotation_angle}°")
return self.rotation_angle
def _apply_transformations(self, img):
"""Apply transformations to image"""
if img is None:
return img
# Apply vertical flip
if self.vertical_flip:
img = cv2.flip(img, 0)
# Apply horizontal flip
if self.horizontal_flip:
img = cv2.flip(img, 1)
# Apply rotation
if self.rotation_angle != 0:
height, width = img.shape[:2]
center = (width // 2, height // 2)
if self.rotation_angle == 90:
img = cv2.rotate(img, cv2.ROTATE_90_CLOCKWISE)
elif self.rotation_angle == 180:
img = cv2.rotate(img, cv2.ROTATE_180)
elif self.rotation_angle == 270:
img = cv2.rotate(img, cv2.ROTATE_90_COUNTERCLOCKWISE)
return img
def get_exposure_range(self):
"""Get exposure time range"""
if not self.cam:
return None
st_float = MVCC_FLOATVALUE()
memset(byref(st_float), 0, sizeof(MVCC_FLOATVALUE))
ret = self.cam.MV_CC_GetFloatValue("ExposureTime", st_float)
if ret == 0:
return (st_float.fMin, st_float.fMax, st_float.fCurValue)
return None
def set_exposure(self, value):
"""Set exposure time"""
if not self.cam:
return False
self.cam.MV_CC_SetEnumValue("ExposureAuto", 0)
ret = self.cam.MV_CC_SetFloatValue("ExposureTime", float(value))
return ret == 0
def get_gain_range(self):
"""Get gain range"""
if not self.cam:
return None
st_float = MVCC_FLOATVALUE()
memset(byref(st_float), 0, sizeof(MVCC_FLOATVALUE))
ret = self.cam.MV_CC_GetFloatValue("Gain", st_float)
if ret == 0:
return (st_float.fMin, st_float.fMax, st_float.fCurValue)
return None
def set_gain(self, value):
"""Set gain"""
if not self.cam:
return False
self.cam.MV_CC_SetEnumValue("GainAuto", 0)
ret = self.cam.MV_CC_SetFloatValue("Gain", float(value))
return ret == 0
def _recording_worker(self):
"""Recording worker thread"""
frame_out = MV_FRAME_OUT()
memset(byref(frame_out), 0, sizeof(frame_out))
input_frame_info = MV_CC_INPUT_FRAME_INFO()
memset(byref(input_frame_info), 0, sizeof(MV_CC_INPUT_FRAME_INFO))
while not self.exit_flag:
# Process commands
try:
cmd_data = self.command_queue.get_nowait()
cmd = cmd_data[0]
if cmd == 'START_RECORDING':
filename = cmd_data[1]
if not self.is_recording and self.is_grabbing:
self.record_params.strFilePath = filename.encode('ascii')
ret = self.cam.MV_CC_StartRecord(self.record_params)
if ret == 0:
self.is_recording = True
with self.stats_lock:
self.stats['frames_captured'] = 0
self.stats['recording_start_time'] = time.time()
self._notify_status(f"Recording started: {filename}")
elif cmd == 'STOP_RECORDING':
if self.is_recording:
ret = self.cam.MV_CC_StopRecord()
if ret == 0:
self.is_recording = False
self._notify_status("Recording stopped")
elif cmd == 'START_OPENCV_RECORDING':
filename = cmd_data[1]
if not self.is_recording and self.is_grabbing and self.record_params:
# Setup OpenCV VideoWriter
fourcc = cv2.VideoWriter_fourcc(*'XVID')
fps = self.record_params.fFrameRate
# Get frame size (considering rotation)
width = self.record_params.nWidth
height = self.record_params.nHeight
if self.rotation_angle in [90, 270]:
width, height = height, width
self.opencv_writer = cv2.VideoWriter(filename, fourcc, fps, (width, height))
if self.opencv_writer.isOpened():
self.is_recording = True
with self.stats_lock:
self.stats['frames_captured'] = 0
self.stats['recording_start_time'] = time.time()
self._notify_status(f"OpenCV recording started: {filename}")
else:
self._notify_status("Failed to start OpenCV recording")
elif cmd == 'STOP_OPENCV_RECORDING':
if self.is_recording and self.opencv_writer:
self.opencv_writer.release()
self.opencv_writer = None
self.is_recording = False
self._notify_status("OpenCV recording stopped")
except queue.Empty:
pass
# Capture frames
if self.is_grabbing:
ret = self.cam.MV_CC_GetImageBuffer(frame_out, 100)
if frame_out.pBufAddr and ret == 0:
# Convert frame for processing
img = self._convert_to_display_format(frame_out.stFrameInfo, frame_out.pBufAddr)
if img is not None:
# Apply transformations
transformed_img = self._apply_transformations(img)
# Handle recording
if self.is_recording:
if self.use_opencv_recording and self.opencv_writer and transformed_img is not None:
# Record transformed frame with OpenCV
self.opencv_writer.write(transformed_img)
with self.stats_lock:
self.stats['frames_captured'] += 1
elif not self.use_opencv_recording:
# Record original frame with SDK
input_frame_info.pData = cast(frame_out.pBufAddr, POINTER(c_ubyte))
input_frame_info.nDataLen = frame_out.stFrameInfo.nFrameLen
ret = self.cam.MV_CC_InputOneFrame(input_frame_info)
if ret == 0:
with self.stats_lock:
self.stats['frames_captured'] += 1
# Send to preview (always transformed)
if self.show_preview and transformed_img is not None:
try:
self.frame_queue.put_nowait(transformed_img)
except queue.Full:
pass
self.cam.MV_CC_FreeImageBuffer(frame_out)
else:
time.sleep(0.01)
def _preview_worker(self):
"""Preview worker thread"""
window_name = "Frame Camera Preview"
cv2.namedWindow(window_name, cv2.WINDOW_NORMAL)
# Position window on TOP HALF of right side of screen
if self.screen_info:
window_width = self.screen_info.get('right_half_width', 640)
window_height = self.screen_info.get('top_half_height', 360) - 50 # Leave space for title bar
window_x = self.screen_info.get('right_half_x', 700)
window_y = 0 # TOP of screen
# Resize and position the window
cv2.resizeWindow(window_name, window_width, window_height)
cv2.moveWindow(window_name, window_x, window_y)
# Set initial always on top state - ALWAYS SET BY DEFAULT
if WINDOWS_AVAILABLE:
# Give window time to be created
time.sleep(0.1)
set_window_always_on_top(window_name, self.preview_always_on_top)
# Enable mouse callback for right-click menu
def mouse_callback(event, x, y, flags, param):
if event == cv2.EVENT_RBUTTONDOWN:
# Create a simple menu using messagebox
self.toggle_preview_always_on_top()
cv2.setMouseCallback(window_name, mouse_callback)
self._notify_status("Frame preview controls: Right-click to toggle always on top, 'T' to toggle, ESC to close")
while not self.exit_flag and self.show_preview:
try:
frame = self.frame_queue.get(timeout=0.1)
# Add overlay text showing transformation status
overlay_frame = frame.copy()
# Create status text
transform_info = []
if self.vertical_flip:
transform_info.append("V-Flip")
if self.horizontal_flip:
transform_info.append("H-Flip")
if self.rotation_angle != 0:
transform_info.append(f"Rot{self.rotation_angle}°")
transform_text = " | ".join(transform_info) if transform_info else "No transforms"
aot_text = f"Always on top: {'ON' if self.preview_always_on_top else 'OFF'}"
cv2.putText(overlay_frame, f"Transforms: {transform_text}", (10, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2)
cv2.putText(overlay_frame, f"{aot_text} (Right-click/T to toggle)", (10, 60),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2)
cv2.imshow(window_name, overlay_frame)
key = cv2.waitKey(1) & 0xFF
if key == 27: # ESC
self.show_preview = False
break
elif key == ord('t') or key == ord('T'): # Toggle always on top
self.toggle_preview_always_on_top()
except queue.Empty:
if cv2.getWindowProperty(window_name, cv2.WND_PROP_VISIBLE) < 1:
self.show_preview = False
break
except Exception as e:
break
cv2.destroyWindow(window_name)
def _convert_to_display_format(self, frame_info, raw_data):
"""Convert raw camera data to displayable format"""
memset(byref(self.convert_param), 0, sizeof(self.convert_param))
self.convert_param.nWidth = frame_info.nWidth
self.convert_param.nHeight = frame_info.nHeight
self.convert_param.pSrcData = raw_data
self.convert_param.nSrcDataLen = frame_info.nFrameLen
self.convert_param.enSrcPixelType = frame_info.enPixelType
self.convert_param.enDstPixelType = PixelType_Gvsp_RGB8_Packed
self.convert_param.pDstBuffer = self.rgb_buffer
self.convert_param.nDstBufferSize = self.rgb_buffer_size
ret = self.cam.MV_CC_ConvertPixelType(self.convert_param)
if ret != 0:
return None
img_buff = np.frombuffer(self.rgb_buffer, dtype=np.uint8, count=self.convert_param.nDstLen)
img = img_buff.reshape((frame_info.nHeight, frame_info.nWidth, 3))
img_bgr = cv2.cvtColor(img, cv2.COLOR_RGB2BGR)
return img_bgr
def _notify_status(self, message):
"""Notify status callback"""
if self.status_callback:
self.status_callback(f"Frame: {message}")
def get_status(self):
"""Get current status"""
with self.stats_lock:
return {
'connected': self.cam is not None,
'grabbing': self.is_grabbing,
'recording': self.is_recording,
'previewing': self.show_preview,
'frames_captured': self.stats['frames_captured'],
'fps': self.stats['last_fps'],
'vertical_flip': self.vertical_flip,
'horizontal_flip': self.horizontal_flip,
'rotation_angle': self.rotation_angle
}
class EventCameraController:
"""Modified version of EventCameraRecorder for GUI integration - simplified connection like old working code"""
def __init__(self, status_callback=None, screen_info=None):
self.device = None
self.mv_iterator = None
self.height = None
self.width = None
self.is_recording = False
self.should_exit = False
self.visualization_running = False
self.capturing = False
self.record_lock = threading.Lock()
self.event_queue = queue.Queue(maxsize=1000)
self.status_callback = status_callback
self.screen_info = screen_info or {}
self.current_log_path = None
# Threads
self.event_thread = None
self.viz_thread = None
# Window reference for proper cleanup and always on top - DEFAULT TO TRUE
self.window = None
self.window_always_on_top = True
def find_cameras(self):
"""Find available event cameras - simplified approach"""
if not EVENT_CAMERA_AVAILABLE:
return []
# Don't actually try to enumerate - just return a placeholder
# The actual connection test will happen in connect_camera()
return [(0, "Event Camera (Auto-detect)", "")]
def connect_camera(self, device_path=""):
"""Connect to event camera - using simple approach from old working code"""
if not EVENT_CAMERA_AVAILABLE:
self._notify_status("Event camera SDK not available")
return False
try:
self._notify_status("Connecting to event camera...")
# Simple cleanup first
self.stop_all()
# Use the exact same approach as the old working code
self.device = initiate_device("")
if self.device is None:
self._notify_status("Event camera connection failed: initiate_device returned None")
return False
# Test the connection by creating iterator - exactly like old code
self.mv_iterator = EventsIterator.from_device(device=self.device, delta_t=1000)
self.height, self.width = self.mv_iterator.get_size()
self._notify_status(f"Event camera connected successfully: {self.width}x{self.height}")
return True
except Exception as e:
self._notify_status(f"Event camera connection failed: {e}")
# Clean up failed attempt
try:
if self.device:
self.device = None
self.mv_iterator = None
except:
pass
return False
def disconnect_camera(self):
"""Disconnect camera gracefully"""
self.stop_all()
self.device = None
self.mv_iterator = None
self._notify_status("Event camera disconnected")
def start_capture(self):
"""Start capturing events"""
if not self.device or self.capturing:
return self.capturing
self.should_exit = False
self.capturing = True
self.event_thread = threading.Thread(target=self._event_processing_thread, daemon=True)
self.event_thread.start()
self._notify_status("Event capture started")
return True
def stop_capture(self):
"""Stop capturing events"""
if self.capturing:
self.should_exit = True
self.capturing = False
if self.event_thread:
self.event_thread.join(timeout=2.0)
self._notify_status("Event capture stopped")
def start_recording(self, output_dir="", filename_prefix=""):
"""Start recording events"""
# Auto-start capture if needed
if not self.capturing:
if not self.start_capture():
return False
with self.record_lock:
if self.is_recording:
return False
if self.device and self.device.get_i_events_stream():
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
if filename_prefix:
self.current_log_path = f"{filename_prefix}_{timestamp}.raw"
else:
self.current_log_path = f"recording_{timestamp}.raw"
if output_dir:
self.current_log_path = os.path.join(output_dir, self.current_log_path)
self.device.get_i_events_stream().log_raw_data(self.current_log_path)
self.is_recording = True
self._notify_status(f"Recording started: {self.current_log_path}")
return True
return False
def stop_recording(self):
"""Stop recording events"""
with self.record_lock:
if not self.is_recording:
return False
try:
self.device.get_i_events_stream().stop_log_raw_data()
except:
pass
self.is_recording = False
self._notify_status(f"Recording stopped: {self.current_log_path}")
return True
def start_visualization(self):
"""Start event visualization with auto-start capture"""
# Auto-start capture if needed
if not self.capturing:
if not self.start_capture():
return False
if not self.device or self.visualization_running:
return False
self.viz_thread = threading.Thread(target=self._visualization_thread, daemon=True)
self.viz_thread.start()
return True
def stop_visualization(self):
"""Stop visualization"""
if self.visualization_running:
self.visualization_running = False
# Force close window
if self.window:
try:
self.window.set_close_flag()
except:
pass
self.window = None
if self.viz_thread:
self.viz_thread.join(timeout=3.0)
def toggle_visualization_always_on_top(self):
"""Toggle always on top for visualization window"""
self.window_always_on_top = not self.window_always_on_top
if WINDOWS_AVAILABLE:
set_window_always_on_top("Event Camera Preview", self.window_always_on_top)
self._notify_status(f"Event preview always on top: {'ON' if self.window_always_on_top else 'OFF'}")
def stop_all(self):
"""Stop all operations"""
if self.is_recording:
self.stop_recording()
self.stop_visualization()
self.stop_capture()
def get_bias_values(self):
"""Get current bias values"""
if not self.device:
return {}
bias_interface = self.device.get_i_ll_biases()
if bias_interface is None:
return {}
try:
return bias_interface.get_all_biases()
except:
return {}
def set_bias_value(self, bias_name, value):
"""Set a bias value"""
if not self.device:
return False
bias_interface = self.device.get_i_ll_biases()
if bias_interface is None:
return False
try:
bias_interface.set(bias_name, int(value))
return True
except:
return False
def _event_processing_thread(self):
"""Thread for reading events from camera - exactly like old working code"""
try:
for evs in self.mv_iterator:
if self.should_exit:
break
try:
self.event_queue.put(evs, block=False)
except queue.Full:
pass
if self.should_exit:
break
except Exception as e:
self._notify_status(f"Event processing error: {e}")
finally:
self.capturing = False
def _visualization_thread(self):
"""Thread for event visualization - like old working code but with always on top"""
try:
event_frame_gen = PeriodicFrameGenerationAlgorithm(
sensor_width=self.width, sensor_height=self.height,
fps=25, palette=ColorPalette.Dark
)
# Calculate window dimensions for BOTTOM half of right side
if self.screen_info:
window_width = min(self.width, self.screen_info.get('right_half_width', self.width))
# Calculate height to fit in bottom portion
available_height = self.screen_info.get('height', 720) - self.screen_info.get('top_half_height', 360)
window_height = min(self.height, available_height - 50) # Leave space for title bar
else:
window_width = self.width
window_height = self.height
with MTWindow(title="Event Camera Preview", width=window_width, height=window_height,
mode=BaseWindow.RenderMode.BGR) as window:
self.window = window
# Position window on BOTTOM half of right side using Windows API
if WINDOWS_AVAILABLE and self.screen_info:
def position_window():
time.sleep(0.3) # Give window time to be created
try:
hwnd = win32gui.FindWindow(None, "Event Camera Preview")
if hwnd:
x = self.screen_info.get('right_half_x', 700)
# Position right after the top window with minimal gap
top_window_height = self.screen_info.get('top_half_height', 360)
y = top_window_height - 20 # Small gap between windows
width = self.screen_info.get('right_half_width', 640)
# Calculate height to fit remaining space
remaining_height = self.screen_info.get('height', 720) - y - 50 # Leave space at bottom
height = min(window_height, remaining_height)
win32gui.SetWindowPos(hwnd, 0, x, y, width, height,