Coverage for openhcs/microscopes/openhcs.py: 36.1%

1"""

2OpenHCS microscope handler implementation for openhcs.

4This module provides the OpenHCSMicroscopeHandler, which reads plates

5that have been pre-processed and standardized into the OpenHCS format.

6The metadata for such plates is defined in an 'openhcs_metadata.json' file.

7"""

9import json

10import logging

11from dataclasses import dataclass, asdict

12from pathlib import Path

13from typing import Any, Dict, List, Optional, Tuple, Union, Type

15from openhcs.constants.constants import Backend, GroupBy, DEFAULT_IMAGE_EXTENSIONS

16from openhcs.io.exceptions import MetadataNotFoundError

17from openhcs.io.filemanager import FileManager

18from openhcs.io.metadata_writer import AtomicMetadataWriter, MetadataWriteError, get_metadata_path, METADATA_CONFIG

19from openhcs.microscopes.microscope_interfaces import MetadataHandler

20logger = logging.getLogger(__name__)

23@dataclass(frozen=True)

24class OpenHCSMetadataFields:

25 """Centralized constants for OpenHCS metadata field names."""

26 # Core metadata structure - use centralized constants

27 SUBDIRECTORIES: str = METADATA_CONFIG.SUBDIRECTORIES_KEY

28 IMAGE_FILES: str = "image_files"

29 AVAILABLE_BACKENDS: str = METADATA_CONFIG.AVAILABLE_BACKENDS_KEY

31 # Required metadata fields

32 GRID_DIMENSIONS: str = "grid_dimensions"

33 PIXEL_SIZE: str = "pixel_size"

34 SOURCE_FILENAME_PARSER_NAME: str = "source_filename_parser_name"

35 MICROSCOPE_HANDLER_NAME: str = "microscope_handler_name"

37 # Optional metadata fields

38 CHANNELS: str = "channels"

39 WELLS: str = "wells"

40 SITES: str = "sites"

41 Z_INDEXES: str = "z_indexes"

42 OBJECTIVES: str = "objectives"

43 ACQUISITION_DATETIME: str = "acquisition_datetime"

44 PLATE_NAME: str = "plate_name"

46 # Default values

47 DEFAULT_SUBDIRECTORY: str = "."

48 DEFAULT_SUBDIRECTORY_LEGACY: str = "images"

50 # Microscope type identifier

51 MICROSCOPE_TYPE: str = "openhcsdata"

54# Global instance for easy access

55FIELDS = OpenHCSMetadataFields()

57def _get_available_filename_parsers():

58 """

59 Lazy import of filename parsers to avoid circular imports.

61 Returns:

62 Dict mapping parser class names to parser classes

63 """

64 # Import parsers only when needed to avoid circular imports

65 from openhcs.microscopes.imagexpress import ImageXpressFilenameParser

66 from openhcs.microscopes.opera_phenix import OperaPhenixFilenameParser

68 return {

69 "ImageXpressFilenameParser": ImageXpressFilenameParser,

70 "OperaPhenixFilenameParser": OperaPhenixFilenameParser,

71 # Add other parsers to this dictionary as they are implemented/imported.

72 # Example: "MyOtherParser": MyOtherParser,

73 }

76class OpenHCSMetadataHandler(MetadataHandler):

77 """

78 Metadata handler for the OpenHCS pre-processed format.

80 This handler reads metadata from an 'openhcs_metadata.json' file

81 located in the root of the plate folder.

82 """

83 METADATA_FILENAME = METADATA_CONFIG.METADATA_FILENAME

85 def __init__(self, filemanager: FileManager):

86 """

87 Initialize the metadata handler.

89 Args:

90 filemanager: FileManager instance for file operations.

91 """

92 super().__init__()

93 self.filemanager = filemanager

94 self.atomic_writer = AtomicMetadataWriter()

95 self._metadata_cache: Optional[Dict[str, Any]] = None

96 self._plate_path_cache: Optional[Path] = None

98 def _load_metadata(self, plate_path: Union[str, Path]) -> Dict[str, Any]:

99 """

100 Loads the JSON metadata file if not already cached or if plate_path changed.

101

102 Args:

103 plate_path: Path to the plate folder.

104

105 Returns:

106 A dictionary containing the parsed JSON metadata.

107

108 Raises:

109 MetadataNotFoundError: If the metadata file cannot be found or parsed.

110 FileNotFoundError: If plate_path does not exist.

111 """

112 current_path = Path(plate_path)

113 if self._metadata_cache is not None and self._plate_path_cache == current_path:

114 return self._metadata_cache

115

116 metadata_file_path = self.find_metadata_file(current_path)

117 if not self.filemanager.exists(str(metadata_file_path), Backend.DISK.value):

118 raise MetadataNotFoundError(f"Metadata file '{self.METADATA_FILENAME}' not found in {plate_path}")

119

120 try:

121 content = self.filemanager.load(str(metadata_file_path), Backend.DISK.value)

122 metadata_dict = json.loads(content.decode('utf-8') if isinstance(content, bytes) else content)

123

124 # Handle subdirectory-keyed format

125 if subdirs := metadata_dict.get(FIELDS.SUBDIRECTORIES):

126 if not subdirs:

127 raise MetadataNotFoundError(f"Empty subdirectories in metadata file '{metadata_file_path}'")

128

129 # Merge all subdirectories: use first as base, combine all image_files

130 base_metadata = next(iter(subdirs.values())).copy()

131 base_metadata[FIELDS.IMAGE_FILES] = [

132 file for subdir in subdirs.values()

133 for file in subdir.get(FIELDS.IMAGE_FILES, [])

134 ]

135 self._metadata_cache = base_metadata

136 else:

137 # Legacy format not supported - use migration script

138 raise MetadataNotFoundError(

139 f"Legacy metadata format detected in '{metadata_file_path}'. "

140 f"Please run the migration script: python scripts/migrate_legacy_metadata.py {current_path}"

141 )

142

143 self._plate_path_cache = current_path

144 return self._metadata_cache

145

146 except json.JSONDecodeError as e:

147 raise MetadataNotFoundError(f"Error decoding JSON from '{metadata_file_path}': {e}") from e

148

149

150

151 def determine_main_subdirectory(self, plate_path: Union[str, Path]) -> str:

152 """Determine main input subdirectory from metadata."""

153 metadata_dict = self._load_metadata_dict(plate_path)

154 subdirs = metadata_dict.get(FIELDS.SUBDIRECTORIES)

155

156 # Legacy format not supported - should have been caught by _load_metadata_dict

157 if not subdirs:

158 raise MetadataNotFoundError(f"No subdirectories found in metadata for {plate_path}")

159

160 # Single subdirectory - use it

161 if len(subdirs) == 1:

162 return next(iter(subdirs.keys()))

163

164 # Multiple subdirectories - find main or fallback

165 main_subdir = next((name for name, data in subdirs.items() if data.get("main")), None)

166 if main_subdir:

167 return main_subdir

168

169 # Fallback hierarchy: legacy default -> first available

170 if FIELDS.DEFAULT_SUBDIRECTORY_LEGACY in subdirs:

171 return FIELDS.DEFAULT_SUBDIRECTORY_LEGACY

172 else:

173 return next(iter(subdirs.keys()))

174

175 def _load_metadata_dict(self, plate_path: Union[str, Path]) -> Dict[str, Any]:

176 """Load and parse metadata JSON, fail-loud on errors."""

177 metadata_file_path = self.find_metadata_file(plate_path)

178 if not self.filemanager.exists(str(metadata_file_path), Backend.DISK.value):

179 raise MetadataNotFoundError(f"Metadata file '{self.METADATA_FILENAME}' not found in {plate_path}")

180

181 try:

182 content = self.filemanager.load(str(metadata_file_path), Backend.DISK.value)

183 return json.loads(content.decode('utf-8') if isinstance(content, bytes) else content)

184 except json.JSONDecodeError as e:

185 raise MetadataNotFoundError(f"Error decoding JSON from '{metadata_file_path}': {e}") from e

186

187 def find_metadata_file(self, plate_path: Union[str, Path], context: Optional[Any] = None) -> Optional[Path]:

188 """Find the OpenHCS JSON metadata file."""

189 plate_p = Path(plate_path)

