|
1 | | -"""Optional compatibility layer for reusing optv py_bind as a native provider.""" |
| 1 | +"""Compatibility layer for selecting between optv and Python/Numba engines.""" |
2 | 2 |
|
3 | 3 | from __future__ import annotations |
4 | 4 |
|
5 | 5 | from importlib import import_module |
6 | 6 | from types import ModuleType |
7 | | -from typing import Any |
| 7 | +from typing import Any, Literal |
| 8 | +import warnings |
| 9 | + |
| 10 | +EngineName = Literal["optv", "python"] |
| 11 | + |
| 12 | +DEFAULT_ENGINE: EngineName = "optv" |
| 13 | +ENGINE_OPTV: EngineName = "optv" |
| 14 | +ENGINE_PYTHON: EngineName = "python" |
| 15 | + |
| 16 | +_ENGINE_PREFERENCE: EngineName = DEFAULT_ENGINE |
| 17 | +_ENGINE_REASON: str = "" |
| 18 | +_ENGINE_WARNING_EMITTED = False |
8 | 19 |
|
9 | 20 |
|
10 | 21 | def _optional_import(module_name: str) -> ModuleType | None: |
@@ -57,6 +68,134 @@ def _optional_import(module_name: str) -> ModuleType | None: |
57 | 68 | ) |
58 | 69 |
|
59 | 70 |
|
| 71 | +def _emit_engine_warning(reason: str) -> None: |
| 72 | + global _ENGINE_WARNING_EMITTED |
| 73 | + if _ENGINE_WARNING_EMITTED: |
| 74 | + return |
| 75 | + |
| 76 | + warnings.warn(reason, RuntimeWarning, stacklevel=3) |
| 77 | + _ENGINE_WARNING_EMITTED = True |
| 78 | + |
| 79 | + |
| 80 | +def _set_engine_state(engine: EngineName, reason: str) -> None: |
| 81 | + global _ENGINE_PREFERENCE, _ENGINE_REASON, _ENGINE_WARNING_EMITTED |
| 82 | + _ENGINE_PREFERENCE = engine |
| 83 | + _ENGINE_REASON = reason |
| 84 | + _ENGINE_WARNING_EMITTED = False |
| 85 | + |
| 86 | + |
| 87 | +def _resolve_engine_request(engine: EngineName | str | None) -> EngineName: |
| 88 | + if engine is None: |
| 89 | + return DEFAULT_ENGINE |
| 90 | + |
| 91 | + normalized = str(engine).strip().lower() |
| 92 | + if normalized in {"optv", "native", "c", "fast"}: |
| 93 | + return ENGINE_OPTV |
| 94 | + if normalized in {"python", "numba", "pure-python", "fallback"}: |
| 95 | + return ENGINE_PYTHON |
| 96 | + |
| 97 | + raise ValueError(f"Unknown engine '{engine}'. Use 'optv' or 'python'.") |
| 98 | + |
| 99 | + |
| 100 | +def _resolve_active_engine() -> tuple[EngineName, str]: |
| 101 | + if _ENGINE_PREFERENCE == ENGINE_PYTHON: |
| 102 | + return ENGINE_PYTHON, "Forced Python engine" |
| 103 | + |
| 104 | + if HAS_OPTV: |
| 105 | + return ENGINE_OPTV, "Using optv engine" |
| 106 | + |
| 107 | + return ENGINE_PYTHON, "optv is unavailable; using Python/Numba fallback" |
| 108 | + |
| 109 | + |
| 110 | +def set_engine(engine: EngineName | str | None = None, *, warn_once: bool = True) -> EngineName: |
| 111 | + """Set the preferred engine for native-backed calls. |
| 112 | +
|
| 113 | + Parameters |
| 114 | + ---------- |
| 115 | + engine: |
| 116 | + Preferred engine. ``optv`` is the default and ``python`` forces the |
| 117 | + Python/Numba code path. |
| 118 | + warn_once: |
| 119 | + Emit a one-time warning when the selected engine cannot be used. |
| 120 | + """ |
| 121 | + |
| 122 | + requested = _resolve_engine_request(engine) |
| 123 | + |
| 124 | + if requested == ENGINE_PYTHON: |
| 125 | + _set_engine_state(requested, "Forced Python engine") |
| 126 | + return ENGINE_PYTHON |
| 127 | + |
| 128 | + if HAS_OPTV: |
| 129 | + _set_engine_state(requested, "Using optv engine") |
| 130 | + return ENGINE_OPTV |
| 131 | + |
| 132 | + reason = "optv is unavailable; using Python/Numba fallback" |
| 133 | + _set_engine_state(requested, reason) |
| 134 | + if warn_once: |
| 135 | + _emit_engine_warning(reason) |
| 136 | + return ENGINE_PYTHON |
| 137 | + |
| 138 | + |
| 139 | +def get_engine_preference() -> EngineName: |
| 140 | + """Return the user-requested engine preference.""" |
| 141 | + |
| 142 | + return _ENGINE_PREFERENCE |
| 143 | + |
| 144 | + |
| 145 | +def get_active_engine() -> EngineName: |
| 146 | + """Return the engine currently used for native-backed calls.""" |
| 147 | + |
| 148 | + active, _ = _resolve_active_engine() |
| 149 | + return active |
| 150 | + |
| 151 | + |
| 152 | +def get_engine_reason() -> str: |
| 153 | + """Return a human-readable explanation of the active engine choice.""" |
| 154 | + |
| 155 | + active, reason = _resolve_active_engine() |
| 156 | + if active == ENGINE_PYTHON and _ENGINE_PREFERENCE == ENGINE_PYTHON: |
| 157 | + return reason |
| 158 | + if active == ENGINE_PYTHON and _ENGINE_PREFERENCE == ENGINE_OPTV: |
| 159 | + return reason |
| 160 | + return reason |
| 161 | + |
| 162 | + |
| 163 | +def get_engine_status() -> str: |
| 164 | + """Return a short status string for GUI and batch reporting.""" |
| 165 | + |
| 166 | + return f"engine={get_active_engine()} ({get_engine_reason()})" |
| 167 | + |
| 168 | + |
| 169 | +def should_use_native(feature_name: str | None = None) -> bool: |
| 170 | + """Return True when the native optv implementation should be used. |
| 171 | +
|
| 172 | + The selector prefers optv by default and falls back to Python/Numba when |
| 173 | + optv is unavailable or the user forced the Python engine. |
| 174 | + """ |
| 175 | + |
| 176 | + if get_engine_preference() == ENGINE_PYTHON: |
| 177 | + return False |
| 178 | + |
| 179 | + if feature_name in {None, "", "preprocess_image"}: |
| 180 | + return HAS_NATIVE_PREPROCESS |
| 181 | + |
| 182 | + if feature_name == "target_recognition": |
| 183 | + return HAS_NATIVE_SEGMENTATION |
| 184 | + |
| 185 | + if feature_name == "calibration": |
| 186 | + return HAS_NATIVE_CALIBRATION |
| 187 | + |
| 188 | + if feature_name == "targets": |
| 189 | + return HAS_NATIVE_TARGETS |
| 190 | + |
| 191 | + return HAS_OPTV |
| 192 | + |
| 193 | + |
| 194 | +# Initialize once so the default preference is explicit and optv availability |
| 195 | +# is reported immediately in sessions where it is missing. |
| 196 | +set_engine(DEFAULT_ENGINE, warn_once=True) |
| 197 | + |
| 198 | + |
60 | 199 | def native_preprocess_image(*args: Any, **kwargs: Any) -> Any: |
61 | 200 | if not HAS_NATIVE_PREPROCESS or optv_image_processing is None: |
62 | 201 | raise RuntimeError("optv native preprocess_image is not available") |
|
0 commit comments