add private variable and copy when return by readinfo

This commit is contained in:
unknown
2025-12-30 14:52:06 +08:00
parent 2a739ed27c
commit 398d2b1706
8 changed files with 134 additions and 158 deletions
Binary file not shown.

Before

Width:  |  Height:  |  Size: 2.0 MiB

After

Width:  |  Height:  |  Size: 2.1 MiB

-31
View File
@@ -1,31 +0,0 @@
{
"name": "prototype",
"version": "1.0",
"camera": {
"resolution_raw": [1920, 1080],
"fps": 60,
"units": 15,
"units_used": 9,
"resolution_stitch": [400, 400]
},
"stitcher": {
"min_area_frac": 0.018,
"max_area_frac": 0.035,
"max_aspect_ratio": 20,
"index": [1,2,3,6,7,8,11,12,13],
"pos_default": [[260, 56], [577, 62], [897, 69], [1214, 75], [1530, 81], [256, 391], [570, 397], [889, 403], [1208, 409], [1523, 416], [248, 724], [563, 730], [882, 738], [1201, 743], [1516, 751]],
"margin": 10,
"nominal_ov_pos": 100
},
"contact": {
"min_contour_area": 3000,
"flow_threshold":1,
"div_threshold":0.02,
"expansion_threshold": 0.01,
"downsample_factor": 1,
"kernel_size": [5,5],
"denoise_sigma": 15
}
}
+1 -1
View File
@@ -28,7 +28,7 @@ def main():
# verbose: 是否打印详细初始化日志 # verbose: 是否打印详细初始化日志
# 方式1: 从视频文件读取 # 方式1: 从视频文件读取
sensor = orisys.Sensor("./s1.mp4", isstitch=True, cuda=False, verbose=True,config_name="prototype") sensor = orisys.Sensor("./s1.mp4", isstitch=True, cuda=True, verbose=True,config_name="prototype")
# 方式2: 从摄像头读取(取消注释使用) # 方式2: 从摄像头读取(取消注释使用)
# sensor = orisys.Sensor(0, isstitch=True, cuda=True, verbose=True) # sensor = orisys.Sensor(0, isstitch=True, cuda=True, verbose=True)
+14 -13
View File
@@ -7,10 +7,10 @@ import time
def main(): def main():
# 方式1: 从视频文件读取 # 方式1: 从视频文件读取
sensor = orisys.Sensor("./s1.mp4", isstitch=True, cuda=True, verbose=True,config_name="./config/prototype.json") # sensor = orisys.Sensor("./s1.mp4", isstitch=True, cuda=True, verbose=True,config_name="prototype")
# 方式2: 从摄像头读取(取消注释使用) # 方式2: 从摄像头读取(取消注释使用)
# sensor = orisys.Sensor(0, isstitch=True, cuda=False, verbose=True,config_name="./config/prototype.json") sensor = orisys.Sensor(0, isstitch=True, cuda=True, verbose=True,config_name="prototype")
print("\n按 'q' 退出程序,按 'r' 重置光流追踪器") print("\n按 'q' 退出程序,按 'r' 重置光流追踪器")
@@ -20,32 +20,33 @@ def main():
# 步骤 2.2: 计算形变场 # 步骤 2.2: 计算形变场
sensor.compute_deformation() sensor.compute_deformation()
# 计算触碰点
# sensor.compute_contact(min_magnitude=0.1,expansion_threshold=0.01,denoise_sigma=15)
sensor.compute_contact()
fps, flow, vnormal = sensor.read_info( #计算后读取信息
fps, flow, vnormal, img, contour, centroid, depth_map = sensor.read_info(
sensor.info.FPS, sensor.info.FPS,
sensor.info.VRAW, sensor.info.VRAW,
sensor.info.VNORMAL sensor.info.VNORMAL,
sensor.info.IMG,
sensor.info.CONTOUR,
sensor.info.CENTROID,
sensor.info.DEPTH
) )
arrows = orisys.util.draw_arrows( arrows = orisys.util.draw_arrows(
sensor.img.copy(), # 使用 copy() 避免修改原图 img, # 使用 copy() 避免修改原图
vnormal, vnormal,
threshold=5, threshold=5,
grid_spacing=10, grid_spacing=10,
arrow_scale=1.0 arrow_scale=1.0
) )
cv2.imshow("形变场 Deformation", arrows) cv2.imshow("形变场 Deformation", arrows)
# 计算触碰点
# sensor.compute_contact(min_magnitude=0.1,expansion_threshold=0.01,denoise_sigma=15)
sensor.compute_contact()
# 在图像上触碰点和区域 # 在图像上触碰点和区域
image_with_foe = orisys.util.draw_contact(sensor.img.copy(), sensor.contour, sensor.centroid) image_with_foe = orisys.util.draw_contact(img, contour, centroid)
cv2.imshow("Contact Centroid", image_with_foe) cv2.imshow("Contact Centroid", image_with_foe)
# 计算触碰区域
depth_map = sensor.depth_map
# 归一化散度值到0-255范围(使用绝对值,然后归一化) # 归一化散度值到0-255范围(使用绝对值,然后归一化)
div_abs = depth_map div_abs = depth_map
+3 -3
View File
@@ -6,7 +6,7 @@ import cv2
class Stitcher: class Stitcher:
def __init__(self,isstitch=False, n=[3,3], config_stitcher=None,verbose=False): def __init__(self,isstitch=False, config_stitcher=None,verbose=False):
# Settings # Settings
self.isstitch = isstitch self.isstitch = isstitch
@@ -22,8 +22,8 @@ class Stitcher:
self.boxes = self.units = None self.boxes = self.units = None
self.index = config_stitcher["index"] self.index = config_stitcher["index"]
self.stitch_calibration = self.tile_list = [] self.stitch_calibration = self.tile_list = []
self.ncol = n[0] self.ncol = config_stitcher["n"][0]
self.nrow = n[1] self.nrow = config_stitcher["n"][1]
self.nominal_ov_pos=config_stitcher["nominal_ov_pos"] self.nominal_ov_pos=config_stitcher["nominal_ov_pos"]
self.margin=config_stitcher["margin"] self.margin=config_stitcher["margin"]
self.bound=self.nominal_ov_pos - self.margin self.bound=self.nominal_ov_pos - self.margin
+4 -4
View File
@@ -47,10 +47,10 @@ def load_config(config_name_or_path):
available = [f.stem for f in CONFIG_DIR.glob("*.json")] available = [f.stem for f in CONFIG_DIR.glob("*.json")]
available_str = ", ".join(available) if available else "none" available_str = ", ".join(available) if available else "none"
raise FileNotFoundError( raise FileNotFoundError(
f"Config file not found: {config_name_or_path}\n" f"Input config file not found with name or path: {config_name_or_path}\n"
f"Tried as file path: {config_path} (not found)\n" f" Tried as file path: {config_path} (not found)\n"
f"Tried as package config: {package_config_path} (not found)\n" f" Tried as package config: {package_config_path} (not found)\n"
f"Available package configs: {available_str}" f" Available package configs: {available_str}"
) )
with open(package_config_path, 'r', encoding='utf-8') as f: with open(package_config_path, 'r', encoding='utf-8') as f:
+2 -1
View File
@@ -15,7 +15,8 @@
"index": [1,2,3,6,7,8,11,12,13], "index": [1,2,3,6,7,8,11,12,13],
"pos_default": [[260, 56], [577, 62], [897, 69], [1214, 75], [1530, 81], [256, 391], [570, 397], [889, 403], [1208, 409], [1523, 416], [248, 724], [563, 730], [882, 738], [1201, 743], [1516, 751]], "pos_default": [[260, 56], [577, 62], [897, 69], [1214, 75], [1530, 81], [256, 391], [570, 397], [889, 403], [1208, 409], [1523, 416], [248, 724], [563, 730], [882, 738], [1201, 743], [1516, 751]],
"margin": 10, "margin": 10,
"nominal_ov_pos": 100 "nominal_ov_pos": 100,
"n": [3,3]
}, },
"contact": { "contact": {
"min_contour_area": 3000, "min_contour_area": 3000,
+110 -105
View File
@@ -18,7 +18,6 @@ from queue import Queue, Empty
from .Stitcher import Stitcher from .Stitcher import Stitcher
from .tracker import Tracker from .tracker import Tracker
from .nHHD import nHHD from .nHHD import nHHD
# from .getDensity import getFilteredDensity
from . import configs from . import configs
from .get_depth import Contact3D from .get_depth import Contact3D
@@ -48,6 +47,7 @@ class Sensor:
CENTROID = auto() CENTROID = auto()
CONTOUR = auto() CONTOUR = auto()
DEPTH = auto() DEPTH = auto()
IMG = auto()
# ========================================================================= # =========================================================================
# Initialization # Initialization
@@ -94,25 +94,24 @@ class Sensor:
self.stitch_calibration = [] self.stitch_calibration = []
# Data Variables # Data Variables
self.img = None self.__img = None
self.img_size_raw = None self.__img_size_raw = None
self.img_filtered = None self.__img_filtered = None
self.flow = None self.__flow = None
self.vnormal = None self.__vnormal = None
self.vshear = None self.__vshear = None
# self.depth = None self.__n_boxes = None
self.n_boxes = None self.__fnormal = self.__fshearx = self.__fsheary = self.__ftor = 0.0
self.fnormal = self.fshearx = self.fsheary = self.ftor = 0.0
# Contact Data Variables # Contact Data Variables
self.centroid = None self.__centroid = None
self.centroid_found = False self.__centroid_found = False
self.contour = None self.__contour = None
self.depth_map = None self.__depth_map = None
# Timing # Timing
self.last_log_time = time.perf_counter() self.last_log_time = time.perf_counter()
self.fps = 0 self._fps = 0
# Async pipeline state # Async pipeline state
self._async_enabled = False self._async_enabled = False
@@ -125,17 +124,17 @@ class Sensor:
self._stop_event = None self._stop_event = None
# Load configuration # Load configuration
self.config_cam = None self.__config_cam = None
self.config_stitcher = None self.__config_stitcher = None
self.config_contact = None self.__config_contact = None
self._load_config() self._load_config()
# Initialize components # Initialize components
self._stitcher = Stitcher(isstitch=isstitch, config_stitcher=self.config_stitcher, verbose=self.verbose) self.__stitcher = Stitcher(isstitch=isstitch, config_stitcher=self.__config_stitcher, verbose=self.verbose)
self.tracker = Tracker(adaptive=True, cuda=cuda, verbose=self.verbose) self.__tracker = Tracker(adaptive=True, cuda=cuda, verbose=self.verbose)
self.contact3d = Contact3D(config_contact=self.config_contact, verbose=self.verbose) self.__contact3d = Contact3D(config_contact=self.__config_contact, verbose=self.verbose)
self.hhd = None self.__hhd = None
# Setup video capture # Setup video capture
self._open_video_source(override, box) self._open_video_source(override, box)
@@ -174,35 +173,35 @@ class Sensor:
# Extract and stitch tiles # Extract and stitch tiles
if log_timing: t0 = time.perf_counter() if log_timing: t0 = time.perf_counter()
self._stitcher.get_stitch_unit(gray) self.__stitcher.get_stitch_unit(gray)
self._extract_tiles(gray) self._extract_tiles(gray)
if log_timing: timings['get_tiles'] = (time.perf_counter() - t0) * 1000 if log_timing: timings['get_tiles'] = (time.perf_counter() - t0) * 1000
if log_timing: t0 = time.perf_counter() if log_timing: t0 = time.perf_counter()
img = np.ascontiguousarray(self._stitcher.stitch_rt_optimized()) img = np.ascontiguousarray(self.__stitcher.stitch_rt_optimized())
if log_timing: timings['stitch'] = (time.perf_counter() - t0) * 1000 if log_timing: timings['stitch'] = (time.perf_counter() - t0) * 1000
self.img_size_raw = img.shape self.__img_size_raw = img.shape
# Resize # Resize
img = cv2.resize(img, tuple(self.config_resolution_stitch), interpolation=cv2.INTER_CUBIC) img = cv2.resize(img, tuple(self._config_resolution_stitch), interpolation=cv2.INTER_CUBIC)
# Temporal filtering (noise reduction) # Temporal filtering (noise reduction)
alpha = 0.7 alpha = 0.7
if self.img_filtered is None: if self.__img_filtered is None:
self.img_filtered = img.astype(np.float32) self.__img_filtered = img.astype(np.float32)
else: else:
self.img_filtered = alpha * img + (1 - alpha) * self.img_filtered self.__img_filtered = alpha * img + (1 - alpha) * self.__img_filtered
self.img = self.img_filtered.astype(np.uint8) self.__img = self.__img_filtered.astype(np.uint8)
self.imgupdate = True self.__imgupdate = True
if log_timing: if log_timing:
timings['total'] = (time.perf_counter() - t_start) * 1000 timings['total'] = (time.perf_counter() - t_start) * 1000
self._print_timings("get_img", timings) self._print_timings("get_img", timings)
if self.mirror: if self.mirror:
self.img = cv2.flip(self.img, 1) self.__img = cv2.flip(self.__img, 1)
return self.img return True
def compute_deformation(self, log_timing=False): def compute_deformation(self, log_timing=False):
""" """
@@ -217,27 +216,27 @@ class Sensor:
# Optical flow # Optical flow
if log_timing: t0 = time.perf_counter() if log_timing: t0 = time.perf_counter()
flow = self.tracker.track(self.img) flow = self.__tracker.track(self.__img)
if hasattr(flow, 'download'): if hasattr(flow, 'download'):
flow = flow.download() flow = flow.download()
if log_timing: timings['optical_flow'] = (time.perf_counter() - t0) * 1000 if log_timing: timings['optical_flow'] = (time.perf_counter() - t0) * 1000
# Initialize HHD solver # Initialize HHD solver
if self.hhd is None: if self.__hhd is None:
self.hhd = nHHD(grid=flow.shape[:2], spacings=(18, 18)) self.__hhd = nHHD(grid=flow.shape[:2], spacings=(18, 18))
# Force computation # Force computation
if log_timing: t0 = time.perf_counter() if log_timing: t0 = time.perf_counter()
if self.cuda and self.CUPY_AVAILABLE: if self.cuda and self.CUPY_AVAILABLE:
_, forces = self.hhd.get_force_gpu_v2(flow) _, forces = self.__hhd.get_force_gpu_v2(flow)
else: else:
_, forces = self.hhd.get_force(flow) _, forces = self.__hhd.get_force(flow)
self.fnormal, self.fshearx, self.fsheary, self.ftor = forces self.__fnormal, self.__fshearx, self.__fsheary, self.__ftor = forces
if log_timing: timings['hhd_decompose'] = (time.perf_counter() - t0) * 1000 if log_timing: timings['hhd_decompose'] = (time.perf_counter() - t0) * 1000
# Get components # Get components
self.vnormal = self.hhd.get_normal() self.__vnormal = self.__hhd.get_normal()
self.vshear = self.hhd.get_shear() self.__vshear = self.__hhd.get_shear()
# # Depth computation # # Depth computation
# if log_timing: t0 = time.perf_counter() # if log_timing: t0 = time.perf_counter()
@@ -245,40 +244,37 @@ class Sensor:
# if log_timing: timings['get_density'] = (time.perf_counter() - t0) * 1000 # if log_timing: timings['get_density'] = (time.perf_counter() - t0) * 1000
# Update state # Update state
self.flow = flow self.__flow = flow
self.imgupdate = False self.imgupdate = False
self.vupdate = True self.vupdate = True
# FPS calculation # FPS calculation
self.fps = 1.0 / (time.perf_counter() - self.last_log_time) self._fps = 1.0 / (time.perf_counter() - self.last_log_time)
self.last_log_time = time.perf_counter() self.last_log_time = time.perf_counter()
if log_timing: if log_timing:
timings['total'] = (time.perf_counter() - t_start) * 1000 timings['total'] = (time.perf_counter() - t_start) * 1000
self._print_timings("compute_deformation", timings) self._print_timings("compute_deformation", timings)
return self.flow return True
def compute_contact(self, multiple_foe=False, min_contour_area=3000, min_magnitude=1.0, expansion_threshold=0.01, downsample_factor=1, kernel_size=(5,5), denoise_sigma=15): def compute_contact(self, multiple_foe=False, min_contour_area=3000, min_magnitude=1.0, expansion_threshold=0.01, downsample_factor=1, kernel_size=(5,5), denoise_sigma=15):
if multiple_foe: if multiple_foe:
# result = compute_multiple_foe(self.vnormal, min_contour_area=min_contour_area, min_magnitude=min_magnitude, expansion_threshold=expansion_threshold, downsample_factor=downsample_factor, kernel_size=kernel_size, denoise_sigma=denoise_sigma) result = self.__contact3d.compute_multiple_foe(self.__vnormal)
result = self.contact3d.compute_multiple_foe(self.vnormal)
else: else:
if self.cuda: if self.cuda:
# result = compute_foe_gpu(self.vnormal, min_magnitude=min_magnitude, expansion_threshold=expansion_threshold, downsample_factor=downsample_factor, kernel_size=kernel_size, denoise_sigma=denoise_sigma) result = self.__contact3d.compute_foe_gpu(self.__vnormal)
result = self.contact3d.compute_foe_gpu(self.vnormal)
else: else:
# result = compute_foe(self.vnormal, min_magnitude=min_magnitude, expansion_threshold=expansion_threshold, downsample_factor=downsample_factor, kernel_size=kernel_size, denoise_sigma=denoise_sigma) result = self.__contact3d.compute_foe(self.__vnormal)
result = self.contact3d.compute_foe(self.vnormal)
self.centroid = result['centroid'] self.__centroid = result['centroid']
self.centroid_found = result['centroid_found'] self.__centroid_found = result['centroid_found']
if 'contour' in result: if 'contour' in result:
self.contour = result['contour'] self.__contour = result['contour']
self.depth_map = result['divergence_map'] self.__depth_map = result['divergence_map']
return result return True
# def get_depth(self, denoise_sigma=15): # def get_depth(self, denoise_sigma=15):
@@ -295,10 +291,10 @@ class Sensor:
self.get_img() self.get_img()
self.compute_deformation() self.compute_deformation()
return { return {
'img': self.img, 'flow': self.flow, 'img': self.__img, 'flow': self.__flow,
'normal': self.vnormal, 'shear': self.vshear, 'depth': self.depth, 'normal': self.__vnormal, 'shear': self.__vshear, 'depth': self.__depth_map,
'fps': self.fps, 'fnormal': self.fnormal, 'fps': self._fps, 'fnormal': self.__fnormal,
'fshearx': self.fshearx, 'fsheary': self.fsheary, 'ftor': self.ftor, 'fshearx': self.__fshearx, 'fsheary': self.__fsheary, 'ftor': self.__ftor,
} }
def read_info(self, *args) -> Tuple: def read_info(self, *args) -> Tuple:
@@ -315,17 +311,27 @@ class Sensor:
print("Warning: Call compute_deformation() first.") print("Warning: Call compute_deformation() first.")
mapping = { mapping = {
self.info.VRAW: lambda: self.flow if self.flow is not None else np.array([]), # Return copies of arrays to prevent external modification of internal state
self.info.VNORMAL: lambda: self.vnormal if self.vnormal is not None else np.array([]), self.info.VRAW: lambda: self.__flow.copy() if self.__flow is not None else np.array([]),
self.info.VSHEAR: lambda: self.vshear if self.vshear is not None else np.array([]), self.info.VNORMAL: lambda: self.__vnormal.copy() if self.__vnormal is not None else np.array([]),
self.info.FNORMAL: lambda: self.fnormal, self.info.VSHEAR: lambda: self.__vshear.copy() if self.__vshear is not None else np.array([]),
self.info.FSHEARX: lambda: self.fshearx, # Scalar values are safe to return directly
self.info.FSHEARY: lambda: self.fsheary, self.info.FNORMAL: lambda: self.__fnormal,
self.info.FPS: lambda: self.fps, self.info.FSHEARX: lambda: self.__fshearx,
self.info.FSHEARY: lambda: self.__fsheary,
self.info.FPS: lambda: self._fps,
self.info.TIMESTAMP: lambda: time.strftime('%H:%M:%S'), self.info.TIMESTAMP: lambda: time.strftime('%H:%M:%S'),
self.info.DEPTH: lambda: self.depth_map if self.depth_map is not None else np.array([]), # Return copies of arrays to prevent external modification
self.info.CENTROID: lambda: self.centroid if self.centroid is not None else np.array([]), self.info.DEPTH: lambda: self.__depth_map.copy() if self.__depth_map is not None else np.array([]),
self.info.CONTOUR: lambda: self.contour if self.contour is not None else np.array([]), # Centroid: can be tuple, list, or array - return a copy to prevent modification
# Return None if centroid is not found or empty
self.info.CENTROID: lambda: (
np.array(self.__centroid).copy()
if self.__centroid is not None and len(self.__centroid) > 0
else None
),
self.info.CONTOUR: lambda: self.__contour.copy() if self.__contour is not None else np.array([]),
self.info.IMG: lambda: self.__img.copy() if self.__img is not None else np.array([]),
} }
return tuple(mapping.get(arg, lambda: None)() for arg in args if isinstance(arg, self.info)) return tuple(mapping.get(arg, lambda: None)() for arg in args if isinstance(arg, self.info))
@@ -365,8 +371,8 @@ class Sensor:
self._async_enabled = True self._async_enabled = True
self._capture_thread = Thread(target=self._capture_worker, daemon=True) self._capture_thread = Thread(target=self._capture_worker, daemon=True)
self._flow_thread = Thread(target=self._flow_worker, daemon=True) self._flow_thread = Thread(target=self.__flow_worker, daemon=True)
self._compute_thread = Thread(target=self._hhd_worker, daemon=True) self._compute_thread = Thread(target=self.__hhd_worker, daemon=True)
self._contact_thread = Thread(target=self._contact_worker, daemon=True) self._contact_thread = Thread(target=self._contact_worker, daemon=True)
self._capture_thread.start() self._capture_thread.start()
@@ -419,7 +425,7 @@ class Sensor:
def export_config(self, path="./config/exported_config.json"): def export_config(self, path="./config/exported_config.json"):
"""Return current configuration dict.""" """Return current configuration dict."""
configs.export_config(self.config, path) configs.export_config(self._config, path)
if self.verbose: if self.verbose:
print(f"Config exported to {path}") print(f"Config exported to {path}")
return True return True
@@ -437,11 +443,11 @@ class Sensor:
if os.path.exists(self.cal_path): if os.path.exists(self.cal_path):
print("Loading calibration from", self.cal_path) print("Loading calibration from", self.cal_path)
self.stitch_calibration = np.load(self.cal_path, allow_pickle=True) self.stitch_calibration = np.load(self.cal_path, allow_pickle=True)
self._stitcher.stitch_calibration = self.stitch_calibration self.__stitcher.stitch_calibration = self.stitch_calibration
return self._stitcher.stitch_rt_optimized() return self.__stitcher.stitch_rt_optimized()
else: else:
print("Creating new calibration...") print("Creating new calibration...")
img, self.stitch_calibration = self._stitcher.stitch_cal() img, self.stitch_calibration = self.__stitcher.stitch_cal()
os.makedirs("./config", exist_ok=True) os.makedirs("./config", exist_ok=True)
np.save(self.cal_path, self.stitch_calibration) np.save(self.cal_path, self.stitch_calibration)
print("Calibration saved to", self.cal_path) print("Calibration saved to", self.cal_path)
@@ -453,30 +459,29 @@ class Sensor:
def _load_config(self): def _load_config(self):
"""Load sensor configuration from JSON file.""" """Load sensor configuration from JSON file."""
self.config = configs.load_config(self.config_name) self._config = configs.load_config(self.config_name)
cam = self.config["camera"] cam = self._config["camera"]
self.config_resolution_raw = cam["resolution_raw"] self._config_resolution_raw = cam["resolution_raw"]
self.config_fps = cam["fps"] self._config_fps = cam["fps"]
self.config_units = cam["units"] self._config_units = cam["units"]
self.config_resolution_stitch = cam["resolution_stitch"] self._config_resolution_stitch = cam["resolution_stitch"]
# self.config_index = cam["index"]
# self.config_pos_default = cam["pos_default"]
self.config_stitcher = self.config["stitcher"] self.__config_stitcher = self._config["stitcher"]
self.config_contact = self.config["contact"] self.__config_contact = self._config["contact"]
if self.verbose: if self.verbose:
print(f"Config: {self.config_name}") print(f"Config: {self.config_name}")
print(f" Resolution: {self.config_resolution_raw} → {self.config_resolution_stitch}") print(f" Resolution: {self._config_resolution_raw} → {self._config_resolution_stitch}")
print(f" FPS: {self.config_fps}, Units: {self.config_units}") print(f" FPS: {self._config_fps}, Units: {self._config_units}")
return True
def _open_video_source(self, override, box): def _open_video_source(self, override, box):
"""Initialize video capture and calibration.""" """Initialize video capture and calibration."""
self.cap = cv2.VideoCapture(self.vid_src) self.cap = cv2.VideoCapture(self.vid_src)
self.cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter.fourcc('M', 'J', 'P', 'G')) self.cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter.fourcc('M', 'J', 'P', 'G'))
self.cap.set(cv2.CAP_PROP_FPS, self.config_fps) self.cap.set(cv2.CAP_PROP_FPS, self._config_fps)
self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.config_resolution_raw[0]) self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, self._config_resolution_raw[0])
self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.config_resolution_raw[1]) self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self._config_resolution_raw[1])
if self.verbose: if self.verbose:
print("Connecting to video source...") print("Connecting to video source...")
@@ -493,8 +498,8 @@ class Sensor:
if override: if override:
if box is not None: if box is not None:
self._stitcher.isstitch = self.isstitch = False self.__stitcher.isstitch = self.isstitch = False
self._stitcher.boxes = box self.__stitcher.boxes = box
else: else:
print("Override=True but no box provided.") print("Override=True but no box provided.")
else: else:
@@ -507,21 +512,21 @@ class Sensor:
n_boxes, v_mean0, timeout = 0, 0, 5 n_boxes, v_mean0, timeout = 0, 0, 5
start_time = cv2.getTickCount() start_time = cv2.getTickCount()
while n_boxes < self.config_units: while n_boxes < self._config_units:
ret, frame = self.cap.read() ret, frame = self.cap.read()
v_mean = np.sum(cv2.mean(frame)) v_mean = np.sum(cv2.mean(frame))
if abs(v_mean - v_mean0) < 5: if abs(v_mean - v_mean0) < 5:
n_boxes = self._stitcher.find_stitch_boxes(frame, show=show) n_boxes = self.__stitcher.find_stitch_boxes(frame, show=show)
if (cv2.getTickCount() - start_time) / cv2.getTickFrequency() > timeout: if (cv2.getTickCount() - start_time) / cv2.getTickFrequency() > timeout:
print("Timeout finding vision units.") print("Timeout finding vision units.")
self._stitcher.isstitch = self.isstitch = False self.__stitcher.isstitch = self.isstitch = False
break break
v_mean0 = v_mean v_mean0 = v_mean
self.n_boxes = n_boxes self.__n_boxes = n_boxes
self.stitch_valid = self.isstitch and n_boxes == self.config_units self.stitch_valid = self.isstitch and n_boxes == self._config_units
status = "valid for stitching" if self.stitch_valid else "non-stitch mode" status = "valid for stitching" if self.stitch_valid else "non-stitch mode"
print(f"Sensor: {status}") print(f"Sensor: {status}")
@@ -529,8 +534,8 @@ class Sensor:
def _extract_tiles(self, img_src): def _extract_tiles(self, img_src):
"""Extract tiles from source image.""" """Extract tiles from source image."""
self._stitcher.get_stitch_unit(img_src) self.__stitcher.get_stitch_unit(img_src)
self._stitcher.extract_tiles() self.__stitcher.extract_tiles()
return True return True
# ========================================================================= # =========================================================================
@@ -554,15 +559,15 @@ class Sensor:
continue continue
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
self._stitcher.get_stitch_unit(gray) self.__stitcher.get_stitch_unit(gray)
self._extract_tiles(gray) self._extract_tiles(gray)
img = np.ascontiguousarray(self._stitcher.stitch_rt_optimized()) img = np.ascontiguousarray(self.__stitcher.stitch_rt_optimized())
img = cv2.resize(img, tuple(self.config_resolution_stitch), interpolation=cv2.INTER_CUBIC) img = cv2.resize(img, tuple(self._config_resolution_stitch), interpolation=cv2.INTER_CUBIC)
if self.img_filtered is None: if self.__img_filtered is None:
self.img_filtered = img.astype(np.float32) self.__img_filtered = img.astype(np.float32)
else: else:
self.img_filtered = alpha * img + (1 - alpha) * self.img_filtered self.__img_filtered = alpha * img + (1 - alpha) * self.__img_filtered
self._queue_put(self._capture_queue, self.img_filtered.astype(np.uint8)) self._queue_put(self._capture_queue, self.img_filtered.astype(np.uint8))
except Exception as e: except Exception as e: