From a8b02105fcd32a15784ebf2bfb27ea436e545b16 Mon Sep 17 00:00:00 2001 From: bluepulsar Date: Sat, 27 Dec 2025 15:05:07 +0800 Subject: [PATCH] add async to speed up 2x --- example4_multiple_sensor.py | 77 +++------------- example5_async.py | 49 +++++++++++ src/orisys/sensor.py | 171 +++++++++++++++++++++++++++++++++++- 3 files changed, 230 insertions(+), 67 deletions(-) create mode 100644 example5_async.py diff --git a/example4_multiple_sensor.py b/example4_multiple_sensor.py index 8334d1b..246609c 100644 --- a/example4_multiple_sensor.py +++ b/example4_multiple_sensor.py @@ -1,8 +1,7 @@ """ Example 4: Multi-Sensor with Multiprocessing -Demonstrates the simplified MultiSensor API for running multiple sensors -in parallel with full CUDA support. +Each sensor runs in its own process with independent CUDA context. """ import os @@ -14,50 +13,26 @@ import multiprocessing as mp def main(): - # Configure sensors 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' - }, + {'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'}, ] with orisys.MultiSensor(configs) as multi: print("\nPress 'q' to quit") for all_data in multi: - for data in all_data: - if data is None: - continue + 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}") - sensor_id = data['sensor_id'] - print(f"Sensor {sensor_id}: FPS={data['fps']:5.1f}, " - f"Fn={data['fnormal']:8.2f}") - - # Draw arrows 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 {sensor_id}", arrows) + arrows = orisys.util.draw_arrows(data['img'].copy(), data['flow'], threshold=5) + cv2.imshow(f"Sensor {sid}", arrows) - # Show depth if data['depth'] is not None: - depth_img = cv2.applyColorMap( - (data['depth'] * 255).astype('uint8'), - cv2.COLORMAP_HOT - ) - cv2.imshow(f"Sensor {sensor_id} - Depth", depth_img) + cv2.imshow(f"Sensor {sid} - Depth", + cv2.applyColorMap((data['depth'] * 255).astype('uint8'), cv2.COLORMAP_HOT)) if cv2.waitKey(1) & 0xFF == ord('q'): break @@ -65,36 +40,6 @@ def main(): cv2.destroyAllWindows() -def main_manual(): - """Alternative: Manual start/stop control""" - configs = [ - {'vid_src': './s1.mp4', 'isstitch': True, 'cal_path': './config/cal.npy'}, - {'vid_src': './s1_copy.mp4', 'isstitch': True, 'cal_path': './config/cal.npy'}, - ] - - multi = orisys.MultiSensor(configs) - multi.start() - - try: - while True: - # Get data from specific sensor - data0 = multi.get(0) # or multi[0] - data1 = multi.get(1) # or multi[1] - - # Or get all at once - # all_data = multi.get_all() - - for data in [data0, data1]: - if data and data['img'] is not None: - 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() # Required for Windows + mp.freeze_support() main() diff --git a/example5_async.py b/example5_async.py new file mode 100644 index 0000000..31b7fe3 --- /dev/null +++ b/example5_async.py @@ -0,0 +1,49 @@ +""" +Example 5: Async/Pipelined Processing + +Overlaps image capture with deformation computation for higher throughput. +Pipeline: [Capture Thread] -> [Queue] -> [Compute Thread] -> [Queue] -> [Main] +""" + +import os +os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" + +import cv2 +import orisys + + +def main(): + sensor = orisys.Sensor("./s1.mp4", isstitch=True, cuda=True, verbose=True) + sensor.enable_async(queue_size=2) + + print("\nPress 'q' to quit, 'r' to reset") + + try: + while True: + data = sensor.get_async(timeout=0.05) + if data is None: + continue + + print(f"FPS: {data['fps']:5.1f}, Fn: {data['fnormal']:8.2f}, " + f"Fx: {data['fshearx']:8.2f}, Fy: {data['fsheary']: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("Deformation", arrows) + + if data['depth'] is not None: + depth_img = cv2.applyColorMap((data['depth'] * 255).astype('uint8'), cv2.COLORMAP_HOT) + cv2.imshow("Depth", depth_img) + + key = cv2.waitKey(1) & 0xFF + if key == ord('q'): + break + elif key == ord('r'): + sensor.reset() + finally: + sensor.disconnect() + cv2.destroyAllWindows() + + +if __name__ == '__main__': + main() diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index 189febe..b55a1a1 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -10,6 +10,9 @@ from enum import Enum, auto from typing import Optional, Tuple, Union, Any from .getDensity import getFilteredDensity from . import configs +from threading import Thread, Event +from queue import Queue, Empty +from concurrent.futures import ThreadPoolExecutor class Sensor: """ @@ -60,6 +63,16 @@ class Sensor: # Logging state self.last_log_time = time.perf_counter() self.fps = 0 + + # Async pipeline state (3-stage: Capture → Flow → HHD) + self._async_enabled = False + self._capture_thread = None + self._flow_thread = None + self._compute_thread = None + self._capture_queue = None + self._flow_queue = None + self._result_queue = None + self._stop_event = None if self.verbose: print(f" Video source: {vid_src}") @@ -282,6 +295,158 @@ class Sensor: self._last_timings = timings return self.flow + + # ========================================================================= + # Async/Pipelined Processing API (3-stage: Capture → Flow → HHD) + # ========================================================================= + + def _queue_put(self, q, item): + """Helper to put item in queue, dropping old if full.""" + if q.full(): + try: q.get_nowait() + except Empty: pass + q.put(item) + + def enable_async(self, queue_size=2): + """ + Enable 3-stage async pipeline for maximum throughput. + + Pipeline: [Capture] → [Optical Flow] → [HHD Compute] + Each stage runs in parallel, overlapping all operations. + """ + if self._async_enabled: + return + + self._capture_queue = Queue(maxsize=queue_size) # img + self._flow_queue = Queue(maxsize=queue_size) # (img, flow) + self._result_queue = Queue(maxsize=queue_size) # final result + self._stop_event = Event() + self._async_enabled = True + + self._capture_thread = Thread(target=self._capture_worker, daemon=True) + self._flow_thread = Thread(target=self._flow_worker, daemon=True) + self._compute_thread = Thread(target=self._hhd_worker, daemon=True) + + self._capture_thread.start() + self._flow_thread.start() + self._compute_thread.start() + + if self.verbose: + print("Async 3-stage pipeline enabled: Capture → Flow → HHD") + + def _capture_worker(self): + """Stage 1: Capture and preprocess frames.""" + alpha = 0.7 + while not self._stop_event.is_set(): + try: + ret, frame = self.cap.read() + if not ret: + continue + + gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) + self._Sensor__stitcher.get_stitch_unit(gray) + self._Sensor__get_tiles(gray) + img = np.ascontiguousarray(self._Sensor__stitcher.stitch_rt_optimized()) + img = cv2.resize(img, tuple(self.config_resolution_stitch), interpolation=cv2.INTER_CUBIC) + + # Temporal filtering + if self.img_filtered is None: + self.img_filtered = img.astype(np.float32) + else: + self.img_filtered = alpha * img + (1 - alpha) * self.img_filtered + + self._queue_put(self._capture_queue, self.img_filtered.astype(np.uint8)) + except Exception as e: + if self.verbose: print(f"Capture error: {e}") + + def _flow_worker(self): + """Stage 2: Compute optical flow.""" + while not self._stop_event.is_set(): + try: + img = self._capture_queue.get(timeout=0.1) + flow = self.tracker.track(img) + if hasattr(flow, 'download'): + flow = flow.download() + + self._queue_put(self._flow_queue, (img, flow)) + except Empty: + continue + except Exception as e: + if self.verbose: print(f"Flow error: {e}") + + def _hhd_worker(self): + """Stage 3: HHD decomposition and force computation.""" + while not self._stop_event.is_set(): + try: + img, flow = self._flow_queue.get(timeout=0.1) + + if self.hhd is None: + self.hhd = nHHD(grid=flow.shape[:2], spacings=(18, 18)) + + if self.cuda and self.CUPY_AVAILABLE: + _, forces = self.hhd.get_force_gpu_v2(flow) + else: + _, forces = self.hhd.get_force(flow) + + self._queue_put(self._result_queue, { + 'img': img, 'flow': flow, + 'normal': self.hhd.get_normal(), 'shear': self.hhd.get_shear(), + 'depth': getFilteredDensity(flow, use_cuda=self.cuda), + 'fnormal': forces[0], 'fshearx': forces[1], + 'fsheary': forces[2], 'ftor': forces[3], + }) + except Empty: + continue + except Exception as e: + if self.verbose: print(f"HHD error: {e}") + + def get_async(self, timeout=0.05): + """Get latest result (non-blocking). Returns dict or None.""" + if not self._async_enabled: + return None + try: + result = self._result_queue.get(timeout=timeout) + t = time.perf_counter() + result['fps'] = 1.0 / (t - self.last_log_time) if self.last_log_time else 0 + self.last_log_time = t + + # Update instance vars + for k in ['img', 'flow', 'normal', 'shear', 'depth', 'fnormal', 'fshearx', 'fsheary', 'ftor', 'fps']: + setattr(self, k, result.get(k)) + self.vupdate = True + return result + except Empty: + return None + + def disable_async(self): + """Stop async pipeline.""" + if not self._async_enabled: + return + self._stop_event.set() + for t in [self._capture_thread, self._flow_thread, self._compute_thread]: + if t: t.join(timeout=2.0) + self._async_enabled = False + self._capture_thread = self._flow_thread = self._compute_thread = None + self._capture_queue = self._flow_queue = self._result_queue = self._stop_event = None + if self.verbose: + print("Async pipeline disabled") + + def update(self): + """Convenience: get_img + compute_deformation in one call.""" + self.get_img() + self.compute_deformation() + return { + 'img': self.img, + 'flow': self.flow, + 'normal': self.normal, + 'shear': self.shear, + 'depth': self.depth, + 'fps': self.fps, + 'fnormal': self.fnormal, + 'fshearx': self.fshearx, + 'fsheary': self.fsheary, + 'ftor': self.ftor, + } def read_info(self, *args: info) -> Tuple: """ @@ -341,8 +506,12 @@ class Sensor: def disconnect(self): """ - Release video capture resources. + Release video capture resources and stop async threads. """ + # Stop async pipeline if running + if self._async_enabled: + self.disable_async() + if self.cap.isOpened(): self.cap.release() if self.verbose: