detect motion by capturing reference image
This commit is contained in:
unknown
Binary file not shown.
|
Before Width: | Height: | Size: 2.0 MiB After Width: | Height: | Size: 2.0 MiB |
@@ -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.4: 读取信息
|
||||
fps, flow, vnormal, img, contour, centroid, depth_map = sensor.read_info(
|
||||
sensor.info.FPS,
|
||||
@@ -88,7 +89,7 @@ def main():
|
||||
if key == ord("r"):
|
||||
# 重置光流追踪器（当追踪漂移时使用）
|
||||
sensor.reset()
|
||||
print("光流追踪器已重置")
|
||||
print("传感器已重置")
|
||||
|
||||
# =========================================================================
|
||||
# 4. 清理资源
|
||||
|
||||
+43
-56
@@ -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."""
|
||||
|
||||
Reference in New Issue
Block a user