Merge remote-tracking branch 'origin/dev' into feature/ffc

Reviewed-on: http://100.106.67.31:3000/orisys-tech/orisys-sdk/pulls/1

roi_polygon.json

@@ -1,44 +0,0 @@
-{
-  "version": "1.0",
-  "type": "polygon",
-  "image_size": [
-    640,
-    480
-  ],
-  "points": [
-    [
-      1,
-      123
-    ],
-    [
-      227,
-      164
-    ],
-    [
-      392,
-      167
-    ],
-    [
-      638,
-      130
-    ],
-    [
-      638,
-      441
-    ],
-    [
-      436,
-      479
-    ],
-    [
-      199,
-      477
-    ],
-    [
-      1,
-      427
-    ]
-  ],
-  "created_at": "2026-07-01T17:27:44",
-  "mask_path": "roi_polygon_mask.png"
-}

src/orisys/Stitcher.py

@@ -3,9 +3,22 @@ import numpy as np
 import scipy.fft
 from scipy import ndimage
 import cv2
-# from numba import njit
 
-# @njit(cache=False)
+try:
+    from numba import njit
+
+    _NUMBA_AVAILABLE = True
+except ImportError:
+    _NUMBA_AVAILABLE = False
+
+    def njit(*args, **kwargs):
+        def decorator(fn):
+            return fn
+
+        return decorator
+
+
+@njit(cache=False)
 def _blend_kernel(blend_patch, wsum_patch, cov_patch, src_patch, w_patch):
     """Fused incremental weighted-average composite (numba-accelerated)."""
     sh = src_patch.shape[0]
@@ -73,6 +86,12 @@ class Stitcher:
         self._feature_stitch_pad = 5
         self._feature_stitch_match_brightness = False
 
+        if self.isstitch and not _NUMBA_AVAILABLE:
+            print(
+                "Warning: numba is not installed; feature stitching will run a slow "
+                "Python blend path (~1 FPS). Install numba for real-time performance."
+            )
+
         if self.verbose:
             print(f"Stitcher initialized with config: {self.config_stitcher}")
 

src/orisys/frame_processors.py

@@ -238,7 +238,7 @@ class FFCFrameProcessor:
     """Flat-field correction with offline weight-map calibration and fast online correction."""
 
     def __init__(self, blur_kernel=151, blur_sigma=0.0, epsilon=1e-6, masked_outside=True):
-        self.blur_kernel = ensure_odd(blur_kernel)
+        self.blur_kernel = self.ensure_odd(blur_kernel)
         self.blur_sigma = float(blur_sigma)
         self.epsilon = float(epsilon)
         self.masked_outside = bool(masked_outside)
@@ -252,8 +252,8 @@ class FFCFrameProcessor:
         self._roi_bbox = None
 
     def calibrate(self, reference_image, roi_mask=None):
-        gray = to_gray_float32(reference_image)
-        roi_mask = normalize_mask(roi_mask, gray.shape)
+        gray = self.to_gray_float32(reference_image)
+        roi_mask = self.normalize_mask(roi_mask, gray.shape)
         active = roi_mask > 0
         if not np.any(active):
             raise ValueError("ROI mask is empty; cannot calibrate FFC.")
@@ -308,7 +308,7 @@ class FFCFrameProcessor:
             cv2.convertScaleAbs(self._multiplied_buffer, dst=dst_view, alpha=1.0, beta=0.0)
             return self._output_buffer
 
-        corrected = saturate_to_dtype(self._multiplied_buffer, image.dtype)
+        corrected = self.saturate_to_dtype(self._multiplied_buffer, image.dtype)
         dst_view[...] = corrected
         return self._output_buffer
 
@@ -379,12 +379,12 @@ class FFCFrameProcessor:
                 x1, y1, x2, y2 = self._roi_bbox
                 self._multiplied_buffer = np.empty((y2 - y1, x2 - x1), dtype=np.float32)
 
-
+    @staticmethod
     def ensure_odd(value, minimum=3):
         value = max(int(value), int(minimum))
         return value if value % 2 == 1 else value + 1
 
-
+    @staticmethod
     def to_gray_float32(image):
         if image.ndim == 2:
             gray = image
@@ -394,7 +394,7 @@ def to_gray_float32(image):
             raise ValueError("image must have shape HxW or HxWxC")
         return gray.astype(np.float32, copy=False)
 
-
+    @staticmethod
     def normalize_mask(mask, shape):
         if mask is None:
             return np.ones(shape, dtype=np.uint8) * 255
@@ -403,7 +403,7 @@ def normalize_mask(mask, shape):
             mask = cv2.resize(mask, (shape[1], shape[0]), interpolation=cv2.INTER_NEAREST)
         return np.where(mask > 0, 255, 0).astype(np.uint8)
 
