orisys-tech/orisys-sdk
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update test and build scripts

Browse Source
This commit is contained in:
ykcheungab
2026-03-09 13:33:18 +08:00
parent 36118ab9cd
commit fa76d66162
19 changed files with 472 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
API文档-0.3.0.pdf
BIN
View File
Binary file not shown.
Readme.docx
BIN
View File
Binary file not shown.
Readme.md
+44 -15
View File
@@ -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` ：箭头光流可视化图像
***
# 配置标定相关
build_nuitka.py
+61 -70
View File
@@ -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()
build_nuitka_linux.py
-139
View File
@@ -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()
config/ddjx01.json
+1 -1
View File
@@ -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,
doc/get_started.docx
BIN
View File
Binary file not shown.
doc/get_started.md
+4 -3
View File
@@ -1,6 +1,6 @@
叠动科技软件使用手册
版本： 0.3.0
编写日期： 2026年3月3日
更新日期： 2026年3月6日
# 简介
本文件介绍如何上手和运行叠动视触觉传感器配套的开发套件。本套件是配合叠动视触觉传感器进行图像处理，计算形变、力以及接触信息。它可以从摄像头或视频文件中读取图像，处理后打印和展示处理后信息。
@@ -33,7 +33,7 @@ mkdir orisys_ws
创建后如果要更换工作区，使用 `cd <path>`命令，例如：
``` bash
cd C:\Users\orisys\orisys_ws
cd "C:\Users\orisys\orisys_ws"
```
<div style="page-break-after: always;"></div>
@@ -112,7 +112,8 @@ print(orisys.__version__)
# 使用方法
## 运行示例程序
套件里面包含了例子展示用法，运行步骤如下：
套件里面包含了例子展示用法，运行步骤如下：
### Windows:
1. 检查是否有其他正在连接的USB摄像头设备：打开powershell, 运行命令`Get-PnpDevice -PresentOnly | Where-Object { $_.Class -match '^Camera' }`，如果没有的话不会有输出。
2. 把传感器的USB头与电脑连接，再次运行步骤一的命令，会新增一行输出
doc/软件使用说明-0.3.0.pdf
BIN
View File
Binary file not shown.
example1_basic_pipeline.py
+1 -1
View File
@@ -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
promote_video.py
+5 -5
View File
@@ -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)
pytest.ini
+5
View File
@@ -0,0 +1,5 @@
[pytest]
testpaths = tests
python_files = test_*.py
markers =
slow: long-running tests for release validation
src/orisys/configs/ddjx01.json
+43
View File
@@ -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
}
}
tests/conftest.py
+116
View File
@@ -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
tests/note.txt
+6
View File
@@ -0,0 +1,6 @@
# set
set ORISYS_TEST_VIDEO=test\idle_1min.mp4
set ORISYS_TEST_CONFIG=ddjx01
# run command
pytest -q -rs
tests/test_backends.py
+66
View File
@@ -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}"
tests/test_errors_and_cleanup.py
+32
View File
@@ -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()
tests/test_longrun_stability.py
+41
View File
@@ -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})."
)
tests/test_smoke_api.py
+47
View File
@@ -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()