190 if not self.filemanager.is_dir(str(plate_p), Backend.DISK.value): 190 ↛ 191line 190 didn't jump to line 191 because the condition on line 190 was never true

191 return None

192

193 expected_file = plate_p / self.METADATA_FILENAME

194 if self.filemanager.exists(str(expected_file), Backend.DISK.value): 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true

195 return expected_file

196

197 # Fallback: recursive search

198 try:

199 if found_files := self.filemanager.find_file_recursive(plate_p, self.METADATA_FILENAME, Backend.DISK.value): 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true

200 if isinstance(found_files, list):

201 # Prioritize root location, then first found

202 return next((Path(f) for f in found_files if Path(f).parent == plate_p), Path(found_files[0]))

203 return Path(found_files)

204 except Exception as e:

205 logger.error(f"Error searching for {self.METADATA_FILENAME} in {plate_path}: {e}")

206

207 return None

208

209

210 def get_grid_dimensions(self, plate_path: Union[str, Path], context: Optional[Any] = None) -> Tuple[int, int]:

211 """Get grid dimensions from OpenHCS metadata."""

212 dims = self._load_metadata(plate_path).get(FIELDS.GRID_DIMENSIONS)

213 if not (isinstance(dims, list) and len(dims) == 2 and all(isinstance(d, int) for d in dims)):

214 raise ValueError(f"'{FIELDS.GRID_DIMENSIONS}' must be a list of two integers in {self.METADATA_FILENAME}")

215 return tuple(dims)

216

217 def get_pixel_size(self, plate_path: Union[str, Path], context: Optional[Any] = None) -> float:

218 """Get pixel size from OpenHCS metadata."""

219 pixel_size = self._load_metadata(plate_path).get(FIELDS.PIXEL_SIZE)

220 if not isinstance(pixel_size, (float, int)):

221 raise ValueError(f"'{FIELDS.PIXEL_SIZE}' must be a number in {self.METADATA_FILENAME}")

222 return float(pixel_size)

223

224 def get_source_filename_parser_name(self, plate_path: Union[str, Path]) -> str:

225 """Get source filename parser name from OpenHCS metadata."""

226 parser_name = self._load_metadata(plate_path).get(FIELDS.SOURCE_FILENAME_PARSER_NAME)

227 if not (isinstance(parser_name, str) and parser_name):

228 raise ValueError(f"'{FIELDS.SOURCE_FILENAME_PARSER_NAME}' must be a non-empty string in {self.METADATA_FILENAME}")

229 return parser_name

230

231 def get_image_files(self, plate_path: Union[str, Path]) -> List[str]:

232 """Get image files list from OpenHCS metadata."""

233 image_files = self._load_metadata(plate_path).get(FIELDS.IMAGE_FILES)

234 if not (isinstance(image_files, list) and all(isinstance(f, str) for f in image_files)):

235 raise ValueError(f"'{FIELDS.IMAGE_FILES}' must be a list of strings in {self.METADATA_FILENAME}")

236 return image_files

237

238 # Optional metadata getters

239 def _get_optional_metadata_dict(self, plate_path: Union[str, Path], key: str) -> Optional[Dict[str, str]]:

240 """Helper to get optional dictionary metadata."""

241 value = self._load_metadata(plate_path).get(key)

242 return {str(k): str(v) for k, v in value.items()} if isinstance(value, dict) else None

243

244 def get_channel_values(self, plate_path: Union[str, Path], context: Optional[Any] = None) -> Optional[Dict[str, Optional[str]]]:

245 return self._get_optional_metadata_dict(plate_path, FIELDS.CHANNELS)

246

247 def get_well_values(self, plate_path: Union[str, Path], context: Optional[Any] = None) -> Optional[Dict[str, Optional[str]]]:

248 return self._get_optional_metadata_dict(plate_path, FIELDS.WELLS)

249

250 def get_site_values(self, plate_path: Union[str, Path], context: Optional[Any] = None) -> Optional[Dict[str, Optional[str]]]:

251 return self._get_optional_metadata_dict(plate_path, FIELDS.SITES)

252

253 def get_z_index_values(self, plate_path: Union[str, Path], context: Optional[Any] = None) -> Optional[Dict[str, Optional[str]]]:

