Coverage for openhcs/microscopes/omero.py: 2.4%
168 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 02:09 +0000
1"""
2OMERO microscope implementations for openhcs.
4This module provides concrete implementations of FilenameParser and MetadataHandler
5for OMERO microscopes using native OMERO metadata.
6"""
8import logging
9import re
10from pathlib import Path
11from typing import Any, Dict, List, Optional, Tuple, Type, Union
13import omero.model
15from openhcs.constants.constants import Backend
16from openhcs.io.filemanager import FileManager
17from openhcs.microscopes.microscope_base import MicroscopeHandler
18from openhcs.microscopes.microscope_interfaces import FilenameParser, MetadataHandler
20logger = logging.getLogger(__name__)
23class OMEROMetadataHandler(MetadataHandler):
24 """
25 Metadata handler that queries OMERO API for native metadata with caching.
27 Does NOT read OpenHCS metadata files - uses OMERO's native metadata.
28 Caches metadata per plate to avoid repeated OMERO queries.
30 Retrieves OMERO connection from backend registry via filemanager (Option B pattern).
31 """
33 def __init__(self, filemanager: FileManager):
34 super().__init__()
35 self.filemanager = filemanager
36 self._metadata_cache: Dict[int, Dict[str, Dict[int, str]]] = {} # plate_id → metadata
38 def _get_omero_conn(self):
39 """
40 Get OMERO connection from backend registry.
42 Retrieves the connection from the OMERO backend instance in the registry,
43 using the backend's _get_connection() method which handles lazy connection
44 management for multiprocessing compatibility.
46 Returns:
47 BlitzGateway connection instance
49 Raises:
50 RuntimeError: If OMERO backend cannot provide a connection
51 """
52 # Get OMERO backend from registry (via filemanager)
53 omero_backend = self.filemanager.registry[Backend.OMERO_LOCAL.value]
55 # Use the backend's connection retrieval method
56 # This handles lazy connection management and multiprocessing
57 try:
58 conn = omero_backend._get_connection()
59 return conn
60 except Exception as e:
61 raise RuntimeError(
62 f"Failed to get OMERO connection from backend: {e}. "
63 "Ensure OMERO backend is properly initialized with connection."
64 ) from e
66 def _load_plate_metadata(self, plate_id: int) -> Dict[str, Dict[int, str]]:
67 """
68 Load all metadata for a plate in one OMERO query (cached).
70 Queries OMERO once per plate and caches results.
71 Subsequent calls return cached data.
72 """
73 if plate_id in self._metadata_cache:
74 return self._metadata_cache[plate_id]
76 # Get connection from backend registry
77 conn = self._get_omero_conn()
79 plate = conn.getObject("Plate", plate_id)
80 if not plate:
81 raise ValueError(f"OMERO Plate not found: {plate_id}")
83 # Query OMERO once for all metadata
84 metadata = {}
86 # Get metadata from all wells to ensure complete coverage
87 # (different wells might have different dimensions)
88 all_channels = {}
89 max_z = 0
90 max_t = 0
92 for well in plate.listChildren():
93 well_sample = well.getWellSample(0)
94 if well_sample:
95 image = well_sample.getImage()
97 # Collect channel names
98 for i, channel in enumerate(image.getChannels()):
99 channel_idx = i + 1
100 if channel_idx not in all_channels:
101 all_channels[channel_idx] = channel.getLabel() or f"Channel {channel_idx}"
103 # Track max dimensions
104 max_z = max(max_z, image.getSizeZ())
105 max_t = max(max_t, image.getSizeT())
107 # Build metadata dict
108 metadata['channel'] = all_channels
109 metadata['z_index'] = {z + 1: f"Z{z + 1}" for z in range(max_z)}
110 metadata['timepoint'] = {t + 1: f"T{t + 1}" for t in range(max_t)}
112 # Cache it
113 self._metadata_cache[plate_id] = metadata
114 return metadata
116 def find_metadata_file(self, plate_path: Union[str, Path]) -> Optional[Path]:
117 """
118 OMERO doesn't use metadata files, but detects based on /omero/ path pattern.
120 Returns the plate_path itself if it matches the OMERO virtual path pattern,
121 otherwise returns None.
122 """
123 plate_path = Path(plate_path)
124 # OMERO plates use virtual paths like /omero/plate_123
125 if str(plate_path).startswith('/omero/plate_'):
126 return plate_path
127 return None
129 def _extract_plate_id(self, plate_path: Union[str, Path, int]) -> int:
130 """
131 Extract plate_id from various path formats.
133 Handles:
134 - int: plate_id directly
135 - Path('/omero/plate_57'): extracts 57 from 'plate_57'
136 - Path('/omero/plate_57_outputs'): extracts 57 from 'plate_57_outputs'
138 Args:
139 plate_path: Plate identifier (int or Path)
141 Returns:
142 Plate ID as integer
144 Raises:
145 ValueError: If path format is invalid
146 """
147 if isinstance(plate_path, int):
148 return plate_path
150 # Extract from path: /omero/plate_57 or /omero/plate_57_outputs
151 path_str = str(Path(plate_path).name) # Get just the filename part
153 # Match 'plate_<id>' or 'plate_<id>_<suffix>'
154 match = re.match(r'plate_(\d+)', path_str)
155 if not match:
156 raise ValueError(f"Invalid OMERO path format: {plate_path}. Expected /omero/plate_<id> or /omero/plate_<id>_<suffix>")
158 return int(match.group(1))
160 def get_channel_values(self, plate_path: Union[str, Path, int]) -> Dict[int, str]:
161 """Get channel metadata (cached)."""
162 plate_id = self._extract_plate_id(plate_path)
163 metadata = self._load_plate_metadata(plate_id)
164 return metadata.get('channel', {})
166 def get_z_index_values(self, plate_path: Union[str, Path, int]) -> Dict[int, str]:
167 """Get Z-index metadata (cached)."""
168 plate_id = self._extract_plate_id(plate_path)
169 metadata = self._load_plate_metadata(plate_id)
170 return metadata.get('z_index', {})
172 def get_timepoint_values(self, plate_path: Union[str, Path, int]) -> Dict[int, str]:
173 """Get timepoint metadata (cached)."""
174 plate_id = self._extract_plate_id(plate_path)
175 metadata = self._load_plate_metadata(plate_id)
176 return metadata.get('timepoint', {})
178 # Other component methods return empty dicts (not applicable for OMERO)
179 def get_site_values(self, plate_path: Union[str, Path, int]) -> Dict[int, str]:
180 return {}
182 def get_well_values(self, plate_path: Union[str, Path, int]) -> Dict[str, str]:
183 return {}
185 def get_grid_dimensions(self, plate_path: Union[str, Path, int]) -> Tuple[int, int]:
186 """
187 Extract grid dimensions from OMERO plate metadata.
189 Grid dimensions should be stored in the plate's MapAnnotation under
190 the key "openhcs.grid_dimensions" as "rows,cols" (e.g., "2,2").
192 Returns:
193 Tuple of (rows, cols) representing the grid dimensions
194 """
195 plate_id = self._extract_plate_id(plate_path)
197 try:
198 conn = self._get_omero_conn()
199 plate = conn.getObject("Plate", plate_id)
201 if not plate:
202 logger.warning(f"Plate {plate_id} not found, using fallback grid_dimensions")
203 return self.FALLBACK_VALUES.get('grid_dimensions', (1, 1))
205 # Try to get grid dimensions from MapAnnotation
206 for ann in plate.listAnnotations():
207 if ann.OMERO_TYPE == omero.model.MapAnnotationI:
208 if ann.getNs() == "openhcs.metadata":
209 # Parse key-value pairs
210 for nv in ann.getMapValue():
211 if nv.name == "openhcs.grid_dimensions":
212 # Parse "rows,cols" format
213 rows, cols = map(int, nv.value.split(','))
214 logger.info(f"Found grid_dimensions ({rows}, {cols}) in OMERO metadata")
215 return (rows, cols)
217 # Grid dimensions not found in metadata
218 logger.warning(f"Grid dimensions not found in OMERO metadata for plate {plate_id}, using fallback")
219 return self.FALLBACK_VALUES.get('grid_dimensions', (1, 1))
221 except Exception as e:
222 logger.warning(f"Error extracting grid_dimensions from OMERO: {e}")
223 return self.FALLBACK_VALUES.get('grid_dimensions', (1, 1))
225 def get_pixel_size(self, plate_path: Union[str, Path, int]) -> float:
226 """
227 Get pixel size from OMERO image metadata.
229 Queries the first image in the plate for pixel size.
230 Falls back to DEFAULT_PIXEL_SIZE if not available.
231 """
232 try:
233 plate_id = plate_path if isinstance(plate_path, int) else int(Path(plate_path).name)
234 conn = self._get_omero_conn()
235 plate = conn.getObject("Plate", plate_id)
237 if plate:
238 # Get first well's first image
239 for well in plate.listChildren():
240 well_sample = well.getWellSample(0)
241 if well_sample:
242 image = well_sample.getImage()
243 pixels = image.getPrimaryPixels()
244 # Get physical pixel size in micrometers
245 pixel_size_x = pixels.getPhysicalSizeX()
246 if pixel_size_x:
247 return float(pixel_size_x)
248 break
249 except Exception:
250 pass
252 # Fallback to default
253 return self.FALLBACK_VALUES.get('pixel_size', 1.0)
255 def get_image_files(self, plate_path: Union[str, Path, int]) -> List[str]:
256 """
257 Get list of virtual filenames from OMERO backend.
259 Delegates to OMEROLocalBackend.list_files() to generate virtual filenames.
260 """
261 plate_id = plate_path if isinstance(plate_path, int) else int(Path(plate_path).name)
263 # Get OMERO backend from registry
264 omero_backend = self.filemanager.registry[Backend.OMERO_LOCAL.value]
266 # Call list_files with plate_id to get virtual filenames
267 virtual_files = omero_backend.list_files(str(plate_id), plate_id=plate_id)
269 # Return just the basenames (they're already basenames from backend)
270 return [Path(f).name for f in virtual_files]
273class OMEROFilenameParser(FilenameParser):
274 """
275 Parser for OMERO virtual filenames.
277 OMERO backend generates filenames in standard format with ALL components:
278 A01_s001_w1_z001_t001.tif
280 This is compatible with ImageXpress format, but OMERO always includes
281 all components (well, site, channel, z_index, timepoint) since it knows
282 the full plate structure from OMERO metadata.
284 For now, this just uses the ImageXpress pattern since they're compatible.
285 In the future, this could enforce that all components are present.
286 """
288 # Use ImageXpress pattern - it's compatible
289 from openhcs.microscopes.imagexpress import ImageXpressFilenameParser
290 _pattern = ImageXpressFilenameParser._pattern
292 @classmethod
293 def can_parse(cls, filename: str) -> bool:
294 """Check if this parser can parse the given filename."""
295 return cls._pattern.match(filename) is not None
297 def parse_filename(self, filename: str) -> Optional[Dict[str, Any]]:
298 """Parse OMERO virtual filename using ImageXpress pattern."""
299 from openhcs.microscopes.imagexpress import ImageXpressFilenameParser
300 parser = ImageXpressFilenameParser()
301 return parser.parse_filename(filename)
303 def construct_filename(self, well, site, channel, z_index, timepoint, extension='.tif', **kwargs) -> str:
304 """
305 Construct OMERO virtual filename.
307 OMERO always generates complete filenames with all components.
308 """
309 from openhcs.microscopes.imagexpress import ImageXpressFilenameParser
310 parser = ImageXpressFilenameParser()
311 return parser.construct_filename(
312 well=well,
313 site=site,
314 channel=channel,
315 z_index=z_index,
316 timepoint=timepoint,
317 extension=extension,
318 **kwargs
319 )
321 def extract_component_coordinates(self, component_value: str) -> Tuple[str, str]:
322 """Extract coordinates from well identifier (e.g., 'A01' → ('A', '01'))."""
323 from openhcs.microscopes.imagexpress import ImageXpressFilenameParser
324 parser = ImageXpressFilenameParser()
325 return parser.extract_component_coordinates(component_value)
328class OMEROHandler(MicroscopeHandler):
329 """OMERO microscope handler - uses OMERO native metadata."""
331 _microscope_type = 'omero'
332 _metadata_handler_class = None # Set after class definition
334 def __init__(self, filemanager: FileManager, pattern_format: Optional[str] = None):
335 """
336 Initialize OMERO handler.
338 OMERO uses OMEROFilenameParser to parse virtual filenames generated by the backend.
339 The orchestrator needs the parser to extract components from filenames.
341 Args:
342 filemanager: FileManager with OMERO backend in registry
343 pattern_format: Unused for OMERO (included for interface compatibility)
344 """
345 parser = OMEROFilenameParser()
346 metadata_handler = OMEROMetadataHandler(filemanager)
347 super().__init__(parser=parser, metadata_handler=metadata_handler)
349 @property
350 def root_dir(self) -> str:
351 """Root directory for OMERO virtual workspace.
353 Returns "Images" for path compatibility with OMERO virtual paths.
354 """
355 return "Images"
357 @property
358 def microscope_type(self) -> str:
359 return "omero"
361 @property
362 def metadata_handler_class(self) -> Type[MetadataHandler]:
363 """Metadata handler class (for interface enforcement only)."""
364 return OMEROMetadataHandler
366 @property
367 def compatible_backends(self) -> List[Backend]:
368 """OMERO is only compatible with OMERO_LOCAL backend."""
369 return [Backend.OMERO_LOCAL]
371 def _prepare_workspace(self, workspace_path: Path, filemanager: FileManager) -> Path:
372 """
373 OMERO doesn't need workspace preparation - it's a virtual filesystem.
375 Args:
376 workspace_path: Path to workspace (unused for OMERO)
377 filemanager: FileManager instance (unused for OMERO)
379 Returns:
380 workspace_path unchanged
381 """
382 return workspace_path
384 def initialize_workspace(self, plate_path: Union[int, Path], filemanager: FileManager) -> Path:
385 """
386 OMERO creates a virtual path for the plate.
388 For OMERO, plate_path is an int (plate_id). We convert it to a virtual
389 Path like "/omero/plate_54/Images" that can be used throughout the system.
390 The backend extracts the plate_id from this virtual path when needed.
392 Args:
393 plate_path: OMERO plate_id (int) or Path (for compatibility)
394 filemanager: Unused for OMERO
396 Returns:
397 Virtual Path for OMERO plate (e.g., "/omero/plate_54/Images")
398 """
399 # Convert int plate_id to virtual path
400 if isinstance(plate_path, int):
401 # Create virtual path: /omero/plate_{id}/Images
402 virtual_path = Path(f"/omero/plate_{plate_path}/Images")
403 return virtual_path
404 else:
405 # Already a Path (shouldn't happen for OMERO, but handle it)
406 return plate_path
409# Set metadata handler class after class definition for automatic registration
410from openhcs.microscopes.microscope_base import register_metadata_handler
411OMEROHandler._metadata_handler_class = OMEROMetadataHandler
412register_metadata_handler(OMEROHandler, OMEROMetadataHandler)