From 0b1b7138c0fd0e5e9d407f5f96337fb73241e6d5 Mon Sep 17 00:00:00 2001 From: bluepulsar Date: Sat, 27 Dec 2025 15:15:42 +0800 Subject: [PATCH] checked 5 sensors run in parralle, cpu max out --- example6_multi_async.py | 71 +++++++++++++++++++++ example7_five_sensors.py | 73 +++++++++++++++++++++ src/orisys/MultiSensor.py | 130 ++++++++++++++++++++++++++------------ 3 files changed, 233 insertions(+), 41 deletions(-) create mode 100644 example6_multi_async.py create mode 100644 example7_five_sensors.py diff --git a/example6_multi_async.py b/example6_multi_async.py new file mode 100644 index 0000000..1b9bb9b --- /dev/null +++ b/example6_multi_async.py @@ -0,0 +1,71 @@ +""" +Example 6: Multi-Sensor with Async Pipeline + +Combines multiprocessing (separate CUDA contexts) with async 3-stage pipeline +(Capture → Flow → HHD) for maximum throughput. + +Architecture: + Process 1 (Sensor 0): [Capture] → [Flow] → [HHD] → Queue + Process 2 (Sensor 1): [Capture] → [Flow] → [HHD] → Queue + ↓ + Main Process: Display ← ─────────────────────┘ +""" + +import os +os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" + +import cv2 +import orisys +import multiprocessing as mp + + +def main(): + configs = [ + {'vid_src': './s1.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, + {'vid_src': './s1_copy.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, + ] + + # Method 1: Using async_context (recommended) + with orisys.MultiSensor.async_context(configs) as multi: + print("\nAsync multi-sensor started. Press 'q' to quit") + + for all_data in multi: + for data in [d for d in all_data if d]: + sid = data['sensor_id'] + print(f"Sensor {sid}: FPS={data['fps']:5.1f}, Fn={data['fnormal']:8.2f}") + + if data['img'] is not None and data['flow'] is not None: + arrows = orisys.util.draw_arrows(data['img'].copy(), data['flow'], threshold=5) + cv2.imshow(f"Sensor {sid}", arrows) + + if cv2.waitKey(1) & 0xFF == ord('q'): + break + + cv2.destroyAllWindows() + + +def main_manual(): + """Alternative: Manual start with async flag""" + configs = [ + {'vid_src': './s1.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, + {'vid_src': './s1_copy.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, + ] + + multi = orisys.MultiSensor(configs) + multi.start(use_async=True) # Enable async mode + + try: + while True: + for data in [d for d in multi.get_all() if d]: + cv2.imshow(f"Sensor {data['sensor_id']}", data['img']) + if cv2.waitKey(1) & 0xFF == ord('q'): + break + finally: + multi.stop() + cv2.destroyAllWindows() + + +if __name__ == '__main__': + mp.freeze_support() + main() + diff --git a/example7_five_sensors.py b/example7_five_sensors.py new file mode 100644 index 0000000..029421a --- /dev/null +++ b/example7_five_sensors.py @@ -0,0 +1,73 @@ +""" +Example 7: Five Sensors with Async Pipeline + +Demonstrates scaling to 5 sensors running in parallel with async mode. +Each sensor runs in its own process with 3-stage async pipeline. +""" + +import os +os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" + +import cv2 +import orisys +import multiprocessing as mp +import numpy as np + + +def main(): + # Configure 5 sensors (using same video for demo - replace with actual sources) + configs = [ + {'vid_src': './s1.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, + {'vid_src': './s1.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, + {'vid_src': './s1.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, + {'vid_src': './s1.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, + {'vid_src': './s1.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, + ] + + with orisys.MultiSensor.async_context(configs) as multi: + print(f"\n{len(configs)} sensors started. Press 'q' to quit") + + for all_data in multi: + valid_data = [d for d in all_data if d and d['img'] is not None] + + # Print stats for all sensors + for data in valid_data: + print(f"S{data['sensor_id']}: FPS={data['fps']:4.1f} Fn={data['fnormal']:7.1f}", end=" | ") + if valid_data: + print() + + # Create grid display (2 rows: 3 + 2) + imgs = [] + for data in valid_data: + if data['flow'] is not None: + img = orisys.util.draw_arrows(data['img'].copy(), data['flow'], threshold=5) + else: + img = data['img'] + # Add sensor ID label + cv2.putText(img, f"S{data['sensor_id']}", (10, 30), + cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2) + imgs.append(img) + + if len(imgs) >= 5: + # Create 2x3 grid (pad with black if needed) + h, w = imgs[0].shape[:2] + black = np.zeros((h, w, 3), dtype=np.uint8) if len(imgs[0].shape) == 3 else np.zeros((h, w), dtype=np.uint8) + + row1 = np.hstack([imgs[0], imgs[1], imgs[2]]) + row2 = np.hstack([imgs[3], imgs[4], black]) + grid = np.vstack([row1, row2]) + + # Resize for display + grid = cv2.resize(grid, (1200, 800)) + cv2.imshow("5 Sensors Grid", grid) + + if cv2.waitKey(1) & 0xFF == ord('q'): + break + + cv2.destroyAllWindows() + + +if __name__ == '__main__': + mp.freeze_support() + main() + diff --git a/src/orisys/MultiSensor.py b/src/orisys/MultiSensor.py index fceb20e..7d099f5 100644 --- a/src/orisys/MultiSensor.py +++ b/src/orisys/MultiSensor.py @@ -9,21 +9,20 @@ independent CUDA contexts. import os -def _sensor_worker(sensor_id, config, data_queue, stop_event): +def _sensor_worker(sensor_id, config, data_queue, stop_event, use_async=False): """ Worker function that runs in a separate PROCESS for each sensor. Each process has its own Python interpreter and CUDA context. - Note: This function must be at module level for multiprocessing to work. + Args: + use_async: If True, use 3-stage async pipeline within the process """ os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" - # Import inside process to get fresh CUDA context from .sensor import Sensor - print(f"[Sensor {sensor_id}] Process started (PID: {os.getpid()})") + print(f"[Sensor {sensor_id}] Process started (PID: {os.getpid()}), async={use_async}") - # Initialize sensor in this process sensor = Sensor( vid_src=config.get('vid_src'), isstitch=config.get('isstitch', True), @@ -33,41 +32,61 @@ def _sensor_worker(sensor_id, config, data_queue, stop_event): cal_path=config.get('cal_path', None) ) + # Enable async pipeline if requested + if use_async: + sensor.enable_async(queue_size=2) + frame_count = 0 + def queue_put(data): + try: + while not data_queue.empty(): + try: data_queue.get_nowait() + except: break + data_queue.put_nowait(data) + except: pass + while not stop_event.is_set(): try: - # Get image and compute deformation - sensor.get_img() - sensor.compute_deformation() - - # Package all data - data = { - 'sensor_id': sensor_id, - 'img': sensor.img.copy() if sensor.img is not None else None, - 'flow': sensor.flow.copy() if sensor.flow is not None else None, - 'normal': sensor.normal.copy() if sensor.normal is not None else None, - 'shear': sensor.shear.copy() if sensor.shear is not None else None, - 'depth': sensor.depth.copy() if sensor.depth is not None else None, - 'fps': sensor.fps, - 'fnormal': sensor.fnormal, - 'fshearx': sensor.fshearx, - 'fsheary': sensor.fsheary, - 'ftor': sensor.ftor, - 'frame_count': frame_count, - } - - # Non-blocking put - drop old frames if queue is full - try: - while not data_queue.empty(): - try: - data_queue.get_nowait() - except: - break - data_queue.put_nowait(data) - except: - pass + if use_async: + # Use async get + result = sensor.get_async(timeout=0.05) + if result is None: + continue + data = { + 'sensor_id': sensor_id, + 'img': result['img'].copy() if result['img'] is not None else None, + 'flow': result['flow'].copy() if result['flow'] is not None else None, + 'normal': result['normal'].copy() if result['normal'] is not None else None, + 'shear': result['shear'].copy() if result['shear'] is not None else None, + 'depth': result['depth'].copy() if result['depth'] is not None else None, + 'fps': result['fps'], + 'fnormal': result['fnormal'], + 'fshearx': result['fshearx'], + 'fsheary': result['fsheary'], + 'ftor': result['ftor'], + 'frame_count': frame_count, + } + else: + # Sync mode + sensor.get_img() + sensor.compute_deformation() + data = { + 'sensor_id': sensor_id, + 'img': sensor.img.copy() if sensor.img is not None else None, + 'flow': sensor.flow.copy() if sensor.flow is not None else None, + 'normal': sensor.normal.copy() if sensor.normal is not None else None, + 'shear': sensor.shear.copy() if sensor.shear is not None else None, + 'depth': sensor.depth.copy() if sensor.depth is not None else None, + 'fps': sensor.fps, + 'fnormal': sensor.fnormal, + 'fshearx': sensor.fshearx, + 'fsheary': sensor.fsheary, + 'ftor': sensor.ftor, + 'frame_count': frame_count, + } + queue_put(data) frame_count += 1 except Exception as e: @@ -136,6 +155,7 @@ class MultiSensor: self.sensor_configs = sensor_configs self.num_sensors = len(sensor_configs) self._mp = mp + self._use_async = False self.processes = [] self.data_queues = [] @@ -147,24 +167,33 @@ class MultiSensor: self.data_queues.append(mp.Queue(maxsize=2)) self.stop_events.append(mp.Event()) - def start(self): - """Start all sensor worker processes.""" + def start(self, use_async=False): + """ + Start all sensor worker processes. + + Args: + use_async: If True, each sensor process uses 3-stage async pipeline + for maximum throughput (Capture → Flow → HHD in parallel) + """ if self._started: print("MultiSensor already started") return + self._use_async = use_async + for i, config in enumerate(self.sensor_configs): process = self._mp.Process( target=_sensor_worker, - args=(i, config, self.data_queues[i], self.stop_events[i]), + args=(i, config, self.data_queues[i], self.stop_events[i], use_async), daemon=True ) process.start() self.processes.append(process) print(f"Started sensor {i} (PID: {process.pid})") + mode = "async (3-stage pipeline)" if use_async else "sync" self._started = True - print(f"All {len(self.processes)} sensors started") + print(f"All {len(self.processes)} sensors started in {mode} mode") def get(self, sensor_idx, timeout=0.01): """ @@ -254,8 +283,8 @@ class MultiSensor: print("All sensors stopped.") def __enter__(self): - """Context manager entry - starts sensors.""" - self.start() + """Context manager entry - starts sensors (sync mode by default).""" + self.start(use_async=False) return self def __exit__(self, exc_type, exc_val, exc_tb): @@ -266,4 +295,23 @@ class MultiSensor: def __del__(self): """Cleanup on deletion.""" self.stop() + + @classmethod + def async_context(cls, sensor_configs): + """ + Create a MultiSensor that uses async mode when used as context manager. + + Example: + with MultiSensor.async_context(configs) as multi: + for all_data in multi: + ... + """ + return _AsyncMultiSensorContext(sensor_configs) + + +class _AsyncMultiSensorContext(MultiSensor): + """Helper class for async context manager.""" + def __enter__(self): + self.start(use_async=True) + return self