254 return self._get_optional_metadata_dict(plate_path, FIELDS.Z_INDEXES)

255

256 def get_objective_values(self, plate_path: Union[str, Path], context: Optional[Any] = None) -> Optional[Dict[str, Any]]:

257 """Get objective lens information if available."""

258 return self._get_optional_metadata_dict(plate_path, FIELDS.OBJECTIVES)

259

260 def get_plate_acquisition_datetime(self, plate_path: Union[str, Path], context: Optional[Any] = None) -> Optional[str]:

261 """Get plate acquisition datetime if available."""

262 return self._get_optional_metadata_str(plate_path, FIELDS.ACQUISITION_DATETIME)

263

264 def get_plate_name(self, plate_path: Union[str, Path], context: Optional[Any] = None) -> Optional[str]:

265 """Get plate name if available."""

266 return self._get_optional_metadata_str(plate_path, FIELDS.PLATE_NAME)

267

268 def _get_optional_metadata_str(self, plate_path: Union[str, Path], field: str) -> Optional[str]:

269 """Helper to get optional string metadata field."""

270 value = self._load_metadata(plate_path).get(field)

271 return value if isinstance(value, str) and value else None

272

273 def get_available_backends(self, input_dir: Union[str, Path]) -> Dict[str, bool]:

274 """

275 Get available storage backends for the input directory.

276

277 This method resolves the plate root from the input directory,

278 loads the OpenHCS metadata, and returns the available backends.

279

280 Args:

281 input_dir: Path to the input directory (may be plate root or subdirectory)

282

283 Returns:

284 Dictionary mapping backend names to availability (e.g., {"disk": True, "zarr": False})

285

286 Raises:

287 MetadataNotFoundError: If metadata file cannot be found or parsed

288 """

289 # Resolve plate root from input directory

290 plate_root = self._resolve_plate_root(input_dir)

291

292 # Load metadata using existing infrastructure

293 metadata = self._load_metadata(plate_root)

294

295 # Extract available backends, defaulting to empty dict if not present

296 available_backends = metadata.get(FIELDS.AVAILABLE_BACKENDS, {})

297

298 if not isinstance(available_backends, dict):

299 logger.warning(f"Invalid available_backends format in metadata: {available_backends}")

300 return {}

301

302 return available_backends

303

304 def _resolve_plate_root(self, input_dir: Union[str, Path]) -> Path:

305 """

306 Resolve the plate root directory from an input directory.

307

308 The input directory may be the plate root itself or a subdirectory.

309 This method walks up the directory tree to find the directory containing

310 the OpenHCS metadata file.

311

312 Args:

313 input_dir: Path to resolve

314

315 Returns:

316 Path to the plate root directory

317

318 Raises:

319 MetadataNotFoundError: If no metadata file is found

320 """

321 current_path = Path(input_dir)

322

323 # Walk up the directory tree looking for metadata file

324 for path in [current_path] + list(current_path.parents):

325 metadata_file = path / self.METADATA_FILENAME

326 if self.filemanager.exists(str(metadata_file), Backend.DISK.value):

327 return path

328

329 # If not found, raise an error

330 raise MetadataNotFoundError(

331 f"Could not find {self.METADATA_FILENAME} in {input_dir} or any parent directory"

332 )

333

334 def update_available_backends(self, plate_path: Union[str, Path], available_backends: Dict[str, bool]) -> None:

335 """Update available storage backends in metadata and save to disk."""

336 metadata_file_path = get_metadata_path(plate_path)

337

338 try:

339 self.atomic_writer.update_available_backends(metadata_file_path, available_backends)

340 # Clear cache to force reload on next access

341 self._metadata_cache = None

342 self._plate_path_cache = None

343 logger.info(f"Updated available backends to {available_backends} in {metadata_file_path}")

344 except MetadataWriteError as e:

345 raise ValueError(f"Failed to update available backends: {e}") from e

346

347

348@dataclass(frozen=True)

349class OpenHCSMetadata:

350 """

351 Declarative OpenHCS metadata structure.

352

353 Fail-loud: All fields are required, no defaults, no fallbacks.

354 """

355 microscope_handler_name: str

356 source_filename_parser_name: str

357 grid_dimensions: List[int]

358 pixel_size: float

359 image_files: List[str]

360 channels: Optional[Dict[str, str]]

361 wells: Optional[Dict[str, str]]

362 sites: Optional[Dict[str, str]]

363 z_indexes: Optional[Dict[str, str]]

364 available_backends: Dict[str, bool]

365 main: Optional[bool] = None # Indicates if this subdirectory is the primary/input subdirectory

366

367

368@dataclass(frozen=True)

369class SubdirectoryKeyedMetadata:

370 """

371 Subdirectory-keyed metadata structure for OpenHCS.

372

373 Organizes metadata by subdirectory to prevent conflicts when multiple

374 steps write to the same plate folder with different subdirectories.

375

376 Structure: {subdirectory_name: OpenHCSMetadata}

377 """

378 subdirectories: Dict[str, OpenHCSMetadata]

379

380 def get_subdirectory_metadata(self, sub_dir: str) -> Optional[OpenHCSMetadata]:

381 """Get metadata for specific subdirectory."""

382 return self.subdirectories.get(sub_dir)

383

384 def add_subdirectory_metadata(self, sub_dir: str, metadata: OpenHCSMetadata) -> 'SubdirectoryKeyedMetadata':

385 """Add or update metadata for subdirectory (immutable operation)."""

386 new_subdirs = {**self.subdirectories, sub_dir: metadata}

387 return SubdirectoryKeyedMetadata(subdirectories=new_subdirs)

388

389 @classmethod

390 def from_single_metadata(cls, sub_dir: str, metadata: OpenHCSMetadata) -> 'SubdirectoryKeyedMetadata':

391 """Create from single OpenHCSMetadata (migration helper)."""

392 return cls(subdirectories={sub_dir: metadata})

393

394 @classmethod

395 def from_legacy_dict(cls, legacy_dict: Dict[str, Any], default_sub_dir: str = FIELDS.DEFAULT_SUBDIRECTORY_LEGACY) -> 'SubdirectoryKeyedMetadata':

396 """Create from legacy single-subdirectory metadata dict."""

397 return cls.from_single_metadata(default_sub_dir, OpenHCSMetadata(**legacy_dict))

398

399

400class OpenHCSMetadataGenerator:

401 """

402 Generator for OpenHCS metadata files.

403

404 Handles creation of openhcs_metadata.json files for processed plates,

405 extracting information from processing context and output directories.

406

407 Design principle: Generate metadata that accurately reflects what exists on disk

408 after processing, not what was originally intended or what the source contained.

409 """

410

411 def __init__(self, filemanager: FileManager):

412 """

413 Initialize the metadata generator.

414

415 Args:

416 filemanager: FileManager instance for file operations

417 """

418 self.filemanager = filemanager

419 self.atomic_writer = AtomicMetadataWriter()

420 self.logger = logging.getLogger(__name__)

421

422 def create_metadata(

423 self,

424 context: 'ProcessingContext',

425 output_dir: str,

426 write_backend: str,

427 is_main: bool = False,

428 plate_root: str = None,

429 sub_dir: str = None

430 ) -> None:

431 """Create or update subdirectory-keyed OpenHCS metadata file."""

432 plate_root_path = Path(plate_root)

433 metadata_path = get_metadata_path(plate_root_path)

434

435 current_metadata = self._extract_metadata_from_disk_state(context, output_dir, write_backend, is_main, sub_dir)

436 metadata_dict = asdict(current_metadata)

437

438 self.atomic_writer.update_subdirectory_metadata(metadata_path, sub_dir, metadata_dict)

439

440

441

442 def _extract_metadata_from_disk_state(self, context: 'ProcessingContext', output_dir: str, write_backend: str, is_main: bool, sub_dir: str) -> OpenHCSMetadata:

443 """Extract metadata reflecting current disk state after processing."""

444 handler = context.microscope_handler

445 cache = context.metadata_cache or {}

446

447 actual_files = self.filemanager.list_image_files(output_dir, write_backend)

448 relative_files = [f"{sub_dir}/{Path(f).name}" for f in actual_files]

449

450 return OpenHCSMetadata(

451 microscope_handler_name=handler.microscope_type,

452 source_filename_parser_name=handler.parser.__class__.__name__,

453 grid_dimensions=handler.metadata_handler._get_with_fallback('get_grid_dimensions', context.input_dir),

454 pixel_size=handler.metadata_handler._get_with_fallback('get_pixel_size', context.input_dir),

455 image_files=relative_files,

456 channels=cache.get(GroupBy.CHANNEL),

457 wells=cache.get(GroupBy.WELL),

458 sites=cache.get(GroupBy.SITE),

459 z_indexes=cache.get(GroupBy.Z_INDEX),

460 available_backends={write_backend: True},

461 main=is_main if is_main else None

462 )

468from openhcs.microscopes.microscope_base import MicroscopeHandler

469from openhcs.microscopes.microscope_interfaces import FilenameParser

470

471

472class OpenHCSMicroscopeHandler(MicroscopeHandler):

473 """

474 MicroscopeHandler for OpenHCS pre-processed format.

475

476 This handler reads plates that have been standardized, with metadata

477 provided in an 'openhcs_metadata.json' file. It dynamically loads the

478 appropriate FilenameParser based on the metadata.

479 """

480

481 # Class attributes for automatic registration

482 _microscope_type = FIELDS.MICROSCOPE_TYPE # Override automatic naming

483 _metadata_handler_class = None # Set after class definition

484

485 def __init__(self, filemanager: FileManager, pattern_format: Optional[str] = None):

486 """

487 Initialize the OpenHCSMicroscopeHandler.

488

489 Args:

490 filemanager: FileManager instance for file operations.

491 pattern_format: Optional pattern format string, passed to dynamically loaded parser.

492 """

493 self.filemanager = filemanager

494 self.metadata_handler = OpenHCSMetadataHandler(filemanager)

495 self._parser: Optional[FilenameParser] = None

496 self.plate_folder: Optional[Path] = None # Will be set by factory or post_workspace

497 self.pattern_format = pattern_format # Store for parser instantiation

498

499 # Initialize super with a None parser. The actual parser is loaded dynamically.

500 # The `parser` property will handle on-demand loading.

501 super().__init__(parser=None, metadata_handler=self.metadata_handler)

502

503 def _load_and_get_parser(self) -> FilenameParser:

504 """

505 Ensures the dynamic filename parser is loaded based on metadata from plate_folder.

506 This method requires self.plate_folder to be set.

507 """

508 if self._parser is None:

509 if self.plate_folder is None:

510 raise RuntimeError(

511 "OpenHCSHandler: plate_folder not set. Cannot determine and load the source filename parser."

512 )

513

514 parser_name = self.metadata_handler.get_source_filename_parser_name(self.plate_folder)

515 available_parsers = _get_available_filename_parsers()

516 ParserClass = available_parsers.get(parser_name)

517

518 if not ParserClass:

519 raise ValueError(

520 f"Unknown or unsupported filename parser '{parser_name}' specified in "

521 f"{OpenHCSMetadataHandler.METADATA_FILENAME} for plate {self.plate_folder}. "

522 f"Available parsers: {list(available_parsers.keys())}"

523 )

524

525 try:

526 # Attempt to instantiate with filemanager and pattern_format

527 self._parser = ParserClass(filemanager=self.filemanager, pattern_format=self.pattern_format)

528 logger.info(f"OpenHCSHandler for plate {self.plate_folder} loaded source filename parser: {parser_name} with filemanager and pattern_format.")

529 except TypeError:

530 try:

531 # Attempt with filemanager only

532 self._parser = ParserClass(filemanager=self.filemanager)

533 logger.info(f"OpenHCSHandler for plate {self.plate_folder} loaded source filename parser: {parser_name} with filemanager.")

534 except TypeError:

535 # Attempt with default constructor

536 self._parser = ParserClass()

537 logger.info(f"OpenHCSHandler for plate {self.plate_folder} loaded source filename parser: {parser_name} with default constructor.")

538

539 return self._parser

540

541 @property

542 def parser(self) -> FilenameParser:

