crosstyan
9c861105f7
- 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
136 lines
3.4 KiB
Python
Executable File
136 lines
3.4 KiB
Python
Executable File
"""
|
|
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())
|