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

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, AllComponents, 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

305

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

307 """

308 Resolve the plate root directory from an input directory.

309

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

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

312 the OpenHCS metadata file.

313

314 Args:

315 input_dir: Path to resolve

316

317 Returns:

318 Path to the plate root directory

319

320 Raises:

321 MetadataNotFoundError: If no metadata file is found

322 """

323 current_path = Path(input_dir)

324

325 # Walk up the directory tree looking for metadata file

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

327 metadata_file = path / self.METADATA_FILENAME

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

329 return path

330

331 # If not found, raise an error

332 raise MetadataNotFoundError(

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

334 )

335

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

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

338 metadata_file_path = get_metadata_path(plate_path)

339

340 try:

341 self.atomic_writer.update_available_backends(metadata_file_path, available_backends)

342 # Clear cache to force reload on next access

343 self._metadata_cache = None

344 self._plate_path_cache = None

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

346 except MetadataWriteError as e:

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

348

349

350@dataclass(frozen=True)

351class OpenHCSMetadata:

352 """

353 Declarative OpenHCS metadata structure.

354

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

356 """

357 microscope_handler_name: str

358 source_filename_parser_name: str

359 grid_dimensions: List[int]

360 pixel_size: float

361 image_files: List[str]

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

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

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

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

366 available_backends: Dict[str, bool]

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

368

369

370@dataclass(frozen=True)

371class SubdirectoryKeyedMetadata:

372 """

373 Subdirectory-keyed metadata structure for OpenHCS.

374

375 Organizes metadata by subdirectory to prevent conflicts when multiple

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

377

378 Structure: {subdirectory_name: OpenHCSMetadata}

379 """

380 subdirectories: Dict[str, OpenHCSMetadata]

381

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

383 """Get metadata for specific subdirectory."""

384 return self.subdirectories.get(sub_dir)

385

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

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

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

389 return SubdirectoryKeyedMetadata(subdirectories=new_subdirs)

390

391 @classmethod

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

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

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

395

396 @classmethod

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

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

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

400

401

402class OpenHCSMetadataGenerator:

403 """

404 Generator for OpenHCS metadata files.

405

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

407 extracting information from processing context and output directories.

408

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

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

411 """

412

413 def __init__(self, filemanager: FileManager):

414 """

415 Initialize the metadata generator.

416

417 Args:

418 filemanager: FileManager instance for file operations

419 """

420 self.filemanager = filemanager

421 self.atomic_writer = AtomicMetadataWriter()

422 self.logger = logging.getLogger(__name__)

423

424 def create_metadata(

425 self,

426 context: 'ProcessingContext',

427 output_dir: str,

428 write_backend: str,

429 is_main: bool = False,

430 plate_root: str = None,

431 sub_dir: str = None

432 ) -> None:

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

434 plate_root_path = Path(plate_root)

435 metadata_path = get_metadata_path(plate_root_path)

436

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

438 metadata_dict = asdict(current_metadata)

439

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

441

442

443

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

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

446 handler = context.microscope_handler

447 cache = context.metadata_cache or {}

448

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

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

451

452 return OpenHCSMetadata(

453 microscope_handler_name=handler.microscope_type,

454 source_filename_parser_name=handler.parser.__class__.__name__,

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

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

457 image_files=relative_files,

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

459 wells=cache.get(AllComponents.WELL), # Use AllComponents for multiprocessing axis

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

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

462 available_backends={write_backend: True},

463 main=is_main if is_main else None

464 )

470from openhcs.microscopes.microscope_base import MicroscopeHandler

471from openhcs.microscopes.microscope_interfaces import FilenameParser

472

473

474class OpenHCSMicroscopeHandler(MicroscopeHandler):

475 """

476 MicroscopeHandler for OpenHCS pre-processed format.

477

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

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

480 appropriate FilenameParser based on the metadata.

481 """

482

483 # Class attributes for automatic registration

484 _microscope_type = FIELDS.MICROSCOPE_TYPE # Override automatic naming

485 _metadata_handler_class = None # Set after class definition

486

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

488 """

489 Initialize the OpenHCSMicroscopeHandler.

490

491 Args:

492 filemanager: FileManager instance for file operations.

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

494 """

495 self.filemanager = filemanager

496 self.metadata_handler = OpenHCSMetadataHandler(filemanager)

497 self._parser: Optional[FilenameParser] = None

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

499 self.pattern_format = pattern_format # Store for parser instantiation

500

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

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

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

504

505 def _load_and_get_parser(self) -> FilenameParser:

506 """

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

508 This method requires self.plate_folder to be set.

509 """

510 if self._parser is None:

511 if self.plate_folder is None:

512 raise RuntimeError(

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

514 )

515

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

517 available_parsers = _get_available_filename_parsers()

518 ParserClass = available_parsers.get(parser_name)

519

520 if not ParserClass:

521 raise ValueError(

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

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

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

525 )

526

527 try:

528 # Attempt to instantiate with filemanager and pattern_format

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

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

531 except TypeError:

532 try:

533 # Attempt with filemanager only

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

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

536 except TypeError:

537 # Attempt with default constructor

538 self._parser = ParserClass()

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

540

541 return self._parser

542

543 @property

544 def parser(self) -> FilenameParser:

545 """

546 Provides the dynamically loaded FilenameParser.

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

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

549 """

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

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

552 if self.plate_folder is None:

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

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

555

556 return self._load_and_get_parser()

557

558 @parser.setter

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

560 """

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

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

563 """

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

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

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

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

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

569 if value is not None:

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

571 self._parser = value

572

573

574 @property

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

576 """

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

578 No common subdirectories are applicable.

579 """

580 return []

581

582 @property

583 def microscope_type(self) -> str:

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

585 return FIELDS.MICROSCOPE_TYPE

586

587 @property

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

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

590 return OpenHCSMetadataHandler

591

592 @property

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

594 """

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

596

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

598 DISK: Standard file operations for compatibility (fallback)

599 """

600 return [Backend.ZARR, Backend.DISK]

601

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

603 """

604 Get available storage backends for OpenHCS plates.

605

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

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

608 """

609 try:

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

611 available_backends_dict = self.metadata_handler.get_available_backends(plate_path)

612

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

614 available_backends = []

615 for backend_enum in self.compatible_backends:

616 backend_name = backend_enum.value

617 if available_backends_dict.get(backend_name, False):

618 available_backends.append(backend_enum)

619

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

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

622 if not available_backends:

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

624 return self.compatible_backends

625

626 return available_backends

627

628 except Exception as e:

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

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

631 return self.compatible_backends

632

633 def get_primary_backend(self, plate_path: Union[str, Path]) -> str:

634 """

635 Get the primary backend name for OpenHCS plates.

636

637 Uses metadata-based detection to determine the primary backend.

638 """

639 available_backends_dict = self.metadata_handler.get_available_backends(plate_path)

640 return next(iter(available_backends_dict.keys()))

641

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

643 """

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

645

646 Args:

647 plate_path: Path to the original plate directory

648 workspace_path: Optional workspace path (ignored for OpenHCS)

649 filemanager: FileManager instance for file operations

650

651 Returns:

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

653 """

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

655

656 # Set plate_folder for this handler

657 self.plate_folder = plate_path

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

659

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

661 main_subdir = self.metadata_handler.determine_main_subdirectory(plate_path)

662 input_dir = plate_path / main_subdir

663

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

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

666 raise FileNotFoundError(

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

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

669 )

670

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

672 return input_dir

673

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

675 """

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

677 This method is a no-op.

678 Args:

679 workspace_path: Path to the symlinked workspace.

680 filemanager: FileManager instance for file operations.

681 Returns:

682 The original workspace_path.

683 """

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

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

686 if self.plate_folder is None:

687 self.plate_folder = Path(workspace_path)

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

689 return workspace_path

690

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

692 """

693 Hook called after workspace symlink creation.

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

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

696 implementation which handles filename normalization using the loaded parser.

697 """

698 current_plate_folder = Path(workspace_path)

699 if self.plate_folder is None:

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

701 self.plate_folder = current_plate_folder

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

703 elif self.plate_folder != current_plate_folder:

704 logger.warning(

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

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

707 )

708 self.plate_folder = current_plate_folder

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

710

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

712 _ = self.parser

713

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

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

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

717

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

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

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

721 # beyond what the dynamically loaded original parser provides.

722 # - parse_filename(self, filename: str)

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

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

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

726

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

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

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

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

731 # These will use our OpenHCSMetadataHandler correctly.

732

733

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

735from openhcs.microscopes.microscope_base import register_metadata_handler

736OpenHCSMicroscopeHandler._metadata_handler_class = OpenHCSMetadataHandler

737register_metadata_handler(OpenHCSMicroscopeHandler, OpenHCSMetadataHandler)