543 """

544 Provides the dynamically loaded FilenameParser.

545 The actual parser is determined from the 'openhcs_metadata.json' file.

546 Requires `self.plate_folder` to be set prior to first access.

547 """

548 # If plate_folder is not set here, it means it wasn't set by the factory

549 # nor by a method like post_workspace before parser access.

550 if self.plate_folder is None:

551 # This situation should ideally be avoided by ensuring plate_folder is set appropriately.

552 raise RuntimeError("OpenHCSHandler: plate_folder must be set before accessing the parser property.")

553

554 return self._load_and_get_parser()

555

556 @parser.setter

557 def parser(self, value: Optional[FilenameParser]):

558 """

559 Allows setting the parser instance. Used by base class __init__ if it attempts to set it,

560 though our dynamic loading means we primarily manage it internally.

561 """

562 # If the base class __init__ tries to set it (e.g. to None as we passed),

563 # this setter will be called. We want our dynamic loading to take precedence.

564 # If an actual parser is passed, we could use it, but it would override dynamic logic.

565 # For now, if None is passed (from our super call), _parser remains None until dynamically loaded.

566 # If a specific parser is passed, it will be set.

567 if value is not None:

568 logger.debug(f"OpenHCSMicroscopeHandler.parser being explicitly set to: {type(value).__name__}")

569 self._parser = value

570

571

572 @property

573 def common_dirs(self) -> List[str]:

574 """

575 OpenHCS format expects images in the root of the plate folder.

576 No common subdirectories are applicable.

577 """

578 return []

579

580 @property

581 def microscope_type(self) -> str:

582 """Microscope type identifier (for interface enforcement only)."""

583 return FIELDS.MICROSCOPE_TYPE

584

585 @property

586 def metadata_handler_class(self) -> Type[MetadataHandler]:

587 """Metadata handler class (for interface enforcement only)."""

588 return OpenHCSMetadataHandler

589

590 @property

591 def compatible_backends(self) -> List[Backend]:

592 """

593 OpenHCS is compatible with ZARR (preferred) and DISK (fallback) backends.

594

595 ZARR: Advanced chunked storage for large datasets (preferred)

596 DISK: Standard file operations for compatibility (fallback)

597 """

598 return [Backend.ZARR, Backend.DISK]

599

600 def get_available_backends(self, plate_path: Union[str, Path]) -> List[Backend]:

601 """

602 Get available storage backends for OpenHCS plates.

603

604 OpenHCS plates can support multiple backends based on what actually exists on disk.

605 This method checks the metadata to see what backends are actually available.

606 """

607 try:

608 # Get available backends from metadata as Dict[str, bool]

609 available_backends_dict = self.metadata_handler.get_available_backends(plate_path)

610

611 # Convert to List[Backend] by filtering compatible backends that are available

612 available_backends = []

613 for backend_enum in self.compatible_backends:

614 backend_name = backend_enum.value

615 if available_backends_dict.get(backend_name, False):

616 available_backends.append(backend_enum)

617

618 # If no backends are available from metadata, fall back to compatible backends

619 # This handles cases where metadata might not have the available_backends field

620 if not available_backends:

621 logger.warning(f"No available backends found in metadata for {plate_path}, using all compatible backends")

622 return self.compatible_backends

623

624 return available_backends

625

626 except Exception as e:

627 logger.warning(f"Failed to get available backends from metadata for {plate_path}: {e}")

628 # Fall back to all compatible backends if metadata reading fails

629 return self.compatible_backends

630

631 def initialize_workspace(self, plate_path: Path, workspace_path: Optional[Path], filemanager: FileManager) -> Path:

632 """

633 OpenHCS format doesn't need workspace - determines the correct input subdirectory from metadata.

634

635 Args:

636 plate_path: Path to the original plate directory

637 workspace_path: Optional workspace path (ignored for OpenHCS)

638 filemanager: FileManager instance for file operations

639

640 Returns:

641 Path to the main subdirectory containing input images (e.g., plate_path/images)

642 """

643 logger.info(f"OpenHCS format: Determining input subdirectory from metadata in {plate_path}")

644

645 # Set plate_folder for this handler

646 self.plate_folder = plate_path

647 logger.debug(f"OpenHCSHandler: plate_folder set to {self.plate_folder}")

