feat: add aruco-svo-calibration plan and utils scripts
- Add comprehensive work plan for ArUco-based multi-camera calibration - Add recording_multi.py for multi-camera SVO recording - Add streaming_receiver.py for network streaming - Add svo_playback.py for synchronized playback - Add zed_network_utils.py for camera configuration - Add AGENTS.md with project context
This commit is contained in:
Executable
+135
@@ -0,0 +1,135 @@
|
||||
"""
|
||||
This sample shows how to record video in Stereolabs SVO format from network cameras.
|
||||
SVO video files can be played with the ZED API and used with its different modules.
|
||||
"""
|
||||
|
||||
import pyzed.sl as sl
|
||||
import threading
|
||||
import signal
|
||||
import time
|
||||
import sys
|
||||
import zed_network_utils
|
||||
|
||||
# Global variable to handle exit
|
||||
exit_app = False
|
||||
|
||||
|
||||
def signal_handler(signal, frame):
|
||||
"""Handle Ctrl+C to properly exit the program"""
|
||||
global exit_app
|
||||
exit_app = True
|
||||
print("\nCtrl+C pressed. Exiting...")
|
||||
|
||||
|
||||
def acquisition(zed):
|
||||
"""Acquisition thread function to continuously grab frames"""
|
||||
infos = zed.get_camera_information()
|
||||
|
||||
while not exit_app:
|
||||
if zed.grab() == sl.ERROR_CODE.SUCCESS:
|
||||
# If needed, add more processing here
|
||||
# But be aware that any processing involving the GiL will slow down the multi threading performance
|
||||
pass
|
||||
|
||||
print(f"{infos.camera_model}[{infos.serial_number}] QUIT")
|
||||
|
||||
# disable Recording
|
||||
zed.disable_recording()
|
||||
# close the Camera
|
||||
zed.close()
|
||||
|
||||
|
||||
def open_camera(zed, config):
|
||||
"""Open a camera with given configuration and enable streaming"""
|
||||
ip, port = zed_network_utils.extract_ip_port(config)
|
||||
|
||||
if not ip or not port:
|
||||
return False
|
||||
|
||||
try:
|
||||
serial = config["FusionConfiguration"]["serial_number"]
|
||||
except KeyError:
|
||||
print("Error: Serial number not found in config")
|
||||
return False
|
||||
|
||||
# Open the remote camera using utility function
|
||||
if not zed_network_utils.open_remote_camera(zed, ip, port):
|
||||
return False
|
||||
|
||||
print(f"ZED SN{serial} Opened from {ip}:{port}")
|
||||
|
||||
# Enable Recording
|
||||
output_svo_file = f"ZED_SN{serial}.svo2"
|
||||
recording_param = sl.RecordingParameters(
|
||||
output_svo_file.replace(" ", ""), sl.SVO_COMPRESSION_MODE.H265
|
||||
)
|
||||
record_err = zed.enable_recording(recording_param)
|
||||
if record_err == sl.ERROR_CODE.SUCCESS:
|
||||
print(f"ZED SN{serial} Enabled recording")
|
||||
else:
|
||||
print(f"ZED SN{serial} Recording initialization error: {record_err}")
|
||||
zed.close()
|
||||
return False
|
||||
|
||||
print(f"Recording SVO file: {recording_param.video_filename}")
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def main():
|
||||
global exit_app
|
||||
|
||||
# Read network configuration using utility
|
||||
network_config = zed_network_utils.parse_network_config()
|
||||
|
||||
if not network_config:
|
||||
return 1
|
||||
|
||||
print(f"Found {len(network_config)} cameras in configuration")
|
||||
|
||||
if len(network_config) == 0:
|
||||
print("No ZED configured, exit program")
|
||||
return 1
|
||||
|
||||
zed_open = False
|
||||
|
||||
# Open all cameras
|
||||
zeds = []
|
||||
threads = []
|
||||
|
||||
for serial, config in network_config.items():
|
||||
zed = sl.Camera()
|
||||
if open_camera(zed, config):
|
||||
zeds.append(zed)
|
||||
zed_open = True
|
||||
|
||||
# Start acquisition thread immediately
|
||||
thread = threading.Thread(target=acquisition, args=(zed,))
|
||||
thread.start()
|
||||
threads.append(thread)
|
||||
|
||||
if not zed_open:
|
||||
print("No ZED opened, exit program")
|
||||
return 1
|
||||
|
||||
# Set up signal handler for Ctrl+C
|
||||
signal.signal(signal.SIGINT, signal_handler)
|
||||
print("Press Ctrl+C to exit")
|
||||
|
||||
# Main loop
|
||||
while not exit_app:
|
||||
time.sleep(0.02)
|
||||
|
||||
# Wait for all threads to finish
|
||||
print("Exit signal, closing ZEDs")
|
||||
time.sleep(0.1)
|
||||
|
||||
for thread in threads:
|
||||
thread.join()
|
||||
|
||||
print("Program exited")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user