-
+    @staticmethod
     def saturate_to_dtype(values, dtype):
         if np.issubdtype(dtype, np.integer):
             info = np.iinfo(dtype)

tests/conftest.py

@@ -1,116 +0,0 @@
-import os
-import sys
-from pathlib import Path
-
-import pytest
-
-REPO_ROOT = Path(__file__).resolve().parents[1]
-SRC_ROOT = REPO_ROOT / "src"
-if str(SRC_ROOT) not in sys.path:
-    sys.path.insert(0, str(SRC_ROOT))
-
-from orisys import Sensor
-from orisys import configs as orisys_configs
-from orisys.backend_availability import check_cupy_available, check_opencl_available
-
-
-def _resolve_test_video() -> Path:
-    env_video = os.getenv("ORISYS_TEST_VIDEO")
-    if env_video:
-        raw = Path(env_video)
-        candidates = [
-            raw,                # as provided (absolute or cwd-relative)
-            REPO_ROOT / raw,    # repo-relative
-        ]
-        # Common typo fallback: "test/..." -> "tests/..."
-        if env_video.startswith("test/") or env_video.startswith("test\\"):
-            fixed = Path("tests", *Path(env_video).parts[1:])
-            candidates.append(REPO_ROOT / fixed)
-
-        for candidate in candidates:
-            if candidate.exists():
-                return candidate
-        return raw
-
-    # Prefer tests/ fixture clip; fallback kept by caller exists check.
-    return REPO_ROOT / "tests" / "idle_1min.mp4"
-
-
-@pytest.fixture(scope="session")
-def test_video_path() -> Path:
-    path = _resolve_test_video()
-    if not path.exists():
-        pytest.skip(
-            "Test video not found. Set ORISYS_TEST_VIDEO to a valid video path."
-        )
-    return path
-
-
-@pytest.fixture(scope="session")
-def test_config_name() -> str:
-    requested = os.getenv("ORISYS_TEST_CONFIG")
-
-    def _is_usable_config(name: str) -> bool:
-        cfg = orisys_configs.load_config(name)
-        return "box_size" in cfg.get("stitcher", {})
-
-    if requested:
-        if not _is_usable_config(requested):
-            raise RuntimeError(
-                f"ORISYS_TEST_CONFIG='{requested}' is missing required key "
-                "'stitcher.box_size'. Use a config like 'ddjx01' or provide a full path."
-            )
-        return requested
-
-    # Auto-pick a stable config for tests (avoid prototype variants missing keys).
-    for name in orisys_configs.list_config():
-        try:
-            if _is_usable_config(name):
-                return name
-        except Exception:
-            continue
-
-    raise RuntimeError(
-        "No usable config found for tests (missing 'stitcher.box_size'). "
-        "Set ORISYS_TEST_CONFIG to a valid config JSON/path."
-    )
-
-
-@pytest.fixture(scope="session")
-def available_backends() -> list[str]:
-    backends = ["cpu"]
-    if check_cupy_available():
-        backends.append("cupy")
-    if check_opencl_available():
-        backends.append("opencl")
-    return backends
-
-
-def make_sensor(
-    video_path: Path,
-    config_name: str = "prototype",
-    backend: str = "cpu",
-    verbose: bool = False,
-) -> Sensor:
-    use_cuda = backend in ("cupy", "opencl")
-    return Sensor(
-        vid_src=str(video_path),
-        cuda=use_cuda,
-        isstitch=True,
-        verbose=verbose,
-        config_name=config_name,
-        backend=backend,
-    )
-
-
-@pytest.fixture
-def sensor_factory(test_video_path, test_config_name):
-    def _factory(backend: str = "cpu", verbose: bool = False):
-        return make_sensor(
-            test_video_path,
-            config_name=test_config_name,
-            backend=backend,
-            verbose=verbose,
-        )
-
-    return _factory

tests/test_ffc.py

@@ -1,12 +1,19 @@
 from __future__ import annotations
-from orisys import FFCFrameProcessor
 
 import argparse
+import sys
 from pathlib import Path
 
 import cv2
 
-from roi_polygon_tool import load_roi_mask, select_polygon_roi
+REPO_ROOT = Path(__file__).resolve().parents[1]
+SRC_ROOT = REPO_ROOT / "src"
+for path in (REPO_ROOT, SRC_ROOT):
+    if str(path) not in sys.path:
+        sys.path.insert(0, str(path))
+
+from orisys import FFCFrameProcessor
+from tools.ffc.roi_polygon_tool import load_roi_mask, select_polygon_roi
 
 
 def parse_source(value):