648

649 # Determine the main subdirectory from metadata - fail-loud on errors

650 main_subdir = self.metadata_handler.determine_main_subdirectory(plate_path)

651 input_dir = plate_path / main_subdir

652

653 # Verify the subdirectory exists - fail-loud if missing

654 if not filemanager.is_dir(str(input_dir), Backend.DISK.value):

655 raise FileNotFoundError(

656 f"Main subdirectory '{main_subdir}' does not exist at {input_dir}. "

657 f"Expected directory structure: {plate_path}/{main_subdir}/"

658 )

659

660 logger.info(f"OpenHCS input directory determined: {input_dir} (subdirectory: {main_subdir})")

661 return input_dir

662

663 def _prepare_workspace(self, workspace_path: Path, filemanager: FileManager) -> Path:

664 """

665 OpenHCS format assumes the workspace is already prepared (e.g., flat structure).

666 This method is a no-op.

667 Args:

668 workspace_path: Path to the symlinked workspace.

669 filemanager: FileManager instance for file operations.

670 Returns:

671 The original workspace_path.

672 """

673 logger.info(f"OpenHCSHandler._prepare_workspace: No preparation needed for {workspace_path} as it's pre-processed.")

674 # Ensure plate_folder is set if this is the first relevant operation knowing the path

675 if self.plate_folder is None:

676 self.plate_folder = Path(workspace_path)

677 logger.debug(f"OpenHCSHandler: plate_folder set to {self.plate_folder} during _prepare_workspace.")

678 return workspace_path

679

680 def post_workspace(self, workspace_path: Union[str, Path], filemanager: FileManager, width: int = 3):

681 """

682 Hook called after workspace symlink creation.

683 For OpenHCS, this ensures the plate_folder is set (if not already) which allows

684 the parser to be loaded using this workspace_path. It then calls the base

685 implementation which handles filename normalization using the loaded parser.

686 """

687 current_plate_folder = Path(workspace_path)

688 if self.plate_folder is None:

689 logger.info(f"OpenHCSHandler.post_workspace: Setting plate_folder to {current_plate_folder}.")

690 self.plate_folder = current_plate_folder

691 self._parser = None # Reset parser if plate_folder changes or is set for the first time

692 elif self.plate_folder != current_plate_folder:

693 logger.warning(

694 f"OpenHCSHandler.post_workspace: plate_folder was {self.plate_folder}, "

695 f"now processing {current_plate_folder}. Re-initializing parser."

696 )

697 self.plate_folder = current_plate_folder

698 self._parser = None # Force re-initialization for the new path

699

700 # Accessing self.parser here will trigger _load_and_get_parser() if not already loaded

701 _ = self.parser

702

703 logger.info(f"OpenHCSHandler (plate: {self.plate_folder}): Files are expected to be pre-normalized. "

704 "Superclass post_workspace will run with the dynamically loaded parser.")

705 return super().post_workspace(workspace_path, filemanager, width)

706

707 # The following methods from MicroscopeHandler delegate to `self.parser`.

708 # The `parser` property will ensure the correct, dynamically loaded parser is used.

709 # No explicit override is needed for them unless special behavior for OpenHCS is required

710 # beyond what the dynamically loaded original parser provides.

711 # - parse_filename(self, filename: str)

712 # - construct_filename(self, well: str, ...)

713 # - auto_detect_patterns(self, folder_path: Union[str, Path], ...)

714 # - path_list_from_pattern(self, directory: Union[str, Path], ...)

715

716 # Metadata handling methods are delegated to `self.metadata_handler` by the base class.

717 # - find_metadata_file(self, plate_path: Union[str, Path])

718 # - get_grid_dimensions(self, plate_path: Union[str, Path])

719 # - get_pixel_size(self, plate_path: Union[str, Path])

720 # These will use our OpenHCSMetadataHandler correctly.

721

722

723# Set metadata handler class after class definition for automatic registration

724from openhcs.microscopes.microscope_base import register_metadata_handler

725OpenHCSMicroscopeHandler._metadata_handler_class = OpenHCSMetadataHandler

726register_metadata_handler(OpenHCSMicroscopeHandler, OpenHCSMetadataHandler)