From 58bb304354c0ae0c582706eb2f03ad39eb041b53 Mon Sep 17 00:00:00 2001 From: C-Achard Date: Tue, 17 Feb 2026 10:58:59 +0100 Subject: [PATCH 1/4] Refactor GenTL CTI discovery and loading Introduce a dedicated gentl_discovery utility to locate .cti GenTL producer files (from explicit files, env vars, glob patterns and extra dirs) and add selection helpers. Refactor GenTLCameraBackend to use the new discovery logic: resolve multiple CTIs, load all viable producers, persist CTI diagnostics into settings (cti_files, cti_files_loaded, cti_files_failed), and provide robust class-level Harvester building for discovery/quick_ping/get_device_count/rebind_settings. Remove legacy glob/search helpers and change behavior to prefer explicit user config while failing only when no producers can be found/loaded. Update tests and fixtures to use tmp_path-created dummy .cti files, adjust patched Harvester setup, and add unit tests covering CTI discovery and selection policies (including newest/raise-if-multiple behavior). --- dlclivegui/cameras/backends/gentl_backend.py | 382 ++++++++++++------ .../cameras/backends/utils/gentl_discovery.py | 244 +++++++++++ tests/cameras/backends/conftest.py | 24 +- tests/cameras/backends/test_gentl_backend.py | 137 ++++++- 4 files changed, 651 insertions(+), 136 deletions(-) create mode 100644 dlclivegui/cameras/backends/utils/gentl_discovery.py diff --git a/dlclivegui/cameras/backends/gentl_backend.py b/dlclivegui/cameras/backends/gentl_backend.py index c74c0ab..e4db409 100644 --- a/dlclivegui/cameras/backends/gentl_backend.py +++ b/dlclivegui/cameras/backends/gentl_backend.py @@ -3,17 +3,16 @@ # dlclivegui/cameras/backends/gentl_backend.py from __future__ import annotations -import glob import logging -import os import time -from collections.abc import Iterable from typing import ClassVar import cv2 import numpy as np from ..base import CameraBackend, SupportLevel, register_backend +from ..factory import DetectedCamera +from .utils import gentl_discovery as cti_finder LOG = logging.getLogger(__name__) @@ -50,12 +49,6 @@ def __init__(self, settings): if not isinstance(ns, dict): ns = {} - # --- CTI / transport configuration --- - self._cti_file: str | None = ns.get("cti_file") or props.get("cti_file") - self._cti_search_paths: tuple[str, ...] = self._parse_cti_paths( - ns.get("cti_search_paths", props.get("cti_search_paths")) - ) - # --- Fast probe mode (CameraProbeWorker sets this) --- # When fast_start=True, open() should avoid starting acquisition if possible. self._fast_start: bool = bool(ns.get("fast_start", False)) @@ -171,7 +164,7 @@ def static_capabilities(cls) -> dict[str, SupportLevel]: @classmethod def get_device_count(cls) -> int: - """Get the actual number of GenTL devices detected by Harvester. + """Get the number of GenTL devices detected by Harvester. Returns the number of devices found, or -1 if detection fails. """ @@ -180,16 +173,11 @@ def get_device_count(cls) -> int: harvester = None try: - harvester = Harvester() - # Use the static helper to find CTI file with default patterns - cti_file = cls._search_cti_file(cls._DEFAULT_CTI_PATTERNS) - - if not cti_file: + harvester, _, _ = cls._build_harvester_for_discovery(strict_single=False) + if harvester is None: return -1 - - harvester.add_file(cti_file) - harvester.update() - return len(harvester.device_info_list) + infos = harvester.device_info_list or [] + return len(infos) except Exception: return -1 finally: @@ -199,8 +187,150 @@ def get_device_count(cls) -> int: except Exception: pass + def _resolve_cti_files_for_settings(self) -> list[str]: + """ + Resolve CTI files to load. + + Policy: + - If the user explicitly provides ctis (cti_files / cti_file), use only those. + - Otherwise, discover all CTIs (env vars + configured patterns/dirs) and return all. + - Never raise just because multiple CTIs exist. + - Raise only when none are found. + """ + props = self.settings.properties if isinstance(self.settings.properties, dict) else {} + ns = props.get(self.OPTIONS_KEY, {}) + if not isinstance(ns, dict): + ns = {} + + # Explicit CTIs (namespace first, then legacy top-level) + ns_cti_files = ns.get("cti_files") + ns_cti_file = ns.get("cti_file") + legacy_cti_files = props.get("cti_files") + legacy_cti_file = props.get("cti_file") + + # 1) If user provided explicit list/file in namespace, use that only + if ns_cti_files or ns_cti_file: + candidates, diag = cti_finder.discover_cti_files( + cti_file=str(ns_cti_file) if ns_cti_file else None, + cti_files=cti_finder.cti_files_as_list(ns_cti_files) if ns_cti_files else None, + include_env=False, + must_exist=True, + ) + if not candidates: + raise RuntimeError( + "No valid GenTL producer (.cti) found from properties.gentl.cti_file/cti_files.\n\n" + f"Discovery details:\n{diag.summarize()}" + ) + return list(candidates) + + # 2) If user provided explicit list/file in legacy top-level, use that only + if legacy_cti_files or legacy_cti_file: + candidates, diag = cti_finder.discover_cti_files( + cti_file=str(legacy_cti_file) if legacy_cti_file else None, + cti_files=cti_finder.cti_files_as_list(legacy_cti_files) if legacy_cti_files else None, + include_env=False, + must_exist=True, + ) + if not candidates: + raise RuntimeError( + "No valid GenTL producer (.cti) found from properties.cti_file/cti_files.\n\n" + f"Discovery details:\n{diag.summarize()}" + ) + return list(candidates) + + # 3) Discovery path: find all CTIs from env vars + configured patterns/dirs + search_paths = ns.get("cti_search_paths", props.get("cti_search_paths")) + extra_dirs = ns.get("cti_dirs", props.get("cti_dirs")) + + candidates, diag = cti_finder.discover_cti_files( + cti_search_paths=cti_finder.cti_files_as_list(search_paths) if search_paths is not None else None, + include_env=True, + extra_dirs=cti_finder.cti_files_as_list(extra_dirs) if extra_dirs is not None else None, + recursive_env_search=False, + recursive_extra_search=False, + must_exist=True, + ) + + if not candidates: + raise RuntimeError( + "Could not locate any GenTL producer (.cti) file.\n\n" + "Fix options:\n" + " - Set camera.properties.gentl.cti_file to the full path of a .cti file\n" + " - Or set GENICAM_GENTL64_PATH / GENICAM_GENTL32_PATH to include the producer directory\n" + " - Or provide camera.properties.gentl.cti_search_paths with glob patterns\n\n" + f"Discovery details:\n{diag.summarize()}" + ) + + # Default: try to load ALL discovered producers + return list(candidates) + + @classmethod + def _build_harvester_for_discovery( + cls, + *, + strict_single: bool = False, # retained for optional future use + ): + """ + Build a Harvester instance and load CTI producers for class-level operations + (discover_devices, quick_ping, get_device_count, rebind_settings). + + Default policy: try to load ALL discovered producers. + """ + if Harvester is None: + return None, [], None + + candidates, diag = cti_finder.discover_cti_files( + include_env=True, + cti_search_paths=list(cls._DEFAULT_CTI_PATTERNS), + must_exist=True, + ) + + if not candidates: + return None, [], diag + + # Default: load all candidates + cti_files = list(candidates) + + # Optional strict mode (off by default) + if strict_single: + # If you ever want strict, use choose_cti_files here; otherwise ignore. + cti_files = cti_finder.choose_cti_files( + cti_files, policy=cti_finder.GenTLDiscoveryPolicy.RAISE_IF_MULTIPLE, max_files=1 + ) + + harvester = Harvester() + loaded: list[str] = [] + failures: list[tuple[str, str]] = [] + + for cti in cti_files: + try: + harvester.add_file(cti) + loaded.append(cti) + except Exception as exc: + failures.append((cti, str(exc))) + LOG.warning("Failed to load CTI '%s': %s", cti, exc) + + if not loaded: + try: + harvester.reset() + except Exception: + pass + return None, [], diag + + try: + harvester.update() + except Exception as exc: + LOG.warning("Harvester.update() failed during discovery: %s", exc) + try: + harvester.reset() + except Exception: + pass + return None, loaded, diag + + return harvester, loaded, diag + def open(self) -> None: - if Harvester is None: # pragma: no cover - optional dependency + if Harvester is None: # pragma: no cover raise RuntimeError( "The 'harvesters' package is required for the GenTL backend. Install it via 'pip install harvesters'." ) @@ -214,23 +344,60 @@ def open(self) -> None: ns = {} props[self.OPTIONS_KEY] = ns + # Resolve CTIs (may return many). This no longer raises just because there are multiple. + cti_files = self._resolve_cti_files_for_settings() + self._harvester = Harvester() - # Resolve CTI file: explicit > configured > search - cti_file = self._cti_file or ns.get("cti_file") or props.get("cti_file") or self._find_cti_file() - self._harvester.add_file(cti_file) + loaded: list[str] = [] + failed: list[tuple[str, str]] = [] + + for cti in cti_files: + try: + self._harvester.add_file(cti) + loaded.append(cti) + except Exception as exc: + failed.append((cti, str(exc))) + LOG.warning("Failed to load CTI '%s': %s", cti, exc) + + # Persist diagnostics for UI / debugging + ns["cti_files"] = [str(p) for p in cti_files] # all resolved candidates + ns["cti_files_loaded"] = [str(p) for p in loaded] # successfully added to harvester + ns["cti_files_failed"] = [{"cti": c, "error": e} for c, e in failed] # load failures + + # Keep single-cti convenience key for backward compatibility / display + if loaded: + ns["cti_file"] = str(loaded[0]) + elif cti_files: + ns["cti_file"] = str(cti_files[0]) # best effort + + if not loaded: + raise RuntimeError( + "No GenTL producer (.cti) could be loaded.\n\n" + f"Resolved CTIs: {cti_files}\n" + f"Failures: {failed}\n" + "Fix: remove/repair incompatible producers or" + " set properties.gentl.cti_file to a known working producer." + ) + + # Update device list after loading producers self._harvester.update() if not self._harvester.device_info_list: - raise RuntimeError("No GenTL cameras detected via Harvesters") + raise RuntimeError( + "No GenTL cameras detected via Harvesters after loading producers.\n\n" + f"Loaded CTIs: {loaded}\n" + f"Failed CTIs: {failed}\n" + "Fix: ensure your camera vendor's GenTL producer is installed and working." + ) infos = list(self._harvester.device_info_list) - # Helper: robustly read device_info fields (supports dict-like or attribute-like entries) + # Helper: robustly read device_info fields (dict-like or attribute-like) def _info_get(info, key: str, default=None): try: if hasattr(info, "get"): - v = info.get(key) # type: ignore[attr-defined] + v = info.get(key) if v is not None: return v except Exception: @@ -250,12 +417,11 @@ def _info_get(info, key: str, default=None): selected_index: int | None = None selected_serial: str | None = None - # 1) Try stable device_id first (supports "serial:..." and "fp:...") target_device_id = self._device_id or ns.get("device_id") or props.get("device_id") if target_device_id: target_device_id = str(target_device_id).strip() - # Match exact against computed device_id_from_info(info) + # Exact match against computed device_id for idx, info in enumerate(infos): try: did = self._device_id_from_info(info) @@ -296,7 +462,7 @@ def _info_get(info, key: str, default=None): f"Ambiguous GenTL serial match for '{serial_target}'. Candidates: {candidates}" ) - # 2) Try legacy serial selection if still not selected + # Legacy serial selection fallback if selected_index is None: serial = self._serial_number if serial: @@ -327,7 +493,7 @@ def _info_get(info, key: str, default=None): available = [str(_info_get(i, "serial_number", "")).strip() for i in infos] raise RuntimeError(f"Camera with serial '{serial}' not found. Available cameras: {available}") - # 3) Fallback to index selection + # Index fallback if selected_index is None: device_count = len(infos) if requested_index < 0 or requested_index >= device_count: @@ -336,20 +502,17 @@ def _info_get(info, key: str, default=None): sn = _info_get(infos[selected_index], "serial_number", "") selected_serial = str(sn).strip() if sn else None - # Update settings.index to the actual selected index (important for UI merge-back + stability) + # Update settings.index to actual selected index (UI stability) self.settings.index = int(selected_index) selected_info = infos[int(selected_index)] - # ------------------------------------------------------------------ - # Create ImageAcquirer using the latest Harvesters API: Harvester.create(...) - # ------------------------------------------------------------------ + # Create ImageAcquirer via Harvester.create(...) try: if selected_serial: self._acquirer = self._harvester.create({"serial_number": str(selected_serial)}) else: self._acquirer = self._harvester.create(int(selected_index)) except TypeError: - # Some versions accept keyword argument; keep as a safety net without reintroducing legacy API. if selected_serial: self._acquirer = self._harvester.create({"serial_number": str(selected_serial)}) else: @@ -358,21 +521,16 @@ def _info_get(info, key: str, default=None): remote = self._acquirer.remote_device node_map = remote.node_map - # Resolve human label for UI self._device_label = self._resolve_device_label(node_map) - # ------------------------------------------------------------------ - # Apply configuration (existing behavior) - # ------------------------------------------------------------------ + # Apply configuration self._configure_pixel_format(node_map) self._configure_resolution(node_map) self._configure_exposure(node_map) self._configure_gain(node_map) self._configure_frame_rate(node_map) - # ------------------------------------------------------------------ - # Capture "actual" telemetry for GUI (existing behavior) - # ------------------------------------------------------------------ + # Read back telemetry try: self._actual_width = int(node_map.Width.value) self._actual_height = int(node_map.Height.value) @@ -394,9 +552,7 @@ def _info_get(info, key: str, default=None): except Exception: self._actual_gain = None - # ------------------------------------------------------------------ - # Persist identity + richer device metadata back into settings for UI merge-back - # ------------------------------------------------------------------ + # Persist identity + metadata computed_id = None try: computed_id = self._device_id_from_info(selected_info) @@ -408,16 +564,13 @@ def _info_get(info, key: str, default=None): elif selected_serial: ns["device_id"] = f"serial:{selected_serial}" - # Canonical serial storage if selected_serial: ns["serial_number"] = str(selected_serial) ns["device_serial_number"] = str(selected_serial) - # UI-friendly name if self._device_label: ns["device_name"] = str(self._device_label) - # Extra metadata from discovery info (helps debugging and stable identity fallbacks) ns["device_display_name"] = str(_info_get(selected_info, "display_name", "") or "") ns["device_info_id"] = str(_info_get(selected_info, "id_", "") or "") ns["device_vendor"] = str(_info_get(selected_info, "vendor", "") or "") @@ -427,12 +580,7 @@ def _info_get(info, key: str, default=None): ns["device_version"] = str(_info_get(selected_info, "version", "") or "") ns["device_access_status"] = _info_get(selected_info, "access_status", None) - # Preserve CTI used (useful for stable operation) - ns["cti_file"] = str(cti_file) - - # ------------------------------------------------------------------ - # Start streaming unless fast_start probe mode is requested - # ------------------------------------------------------------------ + # Start acquisition unless fast_start if getattr(self, "_fast_start", False): LOG.info("GenTL open() in fast_start probe mode: acquisition not started.") return @@ -505,13 +653,15 @@ def discover_devices( """ Rich discovery path for CameraFactory.detect_cameras(). Returns a list of DetectedCamera with device_id filled when possible. + + Cross-platform CTI discovery: + - Uses GENICAM_GENTL64_PATH / GENICAM_GENTL32_PATH when available + - Falls back to built-in Windows patterns + - Best-effort loads multiple CTI producers """ if Harvester is None: return [] - # Local import to avoid circulars at import time - from ..factory import DetectedCamera - def _canceled() -> bool: return bool(should_cancel and should_cancel()) @@ -520,17 +670,15 @@ def _canceled() -> bool: if progress_cb: progress_cb("Initializing GenTL discovery…") - harvester = Harvester() + harvester, loaded, _ = cls._build_harvester_for_discovery(strict_single=False) - # Use default CTI search; we don't have per-camera settings here. - cti_file = cls._search_cti_file(cls._DEFAULT_CTI_PATTERNS) - if not cti_file: + if harvester is None or not loaded: if progress_cb: - progress_cb("No .cti found (GenTL producer missing).") + progress_cb("No GenTL producers could be loaded.") return [] - harvester.add_file(cti_file) - harvester.update() + if progress_cb: + progress_cb(f"Loaded {len(loaded)} GenTL producer(s). Scanning devices…") infos = list(harvester.device_info_list or []) if not infos: @@ -543,8 +691,8 @@ def _canceled() -> bool: if _canceled(): break - # Create a label for the UI, using display_name if available, otherwise vendor/model/serial. info = infos[idx] + display_name = None try: display_name = ( @@ -565,10 +713,7 @@ def _canceled() -> bool: or (info.get("serial_number") if hasattr(info, "get") else None) or "" ) - vendor = str(vendor).strip() - model = str(model).strip() - serial = str(serial).strip() - + vendor, model, serial = str(vendor).strip(), str(model).strip(), str(serial).strip() label = f"{vendor} {model}".strip() if (vendor or model) else f"GenTL device {idx}" if serial: label = f"{label} ({serial})" @@ -580,7 +725,6 @@ def _canceled() -> bool: index=idx, label=label, device_id=device_id, - # GenTL usually doesn't expose vid/pid/path consistently; leave None unless you have it vid=None, pid=None, path=None, @@ -595,8 +739,6 @@ def _canceled() -> bool: return out except Exception: - # Returning None would trigger probing fallback; but since you declared discovery supported, - # returning [] is usually less surprising than a slow probe storm. return [] finally: if harvester is not None: @@ -610,6 +752,10 @@ def rebind_settings(cls, settings): """ If a stable identity exists in settings.properties['gentl'], map it to the correct current index (and serial_number if available). + + Strategy: + - If ctis were previously persisted (cti_files/cti_file), prefer those. + - Otherwise, fall back to env-var + pattern discovery (best-effort). """ if Harvester is None: return settings @@ -625,25 +771,48 @@ def rebind_settings(cls, settings): harvester = None try: - harvester = Harvester() - cti_file = ns.get("cti_file") or props.get("cti_file") or cls._search_cti_file(cls._DEFAULT_CTI_PATTERNS) - if not cti_file: - return settings - - harvester.add_file(cti_file) - harvester.update() + # Prefer persisted CTIs for stability if present + explicit_files = ns.get("cti_files") or props.get("cti_files") + explicit_file = ns.get("cti_file") or props.get("cti_file") + + if explicit_files or explicit_file: + candidates, diag = cti_finder.discover_cti_files( + cti_file=explicit_file, + cti_files=cti_finder.cti_files_as_list(explicit_files), + include_env=False, + must_exist=True, + ) + if not candidates: + return settings + + harvester = Harvester() + loaded: list[str] = [] + for cti in candidates: + try: + harvester.add_file(cti) + loaded.append(cti) + except Exception: + continue + + if not loaded: + return settings + + harvester.update() + else: + harvester, loaded, diag = cls._build_harvester_for_discovery(strict_single=False) + if harvester is None: + return settings infos = list(harvester.device_info_list or []) if not infos: return settings - # Try exact match by computed device_id first + target_id_str = str(target_id).strip() + match_index = None match_serial = None - # Normalize - target_id_str = str(target_id).strip() - + # 1) Exact match by computed device_id for idx, info in enumerate(infos): dev_id = cls._device_id_from_info(info) if dev_id and dev_id == target_id_str: @@ -651,7 +820,7 @@ def rebind_settings(cls, settings): match_serial = getattr(info, "serial_number", None) break - # If not found, fallback: treat target as serial-ish substring (legacy behavior) + # 2) Fallback: treat target as serial-ish substring if match_index is None: for idx, info in enumerate(infos): serial = getattr(info, "serial_number", None) @@ -666,7 +835,7 @@ def rebind_settings(cls, settings): # Apply rebinding settings.index = int(match_index) - # Keep namespace consistent for open() + # Ensure namespace exists if not isinstance(settings.properties, dict): settings.properties = {} ns2 = settings.properties.setdefault(cls.OPTIONS_KEY, {}) @@ -674,7 +843,6 @@ def rebind_settings(cls, settings): ns2 = {} settings.properties[cls.OPTIONS_KEY] = ns2 - # If we got a serial, save it for open() selection (backward compatible) if match_serial: ns2["serial_number"] = str(match_serial) ns2["device_id"] = target_id_str @@ -682,7 +850,6 @@ def rebind_settings(cls, settings): return settings except Exception: - # Any failure should not prevent fallback to index-based open return settings finally: if harvester is not None: @@ -702,12 +869,9 @@ def quick_ping(cls, index: int, _unused=None) -> bool: harvester = None try: - harvester = Harvester() - cti_file = cls._search_cti_file(cls._DEFAULT_CTI_PATTERNS) - if not cti_file: + harvester, _, _ = cls._build_harvester_for_discovery(strict_single=False) + if harvester is None: return False - harvester.add_file(cti_file) - harvester.update() infos = harvester.device_info_list or [] return 0 <= int(index) < len(infos) except Exception: @@ -795,15 +959,6 @@ def close(self) -> None: # Helpers # ------------------------------------------------------------------ - def _parse_cti_paths(self, value) -> tuple[str, ...]: - if value is None: - return self._DEFAULT_CTI_PATTERNS - if isinstance(value, str): - return (value,) - if isinstance(value, Iterable): - return tuple(str(item) for item in value) - return self._DEFAULT_CTI_PATTERNS - def _parse_crop(self, crop) -> tuple[int, int, int, int] | None: if isinstance(crop, (list, tuple)) and len(crop) == 4: return tuple(int(v) for v in crop) @@ -881,31 +1036,6 @@ def _configure_resolution(self, node_map) -> None: else: LOG.info(f"Resolution set to {actual_width}x{actual_height}") - @staticmethod - def _search_cti_file(patterns: tuple[str, ...]) -> str | None: - """Search for a CTI file using the given patterns. - - Returns the first CTI file found, or None if none found. - """ - for pattern in patterns: - for file_path in glob.glob(pattern): - if os.path.isfile(file_path): - return file_path - return None - - def _find_cti_file(self) -> str: - """Find a CTI file using configured or default search paths. - - Raises RuntimeError if no CTI file is found. - """ - cti_file = self._search_cti_file(self._cti_search_paths) - if cti_file is None: - raise RuntimeError( - "Could not locate a GenTL producer (.cti) file. Set 'cti_file' in " - "camera.properties or provide search paths via 'cti_search_paths'." - ) - return cti_file - def _available_serials(self) -> list[str]: assert self._harvester is not None serials: list[str] = [] diff --git a/dlclivegui/cameras/backends/utils/gentl_discovery.py b/dlclivegui/cameras/backends/utils/gentl_discovery.py new file mode 100644 index 0000000..ca73662 --- /dev/null +++ b/dlclivegui/cameras/backends/utils/gentl_discovery.py @@ -0,0 +1,244 @@ +"""Helpers to locate .cti GenTL producer files from various sources +(explicit, env vars, glob patterns, etc.) for GenTL-based camera backends.""" + +# dlclivegui/cameras/backends/utils/gentl_discovery.py +from __future__ import annotations + +import glob +import os +from collections.abc import Iterable, Sequence +from dataclasses import dataclass, field +from enum import Enum, auto +from pathlib import Path + + +class GenTLDiscoveryPolicy(Enum): + FIRST = auto() # default: take first N candidates in order found + NEWEST = auto() # take N candidates with most recent modification time (mtime) + RAISE_IF_MULTIPLE = auto() # if > N candidates, raise an error to avoid ambiguity (forces explicit config) + + +@dataclass +class CTIDiscoveryDiagnostics: + explicit_files: list[str] = field(default_factory=list) + glob_patterns: list[str] = field(default_factory=list) + env_vars_used: dict[str, str] = field(default_factory=dict) # name -> raw value + env_paths_expanded: list[str] = field(default_factory=list) # directories/files derived from env vars + extra_dirs: list[str] = field(default_factory=list) + + candidates: list[str] = field(default_factory=list) + rejected: list[tuple[str, str]] = field(default_factory=list) # (path, reason) + + def summarize(self) -> str: + lines = [] + if self.explicit_files: + lines.append(f"Explicit CTI file(s): {self.explicit_files}") + if self.glob_patterns: + lines.append(f"CTI glob pattern(s): {self.glob_patterns}") + if self.env_vars_used: + # Keep raw env var values in summary; you can redact if needed + lines.append(f"Env vars: {self.env_vars_used}") + if self.env_paths_expanded: + lines.append(f"Env-derived path entries: {self.env_paths_expanded}") + if self.extra_dirs: + lines.append(f"Extra CTI dirs: {self.extra_dirs}") + if self.candidates: + lines.append(f"CTI candidate(s) ({len(self.candidates)}): {self.candidates}") + if self.rejected: + lines.append(f"Rejected ({len(self.rejected)}): " + "; ".join([f"{p} ({r})" for p, r in self.rejected])) + return "\n".join(lines) + + +def cti_files_as_list(value) -> list[str]: + if value is None: + return [] + if isinstance(value, (list, tuple, set)): + return [str(v) for v in value if v is not None and str(v).strip()] + s = str(value).strip() + return [s] if s else [] + + +def _normalize_path(p: str) -> str: + """ + Normalize a filesystem path in a cross-platform way: + - expands ~ and environment variables + - resolves to absolute where possible (without requiring existence) + """ + pp = Path(os.path.expandvars(os.path.expanduser(p))) + try: + # resolve(False) avoids raising if parts don't exist (py>=3.9) + return str(pp.resolve(strict=False)) + except Exception: + return str(pp.absolute()) + + +def _iter_cti_files_in_dir(directory: str, recursive: bool = False) -> Iterable[str]: + """ + Yield *.cti files in directory. Non-recursive by default (faster, safer). + """ + d = Path(directory) + if not d.is_dir(): + return + if recursive: + yield from (str(p) for p in d.rglob("*.cti")) + else: + yield from (str(p) for p in d.glob("*.cti")) + + +def _split_env_paths(raw: str) -> list[str]: + """ + Split environment variable paths using os.pathsep (cross-platform). + Also trims whitespace and strips surrounding quotes. + """ + out: list[str] = [] + for item in (raw or "").split(os.pathsep): + s = item.strip().strip('"').strip("'") + if s: + out.append(s) + return out + + +def discover_cti_files( + *, + cti_file: str | None = None, + cti_files: Sequence[str] | None = None, + cti_search_paths: Sequence[str] | None = None, + include_env: bool = True, + env_vars: Sequence[str] = ("GENICAM_GENTL64_PATH", "GENICAM_GENTL32_PATH"), + extra_dirs: Sequence[str] | None = None, + recursive_env_search: bool = False, + recursive_extra_search: bool = False, + must_exist: bool = True, +) -> tuple[list[str], CTIDiscoveryDiagnostics]: + """ + Discover candidate GenTL producer (.cti) files from multiple sources. + + Returns: + (candidates, diagnostics) + + Notes: + - If must_exist=True (recommended), only existing files are returned. + - Env vars are parsed as path lists; each entry may be a directory OR a .cti file. + """ + diag = CTIDiscoveryDiagnostics() + + # 1) Explicit CTI file(s) + explicit = [] + explicit += cti_files_as_list(cti_file) + explicit += cti_files_as_list(cti_files) + diag.explicit_files = explicit[:] + + # 2) Glob patterns + patterns = cti_files_as_list(cti_search_paths) + diag.glob_patterns = patterns[:] + + # 3) Env var paths + env_entries: list[str] = [] + if include_env: + for name in env_vars: + raw = os.environ.get(name, "") + if raw: + diag.env_vars_used[name] = raw + env_entries.extend(_split_env_paths(raw)) + diag.env_paths_expanded = env_entries[:] + + # 4) Extra directories + extras = cti_files_as_list(extra_dirs) + diag.extra_dirs = extras[:] + + candidates: list[str] = [] + rejected: list[tuple[str, str]] = [] + + def _add_candidate(path: str, reason_ctx: str) -> None: + norm = _normalize_path(path) + if must_exist and not Path(norm).is_file(): + rejected.append((norm, f"not a file ({reason_ctx})")) + return + if not norm.lower().endswith(".cti"): + rejected.append((norm, f"not a .cti ({reason_ctx})")) + return + candidates.append(norm) + + # Process explicit files + for p in explicit: + _add_candidate(p, "explicit") + + # Process glob patterns + for pat in patterns: + # Normalize only for readability; glob needs pattern semantics, so we expanduser/vars but keep globbing + expanded_pat = os.path.expandvars(os.path.expanduser(pat)) + for hit in glob.glob(expanded_pat): + _add_candidate(hit, f"glob:{pat}") + + # Process env var entries + for entry in env_entries: + norm_entry = _normalize_path(entry) + p = Path(norm_entry) + if p.is_file() and p.suffix.lower() == ".cti": + _add_candidate(norm_entry, "env:file") + elif p.is_dir(): + for f in _iter_cti_files_in_dir(norm_entry, recursive=recursive_env_search): + _add_candidate(f, "env:dir") + else: + rejected.append((norm_entry, "env entry missing (not file/dir)")) + + # Process extra dirs + for d in extras: + norm_d = _normalize_path(d) + if Path(norm_d).is_dir(): + for f in _iter_cti_files_in_dir(norm_d, recursive=recursive_extra_search): + _add_candidate(f, "extra:dir") + elif Path(norm_d).is_file(): + _add_candidate(norm_d, "extra:file") + else: + rejected.append((norm_d, "extra entry missing (not file/dir)")) + + # Deduplicate while preserving order + seen = set() + unique: list[str] = [] + for c in candidates: + key = c.lower() if os.name == "nt" else c # Windows case-insensitive + if key in seen: + continue + seen.add(key) + unique.append(c) + + diag.candidates = unique[:] + diag.rejected = rejected[:] + return unique, diag + + +def choose_cti_files( + candidates: Sequence[str], + *, + policy: GenTLDiscoveryPolicy = GenTLDiscoveryPolicy.FIRST, + max_files: int = 1, +) -> list[str]: + """ + Choose which CTI file(s) to load from candidates. + + policy: + - FIRST: take the first N candidates (default) + - NEWEST: take the N most recently modified candidates + - RAISE_IF_MULTIPLE: if more than N candidates, raise an error (to avoid ambiguity) + """ + cand = [str(c) for c in candidates if c] + if not cand: + return [] + + if policy == GenTLDiscoveryPolicy.NEWEST: + cand_sorted = sorted(cand, key=lambda p: Path(p).stat().st_mtime if Path(p).exists() else 0.0, reverse=True) + return cand_sorted[:max_files] + + if policy == GenTLDiscoveryPolicy.FIRST: + return cand[:max_files] + + if policy == GenTLDiscoveryPolicy.RAISE_IF_MULTIPLE: + if len(cand) > max_files: + raise RuntimeError( + f"Multiple GenTL producers (.cti) found ({len(cand)}). " + f"Please set properties.gentl.cti_file explicitly. Candidates: {cand}" + ) + return cand[:max_files] + + raise ValueError(f"Unknown policy: {policy!r}") diff --git a/tests/cameras/backends/conftest.py b/tests/cameras/backends/conftest.py index 0fb3ded..57a049a 100644 --- a/tests/cameras/backends/conftest.py +++ b/tests/cameras/backends/conftest.py @@ -830,10 +830,11 @@ def _make(): @pytest.fixture() -def patch_gentl_sdk(monkeypatch, fake_harvester_factory): +def patch_gentl_sdk(monkeypatch, fake_harvester_factory, tmp_path): """ Patch dlclivegui.cameras.backends.gentl_backend to use FakeHarvester + Fake timeout. - Also bypass CTI search logic so tests never hit filesystem/SDK paths. + Also ensure CTI discovery succeeds for classmethods (discover_devices/quick_ping) + by creating a real dummy .cti and exposing it via GENICAM_GENTL64_PATH. """ import dlclivegui.cameras.backends.gentl_backend as gb @@ -843,17 +844,19 @@ def patch_gentl_sdk(monkeypatch, fake_harvester_factory): # Keep your backend timeout contract as-is: it catches HarvesterTimeoutError monkeypatch.setattr(gb, "HarvesterTimeoutError", FakeGenTLTimeoutException, raising=False) - # Avoid filesystem CTI searching - monkeypatch.setattr(gb.GenTLCameraBackend, "_find_cti_file", lambda self: "dummy.cti", raising=False) - monkeypatch.setattr( - gb.GenTLCameraBackend, "_search_cti_file", staticmethod(lambda patterns: "dummy.cti"), raising=False - ) + # Create a real CTI file and advertise it via env var (cross-platform via os.pathsep) + cti_file = tmp_path / "dummy.cti" + if not cti_file.exists(): + cti_file.write_text("fake", encoding="utf-8") + + monkeypatch.setenv("GENICAM_GENTL64_PATH", str(tmp_path)) + monkeypatch.delenv("GENICAM_GENTL32_PATH", raising=False) return gb @pytest.fixture() -def gentl_settings_factory(): +def gentl_settings_factory(tmp_path): """ Convenience factory for CameraSettings for gentl backend tests. """ @@ -871,8 +874,13 @@ def _make( enabled=True, properties=None, ): + cti = tmp_path / "dummy.cti" + if not cti.exists(): + cti.write_text("fake", encoding="utf-8") props = properties if isinstance(properties, dict) else {} props.setdefault("gentl", {}) + props["gentl"] = dict(props["gentl"]) + props["gentl"].setdefault("cti_file", str(cti)) return CameraSettings( name=name, index=index, diff --git a/tests/cameras/backends/test_gentl_backend.py b/tests/cameras/backends/test_gentl_backend.py index d31bc9c..7ee8ac1 100644 --- a/tests/cameras/backends/test_gentl_backend.py +++ b/tests/cameras/backends/test_gentl_backend.py @@ -1,11 +1,20 @@ # tests/cameras/backends/test_gentl_backend.py from __future__ import annotations +import os +import time import types +from pathlib import Path import numpy as np import pytest +from dlclivegui.cameras.backends.utils.gentl_discovery import ( + GenTLDiscoveryPolicy, + choose_cti_files, + discover_cti_files, +) + # --------------------------------------------------------------------- # Core lifecycle + strict transaction model @@ -235,10 +244,12 @@ def test_open_persists_rich_metadata_in_namespace(patch_gentl_sdk, gentl_setting be.close() -def test_open_persists_cti_file_even_when_provided_in_props(patch_gentl_sdk, gentl_settings_factory): +def test_open_persists_cti_file_even_when_provided_in_props(patch_gentl_sdk, gentl_settings_factory, tmp_path): gb = patch_gentl_sdk - settings = gentl_settings_factory(properties={"cti_file": "from-props.cti", "gentl": {}}) + cti = tmp_path / "from-props.cti" + cti.write_text("dummy", encoding="utf-8") + settings = gentl_settings_factory(properties={"cti_file": str(cti), "gentl": {}}) be = gb.GenTLCameraBackend(settings) be.open() @@ -529,3 +540,125 @@ def create_image_acquirer(self, *args, **kwargs): # Error message should include some context about attempted creation methods msg = str(ei.value).lower() assert "failed to initialise gentl image acquirer" in msg + + +# ---------------------------------- +# CTI discovery and selection logic +# ---------------------------------- + + +def _make_cti(tmp_path: Path, name: str = "Producer.cti") -> Path: + p = tmp_path / name + p.write_text("dummy", encoding="utf-8") + return p + + +def test_discover_explicit_cti_file(tmp_path): + cti = _make_cti(tmp_path, "A.cti") + + candidates, diag = discover_cti_files(cti_file=str(cti), include_env=False, must_exist=True) + + assert len(candidates) == 1 + assert Path(candidates[0]).name == "A.cti" + assert diag.explicit_files == [str(cti)] + + +def test_discover_missing_explicit_cti_is_rejected(tmp_path): + missing = tmp_path / "Missing.cti" + candidates, diag = discover_cti_files(cti_file=str(missing), include_env=False, must_exist=True) + + assert candidates == [] + assert any("not a file" in reason for _, reason in diag.rejected) + + +def test_discover_glob_patterns(tmp_path): + _make_cti(tmp_path, "One.cti") + _make_cti(tmp_path, "Two.cti") + + pattern = str(tmp_path / "*.cti") + candidates, diag = discover_cti_files(cti_search_paths=[pattern], include_env=False, must_exist=True) + + names = sorted(Path(c).name for c in candidates) + assert names == ["One.cti", "Two.cti"] + assert pattern in diag.glob_patterns + + +def test_discover_env_var_directory(monkeypatch, tmp_path): + _make_cti(tmp_path, "Env.cti") + + monkeypatch.setenv("GENICAM_GENTL64_PATH", str(tmp_path)) + + candidates, diag = discover_cti_files(include_env=True, must_exist=True) + + assert any(Path(c).name == "Env.cti" for c in candidates) + assert "GENICAM_GENTL64_PATH" in diag.env_vars_used + + +def test_discover_env_var_direct_file(monkeypatch, tmp_path): + cti = _make_cti(tmp_path, "Direct.cti") + + monkeypatch.setenv("GENICAM_GENTL64_PATH", str(cti)) + + candidates, diag = discover_cti_files(include_env=True, must_exist=True) + + assert candidates == [str(cti.resolve())] or Path(candidates[0]).name == "Direct.cti" + assert diag.env_paths_expanded # should include the raw env entry + + +def test_discover_env_var_multiple_entries(monkeypatch, tmp_path): + d1 = tmp_path / "d1" + d2 = tmp_path / "d2" + d1.mkdir() + d2.mkdir() + + _make_cti(d1, "A.cti") + _make_cti(d2, "B.cti") + + combined = f"{d1}{os.pathsep}{d2}" + monkeypatch.setenv("GENICAM_GENTL64_PATH", combined) + + candidates, _ = discover_cti_files(include_env=True, must_exist=True) + names = sorted(Path(c).name for c in candidates) + + assert names == ["A.cti", "B.cti"] + + +def test_discover_deduplicates_same_file_from_multiple_sources(monkeypatch, tmp_path): + cti = _make_cti(tmp_path, "Dup.cti") + + # Discover it twice: explicit + env dir + monkeypatch.setenv("GENICAM_GENTL64_PATH", str(tmp_path)) + + candidates, _ = discover_cti_files( + cti_file=str(cti), + include_env=True, + must_exist=True, + ) + + # Should appear only once + assert len(candidates) == 1 + assert Path(candidates[0]).name == "Dup.cti" + + +def test_choose_cti_files_raises_if_multiple_candidates(tmp_path): + c1 = _make_cti(tmp_path, "One.cti") + c2 = _make_cti(tmp_path, "Two.cti") + + # This is the key test: "duplicates should raise" (i.e. >1 CTI found) + with pytest.raises(RuntimeError) as exc: + choose_cti_files([str(c1), str(c2)], policy=GenTLDiscoveryPolicy.RAISE_IF_MULTIPLE, max_files=1) + + assert "Multiple GenTL producers" in str(exc.value) + + +def test_choose_cti_files_newest_policy(tmp_path): + old = _make_cti(tmp_path, "Old.cti") + new = _make_cti(tmp_path, "New.cti") + + # Ensure distinct mtimes + time.sleep(0.01) + new.write_text("dummy2", encoding="utf-8") + + selected = choose_cti_files([str(old), str(new)], policy=GenTLDiscoveryPolicy.NEWEST, max_files=1) + assert len(selected) == 1 + assert Path(selected[0]).name == "New.cti" From d21cd3e5200e378d6761d8e47170ae557389765b Mon Sep 17 00:00:00 2001 From: C-Achard Date: Thu, 19 Feb 2026 09:40:00 +0100 Subject: [PATCH 2/4] Address review comments --- dlclivegui/cameras/backends/gentl_backend.py | 2 ++ .../cameras/backends/utils/gentl_discovery.py | 13 +++++++++++-- tests/cameras/backends/test_gentl_backend.py | 8 +++++--- 3 files changed, 18 insertions(+), 5 deletions(-) diff --git a/dlclivegui/cameras/backends/gentl_backend.py b/dlclivegui/cameras/backends/gentl_backend.py index e4db409..2dd6551 100644 --- a/dlclivegui/cameras/backends/gentl_backend.py +++ b/dlclivegui/cameras/backends/gentl_backend.py @@ -372,6 +372,7 @@ def open(self) -> None: ns["cti_file"] = str(cti_files[0]) # best effort if not loaded: + self._harvester = None raise RuntimeError( "No GenTL producer (.cti) could be loaded.\n\n" f"Resolved CTIs: {cti_files}\n" @@ -384,6 +385,7 @@ def open(self) -> None: self._harvester.update() if not self._harvester.device_info_list: + self._harvester = None raise RuntimeError( "No GenTL cameras detected via Harvesters after loading producers.\n\n" f"Loaded CTIs: {loaded}\n" diff --git a/dlclivegui/cameras/backends/utils/gentl_discovery.py b/dlclivegui/cameras/backends/utils/gentl_discovery.py index ca73662..a8c2006 100644 --- a/dlclivegui/cameras/backends/utils/gentl_discovery.py +++ b/dlclivegui/cameras/backends/utils/gentl_discovery.py @@ -174,7 +174,7 @@ def _add_candidate(path: str, reason_ctx: str) -> None: for entry in env_entries: norm_entry = _normalize_path(entry) p = Path(norm_entry) - if p.is_file() and p.suffix.lower() == ".cti": + if p.is_file(): # let _add_candidate check .cti extension and existence _add_candidate(norm_entry, "env:file") elif p.is_dir(): for f in _iter_cti_files_in_dir(norm_entry, recursive=recursive_env_search): @@ -227,7 +227,16 @@ def choose_cti_files( return [] if policy == GenTLDiscoveryPolicy.NEWEST: - cand_sorted = sorted(cand, key=lambda p: Path(p).stat().st_mtime if Path(p).exists() else 0.0, reverse=True) + + def _newest_mtime(p: str) -> float: + try: + if not Path(p).exists(): + return 0.0 + return Path(p).stat().st_mtime + except OSError: + return 0.0 + + cand_sorted = sorted(cand, key=_newest_mtime, reverse=True) return cand_sorted[:max_files] if policy == GenTLDiscoveryPolicy.FIRST: diff --git a/tests/cameras/backends/test_gentl_backend.py b/tests/cameras/backends/test_gentl_backend.py index 7ee8ac1..8e9a40f 100644 --- a/tests/cameras/backends/test_gentl_backend.py +++ b/tests/cameras/backends/test_gentl_backend.py @@ -2,7 +2,6 @@ from __future__ import annotations import os -import time import types from pathlib import Path @@ -655,9 +654,12 @@ def test_choose_cti_files_newest_policy(tmp_path): old = _make_cti(tmp_path, "Old.cti") new = _make_cti(tmp_path, "New.cti") - # Ensure distinct mtimes - time.sleep(0.01) + # Ensure distinct mtimes (platform agnostic) new.write_text("dummy2", encoding="utf-8") + old_stat = old.stat() + new_stat = new.stat() + if new_stat.st_mtime <= old_stat.st_mtime: + os.utime(new, (new_stat.st_atime, old_stat.st_mtime + 1)) selected = choose_cti_files([str(old), str(new)], policy=GenTLDiscoveryPolicy.NEWEST, max_files=1) assert len(selected) == 1 From 70284dbc4aab56defbe00f0c51a6412022f51d7e Mon Sep 17 00:00:00 2001 From: C-Achard Date: Thu, 19 Feb 2026 16:08:10 +0100 Subject: [PATCH 3/4] gentl: reset harvester and CTI failure tests Add a _reset_harvester helper on GenTLCameraBackend and call it where the harvester must be torn down on failure to ensure proper cleanup. Enhance FakeHarvester to support deterministic add_file failures and to record reset/add/update/create calls; keep create_image_acquirer for compatibility. Update test fixtures (conftest) to expose gentl_fail_add_file_for control, inject a dummy CTI only when none are explicitly provided, and expose gb.fail_add_file_for from patch_gentl_sdk. Add test helpers and new tests to isolate GENICAM env vars and verify CTI load diagnostics for all-success, partial-failure, and complete-failure scenarios. Also adjust some discovery tests to explicitly isolate the environment and tighten assertions. --- dlclivegui/cameras/backends/gentl_backend.py | 19 +++- tests/cameras/backends/conftest.py | 68 ++++++++--- tests/cameras/backends/test_gentl_backend.py | 114 ++++++++++++++++++- 3 files changed, 178 insertions(+), 23 deletions(-) diff --git a/dlclivegui/cameras/backends/gentl_backend.py b/dlclivegui/cameras/backends/gentl_backend.py index 2dd6551..405a7ab 100644 --- a/dlclivegui/cameras/backends/gentl_backend.py +++ b/dlclivegui/cameras/backends/gentl_backend.py @@ -372,7 +372,7 @@ def open(self) -> None: ns["cti_file"] = str(cti_files[0]) # best effort if not loaded: - self._harvester = None + self._reset_harvester() raise RuntimeError( "No GenTL producer (.cti) could be loaded.\n\n" f"Resolved CTIs: {cti_files}\n" @@ -385,7 +385,7 @@ def open(self) -> None: self._harvester.update() if not self._harvester.device_info_list: - self._harvester = None + self._reset_harvester() raise RuntimeError( "No GenTL cameras detected via Harvesters after loading producers.\n\n" f"Loaded CTIs: {loaded}\n" @@ -797,6 +797,7 @@ def rebind_settings(cls, settings): continue if not loaded: + cls._reset_select_harvester(harvester) return settings harvester.update() @@ -936,6 +937,20 @@ def stop(self) -> None: except Exception: pass + @staticmethod + def _reset_select_harvester(harvester) -> None: + if harvester is not None: + try: + harvester.reset() + except Exception: + pass + + def _reset_harvester(self) -> None: + try: + self._reset_select_harvester(self._harvester) + finally: + self._harvester = None + def close(self) -> None: if self._acquirer is not None: try: diff --git a/tests/cameras/backends/conftest.py b/tests/cameras/backends/conftest.py index 57a049a..2d965a2 100644 --- a/tests/cameras/backends/conftest.py +++ b/tests/cameras/backends/conftest.py @@ -2,6 +2,7 @@ from __future__ import annotations import importlib +import logging import os from dataclasses import dataclass from typing import Any @@ -9,6 +10,8 @@ import numpy as np import pytest +logger = logging.getLogger(__name__) + # ----------------------------- # Dependency detection helpers @@ -720,21 +723,20 @@ class FakeHarvester: Inventory-driven so tests can control enumeration. """ - def __init__(self, inventory: list[dict[str, Any]] | None = None): + def __init__(self, inventory: list[dict[str, Any]] | None = None, *, fail_add_file_for: set[str] | None = None): self._files: list[str] = [] self._inventory: list[dict[str, Any]] = list(inventory or []) self.device_info_list: list[Any] = [] + # NEW: failure control + self._fail_add_file_for = set(fail_add_file_for or []) + # Call tracing self.add_file_calls: list[str] = [] self.update_calls = 0 self.reset_calls = 0 self.create_calls: list[Any] = [] - def add_file(self, file_path: str): - self._files.append(str(file_path)) - self.add_file_calls.append(str(file_path)) - def update(self): self.update_calls += 1 # If not provided, default to a single fake device @@ -788,6 +790,16 @@ def create(self, selector=None, index: int | None = None, *args, **kwargs): def create_image_acquirer(self, *args, **kwargs): return self.create(*args, **kwargs) + def add_file(self, file_path: str): + p = str(file_path) + self.add_file_calls.append(p) + + # NEW: fail deterministically if requested + if p in self._fail_add_file_for: + raise RuntimeError(f"Simulated CTI load failure for: {p}") + + self._files.append(p) + # ----------------------------------------------------------------------------- # GentL fixtures: inventory, patching, settings factory @@ -817,34 +829,43 @@ def gentl_inventory(): @pytest.fixture() -def fake_harvester_factory(gentl_inventory): +def fake_harvester_factory(gentl_inventory, gentl_fail_add_file_for): """ - Factory that returns a FakeHarvester bound to the current gentl_inventory. - Allows tests to mutate gentl_inventory before calling backend.open(). + Factory that returns a FakeHarvester bound to the current gentl_inventory and + gentl_fail_add_file_for. Tests can mutate both before calling backend.open(). """ def _make(): - return FakeHarvester(inventory=gentl_inventory) + return FakeHarvester(inventory=gentl_inventory, fail_add_file_for=gentl_fail_add_file_for) return _make @pytest.fixture() -def patch_gentl_sdk(monkeypatch, fake_harvester_factory, tmp_path): +def gentl_fail_add_file_for(): + """ + Mutable set of CTI file paths that FakeHarvester.add_file should fail on. + Tests can add/remove paths to simulate partial/complete CTI load failures. + """ + return set() + + +@pytest.fixture() +def patch_gentl_sdk(monkeypatch, fake_harvester_factory, gentl_fail_add_file_for, tmp_path): """ Patch dlclivegui.cameras.backends.gentl_backend to use FakeHarvester + Fake timeout. - Also ensure CTI discovery succeeds for classmethods (discover_devices/quick_ping) - by creating a real dummy .cti and exposing it via GENICAM_GENTL64_PATH. + Ensure CTI discovery succeeds for classmethods by creating a real dummy .cti and + exposing it via GENICAM_GENTL64_PATH. """ import dlclivegui.cameras.backends.gentl_backend as gb # Patch Harvester symbol (the backend calls Harvester() directly) monkeypatch.setattr(gb, "Harvester", lambda: fake_harvester_factory(), raising=False) - # Keep your backend timeout contract as-is: it catches HarvesterTimeoutError + # Keep timeout contract monkeypatch.setattr(gb, "HarvesterTimeoutError", FakeGenTLTimeoutException, raising=False) - # Create a real CTI file and advertise it via env var (cross-platform via os.pathsep) + # Create a real CTI file and advertise it via env var cti_file = tmp_path / "dummy.cti" if not cti_file.exists(): cti_file.write_text("fake", encoding="utf-8") @@ -852,6 +873,9 @@ def patch_gentl_sdk(monkeypatch, fake_harvester_factory, tmp_path): monkeypatch.setenv("GENICAM_GENTL64_PATH", str(tmp_path)) monkeypatch.delenv("GENICAM_GENTL32_PATH", raising=False) + # OPTIONAL: expose failure control so tests can do gb.fail_add_file_for.add(...) + gb.fail_add_file_for = gentl_fail_add_file_for + return gb @@ -877,10 +901,22 @@ def _make( cti = tmp_path / "dummy.cti" if not cti.exists(): cti.write_text("fake", encoding="utf-8") + props = properties if isinstance(properties, dict) else {} props.setdefault("gentl", {}) - props["gentl"] = dict(props["gentl"]) - props["gentl"].setdefault("cti_file", str(cti)) + props["gentl"] = dict(props["gentl"]) # copy to avoid mutating caller dict unexpectedly + + ns = props["gentl"] + + # Detect whether CTIs were explicitly provided in *either* namespace or legacy keys + explicit_ns = bool(ns.get("cti_file") or ns.get("cti_files")) + explicit_legacy = bool(props.get("cti_file") or props.get("cti_files")) + + # Only inject a default dummy.cti if nothing explicit was provided + if not explicit_ns and not explicit_legacy: + logger.debug("No CTI file(s) explicitly provided in settings; injecting dummy CTI for gentl tests.") + ns.setdefault("cti_file", str(cti)) + return CameraSettings( name=name, index=index, diff --git a/tests/cameras/backends/test_gentl_backend.py b/tests/cameras/backends/test_gentl_backend.py index 8e9a40f..8d4094d 100644 --- a/tests/cameras/backends/test_gentl_backend.py +++ b/tests/cameras/backends/test_gentl_backend.py @@ -14,6 +14,31 @@ discover_cti_files, ) +# ---------------------------------------------------------------------- +# Helper functions +# ---------------------------------------------------------------------- + + +@pytest.fixture +def isolate_gentl_env(monkeypatch): + monkeypatch.delenv("GENICAM_GENTL64_PATH", raising=False) + monkeypatch.delenv("GENICAM_GENTL32_PATH", raising=False) + yield + + +def _force_only_these_ctis(settings, ctis: list[str]) -> None: + # Ensure namespace exists + props = settings.properties + props.setdefault("gentl", {}) + ns = props["gentl"] + + # Make sure no default single-cti sneaks in (dummy.cti) + ns.pop("cti_file", None) + props.pop("cti_file", None) + + # Explicit list + ns["cti_files"] = ctis + # --------------------------------------------------------------------- # Core lifecycle + strict transaction model @@ -582,7 +607,7 @@ def test_discover_glob_patterns(tmp_path): assert pattern in diag.glob_patterns -def test_discover_env_var_directory(monkeypatch, tmp_path): +def test_discover_env_var_directory(monkeypatch, tmp_path, isolate_gentl_env): _make_cti(tmp_path, "Env.cti") monkeypatch.setenv("GENICAM_GENTL64_PATH", str(tmp_path)) @@ -593,18 +618,19 @@ def test_discover_env_var_directory(monkeypatch, tmp_path): assert "GENICAM_GENTL64_PATH" in diag.env_vars_used -def test_discover_env_var_direct_file(monkeypatch, tmp_path): +def test_discover_env_var_direct_file(monkeypatch, tmp_path, isolate_gentl_env): cti = _make_cti(tmp_path, "Direct.cti") monkeypatch.setenv("GENICAM_GENTL64_PATH", str(cti)) candidates, diag = discover_cti_files(include_env=True, must_exist=True) - assert candidates == [str(cti.resolve())] or Path(candidates[0]).name == "Direct.cti" + assert len(candidates) == 1 + assert Path(candidates[0]).name == "Direct.cti" assert diag.env_paths_expanded # should include the raw env entry -def test_discover_env_var_multiple_entries(monkeypatch, tmp_path): +def test_discover_env_var_multiple_entries(monkeypatch, tmp_path, isolate_gentl_env): d1 = tmp_path / "d1" d2 = tmp_path / "d2" d1.mkdir() @@ -622,7 +648,7 @@ def test_discover_env_var_multiple_entries(monkeypatch, tmp_path): assert names == ["A.cti", "B.cti"] -def test_discover_deduplicates_same_file_from_multiple_sources(monkeypatch, tmp_path): +def test_discover_deduplicates_same_file_from_multiple_sources(monkeypatch, tmp_path, isolate_gentl_env): cti = _make_cti(tmp_path, "Dup.cti") # Discover it twice: explicit + env dir @@ -664,3 +690,81 @@ def test_choose_cti_files_newest_policy(tmp_path): selected = choose_cti_files([str(old), str(new)], policy=GenTLDiscoveryPolicy.NEWEST, max_files=1) assert len(selected) == 1 assert Path(selected[0]).name == "New.cti" + + +def test_open_persists_cti_load_diagnostics_all_success(patch_gentl_sdk, gentl_settings_factory, tmp_path): + gb = patch_gentl_sdk + + c1 = _make_cti(tmp_path, "A.cti") + c2 = _make_cti(tmp_path, "B.cti") + + # Provide multiple CTIs (how your backend reads these may vary) + settings = gentl_settings_factory(properties={"gentl": {"cti_files": [str(c1), str(c2)]}}) + _force_only_these_ctis(settings, [str(c1), str(c2)]) + be = gb.GenTLCameraBackend(settings) + + be.open() + + ns = settings.properties["gentl"] + assert ns["cti_files"] == [str(c1), str(c2)] + assert ns["cti_files_loaded"] == [str(c1), str(c2)] + # If dict: + assert ns["cti_files_failed"] == [] + + be.close() + + +def test_open_persists_cti_load_diagnostics_partial_failure(patch_gentl_sdk, gentl_settings_factory, tmp_path): + gb = patch_gentl_sdk + + ok = _make_cti(tmp_path, "OK.cti") + bad = _make_cti(tmp_path, "BAD.cti") + + gb.fail_add_file_for.clear() + gb.fail_add_file_for.add(str(bad)) + + settings = gentl_settings_factory(properties={"gentl": {"cti_files": [str(ok), str(bad)]}}) + _force_only_these_ctis(settings, [str(ok), str(bad)]) + + be = gb.GenTLCameraBackend(settings) + be.open() + + ns = settings.properties["gentl"] + assert ns["cti_files"] == [str(ok), str(bad)] + assert ns["cti_files_loaded"] == [str(ok)] + + failed = ns["cti_files_failed"] + assert isinstance(failed, list) + assert len(failed) == 1 + assert failed[0]["cti"] == str(bad) + assert isinstance(failed[0]["error"], str) and failed[0]["error"] + + be.close() + + +def test_open_persists_cti_load_diagnostics_complete_failure(patch_gentl_sdk, gentl_settings_factory, tmp_path): + gb = patch_gentl_sdk + + b1 = _make_cti(tmp_path, "B1.cti") + b2 = _make_cti(tmp_path, "B2.cti") + + gb.fail_add_file_for.clear() + gb.fail_add_file_for.update({str(b1), str(b2)}) + + settings = gentl_settings_factory(properties={"gentl": {"cti_files": [str(b1), str(b2)]}}) + _force_only_these_ctis(settings, [str(b1), str(b2)]) + be = gb.GenTLCameraBackend(settings) + + with pytest.raises(RuntimeError): + be.open() + + # Keys should still be persisted for debugging even though open failed. + ns = settings.properties.get("gentl", {}) + assert ns.get("cti_files") == [str(b1), str(b2)] + assert ns.get("cti_files_loaded") == [] + + failed = ns.get("cti_files_failed") + assert isinstance(failed, list) + assert sorted(d["cti"] for d in failed) == sorted([str(b1), str(b2)]) + for d in failed: + assert isinstance(d.get("error"), str) and d["error"] From a42d4961b43024e77f3d3ff4dbdf915617648b42 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 19 Feb 2026 17:23:33 +0100 Subject: [PATCH 4/4] Apply suggestions from code review Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- dlclivegui/cameras/backends/gentl_backend.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/dlclivegui/cameras/backends/gentl_backend.py b/dlclivegui/cameras/backends/gentl_backend.py index 405a7ab..ba08b6a 100644 --- a/dlclivegui/cameras/backends/gentl_backend.py +++ b/dlclivegui/cameras/backends/gentl_backend.py @@ -377,8 +377,7 @@ def open(self) -> None: "No GenTL producer (.cti) could be loaded.\n\n" f"Resolved CTIs: {cti_files}\n" f"Failures: {failed}\n" - "Fix: remove/repair incompatible producers or" - " set properties.gentl.cti_file to a known working producer." + "Fix: remove/repair incompatible producers or set properties.gentl.cti_file to a known working producer." ) # Update device list after loading producers