commit a5a73d76d1b4914ce76ab8cb58370fce1ac5e03d Author: ykcheungab Date: Thu Aug 6 15:18:27 2026 +0800 first commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bb07c7f --- /dev/null +++ b/.gitignore @@ -0,0 +1,14 @@ +__pycache__/ +*.py[cod] +.pytest_cache/ +*.egg-info/ +.venv/ +venv/ +/dist/ +/build/pyinstaller_work/ +/recordings/* +!/recordings/.gitkeep +/tools/ffmpeg/bin/ +*.exe +*.log +.DS_Store diff --git a/README.md b/README.md new file mode 100644 index 0000000..78eb6bc --- /dev/null +++ b/README.md @@ -0,0 +1,68 @@ +# Sensor Bench + +Sensor Bench 是一款用于触觉传感器测试的 Windows 桌面工具。它基于 Orisys SDK,可预览原始相机画面、显示处理后的传感器图像、录制原始视频,并在各测试步骤中监控 FN / FX / FY 力曲线。 + +## 界面示意 + +![Sensor Bench GUI](bench_snapshot.png) + +应用默认使用简体中文界面,可在语言选择器中切换到英文。 + +## 功能 + +- 通过左侧流程引导操作员完成检测,支持 **下一步**、**上一步**、**新样品 / 重置**。 +- 从 `config/options.json` 加载可选项:传感器类型、弹性体、CMOS、标记、压头、平台。 +- 默认使用 DDJX01 Orisys 采集:原始画面 `1920x1080 @ 30 FPS`,处理后输出约 `365x365`。 +- 显示原始 `sensor.frame`、处理后的 `Sensor.get_img()`,以及实时力曲线。 +- 将 SDK 原始相机帧录制为 MP4,保存到 `recordings/` 或自定义文件夹。 +- 界面文案存放在 `config/locales/zh_CN.json` 与 `config/locales/en_US.json`。 + +## 运行环境 + +推荐环境: + +- Windows x64 +- Python 3.10 +- 从 PyPI 安装 `orisys`,或在开发时使用本地 `orisys-sdk` 源码 + +源码运行时安装依赖: + +```powershell +cd D:\repo\sensor-bench +python -m pip install numpy opencv-python PyQt5 pyqtgraph +python -m pip install orisys +$env:PYTHONPATH = "D:\repo\orisys-sdk\src;$env:PYTHONPATH" +python run.py +``` + +使用 PyPI 包时可不设置 `PYTHONPATH`。若存在同级目录 `D:\repo\orisys-sdk\src`,`run.py` 也会自动检测并加入路径。 + +运行冒烟测试: + +```powershell +python tests/test_smoke.py +``` + +## 打包 EXE + +发布包使用 PyInstaller `onedir` 模式,并依赖 Python 3.10 的 conda 环境。 + +1. 创建构建环境: + +```bat +.\build\create_env.bat +``` + +2. 构建可执行文件: + +```bat +.\build\build_exe.bat +``` + +输出路径: + +```text +dist/SensorBench/SensorBench.exe +``` + +打包结果包含 `config/`、`config/locales/` 和 `steps/`,操作员无需改代码即可编辑选项、翻译文案和工作流程。 diff --git a/bench_snapshot.png b/bench_snapshot.png new file mode 100644 index 0000000..a172ba6 Binary files /dev/null and b/bench_snapshot.png differ diff --git a/build/README.md b/build/README.md new file mode 100644 index 0000000..58d25b9 --- /dev/null +++ b/build/README.md @@ -0,0 +1,58 @@ +# Sensor Bench EXE Build + +This folder contains the Windows PyInstaller build setup for Sensor Bench. + +## Requirements + +- Windows x64 +- Anaconda or Miniconda available as `conda` +- Python 3.10 build environment +- Internet access to install `orisys` from PyPI + +The build installs `orisys` from PyPI. Do not use `PYTHONPATH` or an editable `orisys-sdk` checkout for release builds. + +## Build + +From the repository root (cmd or PowerShell): + +```bat +.\build\create_env.bat +.\build\build_exe.bat +``` + +The `.bat` wrappers bypass the local PowerShell execution policy. If you prefer calling PowerShell directly: + +```bat +powershell -NoProfile -ExecutionPolicy Bypass -File .\build\create_env.ps1 +powershell -NoProfile -ExecutionPolicy Bypass -File .\build\build_exe.ps1 +``` + +Output: + +```text +dist/SensorBench/SensorBench.exe +``` + +The build is `onedir` rather than `onefile` so startup is faster and bundled DLL/resource issues are easier to diagnose. + +## Validation + +The build script runs: + +```powershell +python tests/test_smoke.py +python build/verify_dist.py +``` + +Manual hardware check: + +1. Run `dist\SensorBench\SensorBench.exe`. +2. Confirm the UI opens without setting `PYTHONPATH`. +3. Confirm the UI defaults to Simplified Chinese and can switch to English. +4. Confirm Orisys status is available and fixed mode shows DDJX01. +5. Use Next / Revert / New Sample Reset on the left workflow panel. +6. Start preview with camera index `0`. +7. Confirm raw preview and processed preview update. +8. Record a short clip and verify it appears under `dist\SensorBench\recordings`. + +Locale files are bundled under `_internal/config/locales/` (and can also be placed beside the EXE under `config/locales/` for overrides). Edit those JSON files to change operator-facing text without rebuilding. diff --git a/build/build_exe.bat b/build/build_exe.bat new file mode 100644 index 0000000..dfdcd82 --- /dev/null +++ b/build/build_exe.bat @@ -0,0 +1,3 @@ +@echo off +cd /d "%~dp0.." +powershell -NoProfile -ExecutionPolicy Bypass -File "%~dp0build_exe.ps1" %* diff --git a/build/build_exe.ps1 b/build/build_exe.ps1 new file mode 100644 index 0000000..ada20b6 --- /dev/null +++ b/build/build_exe.ps1 @@ -0,0 +1,42 @@ +param( + [string]$EnvName = "sensor-bench-build-py310" +) + +$ErrorActionPreference = "Stop" + +$RepoRoot = Resolve-Path (Join-Path $PSScriptRoot "..") +$DistDir = Join-Path $RepoRoot "dist" +$WorkDir = Join-Path $RepoRoot "build\pyinstaller_work" +$SpecPath = Join-Path $PSScriptRoot "sensor_bench.spec" + +if (-not (Get-Command conda -ErrorAction SilentlyContinue)) { + throw "conda was not found on PATH. Open an Anaconda Prompt or initialize conda for PowerShell." +} + +Push-Location $RepoRoot +try { + conda run -n $EnvName python -m pip install --upgrade pip + conda run -n $EnvName python -m pip install -e ".[pack]" + conda run -n $EnvName python -m pip install --upgrade orisys + + conda run -n $EnvName python tests/test_smoke.py + + if (Test-Path $WorkDir) { + Remove-Item -Recurse -Force $WorkDir + } + if (Test-Path (Join-Path $DistDir "SensorBench")) { + Remove-Item -Recurse -Force (Join-Path $DistDir "SensorBench") + } + + conda run -n $EnvName python -m PyInstaller --noconfirm --clean --distpath "$DistDir" --workpath "$WorkDir" "$SpecPath" + + Copy-Item -Recurse -Force (Join-Path $RepoRoot "config") (Join-Path $DistDir "SensorBench") + Copy-Item -Recurse -Force (Join-Path $RepoRoot "steps") (Join-Path $DistDir "SensorBench") + conda run -n $EnvName python build/verify_dist.py + + Write-Host "Build complete: dist\SensorBench\SensorBench.exe" +} +finally { + Pop-Location +} + diff --git a/build/create_env.bat b/build/create_env.bat new file mode 100644 index 0000000..706d77c --- /dev/null +++ b/build/create_env.bat @@ -0,0 +1,3 @@ +@echo off +cd /d "%~dp0.." +powershell -NoProfile -ExecutionPolicy Bypass -File "%~dp0create_env.ps1" %* diff --git a/build/create_env.ps1 b/build/create_env.ps1 new file mode 100644 index 0000000..602f340 --- /dev/null +++ b/build/create_env.ps1 @@ -0,0 +1,30 @@ +param( + [string]$EnvName = "sensor-bench-build-py310", + [string]$PythonVersion = "3.10" +) + +$ErrorActionPreference = "Stop" + +$RepoRoot = Resolve-Path (Join-Path $PSScriptRoot "..") + +if (-not (Get-Command conda -ErrorAction SilentlyContinue)) { + throw "conda was not found on PATH. Open an Anaconda Prompt or initialize conda for PowerShell." +} + +$envList = conda env list | Out-String +if ($envList -notmatch "(?m)^\s*$([regex]::Escape($EnvName))\s") { + conda create -y -n $EnvName python=$PythonVersion +} + +conda run -n $EnvName python -m pip install --upgrade pip +Push-Location $RepoRoot +try { + conda run -n $EnvName python -m pip install -e ".[pack]" +} +finally { + Pop-Location +} + +Write-Host "Build environment ready: $EnvName" +Write-Host "Orisys will be installed from PyPI by build_exe.ps1." + diff --git a/build/sensor_bench.spec b/build/sensor_bench.spec new file mode 100644 index 0000000..9a6e13b --- /dev/null +++ b/build/sensor_bench.spec @@ -0,0 +1,98 @@ +# -*- mode: python ; coding: utf-8 -*- + +from pathlib import Path + +from PyInstaller.utils.hooks import collect_all, collect_submodules + + +repo_root = Path.cwd() +src_dir = repo_root / "src" +work_dir = repo_root / "build" / "pyinstaller_work" + +datas = [ + (str(repo_root / "config"), "config"), + (str(repo_root / "steps"), "steps"), +] +binaries = [] +hiddenimports = [ + "sensor_bench.main", + "sensor_bench.app", + "sensor_bench.acquisition", + "sensor_bench.config_loader", + "sensor_bench.i18n", + "sensor_bench.paths", + "sensor_bench.plot_history", + "sensor_bench.recorder", + "orisys", + "orisys_blend", + "scipy.fft", + "scipy.ndimage", + "scipy.sparse.linalg", + "numba", + "llvmlite", +] + +for package in ("PyQt5", "pyqtgraph", "cv2", "orisys", "orisys_blend", "numba", "llvmlite"): + pkg_datas, pkg_binaries, pkg_hidden = collect_all(package) + datas += pkg_datas + binaries += pkg_binaries + hiddenimports += pkg_hidden + +hiddenimports += collect_submodules("orisys_blend") + +excludes = [ + "cupy", + "cupyx", + "pyopencl", + "open3d", + "matplotlib", + "tkinter", + "pytest", + "pip", + "wheel", +] + +a = Analysis( + [str(repo_root / "run.py")], + pathex=[str(repo_root), str(src_dir)], + binaries=binaries, + datas=datas, + hiddenimports=hiddenimports, + hookspath=[], + hooksconfig={}, + runtime_hooks=[], + excludes=excludes, + noarchive=False, + optimize=0, +) + +pyz = PYZ(a.pure) + +exe = EXE( + pyz, + a.scripts, + [], + exclude_binaries=True, + name="SensorBench", + debug=False, + bootloader_ignore_signals=False, + strip=False, + upx=True, + console=False, + disable_windowed_traceback=False, + argv_emulation=False, + target_arch=None, + codesign_identity=None, + entitlements_file=None, +) + +coll = COLLECT( + exe, + a.binaries, + a.datas, + strip=False, + upx=True, + upx_exclude=[], + name="SensorBench", +) + diff --git a/build/vendor/.gitkeep b/build/vendor/.gitkeep new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/build/vendor/.gitkeep @@ -0,0 +1 @@ + diff --git a/build/vendor/orisys-0.4.2-cp310-cp310-win_amd64.whl b/build/vendor/orisys-0.4.2-cp310-cp310-win_amd64.whl new file mode 100644 index 0000000..da7e87c Binary files /dev/null and b/build/vendor/orisys-0.4.2-cp310-cp310-win_amd64.whl differ diff --git a/build/verify_dist.py b/build/verify_dist.py new file mode 100644 index 0000000..17e68eb --- /dev/null +++ b/build/verify_dist.py @@ -0,0 +1,31 @@ +"""Verify the PyInstaller onedir output has the expected portable layout.""" + +from __future__ import annotations + +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +DIST = ROOT / "dist" / "SensorBench" + + +def require(path: Path) -> None: + if not path.exists(): + raise FileNotFoundError(f"Missing build artifact: {path}") + + +def main() -> int: + require(DIST / "SensorBench.exe") + require(DIST / "config" / "options.json") + require(DIST / "config" / "locales" / "zh_CN.json") + require(DIST / "config" / "locales" / "en_US.json") + require(DIST / "steps" / "default.json") + require(DIST / "_internal") + print(f"Verified SensorBench dist layout: {DIST}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) + diff --git a/config/compound_eye.png b/config/compound_eye.png new file mode 100644 index 0000000..ab961f4 Binary files /dev/null and b/config/compound_eye.png differ diff --git a/config/ddjx01.json b/config/ddjx01.json new file mode 100644 index 0000000..6b66456 --- /dev/null +++ b/config/ddjx01.json @@ -0,0 +1,46 @@ +{ + "name": "prototype", + "version": "1.0", + "camera": { + "resolution_raw": [1920, 1080], + "fps": 30, + "units": 15, + "units_used": 9, + "resolution_stitch": [365, 365], + "stitch_res_list": [365,263,123,63] + }, + "stitcher": { + "min_area_frac": 0.01, + "max_area_frac": 0.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], + "blend":"numba", + "blend_list": ["numpy", "numba","none"] + }, + "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": 1 + }, + "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/config/locales/en_US.json b/config/locales/en_US.json new file mode 100644 index 0000000..29bf5ec --- /dev/null +++ b/config/locales/en_US.json @@ -0,0 +1,123 @@ +{ + "app": { + "title": "Sensor Bench", + "language": "Language", + "lang_zh": "中文", + "lang_en": "English" + }, + "panels": { + "instructions": "Instructions", + "options": "Sensor Component Types", + "camera": "Camera (Orisys SDK)", + "recording": "Recording", + "raw": "Raw Image (sensor.frame)", + "processed": "Processed Image (Sensor.get_img)", + "force": "Force Curves" + }, + "labels": { + "sensor_type": "Sensor type", + "elastomer": "Elastomer", + "cmos": "CMOS", + "marker": "Marker", + "indentor": "Indenter", + "platform": "Platform", + "camera_index": "Camera index", + "backend": "Backend", + "fixed_mode": "Fixed mode: {mode}", + "fixed_mode_unset": "Fixed mode: set config path", + "fixed_mode_invalid": "Fixed mode: invalid config ({error})", + "requested": "Requested", + "actual": "Actual", + "capture_fps": "Capture FPS", + "render_fps": "Render FPS", + "proc_ms": "Proc ms", + "fn": "FN", + "fx": "FX", + "fy": "FY", + "no_preview": "No preview", + "processed_placeholder": "Start Preview to show Orisys output", + "not_recording": "Not recording", + "force_plot": "Force" + }, + "buttons": { + "config": "Config…", + "cal": "Cal…", + "start_preview": "Start Preview", + "stop_preview": "Stop Preview", + "browse": "Browse…", + "start_record": "Start Record", + "stop_record": "Stop Record", + "next": "Next", + "revert": "Revert", + "reset": "New Sample / Reset" + }, + "placeholders": { + "config": "DDJX01 config JSON path or name", + "cal": "Optional calibration .npy" + }, + "dialogs": { + "select_config": "Select config JSON", + "select_cal": "Select calibration", + "select_recordings": "Select recordings folder", + "orisys_title": "Orisys", + "orisys_required": "orisys SDK is required for preview.\n\n{detail}", + "config_title": "Config", + "config_required": "Select a DDJX01 config JSON first.", + "config_invalid": "Config could not be loaded.", + "capture_title": "Capture", + "recording_title": "Recording", + "no_frame": "No frame yet." + }, + "status": { + "ready": "Ready — Start Preview uses Orisys DDJX01 fixed format.", + "orisys_ok": "orisys available (version {version})", + "orisys_missing": "orisys not importable — add orisys-sdk/src to PYTHONPATH.", + "opening": "Opening Orisys ({mode})…", + "preview_stopped": "Preview stopped.", + "capture_error": "Capture error: {message}", + "frame_info": "{raw_w}x{raw_h} raw / {proc_w}x{proc_h} processed @ {fps:.1f} FPS", + "recording": "Recording… {name}\nOptions: sensor={sensor_type}, elastomer={elastomer}, cmos={cmos}, marker={marker}, indenter={indentor}, platform={platform}", + "recording_frames": "Recording… {name}\nFrames: {count}", + "saved": "Saved {count} frames → {path}", + "workflow_done": "All steps completed. Click New Sample / Reset for the next unit." + }, + "workflow": { + "title": "Sensor Bench Inspection Workflow", + "current_marker": "Current", + "done_marker": "Done", + "steps": { + "power_on": { + "title": "Power on and connect", + "body": "Connect the sensor USB cable and wait until Windows recognizes the device." + }, + "select_camera": { + "title": "Start Orisys preview", + "body": "Confirm camera index (usually 0) and DDJX01 config, then click Start Preview. Format is fixed by the Orisys config (1920x1080 @ 30)." + }, + "set_options": { + "title": "Set sensor options", + "body": "Choose elastomer type, CMOS type, and marker type that match the physical unit under test." + }, + "check_preview": { + "title": "Check raw and processed preview", + "body": "Confirm raw frame is ~1920x1080 and processed Orisys image is ~365x365. Inspect focus, lighting, and image cleanliness." + }, + "record": { + "title": "Record raw data", + "body": "Choose a storage folder, start recording, exercise the sensor, then stop recording." + }, + "review_curves": { + "title": "Review performance curves", + "body": "Watch capture/render FPS and force curves for drops or unstable exposure before signing off." + }, + "sensitivity": { + "title": "Sensitivity test", + "body": "Choose a small indenter and apply a 2g weight, then watch for a clear force-curve change." + }, + "max_load": { + "title": "Maximum load test", + "body": "Choose a flat plate indenter and apply weights from 10g to 500g to watch for curve changes. Use 10g to 500g for MINI and 2kg for BOX." + } + } + } +} diff --git a/config/locales/zh_CN.json b/config/locales/zh_CN.json new file mode 100644 index 0000000..bd7e432 --- /dev/null +++ b/config/locales/zh_CN.json @@ -0,0 +1,123 @@ +{ + "app": { + "title": "传感器检测台", + "language": "语言", + "lang_zh": "中文", + "lang_en": "English" + }, + "panels": { + "instructions": "操作步骤", + "options": "传感器组件类型", + "camera": "相机(Orisys SDK)", + "recording": "录制", + "raw": "原始图像(sensor.frame)", + "processed": "处理后图像(Sensor.get_img)", + "force": "力曲线" + }, + "labels": { + "sensor_type": "传感器类型", + "elastomer": "弹性体", + "cmos": "CMOS", + "marker": "图案", + "indentor": "压头", + "platform": "平台", + "camera_index": "相机索引", + "backend": "后端", + "fixed_mode": "固定模式:{mode}", + "fixed_mode_unset": "固定模式:请设置配置路径", + "fixed_mode_invalid": "固定模式:配置无效({error})", + "requested": "请求", + "actual": "实际", + "capture_fps": "采集 FPS", + "render_fps": "渲染 FPS", + "proc_ms": "处理 ms", + "fn": "FN", + "fx": "FX", + "fy": "FY", + "no_preview": "暂无预览", + "processed_placeholder": "点击开始预览以显示 Orisys 输出", + "not_recording": "未在录制", + "force_plot": "力" + }, + "buttons": { + "config": "配置…", + "cal": "标定…", + "start_preview": "开始预览", + "stop_preview": "停止预览", + "browse": "浏览…", + "start_record": "开始录制", + "stop_record": "停止录制", + "next": "下一步", + "revert": "上一步", + "reset": "新样品 / 重置" + }, + "placeholders": { + "config": "DDJX01 配置 JSON 路径或名称", + "cal": "可选标定文件 .npy" + }, + "dialogs": { + "select_config": "选择配置 JSON", + "select_cal": "选择标定文件", + "select_recordings": "选择录制文件夹", + "orisys_title": "Orisys", + "orisys_required": "预览需要 Orisys SDK。\n\n{detail}", + "config_title": "配置", + "config_required": "请先选择 DDJX01 配置 JSON。", + "config_invalid": "无法加载配置。", + "capture_title": "采集", + "recording_title": "录制", + "no_frame": "尚无可用帧。" + }, + "status": { + "ready": "就绪 — 开始预览将使用 Orisys DDJX01 固定格式。", + "orisys_ok": "orisys 可用(版本 {version})", + "orisys_missing": "无法导入 orisys — 请将 orisys-sdk/src 加入 PYTHONPATH。", + "opening": "正在打开 Orisys({mode})…", + "preview_stopped": "预览已停止。", + "capture_error": "采集错误:{message}", + "frame_info": "{raw_w}x{raw_h} 原始 / {proc_w}x{proc_h} 处理 @ {fps:.1f} FPS", + "recording": "正在录制… {name}\n选项:传感器={sensor_type},弹性体={elastomer},CMOS={cmos},标记={marker},压头={indentor},平台={platform}", + "recording_frames": "正在录制… {name}\n帧数:{count}", + "saved": "已保存 {count} 帧 → {path}", + "workflow_done": "全部步骤已完成。可点击“新样品 / 重置”开始下一个样品。" + }, + "workflow": { + "title": "传感器检测流程", + "current_marker": "当前", + "done_marker": "已完成", + "steps": { + "power_on": { + "title": "开机并连接", + "body": "连接传感器 USB 线缆,等待 Windows 识别设备。" + }, + "select_camera": { + "title": "启动 Orisys 预览", + "body": "确认相机索引(通常为 0)和 DDJX01 配置,然后点击开始预览。格式由 Orisys 配置固定(1920x1080 @ 30)。" + }, + "set_options": { + "title": "设置传感器选项", + "body": "选择与当前被测实物一致的弹性体类型、CMOS 类型和标记类型。" + }, + "check_preview": { + "title": "检查原始与处理后预览", + "body": "确认原始画面约为 1920x1080,处理后 Orisys 图像约为 365x365。检查对焦、光照和画面清洁度。" + }, + "record": { + "title": "录制原始数据", + "body": "选择存储文件夹,开始录制,操作传感器,然后停止录制。" + }, + "review_curves": { + "title": "查看性能曲线", + "body": "观察采集/渲染 FPS 与力曲线,确认无明显掉帧或曝光不稳定后再签字确认。" + }, + "sensitivity": { + "title": "灵敏度测试", + "body": "选择小压头,并加载 2g 砝码,观察力曲线是否有清晰变化。" + }, + "max_load": { + "title": "最大载荷测试", + "body": "选择平板压头,并加载 10g 到 500g 的砝码观察曲线变化;MINI 使用 10g 到 500g,BOX 使用 2kg。" + } + } + } +} diff --git a/config/options.json b/config/options.json new file mode 100644 index 0000000..08f6eeb --- /dev/null +++ b/config/options.json @@ -0,0 +1,27 @@ +{ + "sensor_types": [ + {"id": "BOX", "label": "BOX"}, + {"id": "MINI", "label": "MINI"} + ], + "elastomer_types": [ + {"id": "shoreA20", "label": "冯总20度"}, + {"id": "shore0030", "label": "0度"}, + {"id": "solaris", "label": "铂金"} + ], + "cmos_types": [ + {"id": "ov2710", "label": "OV2710"}, + {"id": "ov9745", "label": "OV9745"} + ], + "marker_types": [ + {"id": "4color", "label": "4Color"} + ], + "indentor_types": [ + {"id": "1mm_circle", "label": "1mm圆柱"}, + {"id": "1mm_ball", "label": "1mm球"} + ], + "platoform_types": [ + {"id": "manual", "label": "手动"}, + {"id": "single_axis", "label": "单轴"}, + {"id": "six_axis", "label": "六轴"} + ] +} diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..a722ba5 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,31 @@ +[build-system] +requires = ["setuptools>=61.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "sensor-bench" +version = "0.1.0" +description = "Operator tool for connecting sensors, viewing raw data, recording, and inspecting performance" +readme = "README.md" +requires-python = ">=3.10" +authors = [{ name = "Orisys-Tech" }] +dependencies = [ + "numpy>=1.24.4", + "opencv-python>=4.10.0.84", + "PyQt5>=5.15", + "pyqtgraph>=0.13", +] + +[project.optional-dependencies] +pack = [ + "pyinstaller>=6.0", +] + +[project.scripts] +sensor-bench = "sensor_bench.main:main" + +[tool.setuptools] +package-dir = { "" = "src" } + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/recordings/.gitkeep b/recordings/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/run.bat b/run.bat new file mode 100644 index 0000000..d30c431 --- /dev/null +++ b/run.bat @@ -0,0 +1,3 @@ +@echo off +cd /d "%~dp0" +python "%~dp0run.py" %* diff --git a/run.py b/run.py new file mode 100644 index 0000000..3387069 --- /dev/null +++ b/run.py @@ -0,0 +1,27 @@ +#!/usr/bin/env python3 +"""Launch Sensor Bench without installing the package. + +Usage (from the repository root): + python run.py +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parent +SRC = ROOT / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +# Prefer sibling orisys-sdk checkout for Sensor-owned preview. +ORISYS_SRC = ROOT.parent / "orisys-sdk" / "src" +if ORISYS_SRC.is_dir() and str(ORISYS_SRC) not in sys.path: + sys.path.insert(0, str(ORISYS_SRC)) + +from sensor_bench.main import main # noqa: E402 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/sensor_bench/__init__.py b/src/sensor_bench/__init__.py new file mode 100644 index 0000000..e75c543 --- /dev/null +++ b/src/sensor_bench/__init__.py @@ -0,0 +1,3 @@ +"""Sensor Bench: operator tooling for raw sensor inspection and recording.""" + +__version__ = "0.1.0" diff --git a/src/sensor_bench/acquisition.py b/src/sensor_bench/acquisition.py new file mode 100644 index 0000000..fd599c7 --- /dev/null +++ b/src/sensor_bench/acquisition.py @@ -0,0 +1,290 @@ +"""Orisys SDK-owned fixed-format acquisition for Sensor Bench. + +Camera open/read/process is owned by ``orisys.Sensor`` using the DDJX01 +config (1920x1080 @ 30 FPS MJPG by default). Sensor Bench only displays +``sensor.frame`` (raw) and ``get_img()`` (processed) plus force metrics. +""" + +from __future__ import annotations + +import json +import threading +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Callable, Optional + +import cv2 +import numpy as np + + +@dataclass(frozen=True) +class FixedCameraMode: + """Stream mode taken from the Orisys config JSON.""" + + width: int + height: int + fps: float + config_ref: str + stitch_width: int = 0 + stitch_height: int = 0 + + def label(self) -> str: + return f"{self.width}x{self.height} @{self.fps:g} (DDJX01 / MJPG)" + + +@dataclass +class SdkPacket: + """One acquisition tick from the Orisys Sensor.""" + + raw_frame: np.ndarray + processed_frame: np.ndarray + timestamp: float + fps: float + process_ms: float + width: int + height: int + processed_width: int + processed_height: int + normal_force: float = 0.0 + shear_x: float = 0.0 + shear_y: float = 0.0 + session_id: int = 0 + mode: Optional[FixedCameraMode] = None + + +def orisys_importable() -> tuple[bool, str]: + try: + import orisys # noqa: F401 + + return True, getattr(__import__("orisys"), "__version__", "unknown") + except Exception as exc: # noqa: BLE001 + return False, str(exc) + + +def load_fixed_mode(config_ref: str | Path) -> FixedCameraMode: + ref = str(config_ref).strip() + path = Path(ref) + if path.exists(): + data = json.loads(path.read_text(encoding="utf-8")) + resolved_ref = str(path.resolve()) + else: + try: + import orisys + + data = orisys.configs.load_config(ref) + resolved_ref = ref + except Exception as exc: # noqa: BLE001 + raise FileNotFoundError( + f"Orisys config not found as a file or bundled config name: {ref}" + ) from exc + camera = data.get("camera") or {} + raw = camera.get("resolution_raw") or [1920, 1080] + stitch = camera.get("resolution_stitch") or [0, 0] + fps = float(camera.get("fps") or 30.0) + if not isinstance(raw, (list, tuple)) or len(raw) < 2: + raise ValueError(f"Invalid camera.resolution_raw in {ref}") + return FixedCameraMode( + width=int(raw[0]), + height=int(raw[1]), + fps=fps, + config_ref=resolved_ref, + stitch_width=int(stitch[0]) if len(stitch) >= 2 else 0, + stitch_height=int(stitch[1]) if len(stitch) >= 2 else 0, + ) + + +def _ensure_bgr(img: np.ndarray) -> np.ndarray: + if img.ndim == 2: + return cv2.cvtColor(img, cv2.COLOR_GRAY2BGR) + if img.ndim == 3 and img.shape[2] == 1: + return cv2.cvtColor(img[:, :, 0], cv2.COLOR_GRAY2BGR) + if img.ndim == 3 and img.shape[2] == 4: + return cv2.cvtColor(img, cv2.COLOR_BGRA2BGR) + return img + + +class SdkAcquisitionWorker: + """Background loop: ``Sensor.get_img()`` + deformation + contact.""" + + def __init__( + self, + *, + on_error: Optional[Callable[[str], None]] = None, + on_status: Optional[Callable[[str], None]] = None, + ): + self._on_error = on_error + self._on_status = on_status + self._thread: Optional[threading.Thread] = None + self._stop = threading.Event() + self._lock = threading.Lock() + self._latest: Optional[SdkPacket] = None + self._session_id = 0 + self._sensor = None + self._mode: Optional[FixedCameraMode] = None + self._video_source: Optional[str | int] = None + self._cal_path: Optional[str] = None + self._backend = "auto" + + @property + def is_running(self) -> bool: + return self._thread is not None and self._thread.is_alive() + + @property + def session_id(self) -> int: + return self._session_id + + @property + def mode(self) -> Optional[FixedCameraMode]: + return self._mode + + def latest(self) -> Optional[SdkPacket]: + with self._lock: + return self._latest + + def start( + self, + video_source: str | int = 0, + *, + config_path: str | Path, + cal_path: Optional[str | Path] = None, + backend: str = "auto", + ) -> FixedCameraMode: + self.stop() + ok, detail = orisys_importable() + if not ok: + raise RuntimeError( + "orisys SDK is not importable. Install/build orisys-sdk and ensure " + f"it is on PYTHONPATH. Detail: {detail}" + ) + mode = load_fixed_mode(config_path) + cal = Path(cal_path) if cal_path else None + if cal is not None and not cal.exists(): + raise FileNotFoundError(f"Calibration not found: {cal}") + + self._mode = mode + self._video_source = video_source + self._cal_path = str(cal) if cal is not None else None + self._backend = backend or "auto" + self._latest = None + self._session_id += 1 + session = self._session_id + self._stop.clear() + self._thread = threading.Thread( + target=self._run, + name=f"SdkAcquisition-{session}", + daemon=True, + args=(session,), + ) + self._thread.start() + return mode + + def stop(self) -> None: + self._stop.set() + thread = self._thread + if thread is not None and thread.is_alive(): + thread.join(timeout=5.0) + self._thread = None + self._close_sensor() + with self._lock: + self._latest = None + + def _close_sensor(self) -> None: + sensor = self._sensor + self._sensor = None + if sensor is None: + return + try: + sensor.disconnect() + except Exception: # noqa: BLE001 + pass + + def _status(self, message: str) -> None: + if self._on_status is not None: + self._on_status(message) + + def _run(self, session: int) -> None: + import orisys + + mode = self._mode + source = self._video_source + if mode is None or source is None: + return + try: + self._status(f"Opening Orisys Sensor ({mode.label()})…") + kwargs = { + "config_name": mode.config_ref, + "verbose": False, + "backend": self._backend, + } + if self._cal_path: + kwargs["cal_path"] = self._cal_path + self._sensor = orisys.Sensor(source, **kwargs) + self._status(f"Streaming {mode.label()}") + except Exception as exc: # noqa: BLE001 + if self._on_error is not None and session == self._session_id: + self._on_error(str(exc)) + return + + frame_times: list[float] = [] + while not self._stop.is_set() and session == self._session_id: + try: + t0 = time.perf_counter() + processed = self._sensor.get_img() + if processed is None: + time.sleep(0.005) + continue + raw = getattr(self._sensor, "frame", None) + if raw is None: + time.sleep(0.005) + continue + + try: + self._sensor.compute_deformation(decompose=True, check_motion=False) + self._sensor.compute_contact() + except Exception: # noqa: BLE001 + pass + + t1 = time.perf_counter() + frame_times.append(t1) + cutoff = t1 - 1.0 + frame_times = [t for t in frame_times if t >= cutoff] + fps = float(len(frame_times)) if frame_times else 0.0 + + fn = fx = fy = 0.0 + try: + fn, fx, fy = self._sensor.read_info( + self._sensor.info.FNORMAL, + self._sensor.info.FSHEARX, + self._sensor.info.FSHEARY, + ) + except Exception: # noqa: BLE001 + pass + + raw_bgr = _ensure_bgr(raw) + proc_bgr = _ensure_bgr(processed) + packet = SdkPacket( + raw_frame=raw_bgr, + processed_frame=proc_bgr, + timestamp=time.time(), + fps=fps, + process_ms=(t1 - t0) * 1000.0, + width=int(raw_bgr.shape[1]), + height=int(raw_bgr.shape[0]), + processed_width=int(proc_bgr.shape[1]), + processed_height=int(proc_bgr.shape[0]), + normal_force=float(fn), + shear_x=float(fx), + shear_y=float(fy), + session_id=session, + mode=mode, + ) + with self._lock: + self._latest = packet + except Exception as exc: # noqa: BLE001 + if self._on_error is not None and session == self._session_id: + self._on_error(str(exc)) + break + + self._close_sensor() + self._status("Acquisition stopped") diff --git a/src/sensor_bench/app.py b/src/sensor_bench/app.py new file mode 100644 index 0000000..e504ea9 --- /dev/null +++ b/src/sensor_bench/app.py @@ -0,0 +1,1014 @@ +"""Main Sensor Bench window: SDK-owned raw/processed preview, recording, curves.""" + +from __future__ import annotations + +import time +from pathlib import Path +from typing import Optional + +import cv2 +import numpy as np +from PyQt5.QtCore import Qt, QObject, QTimer, pyqtSignal +from PyQt5.QtGui import QImage, QPixmap +from PyQt5.QtWidgets import ( + QComboBox, + QFileDialog, + QFrame, + QGroupBox, + QHBoxLayout, + QLabel, + QLineEdit, + QMainWindow, + QMessageBox, + QPushButton, + QScrollArea, + QSizePolicy, + QSpinBox, + QSplitter, + QStatusBar, + QVBoxLayout, + QWidget, +) + +try: + import pyqtgraph as pg +except Exception as exc: # pragma: no cover + raise ImportError( + "PyQtGraph is required. Install with: pip install pyqt5 pyqtgraph" + ) from exc + +from .acquisition import ( + FixedCameraMode, + SdkAcquisitionWorker, + SdkPacket, + load_fixed_mode, + orisys_importable, +) +from .config_loader import SensorOptions, WorkflowSteps, load_sensor_options, load_workflow_steps +from .i18n import DEFAULT_LOCALE, Translator +from .paths import ( + default_ddjx01_calibration, + default_ddjx01_config, + default_locales_dir, + default_options_path, + default_recordings_dir, + default_steps_path, + project_root, +) +from .plot_history import MetricHistory +from .recorder import RawRecorder + + +STYLE = """ +QMainWindow, QWidget { + background-color: #1e1f22; + color: #e8eaed; + font-family: "Segoe UI", "Microsoft YaHei", Arial, sans-serif; + font-size: 13px; +} +QGroupBox { + border: 1px solid #3c4048; + border-radius: 6px; + margin-top: 10px; + padding-top: 8px; + font-weight: 600; +} +QGroupBox::title { + subcontrol-origin: margin; + left: 10px; + padding: 0 4px; + color: #9aa0a6; +} +QPushButton { + background-color: #2d3139; + border: 1px solid #4a505a; + border-radius: 4px; + padding: 6px 12px; + min-height: 28px; +} +QPushButton:hover { background-color: #3a404c; } +QPushButton:pressed { background-color: #252930; } +QPushButton:disabled { color: #6b7280; background-color: #25262a; } +QPushButton#btnPrimary { + background-color: #1a73e8; + border-color: #1a73e8; + color: white; + font-weight: 600; +} +QPushButton#btnPrimary:hover { background-color: #1967d2; } +QPushButton#btnDanger { + background-color: #c5221f; + border-color: #c5221f; + color: white; + font-weight: 600; +} +QPushButton#btnDanger:hover { background-color: #a50e0e; } +QComboBox, QLineEdit, QSpinBox { + background-color: #2a2d33; + border: 1px solid #4a505a; + border-radius: 4px; + padding: 4px; +} +QLabel#metaValue { color: #8ab4f8; font-weight: 600; } +QLabel#previewPlaceholder { + color: #6b7280; + background-color: #121316; + border: 1px dashed #3c4048; + border-radius: 6px; +} +QScrollArea { border: none; } +QStatusBar { color: #9aa0a6; } +QFrame#stepCurrent { + background-color: #243447; + border: 1px solid #8ab4f8; + border-radius: 6px; + padding: 6px; +} +QFrame#stepDone { + background-color: #23262b; + border: 1px solid #3c4048; + border-radius: 6px; + padding: 6px; +} +QFrame#stepUpcoming { + background-color: #2a2d33; + border: 1px solid #3c4048; + border-radius: 6px; + padding: 6px; +} +""" + + +def ndarray_to_pixmap(frame: np.ndarray) -> QPixmap: + if frame.ndim == 2 or (frame.ndim == 3 and frame.shape[2] == 1): + gray = frame if frame.ndim == 2 else frame[:, :, 0] + h, w = gray.shape[:2] + image = QImage(gray.data, w, h, w, QImage.Format_Grayscale8) + return QPixmap.fromImage(image.copy()) + if frame.ndim == 3 and frame.shape[2] == 4: + rgba = cv2.cvtColor(frame, cv2.COLOR_BGRA2RGBA) + h, w = rgba.shape[:2] + image = QImage(rgba.data, w, h, w * 4, QImage.Format_RGBA8888) + return QPixmap.fromImage(image.copy()) + rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) + h, w = rgb.shape[:2] + image = QImage(rgb.data, w, h, w * 3, QImage.Format_RGB888) + return QPixmap.fromImage(image.copy()) + + +class _AcquisitionBridge(QObject): + error = pyqtSignal(str) + status = pyqtSignal(str) + + +class MainWindow(QMainWindow): + def __init__( + self, + *, + options_path: Optional[Path] = None, + steps_path: Optional[Path] = None, + locales_dir: Optional[Path] = None, + locale: str = DEFAULT_LOCALE, + ): + super().__init__() + self.resize(1480, 900) + self.setStyleSheet(STYLE) + + self.options_path = options_path or default_options_path() + self.steps_path = steps_path or default_steps_path() + self.recordings_dir = default_recordings_dir() + self.tr = Translator(locales_dir or default_locales_dir(), locale=locale) + + self.options: SensorOptions = load_sensor_options(self.options_path) + self.workflow: WorkflowSteps = load_workflow_steps(self.steps_path) + + self._acq_bridge = _AcquisitionBridge() + self._acq_bridge.error.connect(self._on_acquisition_error) + self._acq_bridge.status.connect(self._on_acquisition_status) + + self.acquisition = SdkAcquisitionWorker( + on_error=lambda m: self._acq_bridge.error.emit(m), + on_status=lambda m: self._acq_bridge.status.emit(m), + ) + self.recorder = RawRecorder() + + self.force_history = MetricHistory(npoints=180, channels=3, window_sec=6.0) + + self._raw_pixmap: Optional[QPixmap] = None + self._proc_pixmap: Optional[QPixmap] = None + self._last_packet: Optional[SdkPacket] = None + self._awaiting_first_frame = False + self._render_times: list[float] = [] + self._active_session = 0 + self._fixed_mode: Optional[FixedCameraMode] = None + self._last_packet_id = 0 + self._current_step_index = 0 + self._step_cards: list[QFrame] = [] + self._step_title_labels: list[QLabel] = [] + self._step_body_labels: list[QLabel] = [] + self._step_state_labels: list[QLabel] = [] + + self._build_ui() + self._create_plots() + self._populate_options() + self._populate_steps() + self._apply_language() + self._refresh_fixed_mode() + self._refresh_orisys_status() + self._update_workflow_controls() + + self.timer = QTimer(self) + self.timer.timeout.connect(self._on_tick) + self.timer.start(16) + + def t(self, key: str, **kwargs) -> str: + return self.tr.t(key, **kwargs) + + # ------------------------------------------------------------------ UI + def _build_ui(self) -> None: + central = QWidget() + self.setCentralWidget(central) + root = QHBoxLayout(central) + root.setContentsMargins(8, 8, 8, 8) + root.setSpacing(8) + + splitter = QSplitter(Qt.Horizontal) + root.addWidget(splitter) + + # Full-height workflow column. + splitter.addWidget(self._build_steps_panel()) + + # Controls + previews column. + right = QWidget() + right_layout = QVBoxLayout(right) + right_layout.setContentsMargins(0, 0, 0, 0) + right_layout.setSpacing(8) + + top = QHBoxLayout() + top.addWidget(self._build_language_panel(), 0) + top.addWidget(self._build_options_panel(), 1) + top.addWidget(self._build_camera_panel(), 2) + top.addWidget(self._build_record_panel(), 1) + right_layout.addLayout(top, 0) + + images = QSplitter(Qt.Horizontal) + images.addWidget(self._build_preview_panel()) + images.addWidget(self._build_processed_panel()) + images.setStretchFactor(0, 1) + images.setStretchFactor(1, 1) + right_layout.addWidget(images, 3) + right_layout.addWidget(self._build_plot_panel(), 1) + splitter.addWidget(right) + + splitter.setStretchFactor(0, 0) + splitter.setStretchFactor(1, 1) + splitter.setSizes([340, 1140]) + self.setStatusBar(QStatusBar()) + + def _build_language_panel(self) -> QWidget: + box = QGroupBox() + self.language_box = box + layout = QVBoxLayout(box) + row = QHBoxLayout() + self.language_label = QLabel() + row.addWidget(self.language_label, 0) + self.language_combo = QComboBox() + self.language_combo.addItem("", "zh_CN") + self.language_combo.addItem("", "en_US") + self.language_combo.setCurrentIndex(0 if self.tr.locale == "zh_CN" else 1) + self.language_combo.currentIndexChanged.connect(self._on_language_changed) + row.addWidget(self.language_combo, 1) + layout.addLayout(row) + return box + + def _build_steps_panel(self) -> QWidget: + box = QGroupBox() + self.steps_box = box + layout = QVBoxLayout(box) + self.steps_title = QLabel() + self.steps_title.setWordWrap(True) + self.steps_title.setStyleSheet("font-weight: 700; color: #8ab4f8;") + layout.addWidget(self.steps_title) + scroll = QScrollArea() + scroll.setWidgetResizable(True) + host = QWidget() + self.steps_list_layout = QVBoxLayout(host) + self.steps_list_layout.setContentsMargins(0, 0, 0, 0) + self.steps_list_layout.setSpacing(6) + self.steps_list_layout.addStretch(1) + scroll.setWidget(host) + layout.addWidget(scroll, 1) + + btn_row = QHBoxLayout() + self.btn_revert = QPushButton() + self.btn_revert.clicked.connect(self._on_step_revert) + btn_row.addWidget(self.btn_revert) + self.btn_next = QPushButton() + self.btn_next.setObjectName("btnPrimary") + self.btn_next.clicked.connect(self._on_step_next) + btn_row.addWidget(self.btn_next) + layout.addLayout(btn_row) + self.btn_reset = QPushButton() + self.btn_reset.clicked.connect(self._on_step_reset) + layout.addWidget(self.btn_reset) + self.workflow_status = QLabel() + self.workflow_status.setWordWrap(True) + self.workflow_status.setStyleSheet("color: #9aa0a6;") + layout.addWidget(self.workflow_status) + return box + + def _build_options_panel(self) -> QWidget: + box = QGroupBox() + self.options_box = box + layout = QVBoxLayout(box) + self.sensor_type_combo = QComboBox() + self.elastomer_combo = QComboBox() + self.cmos_combo = QComboBox() + self.marker_combo = QComboBox() + self.indentor_combo = QComboBox() + self.platform_combo = QComboBox() + self.sensor_type_label = QLabel() + self.elastomer_label = QLabel() + self.cmos_label = QLabel() + self.marker_label = QLabel() + self.indentor_label = QLabel() + self.platform_label = QLabel() + for label, widget in ( + (self.sensor_type_label, self.sensor_type_combo), + (self.elastomer_label, self.elastomer_combo), + (self.cmos_label, self.cmos_combo), + (self.marker_label, self.marker_combo), + (self.indentor_label, self.indentor_combo), + (self.platform_label, self.platform_combo), + ): + row = QHBoxLayout() + row.addWidget(label, 0) + row.addWidget(widget, 1) + layout.addLayout(row) + return box + + def _build_camera_panel(self) -> QWidget: + box = QGroupBox() + self.camera_box = box + layout = QVBoxLayout(box) + + cam_row = QHBoxLayout() + self.camera_index_label = QLabel() + cam_row.addWidget(self.camera_index_label, 0) + self.camera_index = QSpinBox() + self.camera_index.setRange(0, 16) + self.camera_index.setValue(0) + cam_row.addWidget(self.camera_index, 1) + layout.addLayout(cam_row) + + cfg_row = QHBoxLayout() + self.config_edit = QLineEdit() + default_cfg = default_ddjx01_config() + self.config_edit.setText(str(default_cfg)) + self.config_edit.editingFinished.connect(self._refresh_fixed_mode) + cfg_row.addWidget(self.config_edit, 1) + self.btn_cfg = QPushButton() + self.btn_cfg.clicked.connect(self._on_browse_config) + cfg_row.addWidget(self.btn_cfg) + layout.addLayout(cfg_row) + + cal_row = QHBoxLayout() + self.cal_edit = QLineEdit() + default_cal = default_ddjx01_calibration() + if default_cal.exists(): + self.cal_edit.setText(str(default_cal)) + cal_row.addWidget(self.cal_edit, 1) + self.btn_cal = QPushButton() + self.btn_cal.clicked.connect(self._on_browse_cal) + cal_row.addWidget(self.btn_cal) + layout.addLayout(cal_row) + + backend_row = QHBoxLayout() + self.backend_label = QLabel() + backend_row.addWidget(self.backend_label, 0) + self.backend_combo = QComboBox() + for item in ("auto", "cpu", "cupy", "opencl"): + self.backend_combo.addItem(item) + backend_row.addWidget(self.backend_combo, 1) + layout.addLayout(backend_row) + + self.mode_label = QLabel() + self.mode_label.setWordWrap(True) + self.mode_label.setStyleSheet("color: #8ab4f8; font-weight: 600;") + layout.addWidget(self.mode_label) + + self.sdk_status = QLabel() + self.sdk_status.setWordWrap(True) + self.sdk_status.setStyleSheet("color: #9aa0a6;") + layout.addWidget(self.sdk_status) + + row2 = QHBoxLayout() + self.btn_start = QPushButton() + self.btn_start.setObjectName("btnPrimary") + self.btn_start.clicked.connect(self._on_start_preview) + row2.addWidget(self.btn_start) + self.btn_stop = QPushButton() + self.btn_stop.setEnabled(False) + self.btn_stop.clicked.connect(self._on_stop_preview) + row2.addWidget(self.btn_stop) + layout.addLayout(row2) + return box + + def _build_record_panel(self) -> QWidget: + box = QGroupBox() + self.record_box = box + layout = QVBoxLayout(box) + path_row = QHBoxLayout() + self.record_path_label = QLabel(str(self.recordings_dir)) + self.record_path_label.setWordWrap(True) + self.record_path_label.setStyleSheet("color: #9aa0a6;") + path_row.addWidget(self.record_path_label, 1) + self.btn_browse = QPushButton() + self.btn_browse.clicked.connect(self._on_browse_recordings) + path_row.addWidget(self.btn_browse, 0) + layout.addLayout(path_row) + btn_row = QHBoxLayout() + self.btn_rec_start = QPushButton() + self.btn_rec_start.setObjectName("btnDanger") + self.btn_rec_start.setEnabled(False) + self.btn_rec_start.clicked.connect(self._on_start_record) + btn_row.addWidget(self.btn_rec_start) + self.btn_rec_stop = QPushButton() + self.btn_rec_stop.setEnabled(False) + self.btn_rec_stop.clicked.connect(self._on_stop_record) + btn_row.addWidget(self.btn_rec_stop) + layout.addLayout(btn_row) + self.record_status = QLabel() + self.record_status.setWordWrap(True) + layout.addWidget(self.record_status) + return box + + def _meta_pair(self) -> tuple[QVBoxLayout, QLabel, QLabel]: + col = QVBoxLayout() + title = QLabel() + value = QLabel("—") + value.setObjectName("metaValue") + col.addWidget(title) + col.addWidget(value) + return col, title, value + + def _build_preview_panel(self) -> QWidget: + box = QGroupBox() + self.raw_box = box + layout = QVBoxLayout(box) + meta = QHBoxLayout() + for value_attr, title_attr in ( + ("meta_requested", "meta_requested_title"), + ("meta_actual", "meta_actual_title"), + ("meta_capture_fps", "meta_capture_fps_title"), + ("meta_render_fps", "meta_render_fps_title"), + ): + col, title, value = self._meta_pair() + setattr(self, title_attr, title) + setattr(self, value_attr, value) + meta.addLayout(col) + meta.addSpacing(10) + meta.addStretch(1) + layout.addLayout(meta) + self.preview = QLabel() + self.preview.setObjectName("previewPlaceholder") + self.preview.setAlignment(Qt.AlignCenter) + self.preview.setMinimumSize(320, 240) + self.preview.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) + layout.addWidget(self.preview, 1) + return box + + def _build_processed_panel(self) -> QWidget: + box = QGroupBox() + self.processed_box = box + layout = QVBoxLayout(box) + meta = QHBoxLayout() + for value_attr, title_attr in ( + ("meta_proc_ms", "meta_proc_ms_title"), + ("meta_fn", "meta_fn_title"), + ("meta_fx", "meta_fx_title"), + ("meta_fy", "meta_fy_title"), + ): + col, title, value = self._meta_pair() + setattr(self, title_attr, title) + setattr(self, value_attr, value) + meta.addLayout(col) + meta.addSpacing(10) + meta.addStretch(1) + layout.addLayout(meta) + self.processed_preview = QLabel() + self.processed_preview.setObjectName("previewPlaceholder") + self.processed_preview.setAlignment(Qt.AlignCenter) + self.processed_preview.setMinimumSize(320, 240) + self.processed_preview.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) + layout.addWidget(self.processed_preview, 1) + return box + + def _build_plot_panel(self) -> QWidget: + box = QGroupBox() + self.force_box = box + layout = QVBoxLayout(box) + self.plot_host = QWidget() + self.plot_layout = QVBoxLayout(self.plot_host) + self.plot_layout.setContentsMargins(0, 0, 0, 0) + self.plot_layout.setSpacing(4) + layout.addWidget(self.plot_host) + return box + + def _create_plots(self) -> None: + pg.setConfigOptions(antialias=True, foreground="#e8eaed", background="#121316") + self.plot_force = pg.PlotWidget() + self.plot_force.showGrid(x=True, y=True, alpha=0.25) + self.plot_force.setMinimumHeight(130) + self.plot_layout.addWidget(self.plot_force) + self.curve_fn = self.plot_force.plot(pen=pg.mkPen("#ea4335", width=2), name="FN") + self.curve_fx = self.plot_force.plot(pen=pg.mkPen("#4285f4", width=2), name="FX") + self.curve_fy = self.plot_force.plot(pen=pg.mkPen("#a142f4", width=2), name="FY") + + def _populate_options(self) -> None: + for combo, items in ( + (self.sensor_type_combo, self.options.sensor_types), + (self.elastomer_combo, self.options.elastomer_types), + (self.cmos_combo, self.options.cmos_types), + (self.marker_combo, self.options.marker_types), + (self.indentor_combo, self.options.indentor_types), + (self.platform_combo, self.options.platform_types), + ): + combo.clear() + for item in items: + combo.addItem(item.label, item.id) + + def _populate_steps(self) -> None: + while self.steps_list_layout.count() > 1: + item = self.steps_list_layout.takeAt(0) + widget = item.widget() + if widget is not None: + widget.deleteLater() + self._step_cards.clear() + self._step_title_labels.clear() + self._step_body_labels.clear() + self._step_state_labels.clear() + for index, step in enumerate(self.workflow.steps, start=1): + card = QFrame() + card_layout = QVBoxLayout(card) + card_layout.setContentsMargins(8, 6, 8, 6) + title = QLabel() + title.setStyleSheet("font-weight: 700;") + body = QLabel() + body.setWordWrap(True) + body.setStyleSheet("color: #9aa0a6;") + state = QLabel() + state.setStyleSheet("color: #8ab4f8; font-size: 11px;") + card_layout.addWidget(title) + card_layout.addWidget(body) + card_layout.addWidget(state) + self.steps_list_layout.insertWidget(index - 1, card) + self._step_cards.append(card) + self._step_title_labels.append(title) + self._step_body_labels.append(body) + self._step_state_labels.append(state) + self._refresh_step_cards() + + def _apply_language(self) -> None: + self.setWindowTitle(self.t("app.title")) + self.language_box.setTitle(self.t("app.language")) + self.language_label.setText(self.t("app.language")) + self.language_combo.blockSignals(True) + self.language_combo.setItemText(0, self.t("app.lang_zh")) + self.language_combo.setItemText(1, self.t("app.lang_en")) + self.language_combo.blockSignals(False) + + self.steps_box.setTitle(self.t("panels.instructions")) + self.options_box.setTitle(self.t("panels.options")) + self.camera_box.setTitle(self.t("panels.camera")) + self.record_box.setTitle(self.t("panels.recording")) + self.raw_box.setTitle(self.t("panels.raw")) + self.processed_box.setTitle(self.t("panels.processed")) + self.force_box.setTitle(self.t("panels.force")) + self.plot_force.setTitle(self.t("labels.force_plot")) + + self.sensor_type_label.setText(self.t("labels.sensor_type")) + self.elastomer_label.setText(self.t("labels.elastomer")) + self.cmos_label.setText(self.t("labels.cmos")) + self.marker_label.setText(self.t("labels.marker")) + self.indentor_label.setText(self.t("labels.indentor")) + self.platform_label.setText(self.t("labels.platform")) + self.camera_index_label.setText(self.t("labels.camera_index")) + self.backend_label.setText(self.t("labels.backend")) + self.meta_requested_title.setText(self.t("labels.requested")) + self.meta_actual_title.setText(self.t("labels.actual")) + self.meta_capture_fps_title.setText(self.t("labels.capture_fps")) + self.meta_render_fps_title.setText(self.t("labels.render_fps")) + self.meta_proc_ms_title.setText(self.t("labels.proc_ms")) + self.meta_fn_title.setText(self.t("labels.fn")) + self.meta_fx_title.setText(self.t("labels.fx")) + self.meta_fy_title.setText(self.t("labels.fy")) + + self.btn_cfg.setText(self.t("buttons.config")) + self.btn_cal.setText(self.t("buttons.cal")) + self.btn_start.setText(self.t("buttons.start_preview")) + self.btn_stop.setText(self.t("buttons.stop_preview")) + self.btn_browse.setText(self.t("buttons.browse")) + self.btn_rec_start.setText(self.t("buttons.start_record")) + self.btn_rec_stop.setText(self.t("buttons.stop_record")) + self.btn_next.setText(self.t("buttons.next")) + self.btn_revert.setText(self.t("buttons.revert")) + self.btn_reset.setText(self.t("buttons.reset")) + + self.config_edit.setPlaceholderText(self.t("placeholders.config")) + self.cal_edit.setPlaceholderText(self.t("placeholders.cal")) + if self._raw_pixmap is None: + self.preview.setText(self.t("labels.no_preview")) + if self._proc_pixmap is None: + self.processed_preview.setText(self.t("labels.processed_placeholder")) + if not self.recorder.is_recording and self.record_status.text() in ( + "", + "Not recording", + "未在录制", + ): + self.record_status.setText(self.t("labels.not_recording")) + + self.steps_title.setText(self.tr.workflow_title()) + self._refresh_step_cards() + self._update_workflow_controls() + self.statusBar().showMessage(self.t("status.ready")) + + def _on_language_changed(self, *_args) -> None: + locale = self.language_combo.currentData() + if not locale: + return + self.tr.set_locale(str(locale)) + self._apply_language() + self._refresh_fixed_mode() + self._refresh_orisys_status() + + def _refresh_step_cards(self) -> None: + total = len(self.workflow.steps) + for index, step in enumerate(self.workflow.steps): + title = self.tr.step_title(step.id) + body = self.tr.step_body(step.id) + self._step_title_labels[index].setText(f"{index + 1}. {title}") + self._step_body_labels[index].setText(body) + if index < self._current_step_index: + self._step_cards[index].setObjectName("stepDone") + self._step_title_labels[index].setStyleSheet("font-weight: 700; color: #6b7280;") + self._step_body_labels[index].setStyleSheet("color: #555b66;") + self._step_state_labels[index].setText(self.t("workflow.done_marker")) + self._step_state_labels[index].setStyleSheet("color: #6b7280; font-size: 11px;") + elif index == self._current_step_index and self._current_step_index < total: + self._step_cards[index].setObjectName("stepCurrent") + self._step_title_labels[index].setStyleSheet("font-weight: 700; color: #e8eaed;") + self._step_body_labels[index].setStyleSheet("color: #9aa0a6;") + self._step_state_labels[index].setText(self.t("workflow.current_marker")) + self._step_state_labels[index].setStyleSheet("color: #8ab4f8; font-size: 11px;") + else: + self._step_cards[index].setObjectName("stepUpcoming") + self._step_title_labels[index].setStyleSheet("font-weight: 700;") + self._step_body_labels[index].setStyleSheet("color: #9aa0a6;") + self._step_state_labels[index].setText("") + # Force style refresh after objectName changes. + self._step_cards[index].style().unpolish(self._step_cards[index]) + self._step_cards[index].style().polish(self._step_cards[index]) + + def _update_workflow_controls(self) -> None: + total = len(self.workflow.steps) + done = self._current_step_index >= total + self.btn_next.setEnabled(not done and total > 0) + self.btn_revert.setEnabled(self._current_step_index > 0) + if done and total > 0: + self.workflow_status.setText(self.t("status.workflow_done")) + else: + self.workflow_status.setText("") + + def _on_step_next(self) -> None: + if self._current_step_index < len(self.workflow.steps): + self._current_step_index += 1 + self._refresh_step_cards() + self._update_workflow_controls() + + def _on_step_revert(self) -> None: + if self._current_step_index > 0: + self._current_step_index -= 1 + self._refresh_step_cards() + self._update_workflow_controls() + + def _on_step_reset(self) -> None: + self._current_step_index = 0 + self._refresh_step_cards() + self._update_workflow_controls() + + # ----------------------------------------------------------- selection + def selected_sensor_options(self) -> dict[str, str]: + return { + "sensor_type": str(self.sensor_type_combo.currentData() or ""), + "elastomer": str(self.elastomer_combo.currentData() or ""), + "cmos": str(self.cmos_combo.currentData() or ""), + "marker": str(self.marker_combo.currentData() or ""), + "indentor": str(self.indentor_combo.currentData() or ""), + "platform": str(self.platform_combo.currentData() or ""), + } + + def _refresh_orisys_status(self) -> None: + ok, detail = orisys_importable() + if ok: + self.sdk_status.setText(self.t("status.orisys_ok", version=detail)) + self.btn_start.setEnabled(True) + else: + self.sdk_status.setText(self.t("status.orisys_missing")) + self.btn_start.setEnabled(False) + + def _refresh_fixed_mode(self) -> None: + text = self.config_edit.text().strip() + if not text: + self._fixed_mode = None + self.mode_label.setText(self.t("labels.fixed_mode_unset")) + return + try: + self._fixed_mode = load_fixed_mode(text) + self.mode_label.setText(self.t("labels.fixed_mode", mode=self._fixed_mode.label())) + except Exception as exc: # noqa: BLE001 + self._fixed_mode = None + self.mode_label.setText(self.t("labels.fixed_mode_invalid", error=exc)) + + # -------------------------------------------------------------- capture + def _on_browse_config(self) -> None: + path, _ = QFileDialog.getOpenFileName( + self, self.t("dialogs.select_config"), str(project_root()), "JSON (*.json);;All (*)" + ) + if path: + self.config_edit.setText(path) + self._refresh_fixed_mode() + + def _on_browse_cal(self) -> None: + path, _ = QFileDialog.getOpenFileName( + self, self.t("dialogs.select_cal"), str(project_root()), "NumPy (*.npy);;All (*)" + ) + if path: + self.cal_edit.setText(path) + + def _on_start_preview(self) -> None: + ok, detail = orisys_importable() + if not ok: + QMessageBox.warning( + self, + self.t("dialogs.orisys_title"), + self.t("dialogs.orisys_required", detail=detail), + ) + return + config = self.config_edit.text().strip() + if not config: + QMessageBox.warning( + self, self.t("dialogs.config_title"), self.t("dialogs.config_required") + ) + return + self._refresh_fixed_mode() + if self._fixed_mode is None: + QMessageBox.warning( + self, self.t("dialogs.config_title"), self.t("dialogs.config_invalid") + ) + return + + self.force_history.reset() + self._render_times.clear() + self._clear_previews() + self._awaiting_first_frame = True + self._active_session = self.acquisition.session_id + 1 + try: + mode = self.acquisition.start( + int(self.camera_index.value()), + config_path=config, + cal_path=self.cal_edit.text().strip() or None, + backend=self.backend_combo.currentText(), + ) + except Exception as exc: # noqa: BLE001 + self._awaiting_first_frame = False + QMessageBox.critical(self, self.t("dialogs.capture_title"), str(exc)) + return + + self._fixed_mode = mode + self.mode_label.setText(self.t("labels.fixed_mode", mode=mode.label())) + self.btn_start.setEnabled(False) + self.btn_stop.setEnabled(True) + self.btn_rec_start.setEnabled(False) + self.camera_index.setEnabled(False) + self.config_edit.setEnabled(False) + self.cal_edit.setEnabled(False) + self.backend_combo.setEnabled(False) + self.statusBar().showMessage(self.t("status.opening", mode=mode.label())) + + def _on_stop_preview(self) -> None: + if self.recorder.is_recording: + self._on_stop_record() + self.acquisition.stop() + self._awaiting_first_frame = False + self._clear_previews() + self.btn_start.setEnabled(orisys_importable()[0]) + self.btn_stop.setEnabled(False) + self.btn_rec_start.setEnabled(False) + self.btn_rec_stop.setEnabled(False) + self.camera_index.setEnabled(True) + self.config_edit.setEnabled(True) + self.cal_edit.setEnabled(True) + self.backend_combo.setEnabled(True) + self.statusBar().showMessage(self.t("status.preview_stopped")) + + def _clear_previews(self) -> None: + self._last_packet = None + self._raw_pixmap = None + self._proc_pixmap = None + self.preview.clear() + self.preview.setText(self.t("labels.no_preview")) + self.processed_preview.clear() + self.processed_preview.setText(self.t("labels.processed_placeholder")) + for attr in ( + "meta_requested", + "meta_actual", + "meta_capture_fps", + "meta_render_fps", + "meta_proc_ms", + "meta_fn", + "meta_fx", + "meta_fy", + ): + getattr(self, attr).setText("—") + + def _on_acquisition_error(self, message: str) -> None: + self._awaiting_first_frame = False + self.statusBar().showMessage(self.t("status.capture_error", message=message)) + self.btn_start.setEnabled(orisys_importable()[0]) + self.btn_stop.setEnabled(False) + self.btn_rec_start.setEnabled(False) + self.camera_index.setEnabled(True) + self.config_edit.setEnabled(True) + self.cal_edit.setEnabled(True) + self.backend_combo.setEnabled(True) + QMessageBox.warning(self, self.t("dialogs.capture_title"), message) + + def _on_acquisition_status(self, message: str) -> None: + self.sdk_status.setText(message) + self.statusBar().showMessage(message) + + # ------------------------------------------------------------ recording + def _on_browse_recordings(self) -> None: + path = QFileDialog.getExistingDirectory( + self, self.t("dialogs.select_recordings"), str(self.recordings_dir) + ) + if path: + self.recordings_dir = Path(path) + self.record_path_label.setText(str(self.recordings_dir)) + + def _on_start_record(self) -> None: + packet = self._last_packet or self.acquisition.latest() + if packet is None: + QMessageBox.warning( + self, self.t("dialogs.recording_title"), self.t("dialogs.no_frame") + ) + return + record_fps = packet.fps or (packet.mode.fps if packet.mode else 30.0) + try: + path = self.recorder.start( + self.recordings_dir, fps=float(record_fps), frame=packet.raw_frame + ) + except Exception as exc: # noqa: BLE001 + QMessageBox.critical(self, self.t("dialogs.recording_title"), str(exc)) + return + self.btn_rec_start.setEnabled(False) + self.btn_rec_stop.setEnabled(True) + opts = self.selected_sensor_options() + self.record_status.setText( + self.t( + "status.recording", + name=path.name, + sensor_type=opts["sensor_type"], + elastomer=opts["elastomer"], + cmos=opts["cmos"], + marker=opts["marker"], + indentor=opts["indentor"], + platform=opts["platform"], + ) + ) + + def _on_stop_record(self) -> None: + path, count = self.recorder.stop() + self.btn_rec_start.setEnabled(self.acquisition.is_running) + self.btn_rec_stop.setEnabled(False) + if path is None: + self.record_status.setText(self.t("labels.not_recording")) + return + self.record_status.setText(self.t("status.saved", count=count, path=path)) + + # ---------------------------------------------------------------- tick + def _on_tick(self) -> None: + packet = self.acquisition.latest() + if packet is None: + return + packet_id = id(packet) + if packet_id == self._last_packet_id and not self._awaiting_first_frame: + now = time.perf_counter() + self._render_times.append(now) + cutoff = now - 1.0 + self._render_times = [t for t in self._render_times if t >= cutoff] + return + + if self._awaiting_first_frame: + self._awaiting_first_frame = False + self._active_session = packet.session_id + self.btn_rec_start.setEnabled(True) + self.statusBar().showMessage( + self.t( + "status.frame_info", + raw_w=packet.width, + raw_h=packet.height, + proc_w=packet.processed_width, + proc_h=packet.processed_height, + fps=packet.fps, + ) + ) + + now = time.perf_counter() + self._render_times.append(now) + cutoff = now - 1.0 + self._render_times = [t for t in self._render_times if t >= cutoff] + render_fps = float(len(self._render_times)) + + self._last_packet = packet + self._last_packet_id = packet_id + self._update_raw_preview(packet) + self._update_processed_preview(packet) + self._update_meta(packet, render_fps) + self.force_history.push( + packet.timestamp, + packet.normal_force, + packet.shear_x, + packet.shear_y, + ) + self._update_force_plots() + + if self.recorder.is_recording: + self.recorder.write(packet.raw_frame) + self.record_status.setText( + self.t( + "status.recording_frames", + name=self.recorder.path.name if self.recorder.path else "", + count=self.recorder.frame_count, + ) + ) + + def _update_raw_preview(self, packet: SdkPacket) -> None: + self._raw_pixmap = ndarray_to_pixmap(packet.raw_frame) + self._scale_raw() + + def _update_processed_preview(self, packet: SdkPacket) -> None: + self._proc_pixmap = ndarray_to_pixmap(packet.processed_frame) + self._scale_processed() + + def _scale_raw(self) -> None: + if self._raw_pixmap is None: + return + scaled = self._raw_pixmap.scaled( + self.preview.size(), Qt.KeepAspectRatio, Qt.SmoothTransformation + ) + self.preview.setPixmap(scaled) + + def _scale_processed(self) -> None: + if self._proc_pixmap is None: + return + scaled = self._proc_pixmap.scaled( + self.processed_preview.size(), Qt.KeepAspectRatio, Qt.SmoothTransformation + ) + self.processed_preview.setPixmap(scaled) + + def resizeEvent(self, event) -> None: # noqa: N802 + super().resizeEvent(event) + self._scale_raw() + self._scale_processed() + + def _update_meta(self, packet: SdkPacket, render_fps: float) -> None: + mode = packet.mode or self._fixed_mode + if mode is not None: + self.meta_requested.setText(mode.label()) + else: + self.meta_requested.setText("DDJX01") + self.meta_actual.setText(f"{packet.width}×{packet.height}") + self.meta_capture_fps.setText(f"{packet.fps:.1f}") + self.meta_render_fps.setText(f"{render_fps:.1f}") + self.meta_proc_ms.setText(f"{packet.process_ms:.1f}") + self.meta_fn.setText(f"{packet.normal_force:.1f}") + self.meta_fx.setText(f"{packet.shear_x:.1f}") + self.meta_fy.setText(f"{packet.shear_y:.1f}") + + def _update_force_plots(self) -> None: + state = self.force_history.state() + if state is None: + return + self.curve_fn.setData(state.times, state.values[:, 0]) + self.curve_fx.setData(state.times, state.values[:, 1]) + self.curve_fy.setData(state.times, state.values[:, 2]) + self.plot_force.setXRange(*state.x_range, padding=0) + + def closeEvent(self, event) -> None: # noqa: N802 + if self.recorder.is_recording: + self.recorder.stop() + self.acquisition.stop() + self.timer.stop() + super().closeEvent(event) diff --git a/src/sensor_bench/config_loader.py b/src/sensor_bench/config_loader.py new file mode 100644 index 0000000..bb2b245 --- /dev/null +++ b/src/sensor_bench/config_loader.py @@ -0,0 +1,74 @@ +"""Load JSON configuration for sensor options and workflow steps.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any + + +@dataclass(frozen=True) +class OptionItem: + id: str + label: str + + +@dataclass(frozen=True) +class SensorOptions: + sensor_types: list[OptionItem] + elastomer_types: list[OptionItem] + cmos_types: list[OptionItem] + marker_types: list[OptionItem] + indentor_types: list[OptionItem] + platform_types: list[OptionItem] + + +@dataclass(frozen=True) +class StepItem: + id: str + + +@dataclass(frozen=True) +class WorkflowSteps: + steps: list[StepItem] + + +def _parse_options(raw: list[Any]) -> list[OptionItem]: + items: list[OptionItem] = [] + for entry in raw: + if isinstance(entry, str): + items.append(OptionItem(id=entry, label=entry)) + continue + item_id = str(entry.get("id", "")).strip() + label = str(entry.get("label", item_id)).strip() or item_id + if item_id: + items.append(OptionItem(id=item_id, label=label)) + return items + + +def load_sensor_options(path: Path) -> SensorOptions: + data = json.loads(path.read_text(encoding="utf-8")) + return SensorOptions( + sensor_types=_parse_options(data.get("sensor_types", [])), + elastomer_types=_parse_options(data.get("elastomer_types", [])), + cmos_types=_parse_options(data.get("cmos_types", [])), + marker_types=_parse_options(data.get("marker_types", [])), + indentor_types=_parse_options(data.get("indentor_types", [])), + platform_types=_parse_options( + data.get("platform_types", data.get("platoform_types", [])) + ), + ) + + +def load_workflow_steps(path: Path) -> WorkflowSteps: + data = json.loads(path.read_text(encoding="utf-8")) + steps: list[StepItem] = [] + for entry in data.get("steps", []): + if isinstance(entry, str): + step_id = entry.strip() + else: + step_id = str(entry.get("id", f"step_{len(steps)+1}")).strip() + if step_id: + steps.append(StepItem(id=step_id)) + return WorkflowSteps(steps=steps) diff --git a/src/sensor_bench/i18n.py b/src/sensor_bench/i18n.py new file mode 100644 index 0000000..855c645 --- /dev/null +++ b/src/sensor_bench/i18n.py @@ -0,0 +1,77 @@ +"""Simple JSON-backed localization for Sensor Bench.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Optional + + +DEFAULT_LOCALE = "zh_CN" +FALLBACK_LOCALE = "en_US" + + +class Translator: + """Lookup translated strings with English fallback.""" + + def __init__(self, locales_dir: Path, *, locale: str = DEFAULT_LOCALE): + self.locales_dir = Path(locales_dir) + self._catalogs: dict[str, dict[str, Any]] = {} + self.locale = locale + self.reload() + + def available_locales(self) -> list[str]: + found = sorted(p.stem for p in self.locales_dir.glob("*.json")) + return found or [DEFAULT_LOCALE, FALLBACK_LOCALE] + + def reload(self) -> None: + self._catalogs.clear() + for path in self.locales_dir.glob("*.json"): + try: + self._catalogs[path.stem] = json.loads(path.read_text(encoding="utf-8")) + except Exception: # noqa: BLE001 + continue + if self.locale not in self._catalogs and DEFAULT_LOCALE in self._catalogs: + self.locale = DEFAULT_LOCALE + elif self.locale not in self._catalogs and FALLBACK_LOCALE in self._catalogs: + self.locale = FALLBACK_LOCALE + + def set_locale(self, locale: str) -> None: + self.locale = locale + if locale not in self._catalogs: + self.reload() + + def t(self, key: str, **kwargs: Any) -> str: + text = self._lookup(self.locale, key) + if text is None: + text = self._lookup(FALLBACK_LOCALE, key) + if text is None: + text = key + if kwargs: + try: + return str(text).format(**kwargs) + except Exception: # noqa: BLE001 + return str(text) + return str(text) + + def step_title(self, step_id: str) -> str: + return self.t(f"workflow.steps.{step_id}.title") + + def step_body(self, step_id: str) -> str: + return self.t(f"workflow.steps.{step_id}.body") + + def workflow_title(self) -> str: + return self.t("workflow.title") + + def _lookup(self, locale: str, key: str) -> Optional[str]: + data: Any = self._catalogs.get(locale) + if not isinstance(data, dict): + return None + node: Any = data + for part in key.split("."): + if not isinstance(node, dict) or part not in node: + return None + node = node[part] + if isinstance(node, str): + return node + return None diff --git a/src/sensor_bench/main.py b/src/sensor_bench/main.py new file mode 100644 index 0000000..afe7b66 --- /dev/null +++ b/src/sensor_bench/main.py @@ -0,0 +1,24 @@ +"""Entry point for the Sensor Bench desktop application.""" + +from __future__ import annotations + +import sys + +from PyQt5.QtWidgets import QApplication + +from .app import MainWindow + + +def main(argv: list[str] | None = None) -> int: + args = list(sys.argv if argv is None else argv) + qt_app = QApplication(args) + qt_app.setApplicationName("Sensor Bench") + qt_app.setOrganizationName("Orisys-Tech") + + window = MainWindow() + window.show() + return qt_app.exec_() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/sensor_bench/paths.py b/src/sensor_bench/paths.py new file mode 100644 index 0000000..cab40f3 --- /dev/null +++ b/src/sensor_bench/paths.py @@ -0,0 +1,80 @@ +"""Resolve project resource paths for editable and packaged runs.""" + +from __future__ import annotations + +import sys +from pathlib import Path + + +def is_frozen() -> bool: + return bool(getattr(sys, "frozen", False)) + + +def app_root() -> Path: + """Return the editable repo root or frozen executable directory.""" + if is_frozen(): + return Path(sys.executable).resolve().parent + return Path(__file__).resolve().parents[2] + + +def resource_root() -> Path: + """Return bundled resource root for editable and PyInstaller runs.""" + if is_frozen(): + return Path(getattr(sys, "_MEIPASS", app_root())).resolve() + return app_root() + + +def project_root() -> Path: + """Return the editable repo root or frozen executable directory.""" + return app_root() + + +def default_options_path() -> Path: + external = app_root() / "config" / "options.json" + if external.exists(): + return external + return resource_root() / "config" / "options.json" + + +def default_steps_path() -> Path: + external = app_root() / "steps" / "default.json" + if external.exists(): + return external + return resource_root() / "steps" / "default.json" + + +def default_locales_dir() -> Path: + external = app_root() / "config" / "locales" + if external.exists(): + return external + return resource_root() / "config" / "locales" + + +def default_recordings_dir() -> Path: + path = app_root() / "recordings" + path.mkdir(parents=True, exist_ok=True) + return path + + +def default_orisys_sdk_root() -> Path: + """Sibling ``orisys-sdk`` checkout when sensor-bench lives next to it.""" + sibling = app_root().parent / "orisys-sdk" + if sibling.exists(): + return sibling + return Path(r"D:\repo\orisys-sdk") + + +def default_ddjx01_config() -> str: + if is_frozen(): + return "ddjx01" + path = default_orisys_sdk_root() / "config" / "ddjx01.json" + if path.exists(): + return str(path) + return "ddjx01" + + +def default_ddjx01_calibration() -> Path: + external = app_root() / "config" / "ddjx01.npy" + if external.exists(): + return external + return default_orisys_sdk_root() / "config" / "ddjx01.npy" diff --git a/src/sensor_bench/plot_history.py b/src/sensor_bench/plot_history.py new file mode 100644 index 0000000..c8e19d9 --- /dev/null +++ b/src/sensor_bench/plot_history.py @@ -0,0 +1,85 @@ +"""Simple rolling history for live performance curves.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional + +import numpy as np + + +@dataclass +class PlotState: + times: np.ndarray + values: np.ndarray # shape (n, channels) + x_range: tuple[float, float] + y_ranges: tuple[tuple[float, float], ...] + + +class MetricHistory: + """Ring-style buffer for FPS / brightness (or other scalar metrics).""" + + def __init__(self, npoints: int = 120, *, channels: int = 2, window_sec: float = 4.0): + self.npoints = int(npoints) + self.channels = int(channels) + self.window_sec = float(window_sec) + self.reset() + + def reset(self) -> None: + self._data = np.full((self.npoints, 1 + self.channels), np.nan, dtype=np.float64) + self.sample_count = 0 + self._t0: Optional[float] = None + + def push(self, timestamp: float, *values: float) -> None: + if len(values) != self.channels: + raise ValueError(f"Expected {self.channels} values, got {len(values)}") + if self._t0 is None: + self._t0 = float(timestamp) + t = float(timestamp) - self._t0 + row = (t, *map(float, values)) + if self.sample_count < self.npoints: + self._data[self.sample_count, :] = row + self.sample_count += 1 + return + self._data[:-1, :] = self._data[1:, :] + self._data[-1, :] = row + + def state(self) -> Optional[PlotState]: + n = self.sample_count + if n == 0: + return None + block = self._data[:n] + times = block[:, 0].copy() + values = block[:, 1:].copy() + if n < self.npoints: + x_range = (0.0, self.window_sec) + else: + x0 = float(times[0]) + x1 = float(times[-1]) + if x1 <= x0: + x1 = x0 + self.window_sec + x_range = (x0, x1) + + y_ranges = [] + for i in range(self.channels): + col = values[:, i] + ymin = float(np.nanmin(col)) + ymax = float(np.nanmax(col)) + if not np.isfinite(ymin) or not np.isfinite(ymax): + ymin, ymax = 0.0, 1.0 + if abs(ymax - ymin) < 1e-6: + pad = 1.0 if abs(ymin) < 1e-6 else abs(ymin) * 0.1 + ymin -= pad + ymax += pad + else: + pad = (ymax - ymin) * 0.12 + ymin -= pad + ymax += pad + y_ranges.append((ymin, ymax)) + + return PlotState( + times=times, + values=values, + x_range=x_range, + y_ranges=tuple(y_ranges), + ) diff --git a/src/sensor_bench/recorder.py b/src/sensor_bench/recorder.py new file mode 100644 index 0000000..2b24a5b --- /dev/null +++ b/src/sensor_bench/recorder.py @@ -0,0 +1,82 @@ +"""Raw video recording helpers.""" + +from __future__ import annotations + +from datetime import datetime +from pathlib import Path +from typing import Optional + +import cv2 +import numpy as np + + +class RawRecorder: + """Write raw camera frames to a timestamped MP4 file.""" + + def __init__(self) -> None: + self._writer: Optional[cv2.VideoWriter] = None + self.path: Optional[Path] = None + self.frame_count = 0 + self._size: Optional[tuple[int, int]] = None + + @property + def is_recording(self) -> bool: + return self._writer is not None + + def start(self, output_dir: Path, *, fps: float, frame: np.ndarray) -> Path: + self.stop() + output_dir.mkdir(parents=True, exist_ok=True) + stamp = datetime.now().strftime("%Y%m%d_%H%M%S") + path = output_dir / f"raw_{stamp}.mp4" + + write_frame = self._to_bgr(frame) + height, width = write_frame.shape[:2] + fps_out = float(fps) if fps and fps > 1.0 else 30.0 + writer = cv2.VideoWriter( + str(path), + cv2.VideoWriter_fourcc(*"mp4v"), + fps_out, + (width, height), + ) + if not writer.isOpened(): + raise RuntimeError(f"Unable to create recording file: {path}") + + self._writer = writer + self.path = path + self.frame_count = 0 + self._size = (width, height) + self.write(frame) + return path + + def write(self, frame: np.ndarray) -> None: + if self._writer is None: + return + write_frame = self._to_bgr(frame) + height, width = write_frame.shape[:2] + if self._size is not None and (width, height) != self._size: + write_frame = cv2.resize(write_frame, self._size, interpolation=cv2.INTER_AREA) + self._writer.write(write_frame) + self.frame_count += 1 + + def stop(self) -> tuple[Optional[Path], int]: + path = self.path + count = self.frame_count + if self._writer is not None: + self._writer.release() + self._writer = None + self.path = None + self.frame_count = 0 + self._size = None + return path, count + + @staticmethod + def _to_bgr(frame: np.ndarray) -> np.ndarray: + if frame.ndim == 2: + return cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR) + if frame.ndim == 3 and frame.shape[2] == 1: + return cv2.cvtColor(frame[:, :, 0], cv2.COLOR_GRAY2BGR) + if frame.ndim == 3 and frame.shape[2] == 4: + return cv2.cvtColor(frame, cv2.COLOR_BGRA2BGR) + if frame.ndim == 3 and frame.shape[2] == 3: + return frame + raise ValueError(f"Unsupported frame shape for recording: {frame.shape}") diff --git a/steps/default.json b/steps/default.json new file mode 100644 index 0000000..3fe1416 --- /dev/null +++ b/steps/default.json @@ -0,0 +1,13 @@ +{ + "title_key": "workflow.title", + "steps": [ + {"id": "power_on"}, + {"id": "select_camera"}, + {"id": "set_options"}, + {"id": "check_preview"}, + {"id": "record"}, + {"id": "review_curves"}, + {"id": "sensitivity"}, + {"id": "max_load"} + ] +} diff --git a/tests/fixtures/ddjx01_min.json b/tests/fixtures/ddjx01_min.json new file mode 100644 index 0000000..9c28ff2 --- /dev/null +++ b/tests/fixtures/ddjx01_min.json @@ -0,0 +1,33 @@ +{ + "name": "prototype", + "version": "1.0", + "camera": { + "resolution_raw": [1920, 1080], + "fps": 30, + "units": 15, + "units_used": 9, + "resolution_stitch": [365, 365] + }, + "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, + "n": [3, 3], + "box_size": [250, 250], + "blend": "numba" + }, + "tracker": { + "adaptive": true, + "lower_threshold": 12.0, + "upper_threshold": 18.0, + "algorithm": "dis", + "dis_preset": "medium" + }, + "contact": { + "min_contour_area": 3000, + "flow_threshold": 5, + "div_threshold": 0.05 + } +} diff --git a/tests/fixtures/ffmpeg_list_devices.txt b/tests/fixtures/ffmpeg_list_devices.txt new file mode 100644 index 0000000..bc11c1d --- /dev/null +++ b/tests/fixtures/ffmpeg_list_devices.txt @@ -0,0 +1,5 @@ +[dshow @ 000001] DirectShow video devices (some may be both video and audio devices) +[dshow @ 000001] "USB Camera" +[dshow @ 000001] Alternative name "@device_pnp_\\\\?\\usb#vid_0000" +[dshow @ 000001] DirectShow audio devices +[dshow @ 000001] "Microphone" diff --git a/tests/fixtures/ffmpeg_list_devices_v9.txt b/tests/fixtures/ffmpeg_list_devices_v9.txt new file mode 100644 index 0000000..89d80f9 --- /dev/null +++ b/tests/fixtures/ffmpeg_list_devices_v9.txt @@ -0,0 +1,7 @@ +[in#0 @ 000001] "USB Camera" (video) +[in#0 @ 000001] Alternative name "@device_pnp_\\?\usb#vid_1bcf&pid_2281&mi_00#b&1d67a3f1&0&0000#{65e8773d-8f56-11d0-a3b9-00a0c9223196}\global" +[in#0 @ 000001] "OBS Virtual Camera" (none) +[in#0 @ 000001] Alternative name "@device_sw_{860BB310-5D01-11D0-BD3B-00A0C911CE86}\{A3FCE0F5-3493-419F-958A-ABA1250EC20B}" +[in#0 @ 000001] "Microphone (SunplusIT Inc)" (audio) +[in#0 @ 000001] Alternative name "@device_cm_{33D9A762-90C8-11D0-BD43-00A0C911CE86}\wave_{47A9A322-5BE2-4304-9CA5-BFEC5748F002}" +Error opening input file dummy. diff --git a/tests/fixtures/ffmpeg_list_options.txt b/tests/fixtures/ffmpeg_list_options.txt new file mode 100644 index 0000000..be0b68f --- /dev/null +++ b/tests/fixtures/ffmpeg_list_options.txt @@ -0,0 +1,5 @@ +[dshow @ 000001] pixel_format=yuyv422 min s=640x480 fps=30 max s=640x480 fps=30 +[dshow @ 000001] pixel_format=yuyv422 min s=1280x720 fps=10 max s=1280x720 fps=10 +[dshow @ 000001] vcodec=mjpeg min s=640x480 fps=30 max s=640x480 fps=30 +[dshow @ 000001] vcodec=mjpeg min s=1280x720 fps=30 max s=1280x720 fps=30 +[dshow @ 000001] vcodec=mjpeg min s=1920x1080 fps=30 max s=1920x1080 fps=30 diff --git a/tests/test_smoke.py b/tests/test_smoke.py new file mode 100644 index 0000000..9340eed --- /dev/null +++ b/tests/test_smoke.py @@ -0,0 +1,172 @@ +"""Smoke tests for Sensor Bench SDK acquisition and UI construction.""" + +from __future__ import annotations + +import os +import sys +import time +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import numpy as np + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + + +def test_load_fixed_mode() -> None: + from sensor_bench.acquisition import load_fixed_mode + + mode = load_fixed_mode(ROOT / "tests" / "fixtures" / "ddjx01_min.json") + assert mode.width == 1920 + assert mode.height == 1080 + assert mode.fps == 30.0 + assert mode.stitch_width == 365 + assert "1920x1080" in mode.label() + + +def test_translator_chinese_default() -> None: + from sensor_bench.i18n import DEFAULT_LOCALE, Translator + from sensor_bench.paths import default_locales_dir + + assert DEFAULT_LOCALE == "zh_CN" + tr = Translator(default_locales_dir()) + assert tr.locale == "zh_CN" + assert "传感器" in tr.t("app.title") + tr.set_locale("en_US") + assert tr.t("app.title") == "Sensor Bench" + assert "Power on" in tr.step_title("power_on") + + +def test_sdk_worker_packet_mapping() -> None: + from sensor_bench.acquisition import SdkAcquisitionWorker + + raw = np.zeros((1080, 1920, 3), dtype=np.uint8) + processed = np.full((365, 365, 3), 40, dtype=np.uint8) + + class FakeSensor: + def __init__(self, *args, **kwargs): + self.frame = raw.copy() + self.info = SimpleNamespace(FNORMAL=1, FSHEARX=2, FSHEARY=3) + + def get_img(self): + self.frame = raw.copy() + return processed.copy() + + def compute_deformation(self, **kwargs): + return True + + def compute_contact(self): + return True + + def read_info(self, *args): + return 1.0, 2.0, 3.0 + + def disconnect(self): + pass + + fake_orisys = MagicMock() + fake_orisys.Sensor = FakeSensor + fake_orisys.__version__ = "test" + + worker = SdkAcquisitionWorker() + with patch.dict(sys.modules, {"orisys": fake_orisys}): + with patch("sensor_bench.acquisition.orisys_importable", return_value=(True, "test")): + mode = worker.start( + 0, + config_path=ROOT / "tests" / "fixtures" / "ddjx01_min.json", + backend="cpu", + ) + assert mode.width == 1920 + deadline = time.time() + 2.0 + packet = None + while time.time() < deadline: + packet = worker.latest() + if packet is not None: + break + time.sleep(0.02) + worker.stop() + + assert packet is not None + assert packet.width == 1920 + assert packet.height == 1080 + assert packet.processed_width == 365 + assert packet.processed_height == 365 + assert packet.normal_force == 1.0 + assert packet.shear_x == 2.0 + assert packet.shear_y == 3.0 + assert packet.raw_frame.shape[:2] == (1080, 1920) + assert packet.processed_frame.shape[:2] == (365, 365) + + +def test_mainwindow_offscreen() -> None: + from PyQt5.QtWidgets import QApplication + from sensor_bench.app import MainWindow + from sensor_bench.acquisition import FixedCameraMode + from sensor_bench.i18n import DEFAULT_LOCALE + + app = QApplication.instance() or QApplication([]) + win = MainWindow() + assert win.tr.locale == DEFAULT_LOCALE == "zh_CN" + assert "传感器" in win.windowTitle() + assert win.btn_start.text() == "开始预览" + assert win.camera_index.value() == 0 + assert win.btn_next is not None + assert win.btn_revert is not None + assert win.btn_reset is not None + assert win.sensor_type_combo.count() == 2 + assert win.indentor_combo.count() == 2 + assert win.platform_combo.count() == 3 + assert win.selected_sensor_options()["sensor_type"] == "BOX" + assert win.selected_sensor_options()["indentor"] == "1mm_circle" + assert win.selected_sensor_options()["platform"] == "manual" + assert not hasattr(win, "resolution_combo") + assert not hasattr(win, "btn_search") + assert win._current_step_index == 0 + assert not win.btn_revert.isEnabled() + assert win.btn_next.isEnabled() + + # Language switch refreshes labels without restart. + win.language_combo.setCurrentIndex(1) + assert win.tr.locale == "en_US" + assert win.windowTitle() == "Sensor Bench" + assert win.btn_start.text() == "Start Preview" + + # Workflow next / revert / reset. + total = len(win.workflow.steps) + win._on_step_next() + assert win._current_step_index == 1 + assert win.btn_revert.isEnabled() + win._on_step_revert() + assert win._current_step_index == 0 + for _ in range(total): + win._on_step_next() + assert win._current_step_index == total + assert not win.btn_next.isEnabled() + assert "completed" in win.workflow_status.text().lower() or "完成" in win.workflow_status.text() + win._on_step_reset() + assert win._current_step_index == 0 + assert win.btn_next.isEnabled() + + win._fixed_mode = FixedCameraMode( + width=1920, + height=1080, + fps=30.0, + config_ref=str(ROOT / "tests" / "fixtures" / "ddjx01_min.json"), + stitch_width=365, + stitch_height=365, + ) + win.mode_label.setText(win.t("labels.fixed_mode", mode=win._fixed_mode.label())) + assert "1920x1080" in win.mode_label.text() + win.close() + + +if __name__ == "__main__": + test_load_fixed_mode() + test_translator_chinese_default() + test_sdk_worker_packet_mapping() + test_mainwindow_offscreen() + print("ok") diff --git a/tools/ffmpeg/README.txt b/tools/ffmpeg/README.txt new file mode 100644 index 0000000..84c2f92 --- /dev/null +++ b/tools/ffmpeg/README.txt @@ -0,0 +1,5 @@ +Place a portable FFmpeg build here: + + tools/ffmpeg/bin/ffmpeg.exe + +Or install ffmpeg system-wide and ensure it is on PATH.