checked 5 sensors run in parralle, cpu max out

2025-12-27 15:15:42 +08:00
parent a8b02105fc
commit 0b1b7138c0
3 changed files with 233 additions and 41 deletions
@@ -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()
+
@@ -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()
+
@@ -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