diff --git a/.gitignore b/.gitignore index ed52d69..01363db 100644 --- a/.gitignore +++ b/.gitignore @@ -12,7 +12,8 @@ orisys.egg-info/ /build_nuitka /data/ /models/ -/doc/.quarto/ -*.npy -*.png -SDK/ \ No newline at end of file +SDK/ + +/patch_generation/exported_data/ +/docs/exported_data/ +/docs/.quarto/ \ No newline at end of file diff --git a/RAFT_test.py b/RAFT_test.py deleted file mode 100644 index 416a4cf..0000000 --- a/RAFT_test.py +++ /dev/null @@ -1,156 +0,0 @@ -import os -os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" -import cv2 -import torch -import torchvision.transforms.functional as F -import torchvision.transforms as T -from torchvision.models.optical_flow import raft_large, raft_small -import numpy as np -import time -import orisys - -def preprocess(batch, resize=None): - """ - Preprocess image batch for RAFT model. - - Args: - batch: Tensor of shape (N, C, H, W) - resize: Optional tuple (height, width) to resize. If None, keeps original size. - - Returns: - Preprocessed tensor - """ - transform_list = [ - T.ConvertImageDtype(torch.float32), - T.Normalize(mean=0.5, std=0.5), # map [0, 1] into [-1, 1] - ] - - if resize is not None: - transform_list.append(T.Resize(size=resize)) - - transforms = T.Compose(transform_list) - batch = transforms(batch) - return batch - -def setup_simple_raft(): - """ - Setup a simpler version using available models - """ - model = raft_small(pretrained=True, progress=False).to('cuda') - model = model.eval() - # Alternative: Use LiteFlowNet (lighter alternative) - print("RAFT loaded successfully") - return model - -def track_raft(prev_frame, frame, model, device): - - gray = frame.copy() - start_time = time.time() - prev_tensor = torch.from_numpy(prev_frame.transpose(2, 0, 1).astype(np.float32) / 255.0) - curr_tensor = torch.from_numpy(gray.transpose(2, 0, 1).astype(np.float32) / 255.0) - prev_tensor = preprocess(prev_tensor.unsqueeze(0), resize=None).to(device) - curr_tensor = preprocess(curr_tensor.unsqueeze(0), resize=None).to(device) - t0 = time.perf_counter() - with torch.no_grad(): - flow = model(prev_tensor, curr_tensor) - predicted_flows = flow[-1] - t1 = time.perf_counter() - print(f"RAFT time: {t1 - t0}") - flow_np = predicted_flows[0].cpu().numpy() # or .squeeze(0) - - # Transpose to (H, W, 2) format expected by draw_arrows - flow_np = flow_np.transpose(1, 2, 0) # (2, H, W) -> (H, W, 2) - - # torch.onnx.export( - # model, # model to export - # (prev_tensor, curr_tensor), # inputs of the model, - # "my_model.onnx", # filename of the ONNX model - # input_names=["prev_tensor", "curr_tensor"], # Rename inputs for the ONNX model - # dynamo=True # True or False to select the exporter to use - # ) - # print("ONNX model exported successfully") - - return flow_np - -def main_simple(): - # Initialize camera - # sensor = orisys.Sensor('s1.mp4', config_name="./config/prototype_of3.json", cal_path= "./config/cal.npy", isstitch=True, cuda=True, verbose=True) - model = setup_simple_raft() - device = "cuda" if torch.cuda.is_available() else "cpu" - print(f"Using device: {device}") - cap = cv2.VideoCapture("s1.mp4") - cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1920) - cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 1080) - - prev_gray = None - fps_buffer = [] - - - while True: - - ret, frame = cap.read() - if not ret: - break - dx = 300 - dy = 0 - frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) - frame1 = frame[420:660, 840:1080, :] - # frame2 = frame[420+dy:660+dy, 840+dx:1080+dx, :] - # frame3 = frame[420-dy:660-dy, 840-dx:1080-dx, :] - - # frame = np.concatenate([frame3, frame1, frame2], axis=1) - frame = frame1 - # cv2.imshow("frame", frame3) - # cv2.waitKey(0) - if prev_gray is None: - prev_gray = frame.copy() - gray = frame - # Compute optical flow - start_time = time.time() - flow_np = track_raft(prev_gray, gray, model, device) - fps = 1.0 / (time.time() - start_time) - fps_buffer.append(fps) - if len(fps_buffer) > 30: - fps_buffer.pop(0) - - avg_fps = np.mean(fps_buffer) - print(f"avg_fps: {avg_fps}") - - - # # Remove batch dimension: (1, 2, H, W) -> (2, H, W) - # flow_np = predicted_flows[0].cpu().numpy() # or .squeeze(0) - - # # Transpose to (H, W, 2) format expected by draw_arrows - # flow_np = flow_np.transpose(1, 2, 0) # (2, H, W) -> (H, W, 2) - - # Flow is already in pixel coordinates, no denormalization needed - # Keep as float32 for draw_arrows (it can handle float values) - print(f"Flow numpy shape: {flow_np.shape}, dtype: {flow_np.dtype}") - print(f"Flow range: x=[{flow_np[:,:,0].min():.2f}, {flow_np[:,:,0].max():.2f}], " - f"y=[{flow_np[:,:,1].min():.2f}, {flow_np[:,:,1].max():.2f}]") - - arrows = orisys.util.draw_arrows( - gray, - flow_np, - threshold=2, - grid_spacing=40, - arrow_scale=1.0 - ) - - arrows = cv2.resize(arrows, (400, 400)) - cv2.imshow('Optical Flow', arrows) - - # prev_gray = gray.copy() - - #步骤 3.5: 键盘控制 - key = cv2.waitKey(1) & 0xFF - if key == ord("q"): - print("\n退出程序...") - break - if key == ord("r"): - # 重置光流追踪器（当追踪漂移时使用） - prev_gray = None - print("传感器已重置") - -if __name__ == "__main__": - main_simple() \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..a9ac9ce --- /dev/null +++ b/README.md @@ -0,0 +1,85 @@ +# Orisys SDK + +Orisys SDK is a Python toolkit for optical tactile sensing. It provides image capture, +stitching, dense optical flow, Helmholtz-Hodge decomposition, contact/depth estimation, +and several runnable examples for camera and video-based workflows. + +## Repository Layout + +- `src/orisys/`: installable SDK package +- `config/`: default and lab/runtime JSON configurations +- `examples/`: user-facing example scripts +- `scripts/`: maintenance, calibration, packaging, and demo scripts +- `tools/`: non-package tooling such as patch-generation utilities +- `tests/`: automated pytest suite +- `docs/`: usage and API documentation + +## Quick Start + +### 1. Create an environment + +```bash +conda create -n orisys python=3.10 +conda activate orisys +``` + +### 2. Install the SDK in editable mode + +```bash +pip install -e . +``` + +Optional GPU extras: + +```bash +pip install -e .[cuda12] +``` + +or + +```bash +pip install -e .[cuda13] +``` + +### 3. Run tests + +```bash +pytest +``` + +### 4. Run an example + +```bash +python examples/basic_pipeline.py -v 0 -c ./config/ddjx01.json +``` + +## Configuration + +The canonical configuration files now live in the repo-level `config/` folder. +The package loader `orisys.configs` resolves configs from: + +1. an explicit JSON file path +2. a config file stem under `config/` +3. a backward-compatible JSON `"name"` match, such as `"prototype"` + +Examples: + +```python +import orisys + +sensor = orisys.Sensor(0, config_name="ddjx01") +cfg = orisys.export_config("ddjx01", "./config/exported_config.json") +``` + +## Examples and Tools + +- Example entry points: `examples/README.md` +- User documentation: `docs/get_started.md` +- Packaging helper: `scripts/build_nuitka.py` +- Specialized utilities: `tools/` + +## Notes for Collaborators + +- Generated build outputs such as `build/`, `dist/`, and `SDK/` are intentionally ignored. +- Large or generated tooling outputs should stay under ignored output folders rather than be committed. +- If you add new images or arrays that should be versioned, review `.gitignore` first. diff --git a/calibration_data_collection.py b/calibration_data_collection.py deleted file mode 100644 index 0ed3600..0000000 --- a/calibration_data_collection.py +++ /dev/null @@ -1,195 +0,0 @@ -import os -os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" -import cv2 -import orisys -import numpy as np -import argparse - -def main(): - parser = argparse.ArgumentParser(description='Calibration Data Collection') - parser.add_argument('-i','--vid_source', type=str, default=0, help='Video Source') - parser.add_argument('-id','--device_id', type=int, default=0, help='Device ID') - - args = parser.parse_args() - device_id = args.device_id - vid_source = args.vid_source - sample_name = f"dd01a_{device_id}" #dd01 (平面款), dd01a (平面款 手搓), dd02 (手指款) - test_type = "normal" # shearX, shearY, normal - print("\n") - print(f"Collecting data for device ID: {device_id} on test type: {test_type}, video source: {vid_source}") - print("\n") - sensor = orisys.Sensor(vid_source, config_name="./config/prototype_of3.json", cal_path= f"./config/{sample_name}.npy", cuda=False, verbose=True) - - print(f"\n按 'q' 退出程序，按 'r' 重置光流追踪器，按 's' 开始保存，按 'e' 停止保存，按 'j' 减小力，按 'l' 增大力") - force = 0 # kgf - force_step = 0.1 # kgf - force_min = 0 # kgf - force_max = 3 # kgf - - is_save = False - save_mode = ["raw", "stitched", "arrows", "contact", "depth"] - save_mode_index = 0 - frame_count = 0 - writer = None - save_path = None - - while True: - sensor.get_img() - - if is_save and writer is not None and sensor.frame is not None: - frame_count += 1 - if save_mode_index == 0: - save_img = sensor.frame - elif save_mode_index == 1: - save_img = sensor.img - elif save_mode_index == 2: - save_img = arrows - elif save_mode_index == 3: - save_img = image_with_foe - elif save_mode_index == 4: - save_img = div_normalized - - # Convert grayscale (1 channel) to BGR (3 channels) if needed - if len(save_img.shape) == 2 or (len(save_img.shape) == 3 and save_img.shape[2] == 1): - if len(save_img.shape) == 3: - save_img = save_img.squeeze(2) - save_img = cv2.cvtColor(save_img, cv2.COLOR_GRAY2BGR) - - writer.write(save_img) - - sensor.compute_deformation() - sensor.compute_contact() - - fps, flow, vnormal, img, contour, centroid, depth_map, fnormal, fshearx, fsheary = sensor.read_info( - sensor.info.FPS, - sensor.info.VRAW, - sensor.info.VNORMAL, - sensor.info.IMG, - sensor.info.CONTOUR, - sensor.info.CENTROID, - sensor.info.DEPTH, - sensor.info.FNORMAL, - sensor.info.FSHEARX, - sensor.info.FSHEARY - ) - - arrows = orisys.util.draw_arrows( - img, - flow, - threshold=4, - grid_spacing=10, - arrow_scale=1.0 - ) - cv2.putText(arrows, f"FPS: {fps:.0f}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2) - cv2.imshow("Deformation", arrows) - - image_with_foe = orisys.util.draw_contact(img, contour, centroid) - cv2.imshow("Contact Centroid", image_with_foe) - - div_abs = depth_map - if div_abs.max() > div_abs.min(): - div_normalized = ((div_abs - div_abs.min()) / (div_abs.max() - div_abs.min()) * 255).astype(np.uint8) - else: - div_normalized = np.zeros_like(div_abs, dtype=np.uint8) - - # 应用颜色映射（使用JET：蓝色=低散度，红色=高散度） - cv2.imshow("Depth Map", cv2.applyColorMap(div_normalized, cv2.COLORMAP_JET)) - - key = cv2.waitKey(1) & 0xFF - - if key == ord("s"): - save_name = f"{sample_name}_{test_type}_{force:.1f}kgf" - save_path = f"./data/{sample_name}/{test_type}/{save_name}.mp4" - # Ensure directory exists - os.makedirs(os.path.dirname(save_path), exist_ok=True) - # Get frame dimensions from the actual frame - save_img = None - if save_mode_index == 0: - save_img = sensor.frame - elif save_mode_index == 1: - save_img = sensor.img - elif save_mode_index == 2: - save_img = arrows - elif save_mode_index == 3: - save_img = image_with_foe - elif save_mode_index == 4: - save_img = div_normalized - - if save_img is not None: - # Convert grayscale (1 channel) to BGR (3 channels) if needed for dimension check - temp_img = save_img.copy() - if len(temp_img.shape) == 2 or (len(temp_img.shape) == 3 and temp_img.shape[2] == 1): - if len(temp_img.shape) == 3: - temp_img = temp_img.squeeze(2) - temp_img = cv2.cvtColor(temp_img, cv2.COLOR_GRAY2BGR) - - frame_height, frame_width = temp_img.shape[:2] - writer = cv2.VideoWriter(save_path, cv2.VideoWriter_fourcc(*'mp4v'), 30, (frame_width, frame_height)) - if writer.isOpened(): - is_save = True - print(f"Start writing to {save_path}") - else: - print(f"Failed to open video writer for {save_path}") - writer = None - else: - print("No frame available to determine video dimensions") - - if key == ord("e"): - is_save = False - if writer is not None: - writer.release() - writer = None - if save_path is not None: - print(f"Stop writing to {save_path}, total frames: {frame_count}") - else: - print(f"Stop writing, total frames: {frame_count}") - frame_count = 0 - - if key == ord("p"): - print(f"Normal Force: {fnormal:.1f} kgf, Shear ForceX: {fshearx:.1f} kgf, Shear ForceY: {fsheary:.1f} kgf") - - if key == ord("i"): - save_mode_index += 1 - if save_mode_index >= len(save_mode): - save_mode_index = 0 - print(f"Save mode: {save_mode[save_mode_index]}") - - if key == ord("k"): - save_mode_index -= 1 - if save_mode_index < 0: - save_mode_index = len(save_mode) - 1 - print(f"Save mode: {save_mode[save_mode_index]}") - if key == ord("j"): - force -= force_step - if force < force_min: - force = force_min - print(f"Force: {force:.1f} kgf") - if key == ord("l"): - force += force_step - if force > force_max: - force = force_max - print(f"Force: {force:.1f} kgf") - - - if key == ord("q"): - print("\n退出程序...") - if writer is not None: - writer.release() - writer = None - break - if key == ord("r"): - # 重置光流追踪器（当追踪漂移时使用） - sensor.reset() - print("传感器已重置") - - - # ========================================================================= - # 4. 清理资源 - # ========================================================================= - if writer is not None: - writer.release() - sensor.disconnect() # 释放视频源和相关资源 - - -if __name__ == '__main__': - main() diff --git a/config/compound_eye.png b/config/compound_eye.png new file mode 100644 index 0000000..1a16011 Binary files /dev/null and b/config/compound_eye.png differ diff --git a/docs/API文档-0.3.0.pdf b/docs/API文档-0.3.0.pdf deleted file mode 100644 index acf4348..0000000 Binary files a/docs/API文档-0.3.0.pdf and /dev/null differ diff --git a/docs/Readme.docx b/docs/Readme.docx deleted file mode 100644 index 11d0506..0000000 Binary files a/docs/Readme.docx and /dev/null differ diff --git a/docs/Readme.md b/docs/api.md similarity index 95% rename from docs/Readme.md rename to docs/api.md index 665bc80..559664f 100644 --- a/docs/Readme.md +++ b/docs/api.md @@ -72,8 +72,8 @@ conda install conda-forge::pyvkfft ### 输入参数 * **vid_src** (`int | str`, 必填): 传感器 ID、相机编号或视频路径。默认为 `0`。 -* **config_name** (`str`，可选): 传感器配置`json`文件路径，用于加载对应的参数配置，默认为 `./config/ddjx01`。 -* * **cal_path** (`str`，可选，已弃用): 该参数已弃用，当前版本中不再使用，传入后也会被忽略。为兼容旧代码暂时保留。: 复眼拼接校准文件路径，默认为 `./config/cal.npy`。有多个传感器同时运行时建议传感器类创建时建议填写，避免校准文件覆写。在首次运行时程序会在找不到文件时自动创建。 +* **config_name** (`str`，可选): 传感器配置文件名称或 `json` 文件路径，用于加载对应的参数配置；例如 `ddjx01` 或 `./config/ddjx01.json`。 +* **cal_path** (`str`，可选，已弃用): 该参数仅为兼容旧代码保留，当前版本中不再使用，传入后也会被忽略。 * **backend** (`str`，可选): 光流场解耦计算后端，默认为 `auto`，可选`cupy`,`opencl`,`cpu`。 * **verbose** (`bool`，可选): 是否输出初始化及运行过程中的详细日志信息，默认为 `False`。 diff --git a/docs/get_started.docx b/docs/get_started.docx deleted file mode 100644 index 013cbab..0000000 Binary files a/docs/get_started.docx and /dev/null differ diff --git a/docs/get_started.md b/docs/get_started.md index 053ce13..ba3107d 100644 --- a/docs/get_started.md +++ b/docs/get_started.md @@ -120,7 +120,7 @@ print(orisys.__version__) 3. 如果第一步没有输出，传感器的ID就是“0”，如果第一步有一个设备，传感器的ID就是“1”，如此类推。 4. 在确定了传感器的ID（下方例子假设为“0”）后，打开Anaconda Prompt, 切换到工作路径，然后在命令行中输入： ``` bash - python example1_basic_pipeline.py -v 0 + python examples/basic_pipeline.py -v 0 ``` 5. 程序会运行，并自动打开可视化窗口和打印资料，在窗口上按键盘”q”键可以停止运行，按”r”键可以重设数据 @@ -130,7 +130,7 @@ print(orisys.__version__) 3. 根据新增的输出，例如”/dev/video0”，确定传感器的ID为“0”, 如果是”/dev/video3”的话ID为“3” ，如此类推。 4. 在确定了传感器的ID（下方例子假设为“0”）后，打开终端, 激活相应的python环境，切换到工作路径，然后在命令行中输入： ``` bash - python example1_basic_pipeline.py -v 0 + python examples/basic_pipeline.py -v 0 ``` 5. 程序会运行，并自动打开可视化窗口和打印资料，在窗口上按键盘”q”键可以停止运行，按”r”键可以重设数据 diff --git a/res/orisys_logo.png b/docs/images/orisys_logo.png similarity index 100% rename from res/orisys_logo.png rename to docs/images/orisys_logo.png diff --git a/res/orisys_logo_notext.png b/docs/images/orisys_logo_notext.png similarity index 100% rename from res/orisys_logo_notext.png rename to docs/images/orisys_logo_notext.png diff --git a/docs/软件使用说明-0.3.0.pdf b/docs/软件使用说明-0.3.0.pdf deleted file mode 100644 index 620aa54..0000000 Binary files a/docs/软件使用说明-0.3.0.pdf and /dev/null differ diff --git a/examples/README.md b/examples/README.md new file mode 100644 index 0000000..b05b54e --- /dev/null +++ b/examples/README.md @@ -0,0 +1,50 @@ +# Examples + +This folder contains user-facing example scripts for the Orisys SDK. + +## Files + +- `basic_pipeline.py`: single-sensor basic processing and visualization flow +- `two_sensor_pipeline.py`: dual-sensor multiprocessing example +- `plot_data.py`: real-time plotting example with PyQtGraph + +## Typical usage + +Run examples from the repository root so relative paths such as `./config/ddjx01.json` +resolve correctly. + +### Basic pipeline + +```bash +python examples/basic_pipeline.py -v 0 -c ./config/ddjx01.json +``` + +### Two-sensor pipeline + +```bash +python examples/two_sensor_pipeline.py +``` + +### Real-time plotting + +```bash +python examples/plot_data.py -v 0 -c ./config/ddjx01.json +``` + +## Requirements + +- Python environment with the SDK dependencies installed +- a supported camera or test video +- valid JSON config in `./config/` + +Some examples also require: + +- `pyqtgraph` +- `PyQt5` +- multiple cameras for multi-sensor demos + +## Notes + +- Most examples assume a tactile sensor camera is connected. +- Press `q` in the display window to quit where applicable. +- If your camera index is not `0`, change `-v` accordingly. diff --git a/example1_basic_pipeline.py b/examples/basic_pipeline.py similarity index 93% rename from example1_basic_pipeline.py rename to examples/basic_pipeline.py index 8faaa95..0cf40b2 100644 --- a/example1_basic_pipeline.py +++ b/examples/basic_pipeline.py @@ -42,10 +42,10 @@ def main(): # 步骤 2：数据处理 # 步骤 2.1：获取并拼接图像（必需） sensor.get_img() - # 步骤 2.2：计算形变场（必需） - is_touched = sensor.compute_deformation(check_motion=True, threshold=6) - # 步骤 2.3：检测到接触后再计算接触区域 - if is_touched: + # 步骤 2.2：计算形变场（必需）；返回值表示当前帧是否检测到接触/运动 + is_contact_now = sensor.compute_deformation(check_motion=True, threshold=6) + # 步骤 2.3：仅在当前帧检测到接触/运动时计算接触区域 + if is_contact_now: sensor.compute_contact() diff --git a/example3_plot_data.py b/examples/plot_data.py similarity index 98% rename from example3_plot_data.py rename to examples/plot_data.py index ca28bb5..49e6fd4 100644 --- a/example3_plot_data.py +++ b/examples/plot_data.py @@ -149,7 +149,7 @@ def main(): writer.write(save_img) - sensor.compute_deformation(check_motion=False) # 计算形变与分解结果 + sensor.compute_deformation(check_motion=True, threshold=0) # 计算形变与分解结果 # 读取信息： # 向量场：VRAW、VNORMAL、VSHEAR -> 原始 / 法向 / 切向 # 标量：FNORMAL、FSHEARX、FSHEARY -> 法向力 / X 向切向力 / Y 向切向力 diff --git a/example2_two_sensor_pipeline.py b/examples/two_sensor_pipeline.py similarity index 100% rename from example2_two_sensor_pipeline.py rename to examples/two_sensor_pipeline.py diff --git a/patch_generation/color_pattern.png b/patch_generation/color_pattern.png deleted file mode 100644 index 221c4aa..0000000 Binary files a/patch_generation/color_pattern.png and /dev/null differ diff --git a/patch_generation/patchgen_DD02C.py b/patch_generation/patchgen_DD02C.py deleted file mode 100755 index ef124f3..0000000 --- a/patch_generation/patchgen_DD02C.py +++ /dev/null @@ -1,217 +0,0 @@ -import numpy as np -import cv2 -from PIL import Image, ImageDraw - -# ---------------------------------------------------------------------- -# Utility functions from the original flat pattern generator -# (unchanged, except we now work with 0‑1 floats) -# ---------------------------------------------------------------------- -def merge_intervals(intervals): - if not intervals: - return [] - intervals.sort(key=lambda x: x[0]) - merged = [intervals[0]] - for current in intervals[1:]: - last = merged[-1] - if current[0] <= last[1]: - merged[-1] = (last[0], max(last[1], current[1])) - else: - merged.append(current) - return merged - -def get_allowed_intervals(neighbor_values, delta): - forbidden = [] - for c in neighbor_values: - low = max(0, c - delta) - high = min(1, c + delta) - forbidden.append((low, high)) - merged = merge_intervals(forbidden) - allowed = [] - if merged: - if merged[0][0] > 0: - allowed.append((0, merged[0][0])) - for i in range(1, len(merged)): - allowed.append((merged[i-1][1], merged[i][0])) - if merged[-1][1] < 1: - allowed.append((merged[-1][1], 1)) - else: - allowed.append((0, 1)) - return allowed - -def sample_from_allowed(allowed): - lengths = [end - start for start, end in allowed] - total_length = sum(lengths) - if total_length <= 0: - return None - point = np.random.rand() * total_length - cumulative = 0 - for (start, end), length in zip(allowed, lengths): - cumulative += length - if point <= cumulative: - return np.random.uniform(start, end) - return allowed[-1][0] - -def sample_channel(neighbor_values, delta): - allowed = get_allowed_intervals(neighbor_values, delta) - if not allowed: - return None - return sample_from_allowed(allowed) - -def generate_valid_color(neighbors, r): - delta = np.sqrt(r) - color = np.zeros(3) - for channel in range(3): - neighbor_channel = [n[channel] for n in neighbors] - sampled = sample_channel(neighbor_channel, delta) - if sampled is None: - sampled = np.random.rand() - color[channel] = sampled - return color - -# ---------------------------------------------------------------------- -# Cylinder‑specific generator -# ---------------------------------------------------------------------- -def generate_color_pattern_cylinder( - R=9.0, # cylinder radius (mm) - H=20.0, # cylinder height (mm) - fov_deg=150.0, # vertical FOV (degrees) - pixels_per_mm=100, # texture resolution - n_azimuth=36, # number of patches around the circumference - n_radial=20, # number of patches vertically (visible part) - r=0.2, # color randomness regulation factor - output_path='cylinder_pattern.png' -): - """ - Generate a texture map for the inner cylinder wall. - Patches are arranged so that, from a camera at the bottom - looking upward with perspective projection, they appear - uniform in angular and radial extent. - """ - - # Camera half‑angle (rad) - half_fov = np.radians(fov_deg / 2.0) - - # Visible height range (z_min = lowest point seen by camera) - z_min = R / np.tan(half_fov) if half_fov < np.pi/2 else 0.0 - z_max = H - - # Image plane radial coordinate r' = R / z - r_top = R / z_max # smallest r' (top) - r_bottom = R / z_min # largest r' (bottom) - - # Define patch grid in image space - theta_step = 2 * np.pi / n_azimuth - r_step = (r_bottom - r_top) / n_radial - - # Texture dimensions - width_px = int(np.round(2 * np.pi * R * pixels_per_mm)) - height_px = int(np.round(H * pixels_per_mm)) - print(f"Texture size: {width_px} x {height_px} pixels") - - # Store patch colors in a 2D array (rows = radial index, columns = azimuth) - patch_colors = np.zeros((n_radial, n_azimuth, 3), dtype=np.float32) - - # Generate colors row by row (top to bottom) - for j in range(n_radial): # j = 0 → top, j = n_radial-1 → bottom - r_start = r_top + j * r_step - r_end = r_start + r_step - # Corresponding z values (note: r increases → z decreases) - z_start = R / r_start # larger z (higher up) - z_end = R / r_end # smaller z (lower down) - v_start = z_start / H - v_end = z_end / H - # Ensure v_start > v_end (top has larger v) - if v_start < v_end: - v_start, v_end = v_end, v_start - - for i in range(n_azimuth): - theta_start = i * theta_step - theta_end = theta_start + theta_step - u_start = theta_start / (2 * np.pi) - u_end = theta_end / (2 * np.pi) - - # Collect neighbor colors (above and left) - neighbors = [] - # Above neighbor (previous row, same column) - if j > 0: - neighbors.append(patch_colors[j-1, i]) - # Left neighbor (same row, previous column) – wrap around - left_i = (i - 1) % n_azimuth - if i > 0 or j > 0: # at least one neighbor exists - neighbors.append(patch_colors[j, left_i]) - - if len(neighbors) == 0: - new_color = np.random.rand(3) - else: - new_color = generate_valid_color(neighbors, r) - - patch_colors[j, i] = new_color - - # Render the texture image - img = Image.new('RGB', (width_px, height_px), color=(255, 255, 255)) - draw = ImageDraw.Draw(img) - - for j in range(n_radial): - r_start = r_top + j * r_step - r_end = r_start + r_step - z_start = R / r_start - z_end = R / r_end - - # Determine v for top and bottom edges - v_top = max(z_start, z_end) / H # larger z -> higher on cylinder - v_bottom = min(z_start, z_end) / H # smaller z -> lower on cylinder - - for i in range(n_azimuth): - u_start = i * theta_step / (2 * np.pi) - u_end = (i+1) * theta_step / (2 * np.pi) - - # Pixel coordinates (ensure integer ordering) - x0 = int(u_start * width_px) - x1 = int(u_end * width_px) - y0 = int(v_bottom * height_px) - y1 = int(v_top * height_px) - - # Sanity: x0 < x1, y0 < y1 - if x0 > x1: - x0, x1 = x1, x0 - if y0 > y1: - y0, y1 = y1, y0 - - # Avoid degenerate rectangles (patch may span seam) - if x1 <= x0: - x1 = x0 + 1 - if y1 <= y0: - y1 = y0 + 1 - - color = tuple((patch_colors[j, i] * 255).astype(np.uint8)) - draw.rectangle([x0, y0, x1, y1], fill=color) - - img.save(output_path) - print(f"Saved texture to {output_path}") - - # Also return as numpy array (BGR for OpenCV) - img_np = np.array(img)[:, :, ::-1] # RGB -> BGR - return img_np - -# ---------------------------------------------------------------------- -# Example usage -# ---------------------------------------------------------------------- -if __name__ == '__main__': - # Generate pattern with 36 azimuth patches and 20 radial patches - patch_size = 12 - R = 9.0 - H = 20.0 - fov_deg = 150.0 - pixels_per_mm = 100 - n_azimuth = int(2*np.pi*R*pixels_per_mm / patch_size) - n_radial = int(H*pixels_per_mm / patch_size) - r = 0.05 - pattern = generate_color_pattern_cylinder( - R=9.0, H=20.0, fov_deg=150.0, pixels_per_mm=100, - n_azimuth=n_azimuth, n_radial=n_radial, r=0.05 - ) - - # Display with OpenCV - cv2.imshow('Cylinder Color Pattern', pattern) - cv2.waitKey(0) - cv2.destroyAllWindows() \ No newline at end of file diff --git a/patch_generation/raw_1.jpg b/patch_generation/raw_1.jpg deleted file mode 100644 index 24bfbdd..0000000 Binary files a/patch_generation/raw_1.jpg and /dev/null differ diff --git a/patch_generation/red_pattern.png b/patch_generation/red_pattern.png deleted file mode 100644 index eaeab93..0000000 Binary files a/patch_generation/red_pattern.png and /dev/null differ diff --git a/patch_generation/s1.jpg b/patch_generation/s1.jpg deleted file mode 100644 index 5b6cce0..0000000 Binary files a/patch_generation/s1.jpg and /dev/null differ diff --git a/patch_generation/test_hue_gray.py b/patch_generation/test_hue_gray.py deleted file mode 100644 index f96defa..0000000 --- a/patch_generation/test_hue_gray.py +++ /dev/null @@ -1,224 +0,0 @@ -import cv2 -import numpy as np -import matplotlib.pyplot as plt -def calculate_local_contrast(image_input, window_size=5): - """ - Calculate Global Average Local Contrast (for fine texture detail evaluation in optical flow). - Args: - image_input: str (image file path) or numpy array (grayscale/RGB image). - window_size: int (odd number, default=5) — sliding window size for fine texture analysis. - Returns: - float: Global average local contrast (higher = better fine texture detail; target ≥0.2 for optical flow). - Raises: - ValueError: If image is invalid, window size is even/too small, or processing fails. - """ - # ---------------------- Step 1: Input Validation & Preprocessing ---------------------- - # Validate window size (must be odd, ≥3 to capture meaningful texture) - if window_size % 2 == 0 or window_size < 3: - raise ValueError(f"Window size must be an odd integer ≥3 (got {window_size}). Use 3/5/7.") - - # Load/convert image to grayscale (consistent with optical flow workflow) - if isinstance(image_input, str): - img = cv2.imread(image_input) - if img is None: - raise ValueError(f"Failed to load image from path: {image_input}") - elif isinstance(image_input, np.ndarray): - img = image_input.copy() - else: - raise ValueError("Input must be an image file path (str) or numpy array (image).") - - # Convert to grayscale (handle RGB/BGR) - if len(img.shape) == 3: - gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) - else: - gray = img # Already grayscale - - # Normalize to 0-255 (avoid non-standard pixel ranges, e.g., float32) - gray = cv2.normalize(gray, None, 0, 255, cv2.NORM_MINMAX).astype(np.uint8) - - # ---------------------- Step 2: Sliding Window Local Contrast Calculation ---------------------- - # Define padding (replicate edge pixels to avoid boundary distortion) - pad_size = window_size // 2 - gray_padded = cv2.copyMakeBorder( - gray, - top=pad_size, bottom=pad_size, left=pad_size, right=pad_size, - borderType=cv2.BORDER_REPLICATE - ) - - # Initialize array to store local contrast for each window - local_contrast_vals = [] - - # Iterate over every pixel (center of the sliding window) - for y in range(pad_size, gray_padded.shape[0] - pad_size): - for x in range(pad_size, gray_padded.shape[1] - pad_size): - # Extract 5x5 (or custom) window around (x,y) - window = gray_padded[y-pad_size : y+pad_size+1, x-pad_size : x+pad_size+1] - - # Compute Michelson contrast for the window (avoid division by zero) - window_max = np.max(window) - window_min = np.min(window) - if window_max + window_min == 0: # All pixels are 0 (black window) - lc = 0.0 - else: - lc = (window_max - window_min) / (window_max + window_min) - - local_contrast_vals.append(lc) - - # ---------------------- Step 3: Global Average Local Contrast ---------------------- - global_local_contrast = np.mean(local_contrast_vals) if local_contrast_vals else 0.0 - - # Round for readability (4 decimal places) - return global_local_contrast - -def calculate_edge_strength_density(image_input): - """ - Calculate Edge Strength and Edge Density for optical flow evaluation. - Args: - image_input: Can be (1) str (image file path) or (2) numpy array (grayscale/RGB image). - Returns: - dict: Contains two metrics: - - edge_strength: Average gradient magnitude of valid edges (higher = sharper edges). - - edge_density: Percentage of valid edge pixels (higher = more texture/edges). - Raises: - ValueError: If image cannot be loaded/processed. - """ - # ---------------------- Step 1: Load/Preprocess Image ---------------------- - - img = image_input - # Convert to grayscale (handle RGB/BGR) - if len(img.shape) == 3: - gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) - else: - gray = img # Already grayscale - - # Normalize to 0-255 (avoid float/non-standard pixel ranges) - gray = cv2.normalize(gray, None, 0, 255, cv2.NORM_MINMAX).astype(np.uint8) - - # ---------------------- Step 2: Compute Gradient Magnitude (Sobel) ---------------------- - # Sobel kernel size = 3 (balance between noise reduction and edge sharpness for optical flow) - sobel_x = cv2.Sobel(gray, cv2.CV_64F, 1, 0, ksize=3) # Horizontal gradient - sobel_y = cv2.Sobel(gray, cv2.CV_64F, 0, 1, ksize=3) # Vertical gradient - - # Compute gradient magnitude (edge strength raw values) - gradient_magnitude = np.sqrt(sobel_x**2 + sobel_y**2) - - # ---------------------- Step 3: Adaptive Threshold for Valid Edges ---------------------- - # Use median gradient to filter noise (robust to outliers) - median_gradient = np.median(gradient_magnitude) - # Threshold = 1.5× median (filters weak noise, keeps meaningful edges for optical flow) - edge_threshold = 1.5 * median_gradient - # Create binary mask for valid edges (1 = valid edge, 0 = non-edge) - valid_edge_mask = (gradient_magnitude > edge_threshold).astype(np.uint8) - - # ---------------------- Step 4: Calculate Metrics ---------------------- - # Edge Strength: Average gradient magnitude of valid edges - valid_gradient_values = gradient_magnitude[valid_edge_mask == 1] - edge_strength = np.mean(valid_gradient_values) if len(valid_gradient_values) > 0 else 0.0 - - # Edge Density: Percentage of valid edge pixels (total pixels = height × width) - total_pixels = gray.shape[0] * gray.shape[1] - valid_edge_pixels = np.sum(valid_edge_mask) - edge_density = (valid_edge_pixels / total_pixels) * 100 if total_pixels > 0 else 0.0 - # print(f"edge_strength: {edge_strength}, edge_density: {edge_density}") - - # Round for readability (3 decimal places) - return edge_strength, edge_density - -def custom_gray_conversion(img, r_weight=0.299, g_weight=0.587, b_weight=0.114, - red_boost=1.2, green_reduce=0.8): - """ - Custom grayscale calculation with tunable RGB weights + red/green contrast enhancement - Args: - image_path: Path to input image - r_weight: Weight for Red channel (default: 0.299) - g_weight: Weight for Green channel (default: 0.587) - b_weight: Weight for Blue channel (default: 0.114) - red_boost: Multiplier to boost red region grayscale (>1 = brighter) - green_reduce: Multiplier to reduce green region grayscale (<1 = darker) - Returns: - default_gray: Grayscale with standard weights - custom_gray: Grayscale with tunable weights + red/green enhancement - """ - # 1. Read image and validate - # img_bgr = cv2.imread(image_path) - img_bgr = img - - if img_bgr is None: - raise ValueError("Failed to read image - check path!") - - # 2. Convert BGR (OpenCV default) to RGB and split channels - img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB) - # Convert to float32 to avoid integer overflow/precision loss - r = img_rgb[:, :, 0].astype(np.float32) - g = img_rgb[:, :, 1].astype(np.float32) - b = img_rgb[:, :, 2].astype(np.float32) - - # 3. Calculate DEFAULT grayscale (standard weights) - default_gray = 0.299 * r + 0.587 * g + 0.114 * b - - # 4. Calculate CUSTOM grayscale (tunable weights) - custom_gray_base = r_weight * r + g_weight * g + b_weight * b - - # 5. Enhance red/green contrast (optional but effective) - # Define red/green regions using RGB value ratios (no HSV needed) - # red_mask = (r > g * 1.2) & (r > b * 1.2) & (r >= 40) # Red-dominant pixels - # green_mask = (g > r * 1.2) & (g > b * 1.2) & (g >= 40) # Green-dominant pixels - - # Apply targeted adjustments to red/green regions - custom_gray = custom_gray_base.copy() - # custom_gray[red_mask] *= red_boost # Boost red region grayscale - # custom_gray[green_mask] *= green_reduce # Reduce green region grayscale - - # 6. Normalize to 0-255 range and convert back to uint8 (OpenCV standard) - # Clip values first to avoid overflow, then scale if needed - default_gray = np.clip(default_gray, 0, 255).astype(np.uint8) - custom_gray = np.clip(custom_gray, 0, 255).astype(np.uint8) - - return default_gray, custom_gray - -# -------------------------- Test the Code -------------------------- -if __name__ == "__main__": - # Replace with your image path - IMAGE_PATH = "./raw_1.jpg" - img = cv2.imread(IMAGE_PATH) - dx = 50 - dy = 0 - tile = img[410+dy:650+dy,940+dx:1180+dx,:] - - # Example 1: Use standard weights (verify correctness) - default_gray, red_gray = custom_gray_conversion( - tile, - r_weight=1, g_weight=0, b_weight=0 - ) - - default_gray, green_gray = custom_gray_conversion( - tile, - r_weight=0, g_weight=1, b_weight=0 - ) - default_gray, blue_gray = custom_gray_conversion( - tile, - r_weight=0, g_weight=0, b_weight=1 - ) - - edge_strength, edge_density = calculate_edge_strength_density(default_gray) - local_contrast = calculate_local_contrast(default_gray) - print(f"标准灰度: 边缘强度: {edge_strength}, 边缘密度: {edge_density}, 局部对比度: {local_contrast}") - edge_strength, edge_density = calculate_edge_strength_density(red_gray) - local_contrast = calculate_local_contrast(red_gray) - print(f"红色灰度: 边缘强度: {edge_strength}, 边缘密度: {edge_density}, 局部对比度: {local_contrast}") - edge_strength, edge_density = calculate_edge_strength_density(green_gray) - local_contrast = calculate_local_contrast(green_gray) - print(f"绿色灰度: 边缘强度: {edge_strength}, 边缘密度: {edge_density}, 局部对比度: {local_contrast}") - edge_strength, edge_density = calculate_edge_strength_density(blue_gray) - local_contrast = calculate_local_contrast(blue_gray) - print(f"蓝色灰度: 边缘强度: {edge_strength}, 边缘密度: {edge_density}, 局部对比度: {local_contrast}") - - plt.subplot(2, 2, 1) - plt.imshow(default_gray, cmap='gray') - plt.subplot(2, 2, 2) - plt.imshow(red_gray, cmap='gray') - plt.subplot(2, 2, 3) - plt.imshow(green_gray, cmap='gray') - plt.subplot(2, 2, 4) - plt.imshow(blue_gray, cmap='gray') - plt.show() diff --git a/patch_generation/yellow_pattern.png b/patch_generation/yellow_pattern.png deleted file mode 100644 index 6dc0e91..0000000 Binary files a/patch_generation/yellow_pattern.png and /dev/null differ diff --git a/benchmark.py b/scripts/benchmark.py similarity index 100% rename from benchmark.py rename to scripts/benchmark.py diff --git a/build_nuitka.py b/scripts/build_nuitka.py similarity index 96% rename from build_nuitka.py rename to scripts/build_nuitka.py index dcf8de9..0e4d286 100644 --- a/build_nuitka.py +++ b/scripts/build_nuitka.py @@ -7,17 +7,17 @@ from importlib.machinery import EXTENSION_SUFFIXES INCLUDE_LAYOUT = { "configs": { - "source_dir": Path("src/orisys/configs"), + "source_dir": Path("config"), "target_subdir": Path("configs"), - "patterns": ["ddjx01.json"], + "patterns": ["*.json"], }, "examples": { - "source_dir": Path("."), + "source_dir": Path("examples"), "target_subdir": Path("examples"), "files": [ - Path("example1_basic_pipeline.py"), - Path("example2_two_sensor_pipeline.py"), - Path("example3_plot_data.py"), + Path("examples/basic_pipeline.py"), + Path("examples/two_sensor_pipeline.py"), + Path("examples/plot_data.py"), ], }, "docs": { diff --git a/calibration_script.py b/scripts/calibration_script.py similarity index 100% rename from calibration_script.py rename to scripts/calibration_script.py diff --git a/demo_wuxi.py b/scripts/demo_wuxi.py similarity index 99% rename from demo_wuxi.py rename to scripts/demo_wuxi.py index bde8dfc..d6cfe1c 100644 --- a/demo_wuxi.py +++ b/scripts/demo_wuxi.py @@ -774,7 +774,7 @@ def main(): if sensor1_active: img1_valid = sensor1.get_img() - if not img1_valid: + if img1_valid is None: panel1_ui["arrow_label"].hide() panel1_ui["plot_widget"].hide() panel1_ui["disconnected_label"].show() @@ -831,7 +831,7 @@ def main(): if sensor2_active: img2_valid = sensor2.get_img() - if not img2_valid: + if img2_valid is None: panel2_ui["arrow_label"].hide() panel2_ui["plot_widget"].hide() panel2_ui["disconnected_label"].show() diff --git a/promote_video.py b/scripts/promote_video.py similarity index 94% rename from promote_video.py rename to scripts/promote_video.py index 2d301b2..56a80b9 100644 --- a/promote_video.py +++ b/scripts/promote_video.py @@ -153,7 +153,9 @@ def main(): frame_count = 0 writer = None while True: - sensor1.get_img() # 提取并拼接复眼图像 + isimg = sensor1.get_img() # 提取并拼接复眼图像 + if isimg is None: + break sensor1.compute_deformation() # 计算形变和分解 @@ -165,8 +167,13 @@ def main(): fps, fn, fx, fy, flow, depth_map = sensor1.read_info(sensor1.info.FPS, sensor1.info.FNORMAL, sensor1.info.FSHEARX, sensor1.info.FSHEARY, sensor1.info.VRAW, sensor1.info.DEPTH) print(fps,fn,fx,fy) # 显示向量场 - black_img = np.zeros_like(sensor1.img) - arrows = orisys.util.draw_arrows(sensor1.img, flow, threshold=2, grid_spacing=20, arrow_scale=0.5) + _bg_bgr = (22, 16, 11) + black_img = np.empty_like(sensor1.img) + black_img[:, :, 0] = _bg_bgr[0] + black_img[:, :, 1] = _bg_bgr[1] + black_img[:, :, 2] = _bg_bgr[2] + + arrows = orisys.util.draw_arrows(black_img, flow, threshold=2, grid_spacing=20, arrow_scale=0.5,below_threshold_color=(160, 160, 160),edge_margin=15,colormap=cv2.COLORMAP_PARULA) cv2.imshow("Deformation", arrows) # 显示深度图 diff --git a/scripts/test_ffmpeg.py b/scripts/test_ffmpeg.py new file mode 100644 index 0000000..1c3396f --- /dev/null +++ b/scripts/test_ffmpeg.py @@ -0,0 +1,159 @@ +import argparse +import av +import cv2 +import platform +import subprocess +import sys +import time + +TARGET_FPS = 30 +WIDTH = 1920 +HEIGHT = 1080 + + +def list_windows_dshow_devices(): + """ + Print DirectShow devices using ffmpeg, if available. + """ + cmd = ["ffmpeg", "-hide_banner", "-list_devices", "true", "-f", "dshow", "-i", "dummy"] + print("Listing Windows DirectShow devices via ffmpeg...") + try: + result = subprocess.run(cmd, capture_output=True, text=True, check=False) + except FileNotFoundError as exc: + raise RuntimeError("ffmpeg was not found in PATH; cannot list DirectShow devices.") from exc + + output = (result.stderr or "") + (result.stdout or "") + print(output.strip()) + + +def open_pyav_capture(system_name, device, pixel_format): + """ + Open a camera device via PyAV on Linux or Windows. + """ + if system_name == "linux": + options = { + "video_size": f"{WIDTH}x{HEIGHT}", + "framerate": str(TARGET_FPS), + "input_format": pixel_format or "mjpeg", + } + container = av.open(device or "/dev/video4", format="v4l2", options=options) + elif system_name == "windows": + if not device: + raise RuntimeError( + "On Windows, PyAV/dshow requires a camera name, for example " + '--device "USB Camera". Use --list-devices first.' + ) + options = { + "video_size": f"{WIDTH}x{HEIGHT}", + "framerate": str(TARGET_FPS), + } + if pixel_format: + options["pixel_format"] = pixel_format + container = av.open(f"video={device}", format="dshow", options=options) + else: + raise RuntimeError( + f"Unsupported OS: {platform.system()}. Only Linux and Windows are supported." + ) + + stream = container.streams.video[0] + return container, stream + + +def main(): + parser = argparse.ArgumentParser( + description="Capture from camera via PyAV on Linux or Windows, optionally save to MP4." + ) + parser.add_argument( + "-d", + "--device", + type=str, + default="", + help=( + "Linux: V4L2 device path like /dev/video4. " + "Windows: DirectShow camera name like 'USB Camera'. " + "If empty, Linux uses /dev/video4; Windows requires --device." + ), + ) + parser.add_argument( + "--pixel-format", + type=str, + default="", + help=( + "Optional input pixel format / format hint. " + "Examples: Linux 'mjpeg'; Windows may accept values such as 'mjpeg' if supported." + ), + ) + parser.add_argument( + "--list-devices", + action="store_true", + help="On Windows, list DirectShow device names via ffmpeg and exit.", + ) + parser.add_argument( + "-o", + "--output", + type=str, + default="", + help="Output MP4 path (e.g. capture.mp4). If empty, only preview, no file.", + ) + args = parser.parse_args() + + system_name = platform.system().lower() + if args.list_devices: + if system_name != "windows": + raise RuntimeError("--list-devices is currently only implemented for Windows.") + list_windows_dshow_devices() + return + + container, stream = open_pyav_capture(system_name, args.device, args.pixel_format) + print( + f"Using PyAV backend on {platform.system()} " + f"with device={args.device or '/dev/video4'}" + ) + + frame_count = 0 + t_start = time.perf_counter() + + writer = None + if args.output: + fourcc = cv2.VideoWriter_fourcc(*"mp4v") + writer = cv2.VideoWriter( + args.output, fourcc, float(TARGET_FPS), (WIDTH, HEIGHT) + ) + if not writer.isOpened(): + container.close() + raise RuntimeError(f"Could not open VideoWriter for {args.output}") + print(f"Recording to {args.output} @ {TARGET_FPS} fps, {WIDTH}x{HEIGHT}") + + try: + for frame in container.decode(stream): + t0 = time.perf_counter() + img = frame.to_ndarray(format="bgr24") # OpenCV-compatible + t1 = time.perf_counter() + + if writer is not None: + writer.write(img) + + cv2.imshow("pyav_cam", img) + key = cv2.waitKey(1) & 0xFF + t2 = time.perf_counter() + + frame_count += 1 + if frame_count % 30 == 0: + elapsed = t2 - t_start + print( + f"Frame {frame_count}: convert={1000*(t1-t0):.1f}ms, " + f"imshow+waitKey={1000*(t2-t1):.1f}ms, avg_fps={frame_count/elapsed:.1f}" + ) + + if key == ord("q"): + break + finally: + container.close() + if writer is not None: + writer.release() + print(f"Saved {frame_count} frames to {args.output}") + cv2.destroyAllWindows() + + +if __name__ == "__main__": + main() diff --git a/setup.py b/setup.py index 6202818..75a00ec 100644 --- a/setup.py +++ b/setup.py @@ -1,11 +1,12 @@ from setuptools import setup, find_packages from pathlib import Path -# Find all JSON files in the configs directory -configs_dir = Path("src/orisys/configs") -json_files = [] -if configs_dir.exists(): - json_files = [str(p.relative_to("src")) for p in configs_dir.glob("*.json")] +config_dir = Path("config") +data_files = [] +if config_dir.exists(): + json_files = [str(p) for p in sorted(config_dir.glob("*.json"))] + if json_files: + data_files.append(("orisys/configs", json_files)) setup( name="orisys", @@ -14,10 +15,8 @@ setup( author="Your Name", package_dir={"": "src"}, packages=find_packages(where="src"), - package_data={ - "orisys": ["configs/*.json"], - }, include_package_data=True, + data_files=data_files, install_requires=[ "numpy>=1.24.4", "opencv-python>=4.10.0.84", diff --git a/src/orisys/configs.py b/src/orisys/configs.py new file mode 100644 index 0000000..ebdfad8 --- /dev/null +++ b/src/orisys/configs.py @@ -0,0 +1,126 @@ +import json +from pathlib import Path + + +def _candidate_config_dirs() -> list[Path]: + """ + Candidate config directories for editable installs and packaged wheels. + """ + here = Path(__file__).resolve().parent + candidates = [ + here / "configs", # packaged wheel layout: orisys/configs/*.json + here.parent.parent / "config", # editable install from repo: repo_root/config/*.json + here.parent / "config", # fallback + Path.cwd() / "config", # script launched from repo root + ] + + unique: list[Path] = [] + for path in candidates: + if path not in unique: + unique.append(path) + return unique + + +def _get_config_dir() -> Path: + for candidate in _candidate_config_dirs(): + if candidate.exists() and any(candidate.glob("*.json")): + return candidate + # Default to the first candidate so error messages are stable. + return _candidate_config_dirs()[0] + + +CONFIG_DIR = _get_config_dir() + + +def _config_files() -> list[Path]: + if not CONFIG_DIR.exists(): + return [] + return sorted(CONFIG_DIR.glob("*.json")) + + +def _load_json(path: Path) -> dict: + with open(path, "r", encoding="utf-8") as f: + return json.load(f) + + +def load_config(config_name_or_path): + """ + Load sensor configuration from a JSON file path or config name. + + Config lookup order: + 1. Exact file path (absolute or relative) + 2. File stem under the config directory (e.g. ``ddjx01`` -> ``config/ddjx01.json``) + 3. JSON field ``"name"`` matching the requested name (keeps ``prototype`` compatible) + """ + config_path = Path(config_name_or_path) + + if config_path.exists() and config_path.is_file(): + if config_path.suffix.lower() != ".json": + raise ValueError(f"Config file must be a JSON file: {config_path}") + return _load_json(config_path) + + if config_path.is_absolute(): + raise FileNotFoundError(f"Config file not found at absolute path: {config_path}") + + direct = CONFIG_DIR / f"{config_name_or_path}.json" + if direct.exists(): + return _load_json(direct) + + # Backward-compatible alias resolution: support old names via JSON "name" field. + for candidate in _config_files(): + try: + data = _load_json(candidate) + except Exception: + continue + if data.get("name") == config_name_or_path: + return data + + available = [f.stem for f in _config_files()] + available_str = ", ".join(available) if available else "none" + raise FileNotFoundError( + 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 config file: {direct} (not found)\n" + f" Available config files: {available_str}" + ) + + +def list_config(): + """ + List all available JSON configuration files by file stem. + """ + return [f.stem for f in _config_files()] + + +def export_config(config_name=None, path="./config/exported_config.json"): + """ + Export a configuration to a JSON file with readable formatting. + """ + config = load_config(config_name) + Path(path).parent.mkdir(parents=True, exist_ok=True) + + with open(path, "w", encoding="utf-8") as f: + formatted = _format_json_value(config) + f.write(formatted) + f.write("\n") + + print(f"Config name: {config_name} exported to {path}") + return config + + +def _format_json_value(value, indent_level=0, indent_str=" "): + if isinstance(value, dict): + if not value: + return "{}" + items = [] + for k, v in value.items(): + key_str = json.dumps(k, ensure_ascii=False) + value_str = _format_json_value(v, indent_level + 1, indent_str) + items.append(f"{indent_str * (indent_level + 1)}{key_str}: {value_str}") + return "{\n" + ",\n".join(items) + "\n" + indent_str * indent_level + "}" + if isinstance(value, list): + if not value: + return "[]" + items = [json.dumps(item, ensure_ascii=False) for item in value] + return "[" + ", ".join(items) + "]" + return json.dumps(value, ensure_ascii=False) diff --git a/src/orisys/configs/__init__.py b/src/orisys/configs/__init__.py deleted file mode 100644 index 97d8b65..0000000 --- a/src/orisys/configs/__init__.py +++ /dev/null @@ -1,172 +0,0 @@ -import json -import sys -from pathlib import Path - -def _get_config_dir(): - """Resolve config directory, robustly handling Nuitka compiled paths.""" - - # 1. Start with the directory of this file - current_path = Path(__file__).parent - - # 2. Check strict relative location (standard case) - # If we are in src/orisys/configs/__init__.py, configs is current_path - if (current_path / "prototype.json").exists(): - return current_path - - # 3. Walk up the directory tree to find a 'configs' folder containing 'prototype.json' - # This handles nested virtual paths like .../orisys/orisys/configs - # We look for a physical directory that matches our expectation - - search_path = current_path - # Limit depth to avoid infinite loops, though unlikely with .parent - for _ in range(5): - # Check if 'configs' exists in this directory - candidate = search_path / "configs" - if candidate.exists() and candidate.is_dir(): - if (candidate / "prototype.json").exists(): - return candidate - - # Also check if the current directory ITSELF is the configs dir we missed - # (e.g. if we are somehow inside configs but missed check 1) - if search_path.name == 'configs' and (search_path / "prototype.json").exists(): - return search_path - - if search_path.parent == search_path: # Reached root - break - search_path = search_path.parent - - # 4. Try finding 'orisys' package root via sys.modules (if loaded) - try: - if 'orisys' in sys.modules: - mod = sys.modules['orisys'] - if hasattr(mod, '__file__') and mod.__file__: - pkg_root = Path(mod.__file__).parent - # pkg_root is .../site-packages/orisys - candidate = pkg_root / "configs" - if candidate.exists() and (candidate / "prototype.json").exists(): - return candidate - except Exception: - pass - - # Fallback to current path if nothing else works - return current_path - -CONFIG_DIR = _get_config_dir() - -def load_config(config_name_or_path): - """ - Load sensor configuration from JSON file. - - Accepts either: - 1. A file path (absolute or relative) to a JSON config file - 2. A config name (without .json extension) to load from package configs directory - - Args: - config_name_or_path: Either: - - A file path string (e.g., "/path/to/config.json" or "./config.json") - - A config name string (e.g., "prototype") to load from package configs - - Returns: - Dictionary containing the configuration - - Raises: - FileNotFoundError: If the config file doesn't exist - ValueError: If the file path is not a JSON file - """ - # Convert to Path object for easier handling - config_path = Path(config_name_or_path) - - # Check if it's an existing file (absolute or relative path) - if config_path.exists() and config_path.is_file(): - # User provided a file path - if config_path.suffix.lower() != '.json': - raise ValueError(f"Config file must be a JSON file: {config_path}") - - with open(config_path, 'r', encoding='utf-8') as f: - return json.load(f) - - # If it's an absolute path but doesn't exist, raise error immediately - if config_path.is_absolute(): - raise FileNotFoundError(f"Config file not found at absolute path: {config_path}") - - # Otherwise, treat it as a config name and look in package directory - package_config_path = CONFIG_DIR / f"{config_name_or_path}.json" - - if not package_config_path.exists(): - # List available configs for better error message - available = [f.stem for f in CONFIG_DIR.glob("*.json")] - available_str = ", ".join(available) if available else "none" - raise FileNotFoundError( - 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 package config: {package_config_path} (not found)\n" - f" Available package configs: {available_str}" - ) - - with open(package_config_path, 'r', encoding='utf-8') as f: - return json.load(f) - -def list_config(): - """ - List all available JSON configuration files in the configs directory. - - Returns: - List of configuration names (without .json extension), sorted alphabetically. - Returns an empty list if no JSON files are found. - """ - config_files = sorted(CONFIG_DIR.glob("*.json")) - return [f.stem for f in config_files] - -def export_config(config_name = None, path="./config/exported_config.json"): - """ - Export configuration to JSON file with readable formatting. - Arrays/lists are kept on a single line, while nested objects are indented. - - Args: - config: Configuration dictionary to export - path: Output file path (default: "./config/exported_config.json") - - Returns: - The exported config dictionary - """ - config = load_config(config_name) - Path(path).parent.mkdir(parents=True, exist_ok=True) - - with open(path, 'w', encoding='utf-8') as f: - formatted = _format_json_value(config) - f.write(formatted) - f.write('\n') # Add trailing newline - - print(f"Config name: {config_name} exported to {path}") - return config - -def _format_json_value(value, indent_level=0, indent_str=" "): - """ - Recursively format JSON value with arrays/lists on single line. - - Args: - value: The value to format (dict, list, or primitive) - indent_level: Current indentation level - indent_str: String used for indentation (default: 4 spaces) - - Returns: - Formatted string representation - """ - if isinstance(value, dict): - if not value: - return "{}" - items = [] - for k, v in value.items(): - key_str = json.dumps(k, ensure_ascii=False) - value_str = _format_json_value(v, indent_level + 1, indent_str) - items.append(f"{indent_str * (indent_level + 1)}{key_str}: {value_str}") - return "{\n" + ",\n".join(items) + "\n" + indent_str * indent_level + "}" - elif isinstance(value, list): - if not value: - return "[]" - # Format arrays/lists on a single line - items = [json.dumps(item, ensure_ascii=False) for item in value] - return "[" + ", ".join(items) + "]" - else: - # Primitive types (string, number, bool, None) - return json.dumps(value, ensure_ascii=False) \ No newline at end of file diff --git a/src/orisys/configs/ddjx01.json b/src/orisys/configs/ddjx01.json deleted file mode 100644 index ac8eba6..0000000 --- a/src/orisys/configs/ddjx01.json +++ /dev/null @@ -1,44 +0,0 @@ -{ - "name": "prototype", - "version": "0.3.1", - "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.05, - "max_aspect_ratio": 50, - "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": [250, 250] - }, - "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"], - "hhd_downsample": 3 - }, - "contact": { - "min_contour_area": 3000, - "flow_threshold":5, - "div_threshold":0.05, - "expansion_threshold": 0.01, - "kernel_size": [5,5], - "denoise_sigma": 12 - } -} - - diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index 577d223..f822787 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -200,7 +200,7 @@ class Sensor: Capture and preprocess a frame. Returns: - np.ndarray: Processed grayscale image + np.ndarray | None: Processed image, or None on capture failure. """ # Simple direct read - buffer size of 1 should minimize frame accumulation # Timing measurement helps identify if capture or processing is the bottleneck @@ -211,7 +211,7 @@ class Sensor: if not ret: print("Failed to get_img from video source.") - return False + return None # Start processing timing t_processing_start = time.perf_counter() @@ -250,41 +250,72 @@ class Sensor: self._fps = 1.0 / (time.perf_counter() - self.last_log_time) self.last_log_time = time.perf_counter() - return True + return self.img.copy() def detect_motion(self, threshold=6): # detect motion by optical flow magnitude - mag = np.linalg.norm(self.flow, axis=1) + mag = np.linalg.norm(self.flow, axis=-1) # print(f"mag: {mag.max()}, {mag.min()}, {mag.mean()}") return mag.mean() > threshold def compute_deformation(self, decompose=True, check_motion=True, threshold=6, tail_frames=10): """ - Compute optical flow and decompose into force components. - - Returns: - np.ndarray: Optical flow field + Compute optical flow and optionally decompose it into force components. + + Parameters + ---------- + decompose : bool + If True, run HHD decomposition (`compute_hhd`) after flow computation. + check_motion : bool + If True, gate decomposition with `detect_motion(threshold)`. When motion + falls below threshold, decomposition may still run for up to + `tail_frames` additional frames to avoid abrupt force drop-off. + threshold : float + Motion-detection threshold on mean flow magnitude. + tail_frames : int + Number of extra non-motion frames allowed to keep running HHD when + `check_motion=True`. + + Returns + ------- + bool | None + - `True` : motion is currently detected (`check_motion=True`) + - `False` : motion is currently not detected (`check_motion=True`) + - `None` : motion gating disabled (`check_motion=False`) """ if self.img is None: raise Exception("Warning: Get image first") - + self.compute_flow() - # self.compute_hhd() - if check_motion: - if self.detect_motion(threshold=threshold): - self.tail_frames_count = 0 + + if not check_motion: + self.tail_frames_count = 0 + if decompose: self.compute_hhd() - return True - else: - if self.tail_frames_count <= tail_frames: - self.tail_frames_count +=1 - self.compute_hhd() - return True - return False - else: return None - # return True + + is_motion = self.detect_motion(threshold=threshold) + try: + tail_frames = max(0, int(tail_frames)) + except (TypeError, ValueError): + tail_frames = 0 + + if is_motion: + self.tail_frames_count = 0 + if decompose: + self.compute_hhd() + return True + + # No motion this frame. Keep HHD alive briefly if requested, but do not + # report this as "motion detected". + if decompose and self.tail_frames_count < tail_frames: + self.tail_frames_count += 1 + self.compute_hhd() + else: + self.tail_frames_count = tail_frames + + return False def compute_flow(self): if self.img is None: diff --git a/test_3Dflow.py b/test_3Dflow.py deleted file mode 100644 index 53bc26f..0000000 --- a/test_3Dflow.py +++ /dev/null @@ -1,449 +0,0 @@ - -""" -Prototype tactile 2.5D motion analysis on top of the existing dense flow output. - -This script compares several motion representations derived from optical flow: -1. Raw flow arrows / magnitude -2. Radial and tangential flow around an estimated contact center -3. Differential flow features: divergence, curl, strain magnitude -4. Pseudo-height reconstructed from divergence (qualitative only) -""" -import os -import sys -import argparse -from pathlib import Path -import time - -os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" - -import cv2 -import numpy as np - -WINDOW_TITLE = "Tactile 2.5D Motion Prototype" - -_local_src = os.path.join(os.path.dirname(__file__), "src") -if _local_src not in sys.path: - sys.path.insert(0, _local_src) - -import orisys - - -def parse_args(): - parser = argparse.ArgumentParser() - parser.add_argument("--video", "-v", type=str, default="0", help="Camera ID / video file path") - parser.add_argument("--config", "-c", type=str, default="./config/ddjx01.json", help="Configuration file path") - parser.add_argument("--cal", "-cal", type=str, default="./config/ddjx01.npy", help="Calibration file path") - parser.add_argument("--verbose", action="store_true", help="Print detailed sensor logs") - parser.add_argument("--no-display", action="store_true", help="Disable OpenCV windows") - parser.add_argument("--save", type=str, default="", help="Optional output video path for the comparison panel") - parser.add_argument("--max-frames", type=int, default=0, help="Stop after N frames (0 = unlimited)") - parser.add_argument("--grid-spacing", type=int, default=20, help="Arrow visualization grid spacing") - parser.add_argument("--flow-threshold", type=float, default=2.0, help="Arrow visualization magnitude threshold") - parser.add_argument("--print-every", type=int, default=1, help="Print scalar metrics every N frames") - parser.add_argument("--panel-size", type=int, default=320, help="Width/height of each panel tile") - parser.add_argument("--plane-width-mm", type=float, default=20.0, help="Physical tactile imaging plane width in mm") - parser.add_argument("--plane-height-mm", type=float, default=20.0, help="Physical tactile imaging plane height in mm") - parser.add_argument("--camera-distance-mm", type=float, default=10.0, help="Distance from camera plane to imaging plane in mm") - parser.add_argument("--display-max-width", type=int, default=1800, help="Maximum preview window width in pixels") - parser.add_argument("--display-max-height", type=int, default=1000, help="Maximum preview window height in pixels") - return parser.parse_args() - - -def create_sensor(args): - try: - video_input = int(args.video) - except ValueError: - video_input = args.video - - return orisys.Sensor( - video_input, - config_name=args.config, - cal_path=args.cal, - verbose=args.verbose, - ) - - -def ensure_bgr(image): - if image.ndim == 2: - return cv2.cvtColor(image, cv2.COLOR_GRAY2BGR) - return image - - -def resize_panel(image, panel_size): - return cv2.resize(ensure_bgr(image), (panel_size, panel_size), interpolation=cv2.INTER_LINEAR) - - -def fit_display_image(image, max_width, max_height): - h, w = image.shape[:2] - scale = min(max_width / max(float(w), 1.0), max_height / max(float(h), 1.0), 1.0) - if scale >= 0.999: - return image - new_size = (max(1, int(w * scale)), max(1, int(h * scale))) - return cv2.resize(image, new_size, interpolation=cv2.INTER_AREA) - - -def signed_heatmap(field, colormap=cv2.COLORMAP_TURBO): - max_abs = float(np.max(np.abs(field))) - if max_abs <= 1e-9: - norm = np.full(field.shape, 127, dtype=np.uint8) - else: - norm = np.clip((field / max_abs + 1.0) * 127.5, 0, 255).astype(np.uint8) - return cv2.applyColorMap(norm, colormap) - - -def magnitude_heatmap(field, colormap=cv2.COLORMAP_TURBO): - max_val = float(np.max(field)) - if max_val <= 1e-9: - norm = np.zeros(field.shape, dtype=np.uint8) - else: - norm = np.clip(field / max_val * 255.0, 0, 255).astype(np.uint8) - return cv2.applyColorMap(norm, colormap) - - -def add_title(image, title): - panel = image.copy() - cv2.rectangle(panel, (0, 0), (panel.shape[1], 28), (0, 0, 0), -1) - cv2.putText(panel, title, (8, 20), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2, cv2.LINE_AA) - return panel - - -def make_panel(image, title, panel_size): - return add_title(resize_panel(image, panel_size), title) - - -def stack_panels(panels, columns=4): - filler = np.zeros_like(panels[0]) - rows = [] - for idx in range(0, len(panels), columns): - row = panels[idx:idx + columns] - if len(row) < columns: - row = row + [filler] * (columns - len(row)) - rows.append(cv2.hconcat(row)) - return cv2.vconcat(rows) - - -def estimate_contact_center(divergence, flow_magnitude): - positive = np.maximum(divergence, 0.0) - weights = positive * (flow_magnitude + 1e-6) - total = float(weights.sum()) - h, w = divergence.shape - - if total <= 1e-9: - return (w // 2, h // 2) - - ys, xs = np.indices(divergence.shape, dtype=np.float32) - cx = int(np.clip((weights * xs).sum() / total, 0, w - 1)) - cy = int(np.clip((weights * ys).sum() / total, 0, h - 1)) - return (cx, cy) - - -def prepare_support_mask(support_map): - if support_map is None: - return None - - support = np.abs(support_map).astype(np.float32) - max_support = float(np.max(support)) - if max_support > 0.0: - support /= max_support - support = (support > 0.05).astype(np.float32) - if float(np.mean(support)) < 0.002: - return None - return support - - -def decompose_radial_tangential(flow, center): - h, w = flow.shape[:2] - ys, xs = np.indices((h, w), dtype=np.float32) - cx, cy = center - - rx = xs - float(cx) - ry = ys - float(cy) - r_norm = np.sqrt(rx * rx + ry * ry) - r_norm = np.where(r_norm < 1e-6, 1.0, r_norm) - - rhat_x = rx / r_norm - rhat_y = ry / r_norm - that_x = -rhat_y - that_y = rhat_x - - u = flow[:, :, 0] - v = flow[:, :, 1] - radial = u * rhat_x + v * rhat_y - tangential = u * that_x + v * that_y - return radial, tangential - - -def compute_flow_jacobian_features(flow): - u = flow[:, :, 0] - v = flow[:, :, 1] - - du_dy, du_dx = np.gradient(u) - dv_dy, dv_dx = np.gradient(v) - - divergence = du_dx + dv_dy - curl = dv_dx - du_dy - strain_xx = du_dx - strain_yy = dv_dy - strain_xy = 0.5 * (du_dy + dv_dx) - strain_mag = np.sqrt(strain_xx * strain_xx + strain_yy * strain_yy + 2.0 * strain_xy * strain_xy) - - return { - "divergence": divergence, - "curl": curl, - "strain_mag": strain_mag, - } - - -def reconstruct_pseudo_height(divergence, flow_magnitude=None, support_map=None): - rhs = divergence.astype(np.float32).copy() - - # No-contact guard: tiny flow noise becomes large low-frequency drift after Poisson solve. - if flow_magnitude is not None: - mean_mag = float(np.mean(flow_magnitude)) - max_mag = float(np.max(flow_magnitude)) - if mean_mag < 0.03 and max_mag < 0.15: - return np.zeros_like(rhs, dtype=np.float32) - - support = prepare_support_mask(support_map) - if support_map is not None: - if support is None: - return np.zeros_like(rhs, dtype=np.float32) - rhs *= support - - noise_floor = max(float(np.std(rhs)) * 1.5, 1e-4) - rhs[np.abs(rhs) < noise_floor] = 0.0 - if not np.any(rhs): - return np.zeros_like(rhs, dtype=np.float32) - - # Zero-pad before FFT solve to reduce periodic wrap-around ghosts, then crop back. - h, w = rhs.shape - pad_y = max(8, h // 2) - pad_x = max(8, w // 2) - rhs_padded = np.pad(rhs, ((pad_y, pad_y), (pad_x, pad_x)), mode="constant") - rhs_padded = rhs_padded - float(rhs_padded.mean()) - - hp, wp = rhs_padded.shape - fy = np.fft.fftfreq(hp).reshape(-1, 1) - fx = np.fft.fftfreq(wp).reshape(1, -1) - denom = (2.0 * np.pi * fx) ** 2 + (2.0 * np.pi * fy) ** 2 - denom[0, 0] = 1.0 - - rhs_hat = np.fft.fft2(rhs_padded) - height_hat = -rhs_hat / denom - height_hat[0, 0] = 0.0 - height_full = np.real(np.fft.ifft2(height_hat)).astype(np.float32) - height = height_full[pad_y:pad_y + h, pad_x:pad_x + w] - height = cv2.GaussianBlur(height, (0, 0), 1.0) - return height - - -def build_visualization(img, flow, center, features, radial, tangential, pseudo_height, depth_map, args): - arrows = orisys.util.draw_arrows( - img, - flow, - threshold=args.flow_threshold, - grid_spacing=args.grid_spacing, - arrow_scale=1.0, - ) - - contact_view = ensure_bgr(img) - cv2.circle(contact_view, center, 6, (0, 255, 0), -1) - cv2.circle(contact_view, center, 12, (0, 255, 0), 2) - - panel_images = [ - (contact_view, "Image / center"), - (arrows, "Raw flow"), - (signed_heatmap(radial), "Radial flow"), - (signed_heatmap(tangential), "Tangential flow"), - (signed_heatmap(features["divergence"]), "Divergence"), - (signed_heatmap(features["curl"]), "Curl"), - (magnitude_heatmap(features["strain_mag"]), "Strain mag"), - (signed_heatmap(depth_map), "Depth map"), - (signed_heatmap(pseudo_height), "Pseudo height"), - ] - panels = [make_panel(image, title, args.panel_size) for image, title in panel_images] - return stack_panels(panels) - - -def make_writer(path, frame_size, fps): - if not path: - return None - parent = Path(path).parent - if parent and not parent.exists(): - parent.mkdir(parents=True, exist_ok=True) - return cv2.VideoWriter(path, cv2.VideoWriter_fourcc(*"mp4v"), fps, frame_size) - - -def approximate_raw_flow_to_mm(flow_px, plane_width_mm, plane_height_mm, camera_distance_mm, image_shape): - h, w = image_shape[:2] - fx = camera_distance_mm * w / max(plane_width_mm, 1e-6) - fy = camera_distance_mm * h / max(plane_height_mm, 1e-6) - flow_mm = flow_px.copy().astype(np.float32) - flow_mm[:, :, 0] *= camera_distance_mm / max(fx, 1e-6) - flow_mm[:, :, 1] *= camera_distance_mm / max(fy, 1e-6) - return flow_mm - - -def compute_metrics(flow_magnitude, raw_flow_mm, radial, tangential, features, depth_map, pseudo_height): - return { - "flow_mean": float(flow_magnitude.mean()), - "flow_mean_mm": float(np.linalg.norm(raw_flow_mm, axis=2).mean()), - "radial_abs_mean": float(np.mean(np.abs(radial))), - "tangential_abs_mean": float(np.mean(np.abs(tangential))), - "divergence_mean": float(features["divergence"].mean()), - "curl_abs_mean": float(np.mean(np.abs(features["curl"]))), - "strain_mean": float(features["strain_mag"].mean()), - "depth_peak": float(np.max(np.abs(depth_map))), - "pseudo_height_peak": float(np.max(np.abs(pseudo_height))), - } - - -def format_metrics(frame_count, fps, metrics): - line = ( - f"frame={frame_count:04d} fps={fps:6.2f} " - f"flow_mean={metrics['flow_mean']:.4f} " - f"flow_mean_mm={metrics['flow_mean_mm']:.4f} " - f"radial_abs={metrics['radial_abs_mean']:.4f} " - f"tangential_abs={metrics['tangential_abs_mean']:.4f} " - f"div_mean={metrics['divergence_mean']:.4f} " - f"curl_abs={metrics['curl_abs_mean']:.4f} " - f"strain={metrics['strain_mean']:.4f} " - f"depth_peak={metrics['depth_peak']:.4f} " - f"height_peak={metrics['pseudo_height_peak']:.4f}" - ) - t = metrics.get("_timings") - if t: - line += ( - f" | analyze {t['total']:.2f}ms" - f" (mag:{t['magnitude']:.2f} jac:{t['jacobian']:.2f}" - f" rad_tan:{t['rad_tan']:.2f} pseudo_h:{t['pseudo_h']:.2f}" - f" metrics:{t['metrics']:.2f} vis:{t['visualize']:.2f})" - ) - return line - - -def read_sensor_frame(sensor): - sensor.compute_flow() - sensor.compute_hhd() - sensor.compute_contact() - return sensor.read_info( - sensor.info.FPS, - sensor.info.VRAW, - sensor.info.IMG, - sensor.info.DEPTH, - ) - - -def analyze_frame(img, flow, depth_map, args): - t = time.perf_counter() - - flow_magnitude = np.linalg.norm(flow, axis=2) - t_mag = time.perf_counter() - - features = compute_flow_jacobian_features(flow) - t_jac = time.perf_counter() - - center = estimate_contact_center(features["divergence"], flow_magnitude) - radial, tangential = decompose_radial_tangential(flow, center) - t_decomp = time.perf_counter() - - pseudo_height = reconstruct_pseudo_height( - features["divergence"], - flow_magnitude=flow_magnitude, - support_map=depth_map, - ) - t_height = time.perf_counter() - - raw_flow_mm = approximate_raw_flow_to_mm( - flow, - args.plane_width_mm, - args.plane_height_mm, - args.camera_distance_mm, - img.shape, - ) - metrics = compute_metrics(flow_magnitude, raw_flow_mm, radial, tangential, features, depth_map, pseudo_height) - t_metrics = time.perf_counter() - - canvas = build_visualization( - img, - flow, - center, - features, - radial, - tangential, - pseudo_height, - depth_map, - args, - ) - t_vis = time.perf_counter() - - metrics["_timings"] = { - "magnitude": (t_mag - t) * 1000, - "jacobian": (t_jac - t_mag) * 1000, - "rad_tan": (t_decomp - t_jac) * 1000, - "pseudo_h": (t_height - t_decomp) * 1000, - "metrics": (t_metrics - t_height) * 1000, - "visualize": (t_vis - t_metrics) * 1000, - "total": (t_vis - t) * 1000, - } - return canvas, metrics - - -def main(): - args = parse_args() - sensor = create_sensor(args) - writer = None - frame_count = 0 - print_every = max(args.print_every, 1) - - try: - print("\nPress 'q' to quit, 'r' to reset tracking.") - - while True: - img = sensor.get_img() - if img is None: - print("No more frames returned from sensor.") - break - - fps, flow, img, depth_map = read_sensor_frame(sensor) - canvas, metrics = analyze_frame(img, flow, depth_map, args) - - if writer is None and args.save: - writer = make_writer(args.save, (canvas.shape[1], canvas.shape[0]), max(float(fps), 1.0)) - - if writer is not None: - writer.write(canvas) - - frame_count += 1 - - if frame_count % print_every == 0: - print(format_metrics(frame_count, fps, metrics)) - - if not args.no_display: - display_canvas = fit_display_image( - canvas, - max_width=args.display_max_width, - max_height=args.display_max_height, - ) - cv2.imshow(WINDOW_TITLE, display_canvas) - key = cv2.waitKey(1) & 0xFF - if key == ord("q"): - print("\nExiting...") - break - if key == ord("r"): - sensor.reset() - print("Sensor reset.") - - if args.max_frames > 0 and frame_count >= args.max_frames: - print(f"Reached max_frames={args.max_frames}.") - break - finally: - if writer is not None: - writer.release() - sensor.disconnect() - if not args.no_display: - cv2.destroyAllWindows() - - -if __name__ == "__main__": - main() diff --git a/test_density.py b/test_density.py deleted file mode 100644 index 2d7f24e..0000000 --- a/test_density.py +++ /dev/null @@ -1,207 +0,0 @@ -""" -示例: 密度图可视化 - -本示例演示如何： -1. 获取拼接图像 -2. 计算光流 -3. 基于光流生成 density map -4. 使用 colormap 实时显示 -""" -import os -import argparse - -import cv2 -import numpy as np -import orisys - -from getDensity import getFilteredDensity - -os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" - - -def density_to_colormap(density): - """Convert density map to a displayable heatmap.""" - if density is None or density.size == 0: - return np.zeros((1, 1, 3), dtype=np.uint8) - - dmin = float(np.min(density)) - dmax = float(np.max(density)) - if dmax > dmin: - density_u8 = ((density - dmin) * (255.0 / (dmax - dmin))).astype(np.uint8) - else: - density_u8 = np.zeros_like(density, dtype=np.uint8) - return cv2.applyColorMap(density_u8, cv2.COLORMAP_TURBO) - - -def main(): - parser = argparse.ArgumentParser() - parser.add_argument("--video", "-v", type=str, default="0", help="Camera ID / Video file path") - parser.add_argument("--config", "-c", type=str, default="./config/ddjx01.json", help="Configuration file path") - parser.add_argument("--cal", "-cal", type=str, default="./config/ddjx01.npy", help="Calibration file path") - parser.add_argument("--verbose", "-verbose", type=bool, default=True, help="Print detailed information") - args = parser.parse_args() - - # 步骤1：创建传感器对象 (必要步骤) - try: - vid_src = int(args.video) - except ValueError: - vid_src = args.video - sensor = orisys.Sensor(vid_src, config_name=args.config, cal_path=args.cal, verbose=args.verbose) - - print("\n按 'q' 退出程序，按 'r' 重置传感器追踪器") - - while True: - # 步骤2：获取图像并计算形变 - img = sensor.get_img() - if img is None: - print("Failed to get image (end of stream or capture error). Exiting.") - break - - sensor.compute_deformation(check_motion=False) - sensor.compute_contact() - (fps, flow, contour, centroid) = sensor.read_info( - sensor.info.FPS, - sensor.info.VRAW, - sensor.info.CONTOUR, - sensor.info.CENTROID, - ) - - # 步骤3：根据光流计算 density map 并显示 - density = getFilteredDensity(flow, use_cuda=sensor.cuda) - density_view = density_to_colormap(density) - cv2.imshow("Density Map", density_view) - - image_with_foe = orisys.util.draw_contact(img, contour, centroid) - cv2.imshow("Contact Centroid", image_with_foe) - - print(f"fps: {fps:.2f}, density_shape: {density.shape}, flow_shape: {flow.shape}") - - # 步骤4：键盘控制 - key = cv2.waitKey(1) & 0xFF - if key == ord("q"): - print("\n退出程序...") - break - if key == ord("r"): - sensor.reset() - print("传感器已重置") - - sensor.disconnect() - - -if __name__ == "__main__": - main() -""" -示例 1: 基础用法 - 单传感器同步处理 - -本示例演示 orisys 的基本使用流程： -1. 初始化传感器（视频文件或摄像头） -2. 循环获取图像并计算形变和触碰点 -3. 读取各类传感器数据 -4. 可视化显示 - - -""" -import os -os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" -import cv2 -import orisys -import numpy as np -import argparse - -def main(): - parser = argparse.ArgumentParser() - parser.add_argument("--video","-v", type=str, default="0", help="Camera ID / Video file path") - parser.add_argument("--config","-c", type=str, default="./config/ddjx01.json", help="Configuration file path") - parser.add_argument("--cal","-cal", type=str, default="./config/ddjx01.npy", help="Calibration file path") - parser.add_argument("--verbose","-verbose", type=bool, default=True, help="Print detailed information") - args = parser.parse_args() - - # 步骤1：创建传感器对象 (必要步骤)， - # config_name为配置文件名（内置配置文件名）或路径（自定义配置文件路径） - # cal_path为校准文件路径, 每一个设备需要创建/使用一个校准文件 - - # 方式1: 从视频文件读取 - try: - input = int(args.video) - is_camera = True - sensor = orisys.Sensor(input, config_name=args.config, cal_path= args.cal, verbose=args.verbose) - except: - is_camera = False - sensor = orisys.Sensor(args.video, config_name=args.config, cal_path= args.cal, verbose=args.verbose) - - print("\n按 'q' 退出程序，按 'r' 重置传感器追踪器") - - while True: - # 步骤2：信息提取 - # 步骤 2.1: 获取并拼接图像 (必要步骤) - sensor.get_img() - # 步骤 2.2: 计算形变场 (必要步骤) - is_touched = sensor.compute_deformation(check_motion=True, threshold=6) - # 步骤 2.3: 计算触碰点 - if is_touched: - sensor.compute_contact() - - - #步骤 2.4: 读取信息 - fps, flow, vnormal, img, contour, centroid, depth_map, fnormal, fshearx, fsheary = sensor.read_info( - sensor.info.FPS, - sensor.info.VRAW, - sensor.info.VNORMAL, - sensor.info.IMG, - sensor.info.CONTOUR, - sensor.info.CENTROID, - sensor.info.DEPTH, - sensor.info.FNORMAL, - sensor.info.FSHEARX, - sensor.info.FSHEARY - ) - - #步骤 3: 可视化显示 - # 步骤 3.1: 绘制形变场 - arrows = orisys.util.draw_arrows( - img, - flow, - threshold=2, - grid_spacing=20, - arrow_scale=1.0 - ) - cv2.imshow("Deformation", arrows) - - # 步骤 3.2: 在图像上触碰点和区域 - image_with_foe = orisys.util.draw_contact(img, contour, centroid) - cv2.imshow("Contact Centroid", image_with_foe) - - # 步骤 3.3: 绘制深度图 - # 归一化散度值到0-255范围（使用绝对值，然后归一化） - div_abs = depth_map - if div_abs.max() > div_abs.min(): - div_normalized = ((div_abs - div_abs.min()) / (div_abs.max() - div_abs.min()) * 255).astype(np.uint8) - else: - div_normalized = np.zeros_like(div_abs, dtype=np.uint8) - - # 应用颜色映射（使用JET：蓝色=低散度，红色=高散度） - 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}, sensor.resolution_raw: {sensor.img_size_raw}") - - #步骤 3.5: 键盘控制 - key = cv2.waitKey(1) & 0xFF - if key == ord("q"): - print("\n退出程序...") - break - if key == ord("r"): - # 重置光流追踪器（当追踪漂移时使用） - sensor.reset() - print("传感器已重置") - - # ========================================================================= - # 4. 清理资源 - # ========================================================================= - sensor.disconnect() # 释放视频源和相关资源 - - -if __name__ == '__main__': - main() diff --git a/test_downsample.py b/test_downsample.py deleted file mode 100644 index d9e1f77..0000000 --- a/test_downsample.py +++ /dev/null @@ -1,252 +0,0 @@ -""" -HHD Downsample Factor Comparison - -Reads a video file and runs the full pipeline at multiple hhd_downsample factors. -Computes per-frame HHD timing and vnormal quality metrics against the baseline (factor=1), -then plots the results. - -Usage: - python test_downsample.py -v jx01_hex1.5_2.mp4 -c ./config/prototype_of3_720p.json -f 1,2,3,4 -n 50 -w 30 -""" -import os -import sys - -os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" - -_local_src = os.path.join(os.path.dirname(__file__), "src") -if _local_src not in sys.path: - sys.path.insert(0, _local_src) - -import cv2 -import orisys -import numpy as np -import time -import argparse -import matplotlib -matplotlib.use("Agg") -import matplotlib.pyplot as plt - - -def run_sensor(vid_src, config, cal, downsample, n_warmup, n_measure): - """Run sensor for n_warmup + n_measure frames. Capture timing and vnormal for every measurement frame.""" - try: - vid = int(vid_src) - except ValueError: - vid = vid_src - - sensor = orisys.Sensor( - vid, config_name=config, cal_path=cal, - isstitch=True, cuda=False, verbose=False, - hhd_downsample=downsample, - ) - - hhd_times = [] - vnormals = [] - frame_idx = 0 - n_total = n_warmup + n_measure - - while frame_idx < n_total: - try: - sensor.get_img() - except Exception: - break - - sensor.compute_flow() - - t0 = time.perf_counter() - sensor.compute_hhd() - t_hhd = time.perf_counter() - t0 - - if frame_idx >= n_warmup: - hhd_times.append(t_hhd) - vn = sensor.vnormal.copy() if sensor.vnormal is not None else None - vnormals.append(vn) - - frame_idx += 1 - - sensor.disconnect() - return {"hhd_times": hhd_times, "vnormals": vnormals} - - -def per_frame_rmse(a, b): - if a is None or b is None: - return float("nan") - diff = a.astype(np.float64) - b.astype(np.float64) - return float(np.sqrt(np.mean(diff ** 2))) - - -def per_frame_max_abs_err(a, b): - if a is None or b is None: - return float("nan") - return float(np.max(np.abs(a.astype(np.float64) - b.astype(np.float64)))) - - -def per_frame_cosine_sim(a, b): - if a is None or b is None: - return float("nan") - a64 = a.astype(np.float64) - b64 = b.astype(np.float64) - dot = np.sum(a64 * b64, axis=-1) - mag_a = np.linalg.norm(a64, axis=-1) - mag_b = np.linalg.norm(b64, axis=-1) - denom = mag_a * mag_b - valid = denom > 1e-12 - if not np.any(valid): - return float("nan") - return float(np.mean(dot[valid] / denom[valid])) - - -def main(): - parser = argparse.ArgumentParser(description="Compare HHD downsample factors") - parser.add_argument("--video", "-v", type=str, default="2", help="Camera ID or video file path") - parser.add_argument("--config", "-c", type=str, default="./config/prototype_of3_720p.json") - parser.add_argument("--cal", "-cal", type=str, default="./config/ddjx01.npy") - parser.add_argument("--factors", "-f", type=str, default="1,2,3", help="Comma-separated downsample factors") - parser.add_argument("--warmup", "-w", type=int, default=5, help="Warmup frames (skipped)") - parser.add_argument("--frames", "-n", type=int, default=30, help="Measurement frames") - parser.add_argument("--output", "-o", type=str, default="downsample_comparison.png", help="Output plot filename") - args = parser.parse_args() - - factors = [int(f) for f in args.factors.split(",")] - - print(f"Video: {args.video}") - print(f"Config: {args.config}") - print(f"Factors: {factors}") - print(f"Warmup: {args.warmup} frames, Measure: {args.frames} frames") - print() - - # --- collect data --- - results = {} - for factor in factors: - print(f"--- Running factor {factor} ---") - results[factor] = run_sensor( - args.video, args.config, args.cal, - factor, args.warmup, args.frames, - ) - n = len(results[factor]["hhd_times"]) - if n > 0: - arr = np.array(results[factor]["hhd_times"]) * 1000 - print(f" {n} frames: hhd mean={arr.mean():.2f}ms, median={np.median(arr):.2f}ms, " - f"min={arr.min():.2f}ms, max={arr.max():.2f}ms") - else: - print(" No frames captured") - print() - - # --- timing summary table --- - print("=" * 72) - print(f"{'Factor':<10} {'Mean HHD (ms)':<16} {'Median (ms)':<14} {'Speedup':<10}") - print("-" * 72) - baseline_median = None - for factor in factors: - arr = np.array(results[factor]["hhd_times"]) * 1000 - if len(arr) == 0: - continue - med = float(np.median(arr)) - if baseline_median is None: - baseline_median = med - speedup = baseline_median / med if med > 0 else 0 - print(f"{factor:<10} {arr.mean():<16.2f} {med:<14.2f} {speedup:<10.2f}x") - print("=" * 72) - print() - - # --- compute per-frame quality metrics vs baseline (factor=1) --- - has_baseline = 1 in factors and len(factors) >= 2 - metrics = {} # factor -> {"rmse": [...], "max_err": [...], "cosine": [...]} - if has_baseline: - base_vn = results[1]["vnormals"] - n_base = len(base_vn) - for factor in factors: - if factor == 1: - continue - test_vn = results[factor]["vnormals"] - n_common = min(n_base, len(test_vn)) - rmse_arr, maxerr_arr, cos_arr = [], [], [] - for i in range(n_common): - rmse_arr.append(per_frame_rmse(base_vn[i], test_vn[i])) - maxerr_arr.append(per_frame_max_abs_err(base_vn[i], test_vn[i])) - cos_arr.append(per_frame_cosine_sim(base_vn[i], test_vn[i])) - metrics[factor] = { - "rmse": np.array(rmse_arr), - "max_err": np.array(maxerr_arr), - "cosine": np.array(cos_arr), - } - - # print per-factor summary - print(f"{'Factor':<10} {'Mean RMSE':<14} {'Mean MaxErr':<14} {'Mean CosSim':<14}") - print("-" * 52) - for factor in sorted(metrics): - m = metrics[factor] - print(f"{factor:<10} {np.nanmean(m['rmse']):<14.6f} {np.nanmean(m['max_err']):<14.4f} " - f"{np.nanmean(m['cosine']):<14.6f}") - print("=" * 52) - print() - - # --- plot --- - n_quality_factors = len(metrics) - n_plots = 1 + (3 if has_baseline else 0) # timing + (rmse, max_err, cosine) - fig, axes = plt.subplots(n_plots, 1, figsize=(12, 3.5 * n_plots), sharex=True) - if n_plots == 1: - axes = [axes] - - colors = plt.cm.tab10.colors - frame_nums = None - - # 1) HHD timing per frame - ax = axes[0] - for i, factor in enumerate(factors): - arr = np.array(results[factor]["hhd_times"]) * 1000 - frames = np.arange(len(arr)) - if frame_nums is None: - frame_nums = frames - ax.plot(frames, arr, label=f"ds={factor}", color=colors[i % len(colors)], linewidth=1) - ax.set_ylabel("HHD time (ms)") - ax.set_title("Per-frame HHD computation time") - ax.legend(loc="upper right") - ax.grid(True, alpha=0.3) - - if has_baseline: - # 2) RMSE - ax = axes[1] - for i, factor in enumerate(sorted(metrics)): - m = metrics[factor] - frames = np.arange(len(m["rmse"])) - ci = factors.index(factor) if factor in factors else i - ax.plot(frames, m["rmse"], label=f"ds={factor}", color=colors[ci % len(colors)], linewidth=1) - ax.set_ylabel("RMSE (vnormal)") - ax.set_title("Per-frame RMSE vs baseline (ds=1)") - ax.legend(loc="upper right") - ax.grid(True, alpha=0.3) - - # 3) Max absolute error - ax = axes[2] - for i, factor in enumerate(sorted(metrics)): - m = metrics[factor] - frames = np.arange(len(m["max_err"])) - ci = factors.index(factor) if factor in factors else i - ax.plot(frames, m["max_err"], label=f"ds={factor}", color=colors[ci % len(colors)], linewidth=1) - ax.set_ylabel("Max Abs Error (vnormal)") - ax.set_title("Per-frame max absolute error vs baseline (ds=1)") - ax.legend(loc="upper right") - ax.grid(True, alpha=0.3) - - # 4) Cosine similarity - ax = axes[3] - for i, factor in enumerate(sorted(metrics)): - m = metrics[factor] - frames = np.arange(len(m["cosine"])) - ci = factors.index(factor) if factor in factors else i - ax.plot(frames, m["cosine"], label=f"ds={factor}", color=colors[ci % len(colors)], linewidth=1) - ax.set_ylabel("Cosine Similarity") - ax.set_title("Per-frame cosine similarity vs baseline (ds=1)") - ax.legend(loc="upper right") - ax.grid(True, alpha=0.3) - ax.set_ylim(bottom=min(0.999, ax.get_ylim()[0])) - - axes[-1].set_xlabel("Frame") - fig.tight_layout() - fig.savefig(args.output, dpi=150) - print(f"Plot saved to: {args.output}") - - -if __name__ == "__main__": - main() diff --git a/test_ffmpeg.py b/test_ffmpeg.py deleted file mode 100644 index 2ed320f..0000000 --- a/test_ffmpeg.py +++ /dev/null @@ -1,81 +0,0 @@ -import argparse -import av -import cv2 -import time - -DEVICE = "/dev/video4" -TARGET_FPS = 30 -WIDTH = 1920 -HEIGHT = 1080 - - -def main(): - parser = argparse.ArgumentParser(description="Capture from V4L2 via PyAV, optionally save to MP4.") - parser.add_argument( - "-o", - "--output", - type=str, - default="", - help="Output MP4 path (e.g. capture.mp4). If empty, only preview, no file.", - ) - args = parser.parse_args() - - # Open v4l2 device with same style options as ffmpeg CLI - container = av.open( - DEVICE, - format="v4l2", - options={ - "input_format": "mjpeg", - "video_size": f"{WIDTH}x{HEIGHT}", - "framerate": str(TARGET_FPS), - }, - ) - - stream = container.streams.video[0] - frame_count = 0 - t_start = time.perf_counter() - - writer = None - if args.output: - fourcc = cv2.VideoWriter_fourcc(*"mp4v") - writer = cv2.VideoWriter( - args.output, fourcc, float(TARGET_FPS), (WIDTH, HEIGHT) - ) - if not writer.isOpened(): - container.close() - raise RuntimeError(f"Could not open VideoWriter for {args.output}") - print(f"Recording to {args.output} @ {TARGET_FPS} fps, {WIDTH}x{HEIGHT}") - - try: - for frame in container.decode(stream): - t0 = time.perf_counter() - img = frame.to_ndarray(format="bgr24") # OpenCV-compatible - t1 = time.perf_counter() - - if writer is not None: - writer.write(img) - - cv2.imshow("pyav_cam", img) - key = cv2.waitKey(1) & 0xFF - t2 = time.perf_counter() - - frame_count += 1 - if frame_count % 30 == 0: - elapsed = t2 - t_start - print( - f"Frame {frame_count}: convert={1000*(t1-t0):.1f}ms, " - f"imshow+waitKey={1000*(t2-t1):.1f}ms, avg_fps={frame_count/elapsed:.1f}" - ) - - if key == ord("q"): - break - finally: - container.close() - if writer is not None: - writer.release() - print(f"Saved {frame_count} frames to {args.output}") - cv2.destroyAllWindows() - - -if __name__ == "__main__": - main() diff --git a/test_of_vis.py b/test_of_vis.py deleted file mode 100644 index 31f6167..0000000 --- a/test_of_vis.py +++ /dev/null @@ -1,121 +0,0 @@ -""" -示例 1: 基础用法 - 单传感器同步处理 - -本示例演示 orisys 的基本使用流程： -1. 初始化传感器（视频文件或摄像头） -2. 循环获取图像并计算形变和触碰点 -3. 读取各类传感器数据 -4. 可视化显示 - - -""" -import os -os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" -import cv2 -import orisys -import numpy as np -import argparse - -def flow_to_colormap(flow): - """Visualize optical flow with hue for direction and value for magnitude.""" - if flow is None or flow.size == 0: - return np.zeros((1, 1, 3), dtype=np.uint8) - - fx = flow[:, :, 0].astype(np.float32, copy=False) - fy = flow[:, :, 1].astype(np.float32, copy=False) - mag, ang = cv2.cartToPolar(fx, fy, angleInDegrees=True) - - hsv = np.zeros(flow.shape[:2] + (3,), dtype=np.uint8) - hsv[:, :, 0] = (ang / 2).astype(np.uint8) - hsv[:, :, 1] = 255 - if mag.max() > 0: - hsv[:, :, 2] = cv2.normalize(mag, None, 0, 255, cv2.NORM_MINMAX).astype(np.uint8) - return cv2.cvtColor(hsv, cv2.COLOR_HSV2BGR) - -def main(): - parser = argparse.ArgumentParser() - parser.add_argument("--video","-v", type=str, default="0", help="Camera ID / Video file path") - parser.add_argument("--config","-c", type=str, default="./config/ddjx01.json", help="Configuration file path") - parser.add_argument("--cal","-cal", type=str, default="./config/ddjx01.npy", help="Calibration file path") - parser.add_argument("--verbose","-verbose", type=bool, default=True, help="Print detailed information") - args = parser.parse_args() - - # 步骤1：创建传感器对象 (必要步骤)， - # config_name为配置文件名（内置配置文件名）或路径（自定义配置文件路径） - # cal_path为校准文件路径, 每一个设备需要创建/使用一个校准文件 - - # 方式1: 从视频文件读取 - try: - input = int(args.video) - is_camera = True - sensor = orisys.Sensor(input, config_name=args.config, cal_path= args.cal, verbose=args.verbose) - except: - is_camera = False - sensor = orisys.Sensor(args.video, config_name=args.config, cal_path= args.cal, verbose=args.verbose) - - print("\n按 'q' 退出程序，按 'r' 重置传感器追踪器") - - while True: - # 步骤2：信息提取 - # 步骤 2.1: 获取并拼接图像 (必要步骤) - sensor.get_img() - # 步骤 2.2: 计算形变场 (必要步骤) - is_touched = sensor.compute_deformation(check_motion=True, threshold=6) - # 步骤 2.3: 计算触碰点 - if is_touched: - sensor.compute_contact() - - - #步骤 2.4: 读取信息 - fps, flow, vnormal, img, contour, centroid, depth_map, fnormal, fshearx, fsheary = sensor.read_info( - sensor.info.FPS, - sensor.info.VRAW, - sensor.info.VNORMAL, - sensor.info.IMG, - sensor.info.CONTOUR, - sensor.info.CENTROID, - sensor.info.DEPTH, - sensor.info.FNORMAL, - sensor.info.FSHEARX, - sensor.info.FSHEARY - ) - - flow_colormap = flow_to_colormap(vnormal) - cv2.imshow("Optical Flow Colormap", flow_colormap) - - # 步骤 3.2: 在图像上触碰点和区域 - image_with_foe = orisys.util.draw_contact(img, contour, centroid) - cv2.imshow("Contact Centroid", image_with_foe) - - # 步骤 3.3: 绘制深度图 - # 归一化散度值到0-255范围（使用绝对值，然后归一化） - div_abs = depth_map - if div_abs.max() > div_abs.min(): - div_normalized = ((div_abs - div_abs.min()) / (div_abs.max() - div_abs.min()) * 255).astype(np.uint8) - else: - div_normalized = np.zeros_like(div_abs, dtype=np.uint8) - - # 应用颜色映射（使用JET：蓝色=低散度，红色=高散度） - 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}, sensor.resolution_raw: {sensor.img_size_raw}") - - #步骤 3.5: 键盘控制 - key = cv2.waitKey(1) & 0xFF - if key == ord("q"): - print("\n退出程序...") - break - if key == ord("r"): - # 重置光流追踪器（当追踪漂移时使用） - sensor.reset() - print("传感器已重置") - - # ========================================================================= - # 4. 清理资源 - # ========================================================================= - sensor.disconnect() # 释放视频源和相关资源 - - -if __name__ == '__main__': - main() diff --git a/test_opencl.py b/test_opencl.py deleted file mode 100644 index 7a0e9a6..0000000 --- a/test_opencl.py +++ /dev/null @@ -1,33 +0,0 @@ -import pyopencl as cl - -# Get all platforms -platforms = cl.get_platforms() -if not platforms: - raise RuntimeError("No OpenCL platforms found") - -# Use the first platform (rusticl in your case) -platform = platforms[0] -print("Using platform:", platform.name) - -# Get devices for that platform -devices = platform.get_devices() -if not devices: - raise RuntimeError("No devices found on platform") - -for platform in cl.get_platforms(): - for device in platform.get_devices(): - print("Device:", device.name) - # if device.type & cl.device_type.GPU == 0: - # print("Skipping non-GPU device:", device.name) - -# Use the first device (your Adreno GPU) -device = devices[0] -print("Using device:", device.name) - -# Create a context with that device -ctx = cl.Context([device]) - -# Create a command queue -queue = cl.CommandQueue(ctx) - -print("Context and command queue created successfully!") \ No newline at end of file diff --git a/test_stitch.py b/test_stitch.py deleted file mode 100644 index 35c4689..0000000 --- a/test_stitch.py +++ /dev/null @@ -1,77 +0,0 @@ -""" -示例 1: 基础用法 - 单传感器同步处理 - -本示例演示 orisys 的基本使用流程： -1. 初始化传感器（视频文件或摄像头） -2. 循环获取图像并计算形变和触碰点 -3. 读取各类传感器数据 -4. 可视化显示 -""" -import os -import argparse - -import cv2 -import orisys - -os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" - - -def main(): - parser = argparse.ArgumentParser() - parser.add_argument("--video", "-v", type=str, default="0", help="Camera ID / Video file path") - parser.add_argument("--config", "-c", type=str, default="./config/ddjx01.json", help="Configuration file path") - parser.add_argument("--cal", "-cal", type=str, default="./config/ddjx01.npy", help="Calibration file path") - parser.add_argument("--verbose", "-verbose", type=bool, default=True, help="Print detailed information") - parser.add_argument( - "--match-brightness", - dest="match_brightness", - action="store_true", - help="Enable per-tile brightness matching", - ) - parser.set_defaults(match_brightness=False) - args = parser.parse_args() - - try: - vid_src = int(args.video) - except ValueError: - vid_src = args.video - - # 步骤1：创建传感器对象 (必要步骤)， - # config_name为配置文件名（内置配置文件名）或路径（自定义配置文件路径） - # cal_path为校准文件路径, 每一个设备需要创建/使用一个校准文件 - sensor = orisys.Sensor(vid_src, config_name=args.config, cal_path=args.cal, verbose=args.verbose) - sensor.stitcher.configure_feature_stitch(match_brightness=args.match_brightness) - - print("\n按 'q' 退出程序，按 'r' 重置传感器追踪器") - - while True: - # 步骤2：信息提取 - # 步骤 2.1: 获取并拼接图像 (必要步骤) - img = sensor.get_img() - if img is None: - print("Failed to get image (end of stream or capture error). Exiting.") - break - - if img is not None: - cv2.imshow("OpenCV Stitch (feature-based)", img) - - # 步骤 2.2: 计算形变场 (必要步骤) — sensor.compute_deformation(...) - # 步骤 2.3: 计算触碰点 - # 步骤 3: 键盘控制 - key = cv2.waitKey(1) & 0xFF - if key == ord("q"): - print("\n退出程序...") - break - if key == ord("r"): - # 重置光流追踪器（当追踪漂移时使用） - sensor.reset() - print("传感器已重置") - - # ========================================================================= - # 4. 清理资源 - # ========================================================================= - sensor.disconnect() # 释放视频源和相关资源 - - -if __name__ == "__main__": - main() diff --git a/test_usb_fps.py b/test_usb_fps.py deleted file mode 100644 index 4560170..0000000 --- a/test_usb_fps.py +++ /dev/null @@ -1,213 +0,0 @@ -#!/usr/bin/env python3 -""" -USB camera format and FPS tester for Jetson. - -- Lists available formats/resolutions via v4l2-ctl. -- For selected modes, opens the camera with OpenCV and measures per-frame - capture time (ms) to show real delivered FPS/latency. -""" - -import argparse -import subprocess -import sys -import time -import statistics -from typing import List, Tuple - -import cv2 - - -def run_v4l2_list_formats(device: str) -> str: - """Return the raw output of v4l2-ctl --list-formats-ext.""" - try: - out = subprocess.check_output( - ["v4l2-ctl", f"--device={device}", "--list-formats-ext"], - stderr=subprocess.STDOUT, - text=True, - ) - return out - except (subprocess.CalledProcessError, FileNotFoundError) as e: - print(f"[WARN] Failed to run v4l2-ctl: {e}") - return "" - - -def parse_formats(output: str) -> List[Tuple[str, int, int, List[float]]]: - """ - Parse v4l2-ctl --list-formats-ext output. - - Returns list of (fourcc, width, height, fps_list). - """ - lines = output.splitlines() - formats: List[Tuple[str, int, int, List[float]]] = [] - - current_fourcc = None - for line in lines: - line = line.strip() - if line.startswith("[") and "):" in line and "'" in line: - # Example: [0]: 'MJPG' (Motion-JPEG, compressed) - try: - fourcc = line.split("'")[1] - current_fourcc = fourcc - except Exception: - current_fourcc = None - elif line.startswith("Size: Discrete") and current_fourcc: - # Example: Size: Discrete 1280x720 - try: - parts = line.split() - wh = parts[-1] - w_str, h_str = wh.split("x") - width = int(w_str) - height = int(h_str) - formats.append((current_fourcc, width, height, [])) - except Exception: - pass - elif "Interval:" in line and "fps" in line and formats: - # Example: Interval: Discrete 0.033s (30.000 fps) - try: - fps_part = line.split("(")[-1].split("fps")[0] - fps = float(fps_part) - fourcc, w, h, fps_list = formats[-1] - fps_list.append(fps) - formats[-1] = (fourcc, w, h, fps_list) - except Exception: - pass - - return formats - - -def fourcc_to_int(code: str) -> int: - return cv2.VideoWriter_fourcc(*code) - - -def benchmark_mode( - device_index: int, - fourcc: str, - width: int, - height: int, - target_fps: float, - warmup: int = 10, - measure_frames: int = 60, -) -> None: - """Open a specific mode and measure per-frame capture time.""" - print(f"\n=== Testing {fourcc} {width}x{height} @ {target_fps:.1f} fps ===") - - cap = cv2.VideoCapture(device_index, cv2.CAP_V4L2) - if not cap.isOpened(): - print(" [ERROR] Failed to open device with V4L2 backend.") - return - - cap.set(cv2.CAP_PROP_FOURCC, fourcc_to_int(fourcc)) - cap.set(cv2.CAP_PROP_FRAME_WIDTH, width) - cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height) - cap.set(cv2.CAP_PROP_FPS, target_fps) - - # Query negotiated parameters - negotiated_w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) - negotiated_h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) - negotiated_fps = cap.get(cv2.CAP_PROP_FPS) - negotiated_fourcc_val = int(cap.get(cv2.CAP_PROP_FOURCC)) - negotiated_fourcc = "".join( - chr((negotiated_fourcc_val >> 8 * i) & 0xFF) for i in range(4) - ) - print( - f" Negotiated: {negotiated_w}x{negotiated_h} {negotiated_fourcc} " - f"@ {negotiated_fps:.1f} fps (backend={cap.getBackendName()})" - ) - - # Warm-up frames (to let auto-exposure, buffers, etc. settle) - for i in range(warmup): - ret, _ = cap.read() - if not ret: - print(f" [WARN] Warmup frame {i} failed.") - time.sleep(0.05) - - times_ms: List[float] = [] - for i in range(measure_frames): - t0 = time.perf_counter() - ret, _ = cap.read() - t1 = time.perf_counter() - if not ret: - print(f" [WARN] Capture failed at frame {i}.") - break - times_ms.append((t1 - t0) * 1000.0) - - cap.release() - - if not times_ms: - print(" [ERROR] No successful frames captured.") - return - - mean_ms = statistics.mean(times_ms) - med_ms = statistics.median(times_ms) - min_ms = min(times_ms) - max_ms = max(times_ms) - eff_fps = 1000.0 / mean_ms - - print( - f" Frames measured: {len(times_ms)}\n" - f" Capture time (ms): mean={mean_ms:.2f}, median={med_ms:.2f}, " - f"min={min_ms:.2f}, max={max_ms:.2f}\n" - f" Effective FPS from capture timing: {eff_fps:.2f}" - ) - - -def main() -> None: - parser = argparse.ArgumentParser( - description="List USB camera formats and benchmark capture times." - ) - parser.add_argument( - "--device", - default="/dev/video4", - help="V4L2 device path (default: /dev/video0)", - ) - parser.add_argument( - "--index", - type=int, - default=0, - help="OpenCV device index (default: 0)", - ) - parser.add_argument( - "--test-modes", - nargs="*", - default=["MJPG:640x480:30", "MJPG:1280x720:30"], - help=( - "Modes to benchmark as FOURCC:WxH:FPS. " - "Example: MJPG:640x480:30 YUYV:640x480:30" - ), - ) - args = parser.parse_args() - - print(f"== v4l2-ctl formats for {args.device} ==") - raw = run_v4l2_list_formats(args.device) - if not raw: - print("No v4l2-ctl output; skipping format list.") - else: - print(raw) - parsed = parse_formats(raw) - if parsed: - print("\n== Parsed formats (FOURCC WxH [fps...]) ==") - for fourcc, w, h, fps_list in parsed: - fps_str = ", ".join(f"{f:.1f}" for f in fps_list) if fps_list else "?" - print(f" {fourcc} {w}x{h} [{fps_str}]") - - print("\n== Benchmarking capture time with OpenCV ==") - for mode_str in args.test_modes: - try: - fourcc_s, wh_s, fps_s = mode_str.split(":") - w_s, h_s = wh_s.lower().split("x") - w = int(w_s) - h = int(h_s) - fps = float(fps_s) - except ValueError: - print(f"[WARN] Cannot parse test mode '{mode_str}', expected FOURCC:WxH:FPS") - continue - - benchmark_mode(args.index, fourcc_s, w, h, fps) - - -if __name__ == "__main__": - try: - main() - except KeyboardInterrupt: - print("\nInterrupted by user.", file=sys.stderr) - diff --git a/patch_generation/3.6mm.svg b/tools/patch_generation/3.6mm.svg similarity index 100% rename from patch_generation/3.6mm.svg rename to tools/patch_generation/3.6mm.svg diff --git a/patch_generation/cmyk_generated_rgb.png b/tools/patch_generation/cmyk_generated_rgb.png similarity index 100% rename from patch_generation/cmyk_generated_rgb.png rename to tools/patch_generation/cmyk_generated_rgb.png diff --git a/patch_generation/cmyk_pattern.tiff b/tools/patch_generation/cmyk_pattern.tiff similarity index 100% rename from patch_generation/cmyk_pattern.tiff rename to tools/patch_generation/cmyk_pattern.tiff diff --git a/patch_generation/dd01.svg b/tools/patch_generation/dd01.svg similarity index 100% rename from patch_generation/dd01.svg rename to tools/patch_generation/dd01.svg diff --git a/patch_generation/patch_generation.py b/tools/patch_generation/patch_generation.py similarity index 100% rename from patch_generation/patch_generation.py rename to tools/patch_generation/patch_generation.py diff --git a/patch_generation/patch_generation_3.py b/tools/patch_generation/patch_generation_3.py similarity index 100% rename from patch_generation/patch_generation_3.py rename to tools/patch_generation/patch_generation_3.py diff --git a/patch_generation/patch_generation_3_org.py b/tools/patch_generation/patch_generation_3_org.py old mode 100755 new mode 100644 similarity index 100% rename from patch_generation/patch_generation_3_org.py rename to tools/patch_generation/patch_generation_3_org.py diff --git a/patch_generation/patchgen_DD01.py b/tools/patch_generation/patchgen_DD01.py old mode 100755 new mode 100644 similarity index 100% rename from patch_generation/patchgen_DD01.py rename to tools/patch_generation/patchgen_DD01.py diff --git a/tools/patch_generation/patchgen_DD02C_dot.py b/tools/patch_generation/patchgen_DD02C_dot.py new file mode 100644 index 0000000..948d6b3 --- /dev/null +++ b/tools/patch_generation/patchgen_DD02C_dot.py @@ -0,0 +1,44 @@ +import numpy as np +from PIL import Image, ImageDraw + +# Parameters +R = 9.0 # mm +H = 20.0 # mm +fov_half = np.radians(75) # 75 degrees half-angle +pixels_per_mm = 100 + +# Texture dimensions +width = int(np.pi * 2 * R * pixels_per_mm) # circumference +height = int(H * pixels_per_mm) + +# Desired dot pattern in image space (normalized) +delta_theta = np.radians(10) # 10° azimuth spacing +delta_r = 0.1 # radial spacing in normalized units +rho = 0.08 # normalized dot radius (from reference) + +# Generate dot centers in image space +r_vals = np.arange(R/H, np.tan(fov_half), delta_r) +theta_vals = np.arange(0, 2*np.pi, delta_theta) + +# Create empty texture (white background) +img = Image.new('L', (width, height), color=255) +draw = ImageDraw.Draw(img) + +for r_img in r_vals: + z = R / r_img + v = z / H + # Ellipse axes in texture coordinates + a_u = (rho * z) / (2 * np.pi * R) + a_v = (rho * z**2) / (R * H) + # Convert to pixels + a_u_px = a_u * width + a_v_px = a_v * height + for theta in theta_vals: + u = theta / (2 * np.pi) + cx = u * width + cy = v * height + # Draw ellipse (bounding box) + bbox = [cx - a_u_px, cy - a_v_px, cx + a_u_px, cy + a_v_px] + draw.ellipse(bbox, fill=0) # black dot + +img.save('cylinder_marker_texture.png') diff --git a/patch_generation/patchgen_DD02M.py b/tools/patch_generation/patchgen_DD02M.py old mode 100755 new mode 100644 similarity index 100% rename from patch_generation/patchgen_DD02M.py rename to tools/patch_generation/patchgen_DD02M.py diff --git a/verify_fft_equivalence.py b/verify_fft_equivalence.py deleted file mode 100644 index a5a17ab..0000000 --- a/verify_fft_equivalence.py +++ /dev/null @@ -1,102 +0,0 @@ -""" -Verify that the OpenCL FFT convolution in solve_opencl / solve_opencl_batch2 -produces the same result as scipy.signal.fftconvolve(f, G, mode='same'). - -This script simulates the OpenCL logic with numpy FFTs so we can compare -directly without needing a GPU. -""" -import numpy as np -from scipy.signal import fftconvolve - -def make_green_2d(H, W, dx=1.0): - """Reproduce PoissonSolver._create_radialGrid for 2D.""" - rdims = (2*H - 1, 2*W - 1) - R = np.indices(rdims, dtype=np.float64) - R[0] += 1 - H - R[1] += 1 - W - R[0] *= dx - R[1] *= dx - np.hypot(R[1], R[0], R[0]) - R = R[0] - zval = 0.5 * (dx + dx) / 2.0 - R[H-1, W-1] = zval - # Apply 2D Green's function: log(r)/(2*pi) - G = np.log(R) * (0.5 / np.pi) - return G - -def test_equivalence(H, W, dx=1.0, seed=42): - np.random.seed(seed) - f = np.random.randn(H, W).astype(np.float64) - G = make_green_2d(H, W, dx) - gdx_prod = dx ** 2 - - # --- CPU reference: scipy fftconvolve --- - p_cpu = fftconvolve(f, G, mode='same') * gdx_prod - - # --- Simulate what OpenCL solve_opencl does --- - f32 = f.astype(np.float32) - G32 = G.astype(np.float32) - full_shape = (2*H - 1, 2*W - 1) - - # G already has shape full_shape, no padding needed - # f is zero-padded to full_shape at [0:H, 0:W] - f_pad = np.zeros(full_shape, dtype=np.float32) - f_pad[:H, :W] = f32 - - # FFT at full_shape, multiply, iFFT - F_f = np.fft.rfftn(f_pad) - F_g = np.fft.rfftn(G32) - conv_circ = np.fft.irfftn(F_f * F_g, s=full_shape) - - # OLD OpenCL extraction (BUGGY): same_starts = [(full_shape[i] - fshape[i]) // 2] - same_starts_old = [(full_shape[0] - H) // 2, (full_shape[1] - W) // 2] - p_old = conv_circ[ - same_starts_old[0]:same_starts_old[0]+H, - same_starts_old[1]:same_starts_old[1]+W - ].astype(np.float32) * np.float32(gdx_prod) - - # FIXED OpenCL extraction: same_starts = [s - 1 for s in fshape] - same_starts_fixed = [H - 1, W - 1] - p_fixed = conv_circ[ - same_starts_fixed[0]:same_starts_fixed[0]+H, - same_starts_fixed[1]:same_starts_fixed[1]+W - ].astype(np.float32) * np.float32(gdx_prod) - - # --- Also test with correct FFT size (3H-2) --- - correct_fft_shape = (H + (2*H-1) - 1, W + (2*W-1) - 1) # (3H-2, 3W-2) - F_f_full = np.fft.rfftn(f32, s=correct_fft_shape) - F_g_full = np.fft.rfftn(G32, s=correct_fft_shape) - conv_full = np.fft.irfftn(F_f_full * F_g_full, s=correct_fft_shape) - # scipy-style 'same' extraction from full linear conv - start_full = [(correct_fft_shape[0] - H) // 2, (correct_fft_shape[1] - W) // 2] - p_full_linear = conv_full[ - start_full[0]:start_full[0]+H, - start_full[1]:start_full[1]+W - ].astype(np.float32) * np.float32(gdx_prod) - - # --- Compare --- - p_cpu32 = p_cpu.astype(np.float32) - scale = max(np.abs(p_cpu32).max(), 1e-10) - - diff_old = np.abs(p_cpu32 - p_old) - diff_fixed = np.abs(p_cpu32 - p_fixed) - diff_full = np.abs(p_cpu32 - p_full_linear) - - print(f"=== Test: f({H},{W}), G({2*H-1},{2*W-1}), full_shape={full_shape} ===") - print(f" scipy fftconvolve full linear conv shape: {correct_fft_shape}") - print(f" p_cpu range: [{p_cpu32.min():.4f}, {p_cpu32.max():.4f}]") - print() - print(f" [OLD BUG] same_starts={same_starts_old}:") - print(f" max_abs_diff = {diff_old.max():.6e} rel_max = {diff_old.max()/scale:.6e}") - print(f" [FIXED] same_starts={same_starts_fixed}:") - print(f" max_abs_diff = {diff_fixed.max():.6e} rel_max = {diff_fixed.max()/scale:.6e}") - print(f" [REF] Full linear conv (FFT size {correct_fft_shape}):") - print(f" max_abs_diff = {diff_full.max():.6e} rel_max = {diff_full.max()/scale:.6e}") - print() - - return diff_old.max(), diff_fixed.max(), diff_full.max() - -if __name__ == '__main__': - for size in [16, 32, 64, 128, 256, 384]: - d_ocl, d_cor, d_full = test_equivalence(size, size) - print()