Coverage for openhcs/microscopes/microscope_base.py: 65.0%
223 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 02:09 +0000
1"""
2Microscope base implementations for openhcs.
4This module provides the base implementations for microscope-specific functionality,
5including filename parsing and metadata handling.
6"""
8import logging
9import os
10from abc import ABC, abstractmethod
11from pathlib import Path
12from typing import Any, Dict, List, Optional, Tuple, Union, Type
14# Import constants
15from openhcs.constants.constants import Backend
16# Import generic metaclass infrastructure
17from openhcs.core.auto_register_meta import (
18 AutoRegisterMeta,
19 SecondaryRegistry,
20 extract_key_from_handler_suffix,
21 PRIMARY_KEY
22)
23# PatternDiscoveryEngine imported locally to avoid circular imports
24from openhcs.io.filemanager import FileManager
25# Import interfaces from the base interfaces module
26from openhcs.microscopes.microscope_interfaces import (FilenameParser,
27 MetadataHandler)
29logger = logging.getLogger(__name__)
31# Dictionary to store registered metadata handlers for auto-detection
32# This will be auto-wrapped with SecondaryRegistryDict by the metaclass
33METADATA_HANDLERS = {}
36def register_metadata_handler(handler_class, metadata_handler_class):
37 """
38 Register a metadata handler for a microscope handler class.
40 This function is called when _metadata_handler_class is set after class definition.
41 """
42 microscope_type = getattr(handler_class, '_microscope_type', None)
43 if microscope_type: 43 ↛ 47line 43 didn't jump to line 47 because the condition on line 43 was always true
44 METADATA_HANDLERS[microscope_type] = metadata_handler_class
45 logger.debug(f"Registered metadata handler {metadata_handler_class.__name__} for '{microscope_type}'")
46 else:
47 logger.warning(f"Could not register metadata handler for {handler_class.__name__} - no microscope type found")
52class MicroscopeHandler(ABC, metaclass=AutoRegisterMeta):
53 """
54 Composed class for handling microscope-specific functionality.
56 Registry auto-created and stored as MicroscopeHandler.__registry__.
57 Subclasses auto-register by setting _microscope_type class attribute.
58 Secondary registry METADATA_HANDLERS populated via _metadata_handler_class.
59 """
60 __registry_key__ = '_microscope_type'
61 __key_extractor__ = extract_key_from_handler_suffix
62 __skip_if_no_key__ = False
63 __secondary_registries__ = [
64 SecondaryRegistry(
65 registry_dict=METADATA_HANDLERS,
66 key_source=PRIMARY_KEY,
67 attr_name='_metadata_handler_class'
68 )
69 ]
71 DEFAULT_MICROSCOPE = 'auto'
72 _handlers_cache = None
74 # Optional class attribute for explicit metadata handler registration
75 _metadata_handler_class: Optional[Type[MetadataHandler]] = None
77 def __init__(self, parser: Optional[FilenameParser],
78 metadata_handler: MetadataHandler):
79 """
80 Initialize the microscope handler.
82 Args:
83 parser: Parser for microscopy filenames. Can be None for virtual backends
84 that don't need workspace preparation or pattern discovery.
85 metadata_handler: Handler for microscope metadata.
86 """
87 self.parser = parser
88 self.metadata_handler = metadata_handler
89 self.plate_folder: Optional[Path] = None # Store workspace path if needed by methods
91 # Pattern discovery engine will be created on demand with the provided filemanager
93 @property
94 @abstractmethod
95 def root_dir(self) -> str:
96 """
97 Root directory where virtual workspace preparation starts.
99 This defines:
100 1. The starting point for virtual workspace operations (flattening, remapping)
101 2. The subdirectory key used when saving virtual workspace metadata
103 Examples:
104 - ImageXpress: "." (plate root - TimePoint/ZStep folders are flattened from plate root)
105 - OperaPhenix: "Images" (field remapping applied to Images/ subdirectory)
106 - OpenHCS: Determined from metadata (e.g., "zarr", "images")
107 """
108 pass
110 @property
111 @abstractmethod
112 def microscope_type(self) -> str:
113 """Microscope type identifier (for interface enforcement only)."""
114 pass
116 @property
117 @abstractmethod
118 def metadata_handler_class(self) -> Type[MetadataHandler]:
119 """Metadata handler class (for interface enforcement only)."""
120 pass
122 @property
123 @abstractmethod
124 def compatible_backends(self) -> List[Backend]:
125 """
126 List of storage backends this microscope handler is compatible with, in priority order.
128 Must be explicitly declared by each handler implementation.
129 The first backend in the list is the preferred/highest priority backend.
130 The compiler will use the first backend for initial step materialization.
132 Common patterns:
133 - [Backend.DISK] - Basic handlers (ImageXpress, Opera Phenix)
134 - [Backend.ZARR, Backend.DISK] - Advanced handlers (OpenHCS: zarr preferred, disk fallback)
135 - [Backend.OMERO_LOCAL] - Virtual backends (OMERO: single required backend)
137 Returns:
138 List of Backend enum values this handler can work with, in priority order
139 """
140 pass
142 def get_required_backend(self) -> Optional['MaterializationBackend']:
143 """
144 Get the required materialization backend if this microscope has only one compatible backend.
146 For microscopes with a single compatible backend (e.g., OMERO with OMERO_LOCAL),
147 this returns the required backend for auto-correction. For microscopes with multiple
148 compatible backends, returns None (user must choose explicitly).
150 Returns:
151 MaterializationBackend if microscope requires a specific backend, None otherwise
152 """
153 from openhcs.core.config import MaterializationBackend
155 if len(self.compatible_backends) == 1:
156 backend_value = self.compatible_backends[0].value
157 # Convert Backend enum value to MaterializationBackend enum
158 try:
159 return MaterializationBackend(backend_value)
160 except ValueError:
161 # Backend not in MaterializationBackend (e.g., MEMORY, VIRTUAL_WORKSPACE)
162 return None
163 return None
165 def get_available_backends(self, plate_path: Union[str, Path]) -> List[Backend]:
166 """
167 Get available storage backends for this specific plate.
169 Default implementation returns all compatible backends.
170 Override this method only if you need to check actual disk state
171 (like OpenHCS which reads from metadata).
173 Args:
174 plate_path: Path to the plate folder
176 Returns:
177 List of Backend enums that are available for this plate.
178 """
179 return self.compatible_backends
181 def get_primary_backend(self, plate_path: Union[str, Path], filemanager: 'FileManager') -> str:
182 """
183 Get the primary backend name for this plate.
185 Checks FileManager registry first for registered backends (like virtual_workspace),
186 then falls back to compatible backends.
188 Override this method only if you need custom backend selection logic
189 (like OpenHCS which reads from metadata).
191 Args:
192 plate_path: Path to the plate folder (or subdirectory)
193 filemanager: FileManager instance to check for registered backends
195 Returns:
196 Backend name string (e.g., 'disk', 'zarr', 'virtual_workspace')
197 """
198 # Check if virtual workspace backend is registered in FileManager
199 # This takes priority over compatible backends
200 if Backend.VIRTUAL_WORKSPACE.value in filemanager.registry: 200 ↛ 205line 200 didn't jump to line 205 because the condition on line 200 was always true
201 logger.info(f"✅ Using backend '{Backend.VIRTUAL_WORKSPACE.value}' from FileManager registry")
202 return Backend.VIRTUAL_WORKSPACE.value
204 # Fall back to compatible backends
205 available_backends = self.get_available_backends(plate_path)
206 if not available_backends:
207 raise RuntimeError(f"No available backends for {self.microscope_type} microscope at {plate_path}")
208 logger.info(f"⚠️ Using backend '{available_backends[0].value}' from compatible backends (virtual workspace not registered)")
209 return available_backends[0].value
211 def _register_virtual_workspace_backend(self, plate_path: Union[str, Path], filemanager: FileManager) -> None:
212 """
213 Register virtual workspace backend if not already registered.
215 Centralized registration logic to avoid duplication across handlers.
217 Args:
218 plate_path: Path to plate directory
219 filemanager: FileManager instance
220 """
221 from openhcs.io.virtual_workspace import VirtualWorkspaceBackend
222 from openhcs.constants.constants import Backend
224 if Backend.VIRTUAL_WORKSPACE.value not in filemanager.registry:
225 backend = VirtualWorkspaceBackend(plate_root=Path(plate_path))
226 filemanager.registry[Backend.VIRTUAL_WORKSPACE.value] = backend
227 logger.info(f"Registered virtual workspace backend for {plate_path}")
229 def initialize_workspace(self, plate_path: Path, filemanager: FileManager) -> Path:
230 """
231 Initialize plate by creating virtual mapping in metadata.
233 No physical workspace directory is created - mapping is purely metadata-based.
234 All paths are relative to plate_path.
236 Args:
237 plate_path: Path to plate directory
238 filemanager: FileManager instance
240 Returns:
241 Path to image directory (determined from plate structure)
242 """
243 plate_path = Path(plate_path)
245 # Set plate_folder for this handler
246 self.plate_folder = plate_path
248 # Call microscope-specific virtual mapping builder
249 # This builds the plate-relative mapping dict and saves to metadata
250 self._build_virtual_mapping(plate_path, filemanager)
252 # Register virtual workspace backend
253 self._register_virtual_workspace_backend(plate_path, filemanager)
255 # Return image directory using post_workspace
256 # skip_preparation=True because _build_virtual_mapping() already ran
257 return self.post_workspace(plate_path, filemanager, skip_preparation=True)
259 def post_workspace(self, plate_path: Union[str, Path], filemanager: FileManager, skip_preparation: bool = False) -> Path:
260 """
261 Apply post-workspace processing using virtual mapping.
263 All operations use plate_path - no workspace directory concept.
265 Args:
266 plate_path: Path to plate directory
267 filemanager: FileManager instance
268 skip_preparation: Skip microscope-specific preparation (default: False)
270 Returns:
271 Path to image directory
272 """
273 # Ensure plate_path is a Path object
274 if isinstance(plate_path, str): 274 ↛ 275line 274 didn't jump to line 275 because the condition on line 274 was never true
275 plate_path = Path(plate_path)
277 # NO existence check - virtual workspaces are metadata-only
279 # Apply microscope-specific preparation logic
280 # _build_virtual_mapping() returns the correct image directory for each microscope:
281 # - ImageXpress: plate_path (images are at plate root with TimePoint_X/ prefix in mapping)
282 # - OperaPhenix: plate_path/Images (images are in Images/ subdirectory)
283 if skip_preparation: 283 ↛ 311line 283 didn't jump to line 311 because the condition on line 283 was always true
284 logger.info("📁 SKIPPING PREPARATION: Virtual mapping already built")
285 # When skipping, we need to determine image_dir from metadata
286 # Read metadata to get the subdirectory key
287 from openhcs.microscopes.openhcs import OpenHCSMetadataHandler
288 from openhcs.io.exceptions import MetadataNotFoundError
289 from openhcs.io.metadata_writer import resolve_subdirectory_path
291 openhcs_metadata_handler = OpenHCSMetadataHandler(filemanager)
292 metadata = openhcs_metadata_handler._load_metadata_dict(plate_path)
293 subdirs = metadata.get("subdirectories", {})
295 # Find the subdirectory with workspace_mapping (should be "." or "Images")
296 subdir_with_mapping = next(
297 (name for name, data in subdirs.items() if "workspace_mapping" in data),
298 None
299 )
301 # Fail if no workspace_mapping found
302 if subdir_with_mapping is None: 302 ↛ 303line 302 didn't jump to line 303 because the condition on line 302 was never true
303 raise MetadataNotFoundError(
304 f"skip_preparation=True but no workspace_mapping found in metadata for {plate_path}. "
305 "Virtual workspace must be prepared before skipping."
306 )
308 # Convert subdirectory name to path (handles "." -> plate_path)
309 image_dir = resolve_subdirectory_path(subdir_with_mapping, plate_path)
310 else:
311 logger.info("🔄 APPLYING PREPARATION: Building virtual mapping")
312 # _build_virtual_mapping() returns the image directory
313 image_dir = self._build_virtual_mapping(plate_path, filemanager)
315 # Determine backend - check if virtual workspace backend is registered
316 if Backend.VIRTUAL_WORKSPACE.value in filemanager.registry: 316 ↛ 319line 316 didn't jump to line 319 because the condition on line 316 was always true
317 backend_type = Backend.VIRTUAL_WORKSPACE.value
318 else:
319 backend_type = Backend.DISK.value
321 # Ensure parser is provided
322 parser = self.parser
324 # Get all image files in the directory
325 image_files = filemanager.list_image_files(image_dir, backend_type)
327 # Map original filenames to reconstructed filenames
328 rename_map = {}
330 for file_path in image_files:
331 # FileManager should return strings, but handle Path objects too
332 if isinstance(file_path, str): 332 ↛ 334line 332 didn't jump to line 334 because the condition on line 332 was always true
333 original_name = os.path.basename(file_path)
334 elif isinstance(file_path, Path):
335 original_name = file_path.name
336 else:
337 # Skip any unexpected types
338 logger.warning("Unexpected file path type: %s", type(file_path).__name__)
339 continue
341 # Parse the filename components
342 metadata = parser.parse_filename(original_name)
343 if not metadata: 343 ↛ 344line 343 didn't jump to line 344 because the condition on line 343 was never true
344 logger.warning("Could not parse filename: %s", original_name)
345 continue
347 # Validate required components
348 if metadata['site'] is None: 348 ↛ 349line 348 didn't jump to line 349 because the condition on line 348 was never true
349 logger.warning("Missing 'site' component in filename: %s", original_name)
350 continue
352 if metadata['channel'] is None: 352 ↛ 353line 352 didn't jump to line 353 because the condition on line 352 was never true
353 logger.warning("Missing 'channel' component in filename: %s", original_name)
354 continue
356 # z_index is optional - default to 1 if not present
357 site = metadata['site']
358 channel = metadata['channel']
359 z_index = metadata['z_index'] if metadata['z_index'] is not None else 1
361 # Log the components for debugging
362 logger.debug(
363 "Parsed components for %s: site=%s, channel=%s, z_index=%s",
364 original_name, site, channel, z_index
365 )
367 # Reconstruct the filename with proper padding
368 metadata['site'] = site
369 metadata['channel'] = channel
370 metadata['z_index'] = z_index
371 new_name = parser.construct_filename(**metadata)
373 # Add to rename map if different
374 if original_name != new_name: 374 ↛ 375line 374 didn't jump to line 375 because the condition on line 374 was never true
375 rename_map[original_name] = new_name
377 # Perform the renaming
378 for original_name, new_name in rename_map.items(): 378 ↛ 380line 378 didn't jump to line 380 because the loop on line 378 never started
379 # Create paths for the source and destination
380 if isinstance(image_dir, str):
381 original_path = os.path.join(image_dir, original_name)
382 new_path = os.path.join(image_dir, new_name)
383 else: # Path object
384 original_path = image_dir / original_name
385 new_path = image_dir / new_name
387 try:
388 # Ensure the parent directory exists
389 # Clause 245: Workspace operations are disk-only by design
390 # This call is structurally hardcoded to use the "disk" backend
391 parent_dir = os.path.dirname(new_path) if isinstance(new_path, str) else new_path.parent
392 filemanager.ensure_directory(parent_dir, Backend.DISK.value)
394 # Rename the file using move operation
395 # Clause 245: Workspace operations are disk-only by design
396 # This call is structurally hardcoded to use the "disk" backend
397 # Use replace_symlinks=True to allow overwriting existing symlinks
398 filemanager.move(original_path, new_path, Backend.DISK.value, replace_symlinks=True)
399 logger.debug("Renamed %s to %s", original_path, new_path)
400 except (OSError, FileNotFoundError) as e:
401 logger.error("Filesystem error renaming %s to %s: %s", original_path, new_path, e)
402 except TypeError as e:
403 logger.error("Type error renaming %s to %s: %s", original_path, new_path, e)
404 except Exception as e:
405 logger.error("Unexpected error renaming %s to %s: %s", original_path, new_path, e)
407 return image_dir
409 def _build_virtual_mapping(self, plate_path: Path, filemanager: FileManager) -> Path:
410 """
411 Build microscope-specific virtual workspace mapping.
413 This method creates a plate-relative mapping dict and saves it to metadata.
414 All paths in the mapping are relative to plate_path.
416 Override in subclasses that need virtual workspace mapping (e.g., ImageXpress, Opera Phenix).
417 Handlers that override initialize_workspace() completely (e.g., OMERO, OpenHCS) don't need
418 to implement this method.
420 Args:
421 plate_path: Path to plate directory
422 filemanager: FileManager instance for file operations
424 Returns:
425 Path: Suggested directory for further processing
427 Raises:
428 NotImplementedError: If called on a handler that doesn't support virtual workspace mapping
429 """
430 raise NotImplementedError(
431 f"{self.__class__.__name__} does not implement _build_virtual_mapping(). "
432 f"This method is only needed for handlers that use the base class initialize_workspace(). "
433 f"Handlers that override initialize_workspace() completely (like OMERO, OpenHCS) don't need this."
434 )
437 # Delegate methods to parser
438 def parse_filename(self, filename: str) -> Optional[Dict[str, Any]]:
439 """Delegate to parser."""
440 return self.parser.parse_filename(filename)
442 def construct_filename(self, extension: str = '.tif', **component_values) -> str:
443 """
444 Delegate to parser using pure generic interface.
445 """
446 return self.parser.construct_filename(extension=extension, **component_values)
448 def auto_detect_patterns(self, folder_path: Union[str, Path], filemanager: FileManager, backend: str,
449 extensions=None, group_by=None, variable_components=None, **kwargs):
450 """
451 Delegate to pattern engine.
453 Args:
454 folder_path: Path to the folder (string or Path object)
455 filemanager: FileManager instance for file operations
456 backend: Backend to use for file operations (required)
457 extensions: Optional list of file extensions to include
458 group_by: GroupBy enum to group patterns by (e.g., GroupBy.CHANNEL, GroupBy.Z_INDEX)
459 variable_components: List of components to make variable (e.g., ['site', 'z_index'])
460 **kwargs: Dynamic filter parameters (e.g., well_filter, site_filter, channel_filter)
462 Returns:
463 Dict[str, Any]: Dictionary mapping axis values to patterns
464 """
465 # Ensure folder_path is a valid path
466 if isinstance(folder_path, str): 466 ↛ 468line 466 didn't jump to line 468 because the condition on line 466 was always true
467 folder_path = Path(folder_path)
468 elif not isinstance(folder_path, Path):
469 raise TypeError(f"Expected string or Path object, got {type(folder_path).__name__}")
471 # Ensure the path exists using FileManager abstraction
472 if not filemanager.exists(str(folder_path), backend): 472 ↛ 473line 472 didn't jump to line 473 because the condition on line 472 was never true
473 raise ValueError(f"Folder path does not exist: {folder_path}")
475 # Set default GroupBy if none provided
476 if group_by is None:
477 from openhcs.constants.constants import GroupBy
478 group_by = GroupBy.CHANNEL
480 # Create pattern engine on demand with the provided filemanager
481 from openhcs.formats.pattern.pattern_discovery import PatternDiscoveryEngine
482 pattern_engine = PatternDiscoveryEngine(self.parser, filemanager)
484 # Get patterns from the pattern engine
485 patterns_by_well = pattern_engine.auto_detect_patterns(
486 folder_path,
487 extensions=extensions,
488 group_by=group_by,
489 variable_components=variable_components,
490 backend=backend,
491 **kwargs # Pass through dynamic filter parameters
492 )
494 # 🔒 Clause 74 — Runtime Behavior Variation
495 # Ensure we always return a dictionary, not a generator
496 if not isinstance(patterns_by_well, dict): 496 ↛ 498line 496 didn't jump to line 498 because the condition on line 496 was never true
497 # Convert to dictionary if it's not already one
498 return dict(patterns_by_well)
500 return patterns_by_well
502 def path_list_from_pattern(self, directory: Union[str, Path], pattern, filemanager: FileManager, backend: str, variable_components: Optional[List[str]] = None):
503 """
504 Delegate to pattern engine.
506 Args:
507 directory: Directory to search (string or Path object)
508 pattern: Pattern to match (str for literal filenames)
509 filemanager: FileManager instance for file operations
510 backend: Backend to use for file operations (required)
511 variable_components: List of components that can vary (will be ignored during matching)
513 Returns:
514 List of matching filenames
516 Raises:
517 TypeError: If a string with braces is passed (pattern paths are no longer supported)
518 ValueError: If directory does not exist
519 """
520 # Ensure directory is a valid path using FileManager abstraction
521 if isinstance(directory, str): 521 ↛ 525line 521 didn't jump to line 525 because the condition on line 521 was always true
522 directory_path = Path(directory)
523 if not filemanager.exists(str(directory_path), backend): 523 ↛ 524line 523 didn't jump to line 524 because the condition on line 523 was never true
524 raise ValueError(f"Directory does not exist: {directory}")
525 elif isinstance(directory, Path):
526 directory_path = directory
527 if not filemanager.exists(str(directory_path), backend):
528 raise ValueError(f"Directory does not exist: {directory}")
529 else:
530 raise TypeError(f"Expected string or Path object, got {type(directory).__name__}")
532 # Allow string patterns with braces - they are used for template matching
533 # The pattern engine will handle template expansion to find matching files
535 # Create pattern engine on demand with the provided filemanager
536 from openhcs.formats.pattern.pattern_discovery import PatternDiscoveryEngine
537 pattern_engine = PatternDiscoveryEngine(self.parser, filemanager)
539 # Delegate to the pattern engine
540 return pattern_engine.path_list_from_pattern(directory_path, pattern, backend=backend, variable_components=variable_components)
542 # Delegate metadata handling methods to metadata_handler with context
544 def find_metadata_file(self, plate_path: Union[str, Path]) -> Optional[Path]:
545 """Delegate to metadata handler."""
546 return self.metadata_handler.find_metadata_file(plate_path)
548 def get_grid_dimensions(self, plate_path: Union[str, Path]) -> Tuple[int, int]:
549 """Delegate to metadata handler."""
550 return self.metadata_handler.get_grid_dimensions(plate_path)
552 def get_pixel_size(self, plate_path: Union[str, Path]) -> float:
553 """Delegate to metadata handler."""
554 return self.metadata_handler.get_pixel_size(plate_path)
557# Import handler classes at module level with explicit mapping
558# No aliases or legacy compatibility layers (Clause 77)
560# Factory function
561def create_microscope_handler(microscope_type: str = 'auto',
562 plate_folder: Optional[Union[str, Path]] = None,
563 filemanager: Optional[FileManager] = None,
564 pattern_format: Optional[str] = None,
565 allowed_auto_types: Optional[List[str]] = None) -> MicroscopeHandler:
566 """
567 Factory function to create a microscope handler.
569 This function enforces explicit dependency injection by requiring a FileManager
570 instance to be provided. This ensures that all components requiring file operations
571 receive their dependencies explicitly, eliminating runtime fallbacks and enforcing
572 declarative configuration.
574 Args:
575 microscope_type: 'auto', 'imagexpress', 'opera_phenix', 'openhcs'.
576 plate_folder: Required for 'auto' detection.
577 filemanager: FileManager instance. Must be provided.
578 pattern_format: Name of the pattern format to use.
579 allowed_auto_types: For 'auto' mode, limit detection to these types.
580 'openhcs' is always included and tried first.
582 Returns:
583 An initialized MicroscopeHandler instance.
585 Raises:
586 ValueError: If filemanager is None or if microscope_type cannot be determined.
587 """
588 if filemanager is None: 588 ↛ 589line 588 didn't jump to line 589 because the condition on line 588 was never true
589 raise ValueError(
590 "FileManager must be provided to create_microscope_handler. "
591 "Default fallback has been removed."
592 )
594 logger.info("Using provided FileManager for microscope handler.")
596 # Auto-detect microscope type if needed
597 if microscope_type == 'auto': 597 ↛ 606line 597 didn't jump to line 606 because the condition on line 597 was always true
598 if not plate_folder: 598 ↛ 599line 598 didn't jump to line 599 because the condition on line 598 was never true
599 raise ValueError("plate_folder is required for auto-detection")
601 plate_folder = Path(plate_folder) if isinstance(plate_folder, str) else plate_folder
602 microscope_type = _auto_detect_microscope_type(plate_folder, filemanager, allowed_types=allowed_auto_types)
603 logger.info("Auto-detected microscope type: %s", microscope_type)
605 # Handlers auto-discovered on first access to MICROSCOPE_HANDLERS
606 from openhcs.microscopes.handler_registry_service import get_all_handler_types
608 # Get the appropriate handler class from the registry
609 # No dynamic imports or fallbacks (Clause 77: Rot Intolerance)
610 handler_class = MICROSCOPE_HANDLERS.get(microscope_type.lower())
611 if not handler_class: 611 ↛ 612line 611 didn't jump to line 612 because the condition on line 611 was never true
612 available_types = get_all_handler_types()
613 raise ValueError(
614 f"Unsupported microscope type: {microscope_type}. "
615 f"Available types: {available_types}"
616 )
618 # Create and configure the handler
619 logger.info(f"Creating {handler_class.__name__}")
621 # Create the handler with the parser and metadata handler
622 # The filemanager will be passed to methods that need it
623 handler = handler_class(filemanager, pattern_format=pattern_format)
625 # If the handler is OpenHCSMicroscopeHandler, set its plate_folder attribute.
626 # This is crucial for its dynamic parser loading mechanism.
627 # Use string comparison to avoid circular import
628 if handler.__class__.__name__ == 'OpenHCSMicroscopeHandler':
629 if plate_folder: 629 ↛ 635line 629 didn't jump to line 635 because the condition on line 629 was always true
630 handler.plate_folder = Path(plate_folder) if isinstance(plate_folder, str) else plate_folder
631 logger.info(f"Set plate_folder for OpenHCSMicroscopeHandler: {handler.plate_folder}")
632 else:
633 # This case should ideally not happen if auto-detection or explicit type setting
634 # implies a plate_folder is known.
635 logger.warning("OpenHCSMicroscopeHandler created without an initial plate_folder. "
636 "Parser will load upon first relevant method call with a path e.g. post_workspace.")
638 return handler
641def validate_backend_compatibility(handler: MicroscopeHandler, backend: Backend) -> bool:
642 """
643 Validate that a microscope handler supports a given storage backend.
645 Args:
646 handler: MicroscopeHandler instance to check
647 backend: Backend to validate compatibility with
649 Returns:
650 bool: True if the handler supports the backend, False otherwise
652 Example:
653 >>> handler = ImageXpressHandler(filemanager)
654 >>> validate_backend_compatibility(handler, Backend.ZARR)
655 False
656 >>> validate_backend_compatibility(handler, Backend.DISK)
657 True
658 """
659 return backend in handler.supported_backends
662def _try_metadata_detection(handler_class, filemanager: FileManager, plate_folder: Path) -> Optional[Path]:
663 """
664 Try metadata detection with a handler, normalizing return types.
666 Args:
667 handler_class: MetadataHandler class to try
668 filemanager: FileManager instance
669 plate_folder: Path to plate directory
671 Returns:
672 Path if metadata found, None if metadata not found
674 Raises:
675 Any exception from the handler (fail-loud behavior)
676 """
677 handler = handler_class(filemanager)
678 result = handler.find_metadata_file(plate_folder)
680 # Normalize return type: convert any truthy result to Path, falsy to None
681 return Path(result) if result else None
684def _auto_detect_microscope_type(plate_folder: Path, filemanager: FileManager,
685 allowed_types: Optional[List[str]] = None) -> str:
686 """
687 Auto-detect microscope type using registry iteration.
689 Args:
690 plate_folder: Path to plate directory
691 filemanager: FileManager instance
692 allowed_types: Optional list of microscope types to try.
693 If None, tries all registered types.
694 'openhcs' is always included and tried first.
696 Returns:
697 Detected microscope type string
699 Raises:
700 ValueError: If microscope type cannot be determined
701 MetadataNotFoundError: If metadata files are missing
702 Any other exception from metadata handlers (fail-loud)
703 """
704 # METADATA_HANDLERS is a SecondaryRegistryDict that auto-triggers discovery
705 from openhcs.io.exceptions import MetadataNotFoundError
707 # Build detection order: openhcsdata first, then filtered/ordered list
708 detection_order = ['openhcsdata'] # Always first, always included (correct registration name)
710 if allowed_types is None: 710 ↛ 715line 710 didn't jump to line 715 because the condition on line 710 was always true
711 # Use all registered handlers in registration order
712 detection_order.extend([name for name in METADATA_HANDLERS.keys() if name != 'openhcsdata'])
713 else:
714 # Use filtered list, but ensure openhcsdata is first
715 filtered_types = [name for name in allowed_types if name != 'openhcsdata' and name in METADATA_HANDLERS]
716 detection_order.extend(filtered_types)
718 # Try detection in order - only catch expected "not found" exceptions
719 for handler_name in detection_order: 719 ↛ 735line 719 didn't jump to line 735 because the loop on line 719 didn't complete
720 handler_class = METADATA_HANDLERS.get(handler_name)
721 if not handler_class: 721 ↛ 722line 721 didn't jump to line 722 because the condition on line 721 was never true
722 continue
724 try:
725 result = _try_metadata_detection(handler_class, filemanager, plate_folder)
726 if result:
727 logger.info(f"Auto-detected {handler_name} microscope type")
728 return handler_name
729 except (FileNotFoundError, MetadataNotFoundError):
730 # Expected - this handler's metadata not found, try next
731 logger.debug(f"{handler_name} metadata not found in {plate_folder}")
732 continue
734 # No handler succeeded - provide detailed error message
735 available_types = list(METADATA_HANDLERS.keys())
736 msg = (f"Could not auto-detect microscope type in {plate_folder}. "
737 f"Tried: {detection_order}. "
738 f"Available types: {available_types}. "
739 f"Ensure metadata files are present for supported formats.")
740 logger.error(msg)
741 raise ValueError(msg)
744# ============================================================================
745# Registry Export
746# ============================================================================
747# Auto-created registry from MicroscopeHandler base class
748MICROSCOPE_HANDLERS = MicroscopeHandler.__registry__