diff --git a/config/compound_eye.png b/config/compound_eye.png index ef3b17b..3f6510b 100644 Binary files a/config/compound_eye.png and b/config/compound_eye.png differ diff --git a/example1_basic_pipeline.py b/example1_basic_pipeline.py index d7b2616..21d291a 100644 --- a/example1_basic_pipeline.py +++ b/example1_basic_pipeline.py @@ -14,15 +14,16 @@ os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" import cv2 import orisys import numpy as np +import time def main(): # 步骤1:创建传感器对象 (必要步骤),config_name为配置文件名(内置配置文件名)或路径(自定义配置文件路径) # 方式1: 从视频文件读取 - sensor = orisys.Sensor("s1.mp4", config_name="prototype", cal_path= "./config/cal.npy", isstitch=True, cuda=True, verbose=True) + # sensor = orisys.Sensor("s1.mp4", config_name="prototype", cal_path= "./config/cal.npy", isstitch=True, cuda=True, verbose=True) # 方式2: 从摄像头读取(取消注释使用) - # sensor = orisys.Sensor(0, config_name="prototype", cal_path= "./config/cal.npy", isstitch=True, cuda=True, verbose=True) + sensor = orisys.Sensor(0, config_name="./config/lens_test.json", cal_path= "./config/cal.npy", isstitch=True, cuda=True, verbose=True) print("\n按 'q' 退出程序,按 'r' 重置光流追踪器") @@ -32,14 +33,14 @@ def main(): # 步骤 2.1: 获取并拼接图像 (必要步骤) sensor.get_img() print("img_size_raw:", sensor.img_size_raw) - if sensor.detect_motion(): + if sensor.detect_motion(threshold=2): print("motion detected") - break - # 步骤 2.2: 计算形变场 (必要步骤) - sensor.compute_deformation() - # 步骤 2.3: 计算触碰点 - sensor.compute_contact() - + # 步骤 2.2: 计算形变场 (必要步骤) + sensor.compute_deformation() + # 步骤 2.3: 计算触碰点 + sensor.compute_contact() + + #步骤 2.4: 读取信息 fps, flow, vnormal, img, contour, centroid, depth_map = sensor.read_info( sensor.info.FPS, @@ -79,7 +80,7 @@ def main(): #步骤 3.4: 打印资料 print(f"fps: {fps}") - + #步骤 3.5: 键盘控制 key = cv2.waitKey(1) & 0xFF if key == ord("q"): @@ -88,8 +89,8 @@ def main(): if key == ord("r"): # 重置光流追踪器(当追踪漂移时使用) sensor.reset() - print("光流追踪器已重置") - + print("传感器已重置") + # ========================================================================= # 4. 清理资源 # ========================================================================= diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index 2d406fb..7c55f48 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -86,6 +86,13 @@ class Sensor: self.cal_path = cal_path self.mirror = mirror + + # Load configuration + self.__config_cam = None + self.__config_stitcher = None + self.__config_contact = None + self._load_config() + # State variables self.stitch_valid = False self.stitch_calibration = [] @@ -93,28 +100,29 @@ class Sensor: self.compute_contact_valid = False # Data Variables (public) + shape = (self.__config_cam["resolution_stitch"][1], self.__config_cam["resolution_stitch"][0]) self.img = None self.img_size_raw = None self.img_filtered = None - self.prev_img = None - self.flow = None - self.vnormal = None - self.vshear = None + self.initial_img = None + self.flow = np.zeros(shape + (2,), dtype=np.float32) + self.vnormal = np.zeros_like(self.flow) + self.vshear = np.zeros_like(self.flow) self.n_boxes = None self.fnormal = self.fshearx = self.fsheary = self.ftor = 0.0 # Contact Data Variables (public) - self.centroid = None + self.centroid = (0, 0) self.centroid_found = False - self.contour = None - self.depth_map = None + self.contour = np.empty(0) + self.depth_map = np.zeros_like(self.img) # Timing self.last_log_time = time.perf_counter() self._fps = 0 # Async pipeline state - self._async_enabled = False + # self._async_enabled = False self._capture_thread = None self._flow_thread = None self._compute_thread = None @@ -123,11 +131,6 @@ class Sensor: self._result_queue = None self._stop_event = None - # Load configuration - self.__config_cam = None - self.__config_stitcher = None - self.__config_contact = None - self._load_config() # Initialize components @@ -150,7 +153,7 @@ class Sensor: # Public API - Core Methods (Most Frequently Used) # ========================================================================= - def get_img(self, log_timing=False): + def get_img(self): """ Capture and preprocess a frame. @@ -162,59 +165,36 @@ class Sensor: print("Failed to get_img from video source.") return None - if log_timing: - timings = {} - t_start = time.perf_counter() # Convert to grayscale - if log_timing: t0 = time.perf_counter() gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) - # gray = frame - if log_timing: timings['cvt_color'] = (time.perf_counter() - t0) * 1000 # Extract and stitch tiles - if log_timing: t0 = time.perf_counter() self.stitcher.get_stitch_unit(gray) self._extract_tiles(gray) - if log_timing: timings['get_tiles'] = (time.perf_counter() - t0) * 1000 - if log_timing: t0 = time.perf_counter() img = np.ascontiguousarray(self.stitcher.stitch_rt_optimized()) - if log_timing: timings['stitch'] = (time.perf_counter() - t0) * 1000 self.img_size_raw = img.shape # Resize img = cv2.resize(img, tuple(self._config_resolution_stitch), interpolation=cv2.INTER_CUBIC) - - # # Temporal filtering (noise reduction) - # alpha = 0.7 - # if self.img_filtered is None: - # self.img_filtered = img.astype(np.float32) - # else: - # self.img_filtered = alpha * img + (1 - alpha) * self.img_filtered - # Update previous image + if self.mirror: img = cv2.flip(img, 1) - if self.img is not None: - self.prev_img = self.img.copy() - self.img = img.astype(np.uint8) - if self.prev_img is None: - self.prev_img = img.copy() - - if log_timing: - timings['total'] = (time.perf_counter() - t_start) * 1000 - self._print_timings("get_img", timings) + # FPS calculation + self._fps = 1.0 / (time.perf_counter() - self.last_log_time) + self.last_log_time = time.perf_counter() return self.img.copy() - def detect_motion(self, threshold=10): - if self.prev_img is None: + def detect_motion(self, threshold=4): + if self.initial_img is None: return False - diff = cv2.absdiff(self.img, self.prev_img) + diff = cv2.absdiff(self.img, self.initial_img) return diff.mean() > threshold def compute_deformation(self): @@ -229,9 +209,6 @@ class Sensor: self.__compute_flow() self.__compute_hhd() - # FPS calculation - self._fps = 1.0 / (time.perf_counter() - self.last_log_time) - self.last_log_time = time.perf_counter() return True @@ -322,11 +299,13 @@ class Sensor: if any(arg in [self.info.VRAW, self.info.VNORMAL, self.info.VSHEAR, self.info.FNORMAL, self.info.FSHEARX, self.info.FSHEARY] for arg in args): if not self.compute_deformation_valid: - raise Exception("Warning: Compute deformation first") + print("Warning: Compute deformation first") + # raise Exception("Warning: Compute deformation first") if any(arg in [self.info.CENTROID, self.info.CONTOUR, self.info.DEPTH] for arg in args): if not self.compute_contact_valid: - raise Exception("Warning: Compute contact first") + print("Warning: Compute contact first") + # raise Exception("Warning: Compute contact first") mapping = { # Return copies of arrays to prevent external modification of internal state @@ -357,11 +336,12 @@ class Sensor: def reset(self): """Reset optical flow tracker.""" self.__tracker.reset() + self.initial_img = self.img.copy() def disconnect(self): """Release resources and stop async threads.""" - if self._async_enabled: - self.disable_async() + # if self._async_enabled: + # self.disable_async() if self.cap.isOpened(): self.cap.release() if self.verbose: @@ -414,11 +394,12 @@ class Sensor: def _load_config(self): """Load sensor configuration from JSON file.""" self._config = configs.load_config(self.config_name) - cam = self._config["camera"] - self._config_resolution_raw = cam["resolution_raw"] - self._config_fps = cam["fps"] - self._config_units = cam["units"] - self._config_resolution_stitch = cam["resolution_stitch"] + self.__config_cam = self._config["camera"] + self._config_resolution_raw = self.__config_cam["resolution_raw"] + self._config_fps = self.__config_cam["fps"] + self._config_units = self.__config_cam["units"] + self._config_resolution_stitch = self.__config_cam["resolution_stitch"] + self._config_dis_preset = self._config["tracker"]["dis_preset"] self.__config_stitcher = self._config["stitcher"] @@ -471,7 +452,13 @@ class Sensor: else: self._find_stitch_boxes(frame) self._extract_tiles(frame) - self.stitch_cal() + img = self.stitch_cal() + img = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) + self.initial_img = cv2.resize(img, tuple(self._config_resolution_stitch), interpolation=cv2.INTER_CUBIC) + if self.mirror: + self.initial_img = cv2.flip(self.initial_img, 1) + self.initial_img = self.initial_img.astype(np.uint8) + def _find_stitch_boxes(self, img_src, show=False): """Detect vision tile boxes with camera stabilization."""