Coverage for openhcs/microscopes/microscope_base.py: 53.3%
248 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-10-01 18:33 +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, ABCMeta, 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# PatternDiscoveryEngine imported locally to avoid circular imports
17from openhcs.io.filemanager import FileManager
18# Import interfaces from the base interfaces module
19from openhcs.microscopes.microscope_interfaces import (FilenameParser,
20 MetadataHandler)
22logger = logging.getLogger(__name__)
24# Dictionary to store registered microscope handlers
25MICROSCOPE_HANDLERS = {}
27# Dictionary to store registered metadata handlers for auto-detection
28METADATA_HANDLERS = {}
31class MicroscopeHandlerMeta(ABCMeta):
32 """Metaclass for automatic registration of microscope handlers."""
34 def __new__(cls, name, bases, attrs):
35 new_class = super().__new__(cls, name, bases, attrs)
37 # Only register concrete handler classes (not the abstract base class)
38 if bases and not getattr(new_class, '__abstractmethods__', None):
39 # Use explicit microscope type if provided, otherwise extract from class name
40 microscope_type = getattr(new_class, '_microscope_type', None)
41 if not microscope_type: 41 ↛ 42line 41 didn't jump to line 42 because the condition on line 41 was never true
42 if name.endswith('Handler'):
43 microscope_type = name[:-7].lower() # ImageXpressHandler -> imagexpress
44 else:
45 microscope_type = name.lower()
47 # Auto-register in MICROSCOPE_HANDLERS
48 MICROSCOPE_HANDLERS[microscope_type] = new_class
50 # Store the microscope type as the standard class attribute
51 new_class._microscope_type = microscope_type
53 # Auto-register metadata handler if the class has one
54 metadata_handler_class = getattr(new_class, '_metadata_handler_class', None)
55 if metadata_handler_class: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true
56 METADATA_HANDLERS[microscope_type] = metadata_handler_class
58 logger.debug(f"Auto-registered {name} as '{microscope_type}'")
60 return new_class
63def register_metadata_handler(handler_class, metadata_handler_class):
64 """
65 Register a metadata handler for a microscope handler class.
67 This function is called when _metadata_handler_class is set after class definition.
68 """
69 microscope_type = getattr(handler_class, '_microscope_type', None)
70 if microscope_type: 70 ↛ 74line 70 didn't jump to line 74 because the condition on line 70 was always true
71 METADATA_HANDLERS[microscope_type] = metadata_handler_class
72 logger.debug(f"Registered metadata handler {metadata_handler_class.__name__} for '{microscope_type}'")
73 else:
74 logger.warning(f"Could not register metadata handler for {handler_class.__name__} - no microscope type found")
79class MicroscopeHandler(ABC, metaclass=MicroscopeHandlerMeta):
80 """Composed class for handling microscope-specific functionality."""
82 DEFAULT_MICROSCOPE = 'auto'
83 _handlers_cache = None
85 # Optional class attribute for explicit metadata handler registration
86 _metadata_handler_class: Optional[Type[MetadataHandler]] = None
88 def __init__(self, parser: FilenameParser,
89 metadata_handler: MetadataHandler):
90 """
91 Initialize the microscope handler.
93 Args:
94 parser: Parser for microscopy filenames.
95 metadata_handler: Handler for microscope metadata.
96 """
97 self.parser = parser
98 self.metadata_handler = metadata_handler
99 self.plate_folder: Optional[Path] = None # Store workspace path if needed by methods
101 # Pattern discovery engine will be created on demand with the provided filemanager
103 @property
104 @abstractmethod
105 def common_dirs(self) -> List[str]:
106 """
107 Canonical subdirectory names where image data may reside.
108 Example: ['Images', 'TimePoint', 'Data']
109 """
110 pass
112 @property
113 @abstractmethod
114 def microscope_type(self) -> str:
115 """Microscope type identifier (for interface enforcement only)."""
116 pass
118 @property
119 @abstractmethod
120 def metadata_handler_class(self) -> Type[MetadataHandler]:
121 """Metadata handler class (for interface enforcement only)."""
122 pass
124 @property
125 @abstractmethod
126 def compatible_backends(self) -> List[Backend]:
127 """
128 List of storage backends this microscope handler is compatible with, in priority order.
130 Must be explicitly declared by each handler implementation.
131 The first backend in the list is the preferred/highest priority backend.
132 The compiler will use the first backend for initial step materialization.
134 Common patterns:
135 - [Backend.DISK] - Basic handlers (ImageXpress, Opera Phenix)
136 - [Backend.ZARR, Backend.DISK] - Advanced handlers (OpenHCS: zarr preferred, disk fallback)
138 Returns:
139 List of Backend enum values this handler can work with, in priority order
140 """
141 pass
143 def get_available_backends(self, plate_path: Union[str, Path]) -> List[Backend]:
144 """
145 Get available storage backends for this specific plate.
147 Default implementation returns all compatible backends.
148 Override this method only if you need to check actual disk state
149 (like OpenHCS which reads from metadata).
151 Args:
152 plate_path: Path to the plate folder
154 Returns:
155 List of Backend enums that are available for this plate.
156 """
157 return self.compatible_backends
159 def get_primary_backend(self, plate_path: Union[str, Path]) -> str:
160 """
161 Get the primary backend name for this plate.
163 Default implementation returns the first compatible backend.
164 Override this method only if you need custom backend selection logic
165 (like OpenHCS which reads from metadata).
167 Args:
168 plate_path: Path to the plate folder
170 Returns:
171 Backend name string (e.g., 'disk', 'zarr')
172 """
173 available_backends = self.get_available_backends(plate_path)
174 if not available_backends: 174 ↛ 175line 174 didn't jump to line 175 because the condition on line 174 was never true
175 raise RuntimeError(f"No available backends for {self.microscope_type} microscope at {plate_path}")
176 return available_backends[0].value
178 def initialize_workspace(self, plate_path: Path, workspace_path: Optional[Path], filemanager: FileManager) -> Path:
179 """
180 Default workspace initialization: create workspace in plate folder and mirror with symlinks.
182 Most microscope handlers need workspace mirroring. Override this method only if different behavior is needed.
184 Args:
185 plate_path: Path to the original plate directory
186 workspace_path: Optional workspace path (creates default in plate folder if None)
187 filemanager: FileManager instance for file operations
189 Returns:
190 Path to the actual directory containing images to process
191 """
192 from openhcs.constants.constants import Backend
194 # Create workspace path in plate folder if not provided
195 if workspace_path is None: 195 ↛ 199line 195 didn't jump to line 199 because the condition on line 195 was always true
196 workspace_path = plate_path / "workspace"
198 # Check if workspace already exists - skip mirroring if it does
199 workspace_already_exists = workspace_path.exists()
200 if workspace_already_exists: 200 ↛ 201line 200 didn't jump to line 201 because the condition on line 200 was never true
201 logger.info(f"📁 EXISTING WORKSPACE FOUND: {workspace_path} - skipping mirror operation")
202 num_links = 0 # No new links created
203 else:
204 # Ensure workspace directory exists
205 filemanager.ensure_directory(str(workspace_path), Backend.DISK.value)
207 # Mirror plate directory with symlinks
208 logger.info(f"Mirroring plate directory {plate_path} to workspace {workspace_path}...")
209 try:
210 num_links = filemanager.mirror_directory_with_symlinks(
211 source_dir=str(plate_path),
212 target_dir=str(workspace_path),
213 backend=Backend.DISK.value,
214 recursive=True,
215 overwrite_symlinks_only=True,
216 )
217 logger.info(f"Created {num_links} symlinks in workspace.")
218 except Exception as mirror_error:
219 # If mirroring fails, clean up and try again with fail-loud
220 logger.warning(f"⚠️ MIRROR FAILED: {mirror_error}. Cleaning workspace and retrying...")
221 try:
222 import shutil
223 shutil.rmtree(workspace_path)
224 logger.info(f"🧹 Cleaned up failed workspace: {workspace_path}")
226 # Recreate directory and try mirroring again
227 filemanager.ensure_directory(str(workspace_path), Backend.DISK.value)
228 num_links = filemanager.mirror_directory_with_symlinks(
229 source_dir=str(plate_path),
230 target_dir=str(workspace_path),
231 backend=Backend.DISK.value,
232 recursive=True,
233 overwrite_symlinks_only=True,
234 )
235 logger.info(f"✅ RETRY SUCCESS: Created {num_links} symlinks in workspace.")
236 except Exception as retry_error:
237 # Fail loud on second attempt
238 error_msg = f"Failed to mirror plate directory to workspace after cleanup: {retry_error}"
239 logger.error(error_msg)
240 raise RuntimeError(error_msg) from retry_error
242 # Set plate_folder for this handler
243 self.plate_folder = workspace_path
245 # Prepare workspace and return final image directory
246 return self.post_workspace(workspace_path, filemanager, skip_preparation=workspace_already_exists)
248 def post_workspace(self, workspace_path: Union[str, Path], filemanager: FileManager, width: int = 3, skip_preparation: bool = False):
249 """
250 Hook called after workspace symlink creation.
251 Applies normalization logic followed by consistent filename padding.
253 This method requires a disk-backed path and should only be called
254 from steps with requires_fs_input=True.
256 Args:
257 workspace_path: Path to the workspace (string or Path object)
258 filemanager: FileManager instance for file operations
259 width: Width for padding (default: 3)
260 skip_preparation: If True, skip microscope-specific preparation (default: False)
262 Returns:
263 Path to the normalized image directory
265 Raises:
266 FileNotFoundError: If workspace_path does not exist
267 """
268 # Ensure workspace_path is a Path object
269 if isinstance(workspace_path, str): 269 ↛ 270line 269 didn't jump to line 270 because the condition on line 269 was never true
270 workspace_path = Path(workspace_path)
272 # Ensure the path exists
273 if not workspace_path.exists(): 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true
274 raise FileNotFoundError(f"Workspace path does not exist: {workspace_path}")
276 # Apply microscope-specific preparation logic (skip if workspace already existed)
277 if skip_preparation: 277 ↛ 278line 277 didn't jump to line 278 because the condition on line 277 was never true
278 logger.info(f"📁 SKIPPING PREPARATION: Workspace already existed - using as-is")
279 prepared_dir = workspace_path
280 else:
281 logger.info(f"🔄 APPLYING PREPARATION: Processing new workspace")
282 prepared_dir = self._prepare_workspace(workspace_path, filemanager)
284 # Deterministically resolve the image directory based on common_dirs
285 # Clause 245: Workspace operations are disk-only by design
286 # This call is structurally hardcoded to use the "disk" backend
287 entries = filemanager.list_dir(workspace_path, Backend.DISK.value)
289 # Filter entries to get only directories
290 subdirs = []
291 for entry in entries:
292 entry_path = Path(workspace_path) / entry
293 if entry_path.is_dir():
294 subdirs.append(entry_path)
296 # Look for a directory matching any of the common_dirs patterns
297 image_dir = None
298 for item in subdirs: 298 ↛ 316line 298 didn't jump to line 316 because the loop on line 298 didn't complete
299 # FileManager should return strings, but handle Path objects too
300 if isinstance(item, str): 300 ↛ 301line 300 didn't jump to line 301 because the condition on line 300 was never true
301 item_name = os.path.basename(item)
302 elif isinstance(item, Path): 302 ↛ 306line 302 didn't jump to line 306 because the condition on line 302 was always true
303 item_name = item.name
304 else:
305 # Skip any unexpected types
306 logger.warning("Unexpected directory path type: %s", type(item).__name__)
307 continue
309 if any(dir_name.lower() in item_name.lower() for dir_name in self.common_dirs): 309 ↛ 298line 309 didn't jump to line 298 because the condition on line 309 was always true
310 # Found a matching directory
311 logger.info("Found directory matching common_dirs pattern: %s", item)
312 image_dir = item
313 break
315 # If no matching directory found, use the prepared directory
316 if image_dir is None: 316 ↛ 317line 316 didn't jump to line 317 because the condition on line 316 was never true
317 logger.info("No directory matching common_dirs found, using prepared directory: %s", prepared_dir)
318 image_dir = prepared_dir
320 # Ensure parser is provided
321 parser = self.parser
323 # Get all image files in the directory
324 # Clause 245: Workspace operations are disk-only by design
325 # This call is structurally hardcoded to use the "disk" backend
326 image_files = filemanager.list_image_files(image_dir, Backend.DISK.value)
328 # Map original filenames to reconstructed filenames
329 rename_map = {}
331 for file_path in image_files:
332 # FileManager should return strings, but handle Path objects too
333 if isinstance(file_path, str): 333 ↛ 335line 333 didn't jump to line 335 because the condition on line 333 was always true
334 original_name = os.path.basename(file_path)
335 elif isinstance(file_path, Path):
336 original_name = file_path.name
337 else:
338 # Skip any unexpected types
339 logger.warning("Unexpected file path type: %s", type(file_path).__name__)
340 continue
342 # Parse the filename components
343 metadata = parser.parse_filename(original_name)
344 if not metadata: 344 ↛ 345line 344 didn't jump to line 345 because the condition on line 344 was never true
345 logger.warning("Could not parse filename: %s", original_name)
346 continue
348 # Validate required components
349 if metadata['site'] is None: 349 ↛ 350line 349 didn't jump to line 350 because the condition on line 349 was never true
350 logger.warning("Missing 'site' component in filename: %s", original_name)
351 continue
353 if metadata['channel'] is None: 353 ↛ 354line 353 didn't jump to line 354 because the condition on line 353 was never true
354 logger.warning("Missing 'channel' component in filename: %s", original_name)
355 continue
357 # z_index is optional - default to 1 if not present
358 site = metadata['site']
359 channel = metadata['channel']
360 z_index = metadata['z_index'] if metadata['z_index'] is not None else 1
362 # Log the components for debugging
363 logger.debug(
364 "Parsed components for %s: site=%s, channel=%s, z_index=%s",
365 original_name, site, channel, z_index
366 )
368 # Reconstruct the filename with proper padding
369 metadata['site'] = site
370 metadata['channel'] = channel
371 metadata['z_index'] = z_index
372 new_name = parser.construct_filename(**metadata)
374 # Add to rename map if different
375 if original_name != new_name: 375 ↛ 376line 375 didn't jump to line 376 because the condition on line 375 was never true
376 rename_map[original_name] = new_name
378 # Perform the renaming
379 for original_name, new_name in rename_map.items(): 379 ↛ 381line 379 didn't jump to line 381 because the loop on line 379 never started
380 # Create paths for the source and destination
381 if isinstance(image_dir, str):
382 original_path = os.path.join(image_dir, original_name)
383 new_path = os.path.join(image_dir, new_name)
384 else: # Path object
385 original_path = image_dir / original_name
386 new_path = image_dir / new_name
388 try:
389 # Ensure the parent directory exists
390 # Clause 245: Workspace operations are disk-only by design
391 # This call is structurally hardcoded to use the "disk" backend
392 parent_dir = os.path.dirname(new_path) if isinstance(new_path, str) else new_path.parent
393 filemanager.ensure_directory(parent_dir, Backend.DISK.value)
395 # Rename the file using move operation
396 # Clause 245: Workspace operations are disk-only by design
397 # This call is structurally hardcoded to use the "disk" backend
398 # Use replace_symlinks=True to allow overwriting existing symlinks
399 filemanager.move(original_path, new_path, Backend.DISK.value, replace_symlinks=True)
400 logger.debug("Renamed %s to %s", original_path, new_path)
401 except (OSError, FileNotFoundError) as e:
402 logger.error("Filesystem error renaming %s to %s: %s", original_path, new_path, e)
403 except TypeError as e:
404 logger.error("Type error renaming %s to %s: %s", original_path, new_path, e)
405 except Exception as e:
406 logger.error("Unexpected error renaming %s to %s: %s", original_path, new_path, e)
408 return image_dir
410 @abstractmethod
411 def _prepare_workspace(self, workspace_path: Path, filemanager: FileManager):
412 """
413 Microscope-specific preparation logic before image directory resolution.
415 This method performs any necessary preprocessing on the workspace but does NOT
416 determine the final image directory. It may return a suggested directory, but
417 the final image directory will be determined by post_workspace() based on
418 common_dirs matching.
420 Override in subclasses. Default implementation just returns the workspace path.
422 This method requires a disk-backed path and should only be called
423 from steps with requires_fs_input=True.
425 Args:
426 workspace_path: Path to the symlinked workspace
427 filemanager: FileManager instance for file operations
429 Returns:
430 Path: A suggested directory for further processing (not necessarily the final image directory)
432 Raises:
433 FileNotFoundError: If workspace_path does not exist
434 """
435 return workspace_path
438 # Delegate methods to parser
439 def parse_filename(self, filename: str) -> Optional[Dict[str, Any]]:
440 """Delegate to parser."""
441 return self.parser.parse_filename(filename)
443 def construct_filename(self, extension: str = '.tif', **component_values) -> str:
444 """
445 Delegate to parser using pure generic interface.
446 """
447 return self.parser.construct_filename(extension=extension, **component_values)
449 def auto_detect_patterns(self, folder_path: Union[str, Path], filemanager: FileManager, backend: str,
450 extensions=None, group_by=None, variable_components=None, **kwargs):
451 """
452 Delegate to pattern engine.
454 Args:
455 folder_path: Path to the folder (string or Path object)
456 filemanager: FileManager instance for file operations
457 backend: Backend to use for file operations (required)
458 extensions: Optional list of file extensions to include
459 group_by: GroupBy enum to group patterns by (e.g., GroupBy.CHANNEL, GroupBy.Z_INDEX)
460 variable_components: List of components to make variable (e.g., ['site', 'z_index'])
461 **kwargs: Dynamic filter parameters (e.g., well_filter, site_filter, channel_filter)
463 Returns:
464 Dict[str, Any]: Dictionary mapping axis values to patterns
465 """
466 # Ensure folder_path is a valid path
467 if isinstance(folder_path, str):
468 folder_path = Path(folder_path)
469 elif not isinstance(folder_path, Path):
470 raise TypeError(f"Expected string or Path object, got {type(folder_path).__name__}")
472 # Ensure the path exists using FileManager abstraction
473 if not filemanager.exists(str(folder_path), backend):
474 raise ValueError(f"Folder path does not exist: {folder_path}")
476 # Set default GroupBy if none provided
477 if group_by is None:
478 from openhcs.constants.constants import GroupBy
479 group_by = GroupBy.CHANNEL
481 # Create pattern engine on demand with the provided filemanager
482 from openhcs.formats.pattern.pattern_discovery import PatternDiscoveryEngine
483 pattern_engine = PatternDiscoveryEngine(self.parser, filemanager)
485 # Get patterns from the pattern engine
486 patterns_by_well = pattern_engine.auto_detect_patterns(
487 folder_path,
488 extensions=extensions,
489 group_by=group_by,
490 variable_components=variable_components,
491 backend=backend,
492 **kwargs # Pass through dynamic filter parameters
493 )
495 # 🔒 Clause 74 — Runtime Behavior Variation
496 # Ensure we always return a dictionary, not a generator
497 if not isinstance(patterns_by_well, dict):
498 # Convert to dictionary if it's not already one
499 return dict(patterns_by_well)
501 return patterns_by_well
503 def path_list_from_pattern(self, directory: Union[str, Path], pattern, filemanager: FileManager, backend: str, variable_components: Optional[List[str]] = None):
504 """
505 Delegate to pattern engine.
507 Args:
508 directory: Directory to search (string or Path object)
509 pattern: Pattern to match (str for literal filenames)
510 filemanager: FileManager instance for file operations
511 backend: Backend to use for file operations (required)
512 variable_components: List of components that can vary (will be ignored during matching)
514 Returns:
515 List of matching filenames
517 Raises:
518 TypeError: If a string with braces is passed (pattern paths are no longer supported)
519 ValueError: If directory does not exist
520 """
521 # Ensure directory is a valid path using FileManager abstraction
522 if isinstance(directory, str):
523 directory_path = Path(directory)
524 if not filemanager.exists(str(directory_path), backend):
525 raise ValueError(f"Directory does not exist: {directory}")
526 elif isinstance(directory, Path):
527 directory_path = directory
528 if not filemanager.exists(str(directory_path), backend):
529 raise ValueError(f"Directory does not exist: {directory}")
530 else:
531 raise TypeError(f"Expected string or Path object, got {type(directory).__name__}")
533 # Allow string patterns with braces - they are used for template matching
534 # The pattern engine will handle template expansion to find matching files
536 # Create pattern engine on demand with the provided filemanager
537 from openhcs.formats.pattern.pattern_discovery import PatternDiscoveryEngine
538 pattern_engine = PatternDiscoveryEngine(self.parser, filemanager)
540 # Delegate to the pattern engine
541 return pattern_engine.path_list_from_pattern(directory_path, pattern, backend=backend, variable_components=variable_components)
543 # Delegate metadata handling methods to metadata_handler with context
545 def find_metadata_file(self, plate_path: Union[str, Path]) -> Optional[Path]:
546 """Delegate to metadata handler."""
547 return self.metadata_handler.find_metadata_file(plate_path)
549 def get_grid_dimensions(self, plate_path: Union[str, Path]) -> Tuple[int, int]:
550 """Delegate to metadata handler."""
551 return self.metadata_handler.get_grid_dimensions(plate_path)
553 def get_pixel_size(self, plate_path: Union[str, Path]) -> float:
554 """Delegate to metadata handler."""
555 return self.metadata_handler.get_pixel_size(plate_path)
558# Import handler classes at module level with explicit mapping
559# No aliases or legacy compatibility layers (Clause 77)
561# Factory function
562def create_microscope_handler(microscope_type: str = 'auto',
563 plate_folder: Optional[Union[str, Path]] = None,
564 filemanager: Optional[FileManager] = None,
565 pattern_format: Optional[str] = None,
566 allowed_auto_types: Optional[List[str]] = None) -> MicroscopeHandler:
567 """
568 Factory function to create a microscope handler.
570 This function enforces explicit dependency injection by requiring a FileManager
571 instance to be provided. This ensures that all components requiring file operations
572 receive their dependencies explicitly, eliminating runtime fallbacks and enforcing
573 declarative configuration.
575 Args:
576 microscope_type: 'auto', 'imagexpress', 'opera_phenix', 'openhcs'.
577 plate_folder: Required for 'auto' detection.
578 filemanager: FileManager instance. Must be provided.
579 pattern_format: Name of the pattern format to use.
580 allowed_auto_types: For 'auto' mode, limit detection to these types.
581 'openhcs' is always included and tried first.
583 Returns:
584 An initialized MicroscopeHandler instance.
586 Raises:
587 ValueError: If filemanager is None or if microscope_type cannot be determined.
588 """
589 if filemanager is None: 589 ↛ 590line 589 didn't jump to line 590 because the condition on line 589 was never true
590 raise ValueError(
591 "FileManager must be provided to create_microscope_handler. "
592 "Default fallback has been removed."
593 )
595 logger.info("Using provided FileManager for microscope handler.")
597 # Auto-detect microscope type if needed
598 if microscope_type == 'auto': 598 ↛ 607line 598 didn't jump to line 607 because the condition on line 598 was always true
599 if not plate_folder: 599 ↛ 600line 599 didn't jump to line 600 because the condition on line 599 was never true
600 raise ValueError("plate_folder is required for auto-detection")
602 plate_folder = Path(plate_folder) if isinstance(plate_folder, str) else plate_folder
603 microscope_type = _auto_detect_microscope_type(plate_folder, filemanager, allowed_types=allowed_auto_types)
604 logger.info("Auto-detected microscope type: %s", microscope_type)
606 # Ensure all handlers are discovered before lookup
607 from openhcs.microscopes.handler_registry_service import discover_all_handlers, get_all_handler_types
608 discover_all_handlers()
610 # Get the appropriate handler class from the registry
611 # No dynamic imports or fallbacks (Clause 77: Rot Intolerance)
612 handler_class = MICROSCOPE_HANDLERS.get(microscope_type.lower())
613 if not handler_class: 613 ↛ 614line 613 didn't jump to line 614 because the condition on line 613 was never true
614 available_types = get_all_handler_types()
615 raise ValueError(
616 f"Unsupported microscope type: {microscope_type}. "
617 f"Available types: {available_types}"
618 )
620 # Create and configure the handler
621 logger.info(f"Creating {handler_class.__name__}")
623 # Create the handler with the parser and metadata handler
624 # The filemanager will be passed to methods that need it
625 handler = handler_class(filemanager, pattern_format=pattern_format)
627 # If the handler is OpenHCSMicroscopeHandler, set its plate_folder attribute.
628 # This is crucial for its dynamic parser loading mechanism.
629 # Use string comparison to avoid circular import
630 if handler.__class__.__name__ == 'OpenHCSMicroscopeHandler': 630 ↛ 631line 630 didn't jump to line 631 because the condition on line 630 was never true
631 if plate_folder:
632 handler.plate_folder = Path(plate_folder) if isinstance(plate_folder, str) else plate_folder
633 logger.info(f"Set plate_folder for OpenHCSMicroscopeHandler: {handler.plate_folder}")
634 else:
635 # This case should ideally not happen if auto-detection or explicit type setting
636 # implies a plate_folder is known.
637 logger.warning("OpenHCSMicroscopeHandler created without an initial plate_folder. "
638 "Parser will load upon first relevant method call with a path e.g. post_workspace.")
640 return handler
643def validate_backend_compatibility(handler: MicroscopeHandler, backend: Backend) -> bool:
644 """
645 Validate that a microscope handler supports a given storage backend.
647 Args:
648 handler: MicroscopeHandler instance to check
649 backend: Backend to validate compatibility with
651 Returns:
652 bool: True if the handler supports the backend, False otherwise
654 Example:
655 >>> handler = ImageXpressHandler(filemanager)
656 >>> validate_backend_compatibility(handler, Backend.ZARR)
657 False
658 >>> validate_backend_compatibility(handler, Backend.DISK)
659 True
660 """
661 return backend in handler.supported_backends
664def _try_metadata_detection(handler_class, filemanager: FileManager, plate_folder: Path) -> Optional[Path]:
665 """
666 Try metadata detection with a handler, normalizing return types and exceptions.
668 Args:
669 handler_class: MetadataHandler class to try
670 filemanager: FileManager instance
671 plate_folder: Path to plate directory
673 Returns:
674 Path if metadata found, None if not found (regardless of handler's native behavior)
675 """
676 try:
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
683 except (FileNotFoundError, Exception) as e:
684 # Expected exceptions for "not found" - convert to None
685 # Note: Using broad Exception catch for now, can be refined based on actual handler exceptions
686 logger.debug(f"Metadata detection failed for {handler_class.__name__}: {e}")
687 return None
690def _auto_detect_microscope_type(plate_folder: Path, filemanager: FileManager,
691 allowed_types: Optional[List[str]] = None) -> str:
692 """
693 Auto-detect microscope type using registry iteration.
695 Args:
696 plate_folder: Path to plate directory
697 filemanager: FileManager instance
698 allowed_types: Optional list of microscope types to try.
699 If None, tries all registered types.
700 'openhcs' is always included and tried first.
702 Returns:
703 Detected microscope type string
705 Raises:
706 ValueError: If microscope type cannot be determined
707 """
708 try:
709 # Ensure all handlers are discovered before auto-detection
710 from openhcs.microscopes.handler_registry_service import discover_all_handlers
711 discover_all_handlers()
713 # Build detection order: openhcsdata first, then filtered/ordered list
714 detection_order = ['openhcsdata'] # Always first, always included (correct registration name)
716 if allowed_types is None: 716 ↛ 721line 716 didn't jump to line 721 because the condition on line 716 was always true
717 # Use all registered handlers in registration order
718 detection_order.extend([name for name in METADATA_HANDLERS.keys() if name != 'openhcsdata'])
719 else:
720 # Use filtered list, but ensure openhcsdata is first
721 filtered_types = [name for name in allowed_types if name != 'openhcsdata' and name in METADATA_HANDLERS]
722 detection_order.extend(filtered_types)
724 # Try detection in order
725 for handler_name in detection_order: 725 ↛ 732line 725 didn't jump to line 732 because the loop on line 725 didn't complete
726 handler_class = METADATA_HANDLERS.get(handler_name)
727 if handler_class and _try_metadata_detection(handler_class, filemanager, plate_folder):
728 logger.info(f"Auto-detected {handler_name} microscope type")
729 return handler_name
731 # No handler succeeded - provide detailed error message
732 available_types = list(METADATA_HANDLERS.keys())
733 msg = (f"Could not auto-detect microscope type in {plate_folder}. "
734 f"Tried: {detection_order}. "
735 f"Available types: {available_types}. "
736 f"Ensure metadata files are present for supported formats.")
737 logger.error(msg)
738 raise ValueError(msg)
740 except Exception as e:
741 # Wrap exception with clear context
742 raise ValueError(f"Error during microscope type auto-detection in {plate_folder}: {e}") from e