Coverage for openhcs/microscopes/microscope_base.py: 58.7%
231 statements
« prev ^ index » next coverage.py v7.10.3, created at 2025-08-14 05:57 +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# Import PatternDiscoveryEngine for MicroscopeHandler initialization
17from openhcs.formats.pattern.pattern_discovery import PatternDiscoveryEngine
18from openhcs.io.filemanager import FileManager
19# Import interfaces from the base interfaces module
20from openhcs.microscopes.microscope_interfaces import (FilenameParser,
21 MetadataHandler)
23logger = logging.getLogger(__name__)
25# Dictionary to store registered microscope handlers
26MICROSCOPE_HANDLERS = {}
28# Dictionary to store registered metadata handlers for auto-detection
29METADATA_HANDLERS = {}
32class MicroscopeHandlerMeta(ABCMeta):
33 """Metaclass for automatic registration of microscope handlers."""
35 def __new__(cls, name, bases, attrs):
36 new_class = super().__new__(cls, name, bases, attrs)
38 # Only register concrete handler classes (not the abstract base class)
39 if bases and not getattr(new_class, '__abstractmethods__', None):
40 # Use explicit microscope type if provided, otherwise extract from class name
41 microscope_type = getattr(new_class, '_microscope_type', None)
42 if not microscope_type: 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true
43 if name.endswith('Handler'):
44 microscope_type = name[:-7].lower() # ImageXpressHandler -> imagexpress
45 else:
46 microscope_type = name.lower()
48 # Auto-register in MICROSCOPE_HANDLERS
49 MICROSCOPE_HANDLERS[microscope_type] = new_class
51 # Store the microscope type as the standard class attribute
52 new_class._microscope_type = microscope_type
54 # Auto-register metadata handler if the class has one
55 metadata_handler_class = getattr(new_class, '_metadata_handler_class', None)
56 if metadata_handler_class: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true
57 METADATA_HANDLERS[microscope_type] = metadata_handler_class
59 logger.debug(f"Auto-registered {name} as '{microscope_type}'")
61 return new_class
64def register_metadata_handler(handler_class, metadata_handler_class):
65 """
66 Register a metadata handler for a microscope handler class.
68 This function is called when _metadata_handler_class is set after class definition.
69 """
70 microscope_type = getattr(handler_class, '_microscope_type', None)
71 if microscope_type: 71 ↛ 75line 71 didn't jump to line 75 because the condition on line 71 was always true
72 METADATA_HANDLERS[microscope_type] = metadata_handler_class
73 logger.debug(f"Registered metadata handler {metadata_handler_class.__name__} for '{microscope_type}'")
74 else:
75 logger.warning(f"Could not register metadata handler for {handler_class.__name__} - no microscope type found")
80class MicroscopeHandler(ABC, metaclass=MicroscopeHandlerMeta):
81 """Composed class for handling microscope-specific functionality."""
83 DEFAULT_MICROSCOPE = 'auto'
84 _handlers_cache = None
86 # Optional class attribute for explicit metadata handler registration
87 _metadata_handler_class: Optional[Type[MetadataHandler]] = None
89 def __init__(self, parser: FilenameParser,
90 metadata_handler: MetadataHandler):
91 """
92 Initialize the microscope handler.
94 Args:
95 parser: Parser for microscopy filenames.
96 metadata_handler: Handler for microscope metadata.
97 """
98 self.parser = parser
99 self.metadata_handler = metadata_handler
100 self.plate_folder: Optional[Path] = None # Store workspace path if needed by methods
102 # Pattern discovery engine will be created on demand with the provided filemanager
104 @property
105 @abstractmethod
106 def common_dirs(self) -> List[str]:
107 """
108 Canonical subdirectory names where image data may reside.
109 Example: ['Images', 'TimePoint', 'Data']
110 """
111 pass
113 @property
114 @abstractmethod
115 def microscope_type(self) -> str:
116 """Microscope type identifier (for interface enforcement only)."""
117 pass
119 @property
120 @abstractmethod
121 def metadata_handler_class(self) -> Type[MetadataHandler]:
122 """Metadata handler class (for interface enforcement only)."""
123 pass
125 @property
126 @abstractmethod
127 def compatible_backends(self) -> List[Backend]:
128 """
129 List of storage backends this microscope handler is compatible with, in priority order.
131 Must be explicitly declared by each handler implementation.
132 The first backend in the list is the preferred/highest priority backend.
133 The compiler will use the first backend for initial step materialization.
135 Common patterns:
136 - [Backend.DISK] - Basic handlers (ImageXpress, Opera Phenix)
137 - [Backend.ZARR, Backend.DISK] - Advanced handlers (OpenHCS: zarr preferred, disk fallback)
139 Returns:
140 List of Backend enum values this handler can work with, in priority order
141 """
142 pass
144 def get_available_backends(self, plate_path: Union[str, Path]) -> List[Backend]:
145 """
146 Get available storage backends for this specific plate.
148 Default implementation returns all compatible backends.
149 Override this method only if you need to check actual disk state
150 (like OpenHCS which reads from metadata).
152 Args:
153 plate_path: Path to the plate folder
155 Returns:
156 List of Backend enums that are available for this plate.
157 """
158 return self.compatible_backends
160 def initialize_workspace(self, plate_path: Path, workspace_path: Optional[Path], filemanager: FileManager) -> Path:
161 """
162 Default workspace initialization: create workspace in plate folder and mirror with symlinks.
164 Most microscope handlers need workspace mirroring. Override this method only if different behavior is needed.
166 Args:
167 plate_path: Path to the original plate directory
168 workspace_path: Optional workspace path (creates default in plate folder if None)
169 filemanager: FileManager instance for file operations
171 Returns:
172 Path to the actual directory containing images to process
173 """
174 from openhcs.constants.constants import Backend
176 # Create workspace path in plate folder if not provided
177 if workspace_path is None: 177 ↛ 181line 177 didn't jump to line 181 because the condition on line 177 was always true
178 workspace_path = plate_path / "workspace"
180 # Check if workspace already exists - skip mirroring if it does
181 workspace_already_exists = workspace_path.exists()
182 if workspace_already_exists: 182 ↛ 183line 182 didn't jump to line 183 because the condition on line 182 was never true
183 logger.info(f"📁 EXISTING WORKSPACE FOUND: {workspace_path} - skipping mirror operation")
184 num_links = 0 # No new links created
185 else:
186 # Ensure workspace directory exists
187 filemanager.ensure_directory(str(workspace_path), Backend.DISK.value)
189 # Mirror plate directory with symlinks
190 logger.info(f"Mirroring plate directory {plate_path} to workspace {workspace_path}...")
191 try:
192 num_links = filemanager.mirror_directory_with_symlinks(
193 source_dir=str(plate_path),
194 target_dir=str(workspace_path),
195 backend=Backend.DISK.value,
196 recursive=True,
197 overwrite_symlinks_only=True,
198 )
199 logger.info(f"Created {num_links} symlinks in workspace.")
200 except Exception as mirror_error:
201 # If mirroring fails, clean up and try again with fail-loud
202 logger.warning(f"⚠️ MIRROR FAILED: {mirror_error}. Cleaning workspace and retrying...")
203 try:
204 import shutil
205 shutil.rmtree(workspace_path)
206 logger.info(f"🧹 Cleaned up failed workspace: {workspace_path}")
208 # Recreate directory and try mirroring again
209 filemanager.ensure_directory(str(workspace_path), Backend.DISK.value)
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"✅ RETRY SUCCESS: Created {num_links} symlinks in workspace.")
218 except Exception as retry_error:
219 # Fail loud on second attempt
220 error_msg = f"Failed to mirror plate directory to workspace after cleanup: {retry_error}"
221 logger.error(error_msg)
222 raise RuntimeError(error_msg) from retry_error
224 # Set plate_folder for this handler
225 self.plate_folder = workspace_path
227 # Prepare workspace and return final image directory
228 return self.post_workspace(workspace_path, filemanager, skip_preparation=workspace_already_exists)
230 def post_workspace(self, workspace_path: Union[str, Path], filemanager: FileManager, width: int = 3, skip_preparation: bool = False):
231 """
232 Hook called after workspace symlink creation.
233 Applies normalization logic followed by consistent filename padding.
235 This method requires a disk-backed path and should only be called
236 from steps with requires_fs_input=True.
238 Args:
239 workspace_path: Path to the workspace (string or Path object)
240 filemanager: FileManager instance for file operations
241 width: Width for padding (default: 3)
242 skip_preparation: If True, skip microscope-specific preparation (default: False)
244 Returns:
245 Path to the normalized image directory
247 Raises:
248 FileNotFoundError: If workspace_path does not exist
249 """
250 # Ensure workspace_path is a Path object
251 if isinstance(workspace_path, str): 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true
252 workspace_path = Path(workspace_path)
254 # Ensure the path exists
255 if not workspace_path.exists(): 255 ↛ 256line 255 didn't jump to line 256 because the condition on line 255 was never true
256 raise FileNotFoundError(f"Workspace path does not exist: {workspace_path}")
258 # Apply microscope-specific preparation logic (skip if workspace already existed)
259 if skip_preparation: 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
260 logger.info(f"📁 SKIPPING PREPARATION: Workspace already existed - using as-is")
261 prepared_dir = workspace_path
262 else:
263 logger.info(f"🔄 APPLYING PREPARATION: Processing new workspace")
264 prepared_dir = self._prepare_workspace(workspace_path, filemanager)
266 # Deterministically resolve the image directory based on common_dirs
267 # Clause 245: Workspace operations are disk-only by design
268 # This call is structurally hardcoded to use the "disk" backend
269 entries = filemanager.list_dir(workspace_path, Backend.DISK.value)
271 # Filter entries to get only directories
272 subdirs = []
273 for entry in entries:
274 entry_path = Path(workspace_path) / entry
275 if entry_path.is_dir():
276 subdirs.append(entry_path)
278 # Look for a directory matching any of the common_dirs patterns
279 image_dir = None
280 for item in subdirs: 280 ↛ 298line 280 didn't jump to line 298 because the loop on line 280 didn't complete
281 # FileManager should return strings, but handle Path objects too
282 if isinstance(item, str): 282 ↛ 283line 282 didn't jump to line 283 because the condition on line 282 was never true
283 item_name = os.path.basename(item)
284 elif isinstance(item, Path): 284 ↛ 288line 284 didn't jump to line 288 because the condition on line 284 was always true
285 item_name = item.name
286 else:
287 # Skip any unexpected types
288 logger.warning("Unexpected directory path type: %s", type(item).__name__)
289 continue
291 if any(dir_name.lower() in item_name.lower() for dir_name in self.common_dirs): 291 ↛ 280line 291 didn't jump to line 280 because the condition on line 291 was always true
292 # Found a matching directory
293 logger.info("Found directory matching common_dirs pattern: %s", item)
294 image_dir = item
295 break
297 # If no matching directory found, use the prepared directory
298 if image_dir is None: 298 ↛ 299line 298 didn't jump to line 299 because the condition on line 298 was never true
299 logger.info("No directory matching common_dirs found, using prepared directory: %s", prepared_dir)
300 image_dir = prepared_dir
302 # Ensure parser is provided
303 parser = self.parser
305 # Get all image files in the directory
306 # Clause 245: Workspace operations are disk-only by design
307 # This call is structurally hardcoded to use the "disk" backend
308 image_files = filemanager.list_image_files(image_dir, Backend.DISK.value)
310 # Map original filenames to reconstructed filenames
311 rename_map = {}
313 for file_path in image_files:
314 # FileManager should return strings, but handle Path objects too
315 if isinstance(file_path, str): 315 ↛ 317line 315 didn't jump to line 317 because the condition on line 315 was always true
316 original_name = os.path.basename(file_path)
317 elif isinstance(file_path, Path):
318 original_name = file_path.name
319 else:
320 # Skip any unexpected types
321 logger.warning("Unexpected file path type: %s", type(file_path).__name__)
322 continue
324 # Parse the filename components
325 metadata = parser.parse_filename(original_name)
326 if not metadata: 326 ↛ 327line 326 didn't jump to line 327 because the condition on line 326 was never true
327 logger.warning("Could not parse filename: %s", original_name)
328 continue
330 # Validate required components
331 if metadata['site'] is None: 331 ↛ 332line 331 didn't jump to line 332 because the condition on line 331 was never true
332 logger.warning("Missing 'site' component in filename: %s", original_name)
333 continue
335 if metadata['channel'] is None: 335 ↛ 336line 335 didn't jump to line 336 because the condition on line 335 was never true
336 logger.warning("Missing 'channel' component in filename: %s", original_name)
337 continue
339 # z_index is optional - default to 1 if not present
340 site = metadata['site']
341 channel = metadata['channel']
342 z_index = metadata['z_index'] if metadata['z_index'] is not None else 1
344 # Log the components for debugging
345 logger.debug(
346 "Parsed components for %s: site=%s, channel=%s, z_index=%s",
347 original_name, site, channel, z_index
348 )
350 # Reconstruct the filename with proper padding
351 new_name = parser.construct_filename(
352 well=metadata['well'],
353 site=site,
354 channel=channel,
355 z_index=z_index,
356 extension=metadata['extension'],
357 site_padding=width,
358 z_padding=width
359 )
361 # Add to rename map if different
362 if original_name != new_name: 362 ↛ 363line 362 didn't jump to line 363 because the condition on line 362 was never true
363 rename_map[original_name] = new_name
365 # Perform the renaming
366 for original_name, new_name in rename_map.items(): 366 ↛ 368line 366 didn't jump to line 368 because the loop on line 366 never started
367 # Create paths for the source and destination
368 if isinstance(image_dir, str):
369 original_path = os.path.join(image_dir, original_name)
370 new_path = os.path.join(image_dir, new_name)
371 else: # Path object
372 original_path = image_dir / original_name
373 new_path = image_dir / new_name
375 try:
376 # Ensure the parent directory exists
377 # Clause 245: Workspace operations are disk-only by design
378 # This call is structurally hardcoded to use the "disk" backend
379 parent_dir = os.path.dirname(new_path) if isinstance(new_path, str) else new_path.parent
380 filemanager.ensure_directory(parent_dir, Backend.DISK.value)
382 # Rename the file using move operation
383 # Clause 245: Workspace operations are disk-only by design
384 # This call is structurally hardcoded to use the "disk" backend
385 # Use replace_symlinks=True to allow overwriting existing symlinks
386 filemanager.move(original_path, new_path, Backend.DISK.value, replace_symlinks=True)
387 logger.debug("Renamed %s to %s", original_path, new_path)
388 except (OSError, FileNotFoundError) as e:
389 logger.error("Filesystem error renaming %s to %s: %s", original_path, new_path, e)
390 except TypeError as e:
391 logger.error("Type error renaming %s to %s: %s", original_path, new_path, e)
392 except Exception as e:
393 logger.error("Unexpected error renaming %s to %s: %s", original_path, new_path, e)
395 return image_dir
397 @abstractmethod
398 def _prepare_workspace(self, workspace_path: Path, filemanager: FileManager):
399 """
400 Microscope-specific preparation logic before image directory resolution.
402 This method performs any necessary preprocessing on the workspace but does NOT
403 determine the final image directory. It may return a suggested directory, but
404 the final image directory will be determined by post_workspace() based on
405 common_dirs matching.
407 Override in subclasses. Default implementation just returns the workspace path.
409 This method requires a disk-backed path and should only be called
410 from steps with requires_fs_input=True.
412 Args:
413 workspace_path: Path to the symlinked workspace
414 filemanager: FileManager instance for file operations
416 Returns:
417 Path: A suggested directory for further processing (not necessarily the final image directory)
419 Raises:
420 FileNotFoundError: If workspace_path does not exist
421 """
422 return workspace_path
425 # Delegate methods to parser
426 def parse_filename(self, filename: str) -> Optional[Dict[str, Any]]:
427 """Delegate to parser."""
428 return self.parser.parse_filename(filename)
430 def construct_filename(self, well: str, site: Optional[Union[int, str]] = None,
431 channel: Optional[int] = None,
432 z_index: Optional[Union[int, str]] = None,
433 extension: str = '.tif',
434 site_padding: int = 3, z_padding: int = 3) -> str:
435 """Delegate to parser."""
436 return self.parser.construct_filename(
437 well, site, channel, z_index, extension, site_padding, z_padding
438 )
440 def auto_detect_patterns(self, folder_path: Union[str, Path], filemanager: FileManager, backend: str,
441 well_filter=None, extensions=None, group_by='channel', variable_components=None):
442 """
443 Delegate to pattern engine.
445 Args:
446 folder_path: Path to the folder (string or Path object)
447 filemanager: FileManager instance for file operations
448 backend: Backend to use for file operations (required)
449 well_filter: Optional list of wells to include
450 extensions: Optional list of file extensions to include
451 group_by: Component to group patterns by (e.g., 'channel', 'z_index', 'well')
452 variable_components: List of components to make variable (e.g., ['site', 'z_index'])
454 Returns:
455 Dict[str, Any]: Dictionary mapping wells to patterns
456 """
457 # Ensure folder_path is a valid path
458 if isinstance(folder_path, str): 458 ↛ 460line 458 didn't jump to line 460 because the condition on line 458 was always true
459 folder_path = Path(folder_path)
460 elif not isinstance(folder_path, Path):
461 raise TypeError(f"Expected string or Path object, got {type(folder_path).__name__}")
463 # Ensure the path exists using FileManager abstraction
464 if not filemanager.exists(str(folder_path), backend): 464 ↛ 465line 464 didn't jump to line 465 because the condition on line 464 was never true
465 raise ValueError(f"Folder path does not exist: {folder_path}")
467 # Create pattern engine on demand with the provided filemanager
468 pattern_engine = PatternDiscoveryEngine(self.parser, filemanager)
470 # Get patterns from the pattern engine
471 patterns_by_well = pattern_engine.auto_detect_patterns(
472 folder_path,
473 well_filter=well_filter,
474 extensions=extensions,
475 group_by=group_by,
476 variable_components=variable_components,
477 backend=backend
478 )
480 # 🔒 Clause 74 — Runtime Behavior Variation
481 # Ensure we always return a dictionary, not a generator
482 if not isinstance(patterns_by_well, dict): 482 ↛ 484line 482 didn't jump to line 484 because the condition on line 482 was never true
483 # Convert to dictionary if it's not already one
484 return dict(patterns_by_well)
486 return patterns_by_well
488 def path_list_from_pattern(self, directory: Union[str, Path], pattern, filemanager: FileManager, backend: str, variable_components: Optional[List[str]] = None):
489 """
490 Delegate to pattern engine.
492 Args:
493 directory: Directory to search (string or Path object)
494 pattern: Pattern to match (str for literal filenames)
495 filemanager: FileManager instance for file operations
496 backend: Backend to use for file operations (required)
497 variable_components: List of components that can vary (will be ignored during matching)
499 Returns:
500 List of matching filenames
502 Raises:
503 TypeError: If a string with braces is passed (pattern paths are no longer supported)
504 ValueError: If directory does not exist
505 """
506 # Ensure directory is a valid path using FileManager abstraction
507 if isinstance(directory, str): 507 ↛ 511line 507 didn't jump to line 511 because the condition on line 507 was always true
508 directory_path = Path(directory)
509 if not filemanager.exists(str(directory_path), backend): 509 ↛ 510line 509 didn't jump to line 510 because the condition on line 509 was never true
510 raise ValueError(f"Directory does not exist: {directory}")
511 elif isinstance(directory, Path):
512 directory_path = directory
513 if not filemanager.exists(str(directory_path), backend):
514 raise ValueError(f"Directory does not exist: {directory}")
515 else:
516 raise TypeError(f"Expected string or Path object, got {type(directory).__name__}")
518 # Allow string patterns with braces - they are used for template matching
519 # The pattern engine will handle template expansion to find matching files
521 # Create pattern engine on demand with the provided filemanager
522 pattern_engine = PatternDiscoveryEngine(self.parser, filemanager)
524 # Delegate to the pattern engine
525 return pattern_engine.path_list_from_pattern(directory_path, pattern, backend=backend, variable_components=variable_components)
527 # Delegate metadata handling methods to metadata_handler with context
529 def find_metadata_file(self, plate_path: Union[str, Path]) -> Optional[Path]:
530 """Delegate to metadata handler."""
531 return self.metadata_handler.find_metadata_file(plate_path)
533 def get_grid_dimensions(self, plate_path: Union[str, Path]) -> Tuple[int, int]:
534 """Delegate to metadata handler."""
535 return self.metadata_handler.get_grid_dimensions(plate_path)
537 def get_pixel_size(self, plate_path: Union[str, Path]) -> float:
538 """Delegate to metadata handler."""
539 return self.metadata_handler.get_pixel_size(plate_path)
542# Import handler classes at module level with explicit mapping
543# No aliases or legacy compatibility layers (Clause 77)
545# Factory function
546def create_microscope_handler(microscope_type: str = 'auto',
547 plate_folder: Optional[Union[str, Path]] = None,
548 filemanager: Optional[FileManager] = None,
549 pattern_format: Optional[str] = None,
550 allowed_auto_types: Optional[List[str]] = None) -> MicroscopeHandler:
551 """
552 Factory function to create a microscope handler.
554 This function enforces explicit dependency injection by requiring a FileManager
555 instance to be provided. This ensures that all components requiring file operations
556 receive their dependencies explicitly, eliminating runtime fallbacks and enforcing
557 declarative configuration.
559 Args:
560 microscope_type: 'auto', 'imagexpress', 'opera_phenix', 'openhcs'.
561 plate_folder: Required for 'auto' detection.
562 filemanager: FileManager instance. Must be provided.
563 pattern_format: Name of the pattern format to use.
564 allowed_auto_types: For 'auto' mode, limit detection to these types.
565 'openhcs' is always included and tried first.
567 Returns:
568 An initialized MicroscopeHandler instance.
570 Raises:
571 ValueError: If filemanager is None or if microscope_type cannot be determined.
572 """
573 if filemanager is None: 573 ↛ 574line 573 didn't jump to line 574 because the condition on line 573 was never true
574 raise ValueError(
575 "FileManager must be provided to create_microscope_handler. "
576 "Default fallback has been removed."
577 )
579 logger.info("Using provided FileManager for microscope handler.")
581 # Auto-detect microscope type if needed
582 if microscope_type == 'auto': 582 ↛ 592line 582 didn't jump to line 592 because the condition on line 582 was always true
583 if not plate_folder: 583 ↛ 584line 583 didn't jump to line 584 because the condition on line 583 was never true
584 raise ValueError("plate_folder is required for auto-detection")
586 plate_folder = Path(plate_folder) if isinstance(plate_folder, str) else plate_folder
587 microscope_type = _auto_detect_microscope_type(plate_folder, filemanager, allowed_types=allowed_auto_types)
588 logger.info("Auto-detected microscope type: %s", microscope_type)
590 # Get the appropriate handler class from the constant mapping
591 # No dynamic imports or fallbacks (Clause 77: Rot Intolerance)
592 handler_class = MICROSCOPE_HANDLERS.get(microscope_type.lower())
593 if not handler_class: 593 ↛ 594line 593 didn't jump to line 594 because the condition on line 593 was never true
594 raise ValueError(
595 f"Unsupported microscope type: {microscope_type}. "
596 f"Supported types: {list(MICROSCOPE_HANDLERS.keys())}"
597 )
599 # Create and configure the handler
600 logger.info(f"Creating {handler_class.__name__}")
602 # Create the handler with the parser and metadata handler
603 # The filemanager will be passed to methods that need it
604 handler = handler_class(filemanager, pattern_format=pattern_format)
606 # If the handler is OpenHCSMicroscopeHandler, set its plate_folder attribute.
607 # This is crucial for its dynamic parser loading mechanism.
608 # Use string comparison to avoid circular import
609 if handler.__class__.__name__ == 'OpenHCSMicroscopeHandler': 609 ↛ 610line 609 didn't jump to line 610 because the condition on line 609 was never true
610 if plate_folder:
611 handler.plate_folder = Path(plate_folder) if isinstance(plate_folder, str) else plate_folder
612 logger.info(f"Set plate_folder for OpenHCSMicroscopeHandler: {handler.plate_folder}")
613 else:
614 # This case should ideally not happen if auto-detection or explicit type setting
615 # implies a plate_folder is known.
616 logger.warning("OpenHCSMicroscopeHandler created without an initial plate_folder. "
617 "Parser will load upon first relevant method call with a path e.g. post_workspace.")
619 return handler
622def validate_backend_compatibility(handler: MicroscopeHandler, backend: Backend) -> bool:
623 """
624 Validate that a microscope handler supports a given storage backend.
626 Args:
627 handler: MicroscopeHandler instance to check
628 backend: Backend to validate compatibility with
630 Returns:
631 bool: True if the handler supports the backend, False otherwise
633 Example:
634 >>> handler = ImageXpressHandler(filemanager)
635 >>> validate_backend_compatibility(handler, Backend.ZARR)
636 False
637 >>> validate_backend_compatibility(handler, Backend.DISK)
638 True
639 """
640 return backend in handler.supported_backends
643def _try_metadata_detection(handler_class, filemanager: FileManager, plate_folder: Path) -> Optional[Path]:
644 """
645 Try metadata detection with a handler, normalizing return types and exceptions.
647 Args:
648 handler_class: MetadataHandler class to try
649 filemanager: FileManager instance
650 plate_folder: Path to plate directory
652 Returns:
653 Path if metadata found, None if not found (regardless of handler's native behavior)
654 """
655 try:
656 handler = handler_class(filemanager)
657 result = handler.find_metadata_file(plate_folder)
659 # Normalize return type: convert any truthy result to Path, falsy to None
660 return Path(result) if result else None
662 except (FileNotFoundError, Exception) as e:
663 # Expected exceptions for "not found" - convert to None
664 # Note: Using broad Exception catch for now, can be refined based on actual handler exceptions
665 logger.debug(f"Metadata detection failed for {handler_class.__name__}: {e}")
666 return None
669def _auto_detect_microscope_type(plate_folder: Path, filemanager: FileManager,
670 allowed_types: Optional[List[str]] = None) -> str:
671 """
672 Auto-detect microscope type using registry iteration.
674 Args:
675 plate_folder: Path to plate directory
676 filemanager: FileManager instance
677 allowed_types: Optional list of microscope types to try.
678 If None, tries all registered types.
679 'openhcs' is always included and tried first.
681 Returns:
682 Detected microscope type string
684 Raises:
685 ValueError: If microscope type cannot be determined
686 """
687 try:
688 # Build detection order: openhcsdata first, then filtered/ordered list
689 detection_order = ['openhcsdata'] # Always first, always included (correct registration name)
691 if allowed_types is None: 691 ↛ 696line 691 didn't jump to line 696 because the condition on line 691 was always true
692 # Use all registered handlers in registration order
693 detection_order.extend([name for name in METADATA_HANDLERS.keys() if name != 'openhcsdata'])
694 else:
695 # Use filtered list, but ensure openhcsdata is first
696 filtered_types = [name for name in allowed_types if name != 'openhcsdata' and name in METADATA_HANDLERS]
697 detection_order.extend(filtered_types)
699 # Try detection in order
700 for handler_name in detection_order: 700 ↛ 707line 700 didn't jump to line 707 because the loop on line 700 didn't complete
701 handler_class = METADATA_HANDLERS.get(handler_name)
702 if handler_class and _try_metadata_detection(handler_class, filemanager, plate_folder):
703 logger.info(f"Auto-detected {handler_name} microscope type")
704 return handler_name
706 # No handler succeeded - provide detailed error message
707 available_types = list(METADATA_HANDLERS.keys())
708 msg = (f"Could not auto-detect microscope type in {plate_folder}. "
709 f"Tried: {detection_order}. "
710 f"Available types: {available_types}. "
711 f"Ensure metadata files are present for supported formats.")
712 logger.error(msg)
713 raise ValueError(msg)
715 except Exception as e:
716 # Wrap exception with clear context
717 raise ValueError(f"Error during microscope type auto-detection in {plate_folder}: {e}") from e