diff --git a/API文档-0.3.0.pdf b/API文档-0.3.0.pdf index bbeb140..acf4348 100644 Binary files a/API文档-0.3.0.pdf and b/API文档-0.3.0.pdf differ diff --git a/Readme.docx b/Readme.docx index 76f19cf..c2b5f4e 100644 Binary files a/Readme.docx and b/Readme.docx differ diff --git a/Readme.md b/Readme.md index b59c8b9..07375c4 100644 --- a/Readme.md +++ b/Readme.md @@ -53,17 +53,29 @@ pip install orisys-0.3.0-cp310-cp310-win_amd64.whl import orisys print(orisys.__version__) ``` + +5 安装opencl（实验）: + +在不带nvidia显卡的设备，可选择通过opencl进行gpu 计算加速，需要确保opencl的库或驱动已经安装，然后在conda环境中安装依赖： + +``` bash +pip install pyopencl +conda install conda-forge::pyvkfft +``` + *** ## `Sensor` 类别 ### 描述 `orisys`套件的核心类，用于从视频文件或实时摄像头中读取传感器图像，创建一个具体的传感器对象（实例）。并完成形变、光流及力信息的计算与管理。 + ### 输入参数 -* **vid_src** (`int | str`, 必填): 传感器 ID、相机编号或视频路径。默认为 `0`。 -* **config_name** (`str`，可选): 传感器配置`json`文件路径，用于加载对应的参数配置，默认为 `./config/ddjx01`。 -* **cal_path** (`str`，可选): 复眼拼接校准文件路径，默认为 `./config/cal.npy`。有多个传感器同时运行时建议传感器类创建时建议填写，避免校准文件覆写。在首次运行时程序会在找不到文件时自动创建。 -* **backend** (`str`，可选): 光流场解耦计算后端，默认为 `auto`，可选`cupy`,`opencl`,`cpu`。 -* **verbose** (`bool`，可选): 是否输出初始化及运行过程中的详细日志信息，默认为 `False`。 + +* **vid_src** (`int | str`, 必填): 传感器 ID、相机编号或视频路径。默认为 `0`。 +* **config_name** (`str`，可选): 传感器配置`json`文件路径，用于加载对应的参数配置，默认为 `./config/ddjx01`。 +* **cal_path** (`str`，可选): 复眼拼接校准文件路径，默认为 `./config/cal.npy`。有多个传感器同时运行时建议传感器类创建时建议填写，避免校准文件覆写。在首次运行时程序会在找不到文件时自动创建。 +* **backend** (`str`，可选): 光流场解耦计算后端，默认为 `auto`，可选`cupy`,`opencl`,`cpu`。 +* **verbose** (`bool`，可选): 是否输出初始化及运行过程中的详细日志信息，默认为 `False`。 ### 返回 @@ -135,6 +147,7 @@ if is_touched: ### 返回 `dict`为接触分析结果字典，至少包含以下键： + * **expansion_mask**: 扩张区域掩码（`np.ndarray`）。 * **contraction_mask**: 收缩区域掩码（`np.ndarray`）。 * **divergence_map**: 散度图（`np.ndarray`）。 @@ -169,13 +182,16 @@ centroid, contour, depth = sensor.read_info( * **args**: 任意数量的 `Sensor.info` 枚举，用于指定需要获取的数据类型。支持的枚举值及对应数据如下： ### 支持的枚 info keys： + 向量/图像类 + * VRAW: 原始光流场, `np.ndarray`，shape=(H, W, 2) * VNORMAL: 法向光流场, `np.ndarray`，shape=(H, W, 2) * VSHEAR: 切向光流场, `np.ndarray`，shape=(H, W,2) * DEPTH: 深度场（可选）, 未启用/未生成时返回空数组`np.array([])`。 标量类 + * FNORMAL: 法向力，`float`。 * FSHEARX: X切向力，`float`。 * FSHEARY: Y切向力，`float`。 @@ -184,6 +200,7 @@ centroid, contour, depth = sensor.read_info( * CONTOUR： 触碰区域轮廓，`np.ndarray`, shape=(H, W), 未生成时返回`np.empty` 时间戳 + * TIMESTAMP: 时间戳字符串（HH:MM:SS），`str`。 注：H/W 由当前`json`文件中的`resolution_stitch`配置分辨率决定。 @@ -230,21 +247,33 @@ flow = sensor.read_info(sensor.info.VRAW) ## 1. `util.draw_arrows` 方法 ### 输入参数 -**frame**：背景图像 -**optical_flow**：光流场 -**threshold**：光流箭头显示阈值 -**grid_spaceing**：光流箭头间距（pixel） -**arrow_scale**：箭头大小 -**margin**：箭头最大长度/间距的比例 -**colormap**：色图，如`cv2.COLORMAP_JET` -**below_threshold_color**：RGB格式的`uint8`颜色，如(0, 0, 255) -**edge_margin**: 不显示箭头边距（pixel） + +* **frame**：背景图像 +* **optical_flow**：光流场 +* **threshold**：光流箭头显示阈值 +* **grid_spacing**：光流箭头间距（pixel） +* **arrow_scale**：箭头大小 +* **margin**：箭头最大长度/间距的比例 +* **colormap**：色图，如`cv2.COLORMAP_JET` +* **below_threshold_color**：RGB格式的`uint8`颜色，如(0, 0, 255) +* **edge_margin**: 不显示箭头边距（pixel） ### 返回 `np.ndarray` ：箭头光流可视化图像 ## 2. `util.draw_contact` 方法 -*** + +### 输入参数 + +* **image**：背景图像 +* **contour**：轮廓 +* **centroid**：触碰点 +* **point_color**：触碰点的颜色 (BGR) +* **region_color**：轮廓的颜色 (BGR) + +### 返回 +`np.ndarray` ：箭头光流可视化图像 + *** # 配置标定相关 diff --git a/build_nuitka.py b/build_nuitka.py index d052c86..323a6e6 100644 --- a/build_nuitka.py +++ b/build_nuitka.py @@ -1,42 +1,63 @@ import os import sys -import subprocess import shutil +import subprocess from pathlib import Path +from importlib.machinery import EXTENSION_SUFFIXES + def install_nuitka(): print("Checking Nuitka installation...") try: - import nuitka + import nuitka # noqa: F401 print("Nuitka is installed.") except ImportError: print("Installing Nuitka...") - subprocess.check_call([sys.executable, "-m", "pip", "install", "nuitka", "ordered-set", "zstandard"]) + subprocess.check_call( + [sys.executable, "-m", "pip", "install", "nuitka", "ordered-set", "zstandard"] + ) + + +def get_version() -> str: + init_file = Path("src/orisys/__init__.py") + if not init_file.exists(): + return "0.3.0" + for line in init_file.read_text(encoding="utf-8").splitlines(): + if line.startswith("__version__"): + return line.split("=", 1)[1].strip().strip('"').strip("'") + return "0.3.0" + + +def find_compiled_extension(output_dir: Path) -> Path: + candidates = [] + for suffix in EXTENSION_SUFFIXES: + candidates.extend(output_dir.glob(f"*{suffix}")) + if not candidates: + all_files = ", ".join(p.name for p in output_dir.iterdir()) if output_dir.exists() else "(none)" + raise RuntimeError(f"Nuitka did not produce a compiled extension. Found: {all_files}") + # Prefer the shortest filename (usually the package extension itself) + return sorted(candidates, key=lambda p: len(p.name))[0] + def build(): install_nuitka() - - version = "0.3.0" + + version = get_version() src_dir = Path("src/orisys") output_dir = Path("build_nuitka") dist_dir = Path("dist_nuitka") - - # Clean previous builds - if output_dir.exists(): shutil.rmtree(output_dir) - if dist_dir.exists(): shutil.rmtree(dist_dir) + + if output_dir.exists(): + shutil.rmtree(output_dir) + if dist_dir.exists(): + shutil.rmtree(dist_dir) output_dir.mkdir() dist_dir.mkdir() - + print("\n=== Compiling with Nuitka ===") - # Set PYTHONPATH to include src so nuitka finds the package env = os.environ.copy() env["PYTHONPATH"] = "src" + os.pathsep + env.get("PYTHONPATH", "") - - # Nuitka arguments - # --module: compile as module/package - # --include-package=orisys: include the whole package - # --output-dir: where to put result - # src/orisys: the target directory + cmd = [ sys.executable, "-m", "nuitka", "--module", @@ -44,76 +65,48 @@ def build(): "--output-dir=" + str(output_dir), "--remove-output", "--no-pyi-file", - "src/orisys" + "src/orisys", ] - print(f"Running: {' '.join(cmd)}") subprocess.check_call(cmd, env=env) - - # Find the generated .pyd file - # It should be something like orisys.cp310-win_amd64.pyd - pyd_files = list(output_dir.glob("*.pyd")) - if not pyd_files: - raise RuntimeError("Nuitka did not produce a .pyd file!") - - pyd_file = pyd_files[0] - print(f"Generated compiled extension: {pyd_file}") - + + ext_file = find_compiled_extension(output_dir) + ext_suffix = "".join(ext_file.suffixes) + print(f"Generated compiled extension: {ext_file}") + print("\n=== Preparing Wheel Structure ===") - # We want the structure: - # orisys/ - # __init__.pyd <-- renamed from orisys.pyd - # configs/ - # *.json - wheel_root = dist_dir / "wheel_root" pkg_dir = wheel_root / "orisys" pkg_dir.mkdir(parents=True) - - # Move and rename pyd - dest_pyd = pkg_dir / "__init__.pyd" - shutil.copy(pyd_file, dest_pyd) - print(f"Copied to {dest_pyd}") - - # Copy JSON configs + + dest_ext = pkg_dir / f"__init__{ext_suffix}" + shutil.copy(ext_file, dest_ext) + print(f"Copied compiled package to {dest_ext}") + configs_src = src_dir / "configs" configs_dst = pkg_dir / "configs" configs_dst.mkdir() - json_count = 0 for json_file in configs_src.glob("*.json"): shutil.copy(json_file, configs_dst) json_count += 1 print(f"Copied {json_count} JSON config files") - - # Create and copy examples folder (as part of the package) + examples_dst = pkg_dir / "examples" examples_dst.mkdir(exist_ok=True) - - # Create __init__.py for examples package - examples_init = examples_dst / "__init__.py" - with open(examples_init, 'w') as f: - f.write("# Examples for orisys package\n") - - example_files = [ - Path("example1_basic_pipeline.py"), - Path("example2_two_sensor_pipeline.py"), - ] - + (examples_dst / "__init__.py").write_text("# Examples for orisys package\n", encoding="utf-8") + example_files = [Path("example1_basic_pipeline.py"), Path("example2_two_sensor_pipeline.py")] example_count = 0 for example_file in example_files: if example_file.exists(): shutil.copy(example_file, examples_dst) example_count += 1 print(f"Copied example: {example_file.name}") - else: - print(f"Warning: Example file not found: {example_file}") print(f"Copied {example_count} example files to orisys/examples/") - - # Create setup.py for packaging + setup_py_path = dist_dir / "setup.py" setup_content = f""" -from setuptools import setup, Distribution, find_packages +from setuptools import setup, Distribution class BinaryDistribution(Distribution): def has_ext_modules(self): @@ -127,7 +120,7 @@ setup( packages=['orisys', 'orisys.examples'], package_dir={{'': 'wheel_root'}}, package_data={{ - 'orisys': ['*.pyd', 'configs/*.json'], + 'orisys': ['*{ext_suffix}', 'configs/*.json'], 'orisys.examples': ['*.py'], }}, include_package_data=True, @@ -139,26 +132,24 @@ setup( extras_require={{ "cuda12": ["cupy-cuda12x>=12.3.0"], "cuda13": ["cupy-cuda13x>=13.0.0"], - "cuda": ["cupy-cuda12x>=12.3.0"], # For backward compatibility, defaults to CUDA 12.x + "cuda": ["cupy-cuda12x>=12.3.0"], }}, distclass=BinaryDistribution, zip_safe=False, ) """ - with open(setup_py_path, "w") as f: - f.write(setup_content) - + setup_py_path.write_text(setup_content, encoding="utf-8") + print("\n=== Building Wheel ===") subprocess.check_call([sys.executable, "setup.py", "bdist_wheel"], cwd=dist_dir) - - # Find the wheel - dist_output = dist_dir / "dist" - wheels = list(dist_output.glob("*.whl")) + + wheels = list((dist_dir / "dist").glob("*.whl")) if wheels: print(f"\n[SUCCESS] Wheel created: {wheels[0]}") else: print("\n[ERROR] Wheel not found.") + if __name__ == "__main__": build() diff --git a/build_nuitka_linux.py b/build_nuitka_linux.py deleted file mode 100644 index 9c76f31..0000000 --- a/build_nuitka_linux.py +++ /dev/null @@ -1,139 +0,0 @@ -import os -import sys -import subprocess -import shutil -from pathlib import Path - -def install_nuitka(): - print("Checking Nuitka installation...") - try: - import nuitka - print("Nuitka is installed.") - except ImportError: - print("Installing Nuitka...") - subprocess.check_call([sys.executable, "-m", "pip", "install", "nuitka", "ordered-set", "zstandard"]) - -def build(): - install_nuitka() - - version = "0.3.0" - src_dir = Path("src/orisys") - output_dir = Path("build_nuitka") - dist_dir = Path("dist_nuitka") - - # Clean previous builds - if output_dir.exists(): shutil.rmtree(output_dir) - if dist_dir.exists(): shutil.rmtree(dist_dir) - output_dir.mkdir() - dist_dir.mkdir() - - print("\n=== Compiling with Nuitka ===") - # Set PYTHONPATH to include src so nuitka finds the package - env = os.environ.copy() - env["PYTHONPATH"] = "src" + os.pathsep + env.get("PYTHONPATH", "") - - # Nuitka arguments - # --module: compile as module/package - # --include-package=orisys: include the whole package - # --output-dir: where to put result - # src/orisys: the target directory - cmd = [ - sys.executable, "-m", "nuitka", - "--module", - "--include-package=orisys", - "--output-dir=" + str(output_dir), - "--remove-output", - "--no-pyi-file", - "src/orisys" - ] - - print(f"Running: {' '.join(cmd)}") - subprocess.check_call(cmd, env=env) - - # Find the generated .so file - # It should be something like orisys.cpython-310-x86_64-linux-gnu.so - so_files = list(output_dir.glob("*.so")) - if not so_files: - raise RuntimeError("Nuitka did not produce a .so file!") - - so_file = so_files[0] - print(f"Generated compiled extension: {so_file}") - - print("\n=== Preparing Wheel Structure ===") - # We want the structure: - # orisys/ - # __init__.so <-- renamed from orisys.so - # configs/ - # *.json - - wheel_root = dist_dir / "wheel_root" - pkg_dir = wheel_root / "orisys" - pkg_dir.mkdir(parents=True) - - # Move and rename so - dest_so = pkg_dir / "__init__.so" - shutil.copy(so_file, dest_so) - print(f"Copied to {dest_so}") - - # Copy JSON configs - configs_src = src_dir / "configs" - configs_dst = pkg_dir / "configs" - configs_dst.mkdir() - - json_count = 0 - for json_file in configs_src.glob("*.json"): - shutil.copy(json_file, configs_dst) - json_count += 1 - print(f"Copied {json_count} JSON config files") - - # Create setup.py for packaging - setup_py_path = dist_dir / "setup.py" - setup_content = f""" -from setuptools import setup, Distribution, find_packages - -class BinaryDistribution(Distribution): - def has_ext_modules(self): - return True - -setup( - name="orisys", - version="{version}", - description="Orisys - Optical Tactile Sensor SDK (Nuitka Compiled)", - author="Orisys", - packages=['orisys'], - package_dir={{'': 'wheel_root'}}, - package_data={{ - 'orisys': ['*.so', 'configs/*.json'], - }}, - include_package_data=True, - install_requires=[ - "numpy>=1.24.4", - "opencv-python>=4.10.0.84", - "scipy>=1.10.1", - ], - extras_require={{ - "cuda12": ["cupy-cuda12x>=12.3.0"], - "cuda13": ["cupy-cuda13x>=13.0.0"], - "cuda": ["cupy-cuda12x>=12.3.0"], # For backward compatibility, defaults to CUDA 12.x - }}, - distclass=BinaryDistribution, - zip_safe=False, -) -""" - with open(setup_py_path, "w") as f: - f.write(setup_content) - - print("\n=== Building Wheel ===") - subprocess.check_call([sys.executable, "setup.py", "bdist_wheel"], cwd=dist_dir) - - # Find the wheel - dist_output = dist_dir / "dist" - wheels = list(dist_output.glob("*.whl")) - if wheels: - print(f"\n[SUCCESS] Wheel created: {wheels[0]}") - else: - print("\n[ERROR] Wheel not found.") - -if __name__ == "__main__": - build() - diff --git a/config/ddjx01.json b/config/ddjx01.json index dc1e174..6c3ddb0 100644 --- a/config/ddjx01.json +++ b/config/ddjx01.json @@ -18,7 +18,7 @@ "nominal_ov_pos": 100, "n": [3,3], "row_tol": 0.6, - "box_size": [220, 220] + "box_size": [250, 250] }, "tracker": { "adaptive": true, diff --git a/doc/get_started.docx b/doc/get_started.docx index bbc941d..5426117 100644 Binary files a/doc/get_started.docx and b/doc/get_started.docx differ diff --git a/doc/get_started.md b/doc/get_started.md index d1993eb..b580714 100644 --- a/doc/get_started.md +++ b/doc/get_started.md @@ -1,6 +1,6 @@ 叠动科技软件使用手册 版本： 0.3.0 -编写日期： 2026年3月3日 +更新日期： 2026年3月6日 # 简介 本文件介绍如何上手和运行叠动视触觉传感器配套的开发套件。本套件是配合叠动视触觉传感器进行图像处理，计算形变、力以及接触信息。它可以从摄像头或视频文件中读取图像，处理后打印和展示处理后信息。 @@ -33,7 +33,7 @@ mkdir orisys_ws 创建后如果要更换工作区，使用 `cd `命令，例如： ``` bash -cd C:\Users\orisys\orisys_ws +cd "C:\Users\orisys\orisys_ws" ```
@@ -112,7 +112,8 @@ print(orisys.__version__) # 使用方法 ## 运行示例程序 -套件里面包含了例子展示用法，运行步骤如下： +套件里面包含了例子展示用法，运行步骤如下： + ### Windows: 1. 检查是否有其他正在连接的USB摄像头设备：打开powershell, 运行命令`Get-PnpDevice -PresentOnly | Where-Object { $_.Class -match '^Camera' }`，如果没有的话不会有输出。 2. 把传感器的USB头与电脑连接，再次运行步骤一的命令，会新增一行输出 diff --git a/doc/软件使用说明-0.3.0.pdf b/doc/软件使用说明-0.3.0.pdf index ae0ab06..620aa54 100644 Binary files a/doc/软件使用说明-0.3.0.pdf and b/doc/软件使用说明-0.3.0.pdf differ diff --git a/example1_basic_pipeline.py b/example1_basic_pipeline.py index b8b340d..d8763c6 100644 --- a/example1_basic_pipeline.py +++ b/example1_basic_pipeline.py @@ -91,7 +91,7 @@ def main(): cv2.imshow("Depth Map", cv2.applyColorMap(div_normalized, cv2.COLORMAP_JET)) #步骤 3.4: 打印资料 - print(f"fps: {fps}, fnormal: {fnormal}, fshearx: {fshearx}, fsheary: {fsheary}, depth_shape: {depth_map.shape}") + print(f"fps: {fps}, fnormal: {fnormal}, fshearx: {fshearx}, fsheary: {fsheary}, depth_shape: {depth_map.shape}, sensor.resolution_raw: {sensor.img_size_raw}") #步骤 3.5: 键盘控制 key = cv2.waitKey(1) & 0xFF diff --git a/promote_video.py b/promote_video.py index 625cc20..fb7feeb 100644 --- a/promote_video.py +++ b/promote_video.py @@ -141,11 +141,11 @@ def main(): parser.add_argument("--video","-v", type=str, default="s1.mp4", help="视频文件路径") args = parser.parse_args() - arrow_writer = cv2.VideoWriter(args.video.replace(".mp4", "_arrows.mp4"), cv2.VideoWriter_fourcc(*'mp4v'), 30, (384, 384)) + arrow_writer = cv2.VideoWriter(args.video.replace(".mp4", "_arrows.mp4"), cv2.VideoWriter_fourcc(*'mp4v'), 30, (365, 365)) pyqt_writer = cv2.VideoWriter(args.video.replace(".mp4", "_pyqt.mp4"), cv2.VideoWriter_fourcc(*'mp4v'), 30, (1600, 1200)) - depth_writer = cv2.VideoWriter(args.video.replace(".mp4", "_depth.mp4"), cv2.VideoWriter_fourcc(*'mp4v'), 30, (384, 384)) + depth_writer = cv2.VideoWriter(args.video.replace(".mp4", "_depth.mp4"), cv2.VideoWriter_fourcc(*'mp4v'), 30, (365, 365)) - sensor1 = orisys.Sensor(args.video, config_name="./config/prototype_of3.json", cal_path= "./config/dd01a_98.npy", isstitch=True, cuda=False, verbose=True) + sensor1 = orisys.Sensor(args.video, config_name="./config/prototype_of3_0306.json", cal_path= "./config/dd01a_0306.npy", isstitch=True, cuda=False, verbose=True) start_time = time.time() is_save = False @@ -165,7 +165,7 @@ def main(): print(fps,fn,fx,fy) # 显示向量场 black_img = np.zeros_like(sensor1.img) - arrows = orisys.util.draw_arrows(black_img, flow, threshold=2, grid_spacing=20, arrow_scale=0.5) + arrows = orisys.util.draw_arrows(sensor1.img, flow, threshold=2, grid_spacing=20, arrow_scale=0.5) cv2.imshow("Deformation", arrows) # 显示深度图 @@ -212,7 +212,7 @@ def main(): arrow_writer.write(arrows) pyqt_writer.write(pyqt_img_resized) - depth_frame = cv2.resize(depth_colored, (384, 384), interpolation=cv2.INTER_LINEAR) + depth_frame = cv2.resize(depth_colored, (365, 365), interpolation=cv2.INTER_LINEAR) depth_writer.write(depth_frame) # depth_writer.write(depth_map) diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..39aeead --- /dev/null +++ b/pytest.ini @@ -0,0 +1,5 @@ +[pytest] +testpaths = tests +python_files = test_*.py +markers = + slow: long-running tests for release validation diff --git a/src/orisys/configs/ddjx01.json b/src/orisys/configs/ddjx01.json new file mode 100644 index 0000000..dc1e174 --- /dev/null +++ b/src/orisys/configs/ddjx01.json @@ -0,0 +1,43 @@ +{ + "name": "prototype", + "version": "1.0", + "camera": { + "resolution_raw": [1920, 1080], + "fps": 30, + "units": 15, + "units_used": 9, + "resolution_stitch": [365, 365], + "stitch_res_list": [365,263,123,63] + }, + "stitcher": { + "min_area_frac": 0.01, + "max_area_frac": 0.04, + "max_aspect_ratio": 100, + "index": [1,2,3,6,7,8,11,12,13], + "margin": 10, + "nominal_ov_pos": 100, + "n": [3,3], + "row_tol": 0.6, + "box_size": [220, 220] + }, + "tracker": { + "adaptive": true, + "lower_threshold": 12.0, + "upper_threshold": 18.0, + "loop_closure_interval": 1, + "algorithm": "dis", + "dis_preset": "medium", + "dis_mode": 3, + "algorithm_list": ["raft", "dis"] + }, + "contact": { + "min_contour_area": 3000, + "flow_threshold":1, + "div_threshold":0.02, + "expansion_threshold": 0.01, + "kernel_size": [5,5], + "denoise_sigma": 15 + } +} + + diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..c6da59a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,116 @@ +import os +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] +SRC_ROOT = REPO_ROOT / "src" +if str(SRC_ROOT) not in sys.path: + sys.path.insert(0, str(SRC_ROOT)) + +from orisys import Sensor +from orisys import configs as orisys_configs +from orisys.backend_availability import check_cupy_available, check_opencl_available + + +def _resolve_test_video() -> Path: + env_video = os.getenv("ORISYS_TEST_VIDEO") + if env_video: + raw = Path(env_video) + candidates = [ + raw, # as provided (absolute or cwd-relative) + REPO_ROOT / raw, # repo-relative + ] + # Common typo fallback: "test/..." -> "tests/..." + if env_video.startswith("test/") or env_video.startswith("test\\"): + fixed = Path("tests", *Path(env_video).parts[1:]) + candidates.append(REPO_ROOT / fixed) + + for candidate in candidates: + if candidate.exists(): + return candidate + return raw + + # Prefer tests/ fixture clip; fallback kept by caller exists check. + return REPO_ROOT / "tests" / "idle_1min.mp4" + + +@pytest.fixture(scope="session") +def test_video_path() -> Path: + path = _resolve_test_video() + if not path.exists(): + pytest.skip( + "Test video not found. Set ORISYS_TEST_VIDEO to a valid video path." + ) + return path + + +@pytest.fixture(scope="session") +def test_config_name() -> str: + requested = os.getenv("ORISYS_TEST_CONFIG") + + def _is_usable_config(name: str) -> bool: + cfg = orisys_configs.load_config(name) + return "box_size" in cfg.get("stitcher", {}) + + if requested: + if not _is_usable_config(requested): + raise RuntimeError( + f"ORISYS_TEST_CONFIG='{requested}' is missing required key " + "'stitcher.box_size'. Use a config like 'ddjx01' or provide a full path." + ) + return requested + + # Auto-pick a stable config for tests (avoid prototype variants missing keys). + for name in orisys_configs.list_config(): + try: + if _is_usable_config(name): + return name + except Exception: + continue + + raise RuntimeError( + "No usable config found for tests (missing 'stitcher.box_size'). " + "Set ORISYS_TEST_CONFIG to a valid config JSON/path." + ) + + +@pytest.fixture(scope="session") +def available_backends() -> list[str]: + backends = ["cpu"] + if check_cupy_available(): + backends.append("cupy") + if check_opencl_available(): + backends.append("opencl") + return backends + + +def make_sensor( + video_path: Path, + config_name: str = "prototype", + backend: str = "cpu", + verbose: bool = False, +) -> Sensor: + use_cuda = backend in ("cupy", "opencl") + return Sensor( + vid_src=str(video_path), + cuda=use_cuda, + isstitch=True, + verbose=verbose, + config_name=config_name, + backend=backend, + ) + + +@pytest.fixture +def sensor_factory(test_video_path, test_config_name): + def _factory(backend: str = "cpu", verbose: bool = False): + return make_sensor( + test_video_path, + config_name=test_config_name, + backend=backend, + verbose=verbose, + ) + + return _factory diff --git a/tests/note.txt b/tests/note.txt new file mode 100644 index 0000000..0aae562 --- /dev/null +++ b/tests/note.txt @@ -0,0 +1,6 @@ +# set +set ORISYS_TEST_VIDEO=test\idle_1min.mp4 +set ORISYS_TEST_CONFIG=ddjx01 + +# run command +pytest -q -rs \ No newline at end of file diff --git a/tests/test_backends.py b/tests/test_backends.py new file mode 100644 index 0000000..9b3adeb --- /dev/null +++ b/tests/test_backends.py @@ -0,0 +1,66 @@ +import os + +import numpy as np + + +def _collect_vnormal_sequence(sensor, frames: int) -> np.ndarray: + seq = [] + for _ in range(frames): + img = sensor.get_img() + assert img is not None + sensor.compute_deformation(check_motion=False) + (vnormal,) = sensor.read_info(sensor.info.VNORMAL) + seq.append(vnormal.copy()) + return np.asarray(seq) + + +def test_available_backends_smoke(sensor_factory, available_backends): + for backend in available_backends: + sensor = sensor_factory(backend=backend) + try: + img = sensor.get_img() + assert img is not None + sensor.compute_deformation(check_motion=False) + flow, vnormal, vshear = sensor.read_info( + sensor.info.VRAW, + sensor.info.VNORMAL, + sensor.info.VSHEAR, + ) + assert flow.shape == vnormal.shape == vshear.shape + assert flow.shape[-1] == 2 + assert np.isfinite(flow).all() + assert np.isfinite(vnormal).all() + assert np.isfinite(vshear).all() + finally: + sensor.disconnect() + + +def test_backend_parity_against_cpu(sensor_factory, available_backends): + if len(available_backends) <= 1: + return + + compare_frames = int(os.getenv("ORISYS_PARITY_FRAMES", "30")) + warmup = int(os.getenv("ORISYS_PARITY_WARMUP", "5")) + tol = float(os.getenv("ORISYS_PARITY_REL_TOL", "2.0")) + + cpu_sensor = sensor_factory(backend="cpu") + try: + cpu_seq = _collect_vnormal_sequence(cpu_sensor, compare_frames) + finally: + cpu_sensor.disconnect() + + cpu_ref = cpu_seq[warmup:].mean(axis=0) + cpu_scale = float(np.mean(np.abs(cpu_ref))) + 1e-6 + + for backend in available_backends: + if backend == "cpu": + continue + sensor = sensor_factory(backend=backend) + try: + seq = _collect_vnormal_sequence(sensor, compare_frames) + finally: + sensor.disconnect() + + cand_ref = seq[warmup:].mean(axis=0) + rel = float(np.mean(np.abs(cand_ref - cpu_ref)) / cpu_scale) + assert rel < tol, f"{backend} parity drift too high: relative_error={rel:.4f}, tol={tol}" diff --git a/tests/test_errors_and_cleanup.py b/tests/test_errors_and_cleanup.py new file mode 100644 index 0000000..e5963e9 --- /dev/null +++ b/tests/test_errors_and_cleanup.py @@ -0,0 +1,32 @@ +from pathlib import Path + +import pytest + +from orisys import Sensor + + +def test_compute_deformation_requires_get_img(sensor_factory): + sensor = sensor_factory(backend="cpu") + try: + with pytest.raises(Exception): + sensor.compute_deformation(check_motion=False) + finally: + sensor.disconnect() + + +def test_invalid_video_source_raises(): + bad_path = Path("this_video_file_should_not_exist_123456.mp4") + with pytest.raises(Exception): + sensor = Sensor(vid_src=str(bad_path), verbose=False, backend="cpu") + sensor.disconnect() + + +def test_repeated_connect_disconnect(sensor_factory): + for _ in range(3): + sensor = sensor_factory(backend="cpu") + try: + img = sensor.get_img() + assert img is not None + sensor.compute_deformation(check_motion=False) + finally: + sensor.disconnect() diff --git a/tests/test_longrun_stability.py b/tests/test_longrun_stability.py new file mode 100644 index 0000000..f8eb3c6 --- /dev/null +++ b/tests/test_longrun_stability.py @@ -0,0 +1,41 @@ +import os + +import numpy as np +import pytest + + +@pytest.mark.slow +def test_vnormal_longrun_no_outlier_jumps(sensor_factory, available_backends): + preferred = os.getenv("ORISYS_LONGRUN_BACKEND", "cupy") + backend = preferred if preferred in available_backends else "cpu" + + frames = int(os.getenv("ORISYS_LONGRUN_FRAMES", "1200")) + ratio_limit = float(os.getenv("ORISYS_LONGRUN_MAX_P99_P95_RATIO", "6.0")) + + sensor = sensor_factory(backend=backend) + deltas = [] + prev = None + try: + for _ in range(frames): + img = sensor.get_img() + assert img is not None + sensor.compute_deformation(check_motion=False) + (vnormal,) = sensor.read_info(sensor.info.VNORMAL) + assert np.isfinite(vnormal).all() + + if prev is not None: + deltas.append(float(np.mean(np.abs(vnormal - prev)))) + prev = vnormal.copy() + finally: + sensor.disconnect() + + if len(deltas) < 30: + pytest.skip("Not enough frames to evaluate stability.") + + p95 = float(np.percentile(deltas, 95)) + p99 = float(np.percentile(deltas, 99)) + ratio = p99 / (p95 + 1e-9) + assert ratio < ratio_limit, ( + f"Long-run jump risk detected on {backend}: p99/p95={ratio:.3f} " + f"(limit={ratio_limit})." + ) diff --git a/tests/test_smoke_api.py b/tests/test_smoke_api.py new file mode 100644 index 0000000..f2649f9 --- /dev/null +++ b/tests/test_smoke_api.py @@ -0,0 +1,47 @@ +import numpy as np + +import orisys + + +def test_import_version(): + assert isinstance(orisys.__version__, str) + assert len(orisys.__version__) > 0 + + +def test_sensor_smoke_pipeline(sensor_factory): + sensor = sensor_factory(backend="cpu") + try: + img = sensor.get_img() + assert img is not None + assert isinstance(img, np.ndarray) + assert img.ndim in (2, 3) + + result = sensor.compute_deformation(check_motion=False, threshold=6) + assert result is None + + flow, vnormal, vshear = sensor.read_info( + sensor.info.VRAW, + sensor.info.VNORMAL, + sensor.info.VSHEAR, + ) + assert flow.ndim == 3 and flow.shape[-1] == 2 + assert vnormal.shape == flow.shape + assert vshear.shape == flow.shape + assert np.isfinite(flow).all() + assert np.isfinite(vnormal).all() + assert np.isfinite(vshear).all() + + fnormal, fsx, fsy = sensor.read_info( + sensor.info.FNORMAL, + sensor.info.FSHEARX, + sensor.info.FSHEARY, + ) + assert np.isfinite([fnormal, fsx, fsy]).all() + + contact = sensor.compute_contact() + assert isinstance(contact, dict) + assert "centroid_found" in contact + assert "divergence_map" in contact + assert isinstance(contact["divergence_map"], np.ndarray) + finally: + sensor.disconnect()