Coverage for openhcs/core/config.py: 67.3%
268 statements
« prev ^ index » next coverage.py v7.10.3, created at 2025-08-14 05:57 +0000
1"""
2Global configuration dataclasses for OpenHCS.
4This module defines the primary configuration objects used throughout the application,
5such as VFSConfig, PathPlanningConfig, and the overarching GlobalPipelineConfig.
6Configuration is intended to be immutable and provided as Python objects.
7"""
9import logging
10import os # For a potentially more dynamic default for num_workers
11import threading
12import dataclasses
13from dataclasses import dataclass, field
14from pathlib import Path
15from typing import Literal, Optional, Union, Dict, Any, List, Type
16from enum import Enum
17from openhcs.constants import Microscope
18from openhcs.constants.constants import Backend
20# Import TilingLayout for TUI configuration
21try:
22 from textual_window import TilingLayout
23except ImportError:
24 # Fallback for when textual-window is not available
25 from enum import Enum
26 class TilingLayout(Enum):
27 FLOATING = "floating"
28 MASTER_DETAIL = "master_detail"
30logger = logging.getLogger(__name__)
33class ZarrCompressor(Enum):
34 """Available compression algorithms for zarr storage."""
35 BLOSC = "blosc"
36 ZLIB = "zlib"
37 LZ4 = "lz4"
38 ZSTD = "zstd"
39 NONE = "none"
41 def create_compressor(self, compression_level: int, shuffle: bool = True) -> Optional[Any]:
42 """Create the actual zarr compressor instance.
44 Args:
45 compression_level: Compression level (1-22 for ZSTD, 1-9 for others)
46 shuffle: Enable byte shuffling for better compression (blosc only)
48 Returns:
49 Configured zarr compressor instance or None for no compression
50 """
51 import zarr
53 match self:
54 case ZarrCompressor.NONE: 54 ↛ 55line 54 didn't jump to line 55 because the pattern on line 54 never matched
55 return None
56 case ZarrCompressor.BLOSC: 56 ↛ 57line 56 didn't jump to line 57 because the pattern on line 56 never matched
57 return zarr.Blosc(cname='lz4', clevel=compression_level, shuffle=shuffle)
58 case ZarrCompressor.ZLIB: 58 ↛ 59line 58 didn't jump to line 59 because the pattern on line 58 never matched
59 return zarr.Zlib(level=compression_level)
60 case ZarrCompressor.LZ4: 60 ↛ 62line 60 didn't jump to line 62 because the pattern on line 60 always matched
61 return zarr.LZ4(acceleration=compression_level)
62 case ZarrCompressor.ZSTD:
63 return zarr.Zstd(level=compression_level)
66class ZarrChunkStrategy(Enum):
67 """Chunking strategies for zarr arrays."""
68 SINGLE = "single" # Single chunk per array (optimal for batch I/O)
69 AUTO = "auto" # Let zarr decide chunk size
70 CUSTOM = "custom" # User-defined chunk sizes
73class MaterializationBackend(Enum):
74 """Available backends for materialization (persistent storage only)."""
75 ZARR = "zarr"
76 DISK = "disk"
79class WellFilterMode(Enum):
80 """Well filtering modes for selective materialization."""
81 INCLUDE = "include" # Materialize only specified wells
82 EXCLUDE = "exclude" # Materialize all wells except specified ones
84@dataclass(frozen=True)
85class ZarrConfig:
86 """Configuration for Zarr storage backend."""
87 store_name: str = "images"
88 """Name of the zarr store directory."""
90 compressor: ZarrCompressor = ZarrCompressor.LZ4
91 """Compression algorithm to use."""
93 compression_level: int = 9
94 """Compression level (1-9 for LZ4, higher = more compression)."""
96 shuffle: bool = True
97 """Enable byte shuffling for better compression (blosc only)."""
99 chunk_strategy: ZarrChunkStrategy = ZarrChunkStrategy.SINGLE
100 """Chunking strategy for zarr arrays."""
102 ome_zarr_metadata: bool = True
103 """Generate OME-ZARR compatible metadata and structure."""
105 write_plate_metadata: bool = True
106 """Write plate-level metadata for HCS viewing (required for OME-ZARR viewers like napari)."""
109@dataclass(frozen=True)
110class VFSConfig:
111 """Configuration for Virtual File System (VFS) related operations."""
112 intermediate_backend: Backend = Backend.MEMORY
113 """Backend for storing intermediate step results that are not explicitly materialized."""
115 materialization_backend: MaterializationBackend = MaterializationBackend.DISK
116 """Backend for explicitly materialized outputs (e.g., final results, user-requested saves)."""
118@dataclass(frozen=True)
119class AnalysisConsolidationConfig:
120 """Configuration for automatic analysis results consolidation."""
121 enabled: bool = True
122 """Whether to automatically run analysis consolidation after pipeline completion."""
124 metaxpress_style: bool = True
125 """Whether to generate MetaXpress-compatible output format with headers."""
127 well_pattern: str = r"([A-Z]\d{2})"
128 """Regex pattern for extracting well IDs from filenames."""
130 file_extensions: tuple[str, ...] = (".csv",)
131 """File extensions to include in consolidation."""
133 exclude_patterns: tuple[str, ...] = (r".*consolidated.*", r".*metaxpress.*", r".*summary.*")
134 """Filename patterns to exclude from consolidation."""
136 output_filename: str = "metaxpress_style_summary.csv"
137 """Name of the consolidated output file."""
140@dataclass(frozen=True)
141class PlateMetadataConfig:
142 """Configuration for plate metadata in MetaXpress-style output."""
143 barcode: Optional[str] = None
144 """Plate barcode. If None, will be auto-generated from plate name."""
146 plate_name: Optional[str] = None
147 """Plate name. If None, will be derived from plate path."""
149 plate_id: Optional[str] = None
150 """Plate ID. If None, will be auto-generated."""
152 description: Optional[str] = None
153 """Experiment description. If None, will be auto-generated."""
155 acquisition_user: str = "OpenHCS"
156 """User who acquired the data."""
158 z_step: str = "1"
159 """Z-step information for MetaXpress compatibility."""
162@dataclass(frozen=True)
163class PathPlanningConfig:
164 """
165 Configuration for pipeline path planning and directory structure.
167 This class handles path construction concerns including plate root directories,
168 output directory suffixes, and subdirectory organization. It does not handle
169 analysis results location, which is controlled at the pipeline level.
170 """
171 output_dir_suffix: str = "_outputs"
172 """Default suffix for general step output directories."""
174 global_output_folder: Optional[Path] = None
175 """
176 Optional global output folder where all plate workspaces and outputs will be created.
177 If specified, plate workspaces will be created as {global_output_folder}/{plate_name}_workspace/
178 and outputs as {global_output_folder}/{plate_name}_workspace_outputs/.
179 If None, uses the current behavior (workspace and outputs in same directory as plate).
180 Example: "/data/results" or "/mnt/hcs_output"
181 """
183 sub_dir: str = "images"
184 """
185 Subdirectory within plate folder for storing processed data.
186 Automatically adds .zarr suffix when using zarr backend.
187 Examples: "images", "processed", "data/images"
188 """
191@dataclass(frozen=True)
192class StepMaterializationConfig(PathPlanningConfig):
193 """
194 Configuration for per-step materialization - configurable in UI.
196 This dataclass appears in the UI like any other configuration, allowing users
197 to set pipeline-level defaults for step materialization behavior. All step
198 materialization instances will inherit these defaults unless explicitly overridden.
200 Inherits from PathPlanningConfig to ensure all required path planning fields
201 (like global_output_folder) are available for the lazy loading system.
203 Well Filtering Options:
204 - well_filter=1 materializes first well only (enables quick checkpointing)
205 - well_filter=None materializes all wells
206 - well_filter=["A01", "B03"] materializes only specified wells
207 - well_filter="A01:A12" materializes well range
208 - well_filter=5 materializes first 5 wells processed
209 - well_filter_mode controls include/exclude behavior
210 """
212 # Well filtering defaults
213 well_filter: Optional[Union[List[str], str, int]] = 1
214 """
215 Well filtering for selective step materialization:
216 - 1: Materialize first well only (default - enables quick checkpointing)
217 - None: Materialize all wells
218 - List[str]: Specific well IDs ["A01", "B03", "D12"]
219 - str: Pattern/range "A01:A12", "row:A", "col:01-06"
220 - int: Maximum number of wells (first N processed)
221 """
223 well_filter_mode: WellFilterMode = WellFilterMode.INCLUDE
224 """
225 Well filtering mode for step materialization:
226 - INCLUDE: Materialize only wells matching the filter
227 - EXCLUDE: Materialize all wells except those matching the filter
228 """
230 # Override PathPlanningConfig defaults to prevent collisions
231 output_dir_suffix: str = "" # Uses same output plate path as main pipeline
232 sub_dir: str = "checkpoints" # vs global "images"
235# Generic thread-local storage for any global config type
236_global_config_contexts: Dict[Type, threading.local] = {}
238def set_current_global_config(config_type: Type, config_instance: Any) -> None:
239 """Set current global config for any dataclass type."""
240 if config_type not in _global_config_contexts:
241 _global_config_contexts[config_type] = threading.local()
242 _global_config_contexts[config_type].value = config_instance
244def get_current_global_config(config_type: Type) -> Optional[Any]:
245 """Get current global config for any dataclass type."""
246 context = _global_config_contexts.get(config_type)
247 return getattr(context, 'value', None) if context else None
249def get_current_materialization_defaults() -> StepMaterializationConfig:
250 """Get current step materialization config from pipeline config."""
251 current_config = get_current_global_config(GlobalPipelineConfig)
252 if current_config:
253 return current_config.materialization_defaults
254 # Fallback to default instance if no pipeline config is set
255 return StepMaterializationConfig()
258# Type registry for lazy dataclass to base class mapping
259_lazy_type_registry: Dict[Type, Type] = {}
261def register_lazy_type_mapping(lazy_type: Type, base_type: Type) -> None:
262 """Register mapping between lazy dataclass type and its base type."""
263 _lazy_type_registry[lazy_type] = base_type
265def get_base_type_for_lazy(lazy_type: Type) -> Optional[Type]:
266 """Get the base type for a lazy dataclass type."""
267 return _lazy_type_registry.get(lazy_type)
270class LazyDefaultPlaceholderService:
271 """
272 Enhanced service supporting factory-created lazy classes with flexible resolution.
274 Provides consistent placeholder pattern for both static and dynamic lazy configuration classes.
275 """
277 # Configurable placeholder prefix - set to empty string for cleaner appearance
278 PLACEHOLDER_PREFIX = ""
280 @staticmethod
281 def has_lazy_resolution(dataclass_type: type) -> bool:
282 """Check if dataclass has lazy resolution methods (created by factory)."""
283 return (hasattr(dataclass_type, '_resolve_field_value') and
284 hasattr(dataclass_type, 'to_base_config'))
286 @staticmethod
287 def get_lazy_resolved_placeholder(
288 dataclass_type: type,
289 field_name: str,
290 app_config: Optional[Any] = None,
291 force_static_defaults: bool = False
292 ) -> Optional[str]:
293 """
294 Get placeholder text for lazy-resolved field with flexible resolution.
296 Args:
297 dataclass_type: The lazy dataclass type (created by factory)
298 field_name: Name of the field to resolve
299 app_config: Optional app config for dynamic resolution
300 force_static_defaults: If True, always use static defaults regardless of thread-local context
302 Returns:
303 Placeholder text with configurable prefix for consistent UI experience.
304 """
305 if not LazyDefaultPlaceholderService.has_lazy_resolution(dataclass_type):
306 return None
308 if force_static_defaults:
309 # For global config editing: always use static defaults
310 if hasattr(dataclass_type, 'to_base_config'):
311 # This is a lazy dataclass - get the base class and create instance with static defaults
312 base_class = LazyDefaultPlaceholderService._get_base_class_from_lazy(dataclass_type)
313 static_instance = base_class()
314 resolved_value = getattr(static_instance, field_name, None)
315 else:
316 # Regular dataclass - create instance with static defaults
317 static_instance = dataclass_type()
318 resolved_value = getattr(static_instance, field_name, None)
319 elif app_config:
320 # For dynamic resolution, create lazy class with current app config
321 from openhcs.core.lazy_config import LazyDataclassFactory
322 dynamic_lazy_class = LazyDataclassFactory.create_lazy_dataclass(
323 defaults_source=app_config, # Use the app_config directly
324 lazy_class_name=f"Dynamic{dataclass_type.__name__}"
325 )
326 temp_instance = dynamic_lazy_class()
327 resolved_value = getattr(temp_instance, field_name)
328 else:
329 # Use existing lazy class (thread-local resolution)
330 temp_instance = dataclass_type()
331 resolved_value = getattr(temp_instance, field_name)
333 if resolved_value is not None:
334 # Format nested dataclasses with key field values
335 if hasattr(resolved_value, '__dataclass_fields__'):
336 # For nested dataclasses, show key field values instead of generic info
337 summary = LazyDefaultPlaceholderService._format_nested_dataclass_summary(resolved_value)
338 return f"{LazyDefaultPlaceholderService.PLACEHOLDER_PREFIX}{summary}"
339 else:
340 return f"{LazyDefaultPlaceholderService.PLACEHOLDER_PREFIX}{resolved_value}"
341 else:
342 return f"{LazyDefaultPlaceholderService.PLACEHOLDER_PREFIX}(none)"
344 @staticmethod
345 def _get_base_class_from_lazy(lazy_class: Type) -> Type:
346 """
347 Extract the base class from a lazy dataclass using type registry.
348 """
349 # First check the type registry
350 base_type = get_base_type_for_lazy(lazy_class)
351 if base_type:
352 return base_type
354 # Check if the lazy class has a to_base_config method
355 if hasattr(lazy_class, 'to_base_config'):
356 # Create a dummy instance to inspect the to_base_config method
357 dummy_instance = lazy_class()
358 base_instance = dummy_instance.to_base_config()
359 return type(base_instance)
361 # If no mapping found, raise an error - this indicates missing registration
362 raise ValueError(
363 f"No base type registered for lazy class {lazy_class.__name__}. "
364 f"Use register_lazy_type_mapping() to register the mapping."
365 )
367 @staticmethod
368 def _format_nested_dataclass_summary(dataclass_instance) -> str:
369 """
370 Format nested dataclass with all field values for user-friendly placeholders.
372 Uses generic dataclass introspection to show all fields with their current values,
373 providing a complete and maintainable summary without hardcoded field mappings.
374 """
375 import dataclasses
377 class_name = dataclass_instance.__class__.__name__
379 # Get all fields from the dataclass using introspection
380 all_fields = [f.name for f in dataclasses.fields(dataclass_instance)]
382 # Extract all field values
383 field_summaries = []
384 for field_name in all_fields:
385 try:
386 value = getattr(dataclass_instance, field_name)
388 # Skip None values to keep summary concise
389 if value is None:
390 continue
392 # Format different value types appropriately
393 if hasattr(value, 'value'): # Enum
394 formatted_value = value.value
395 elif hasattr(value, 'name'): # Enum with name
396 formatted_value = value.name
397 elif isinstance(value, str) and len(value) > 20: # Long strings
398 formatted_value = f"{value[:17]}..."
399 elif dataclasses.is_dataclass(value): # Nested dataclass
400 formatted_value = f"{value.__class__.__name__}(...)"
401 else:
402 formatted_value = str(value)
404 field_summaries.append(f"{field_name}={formatted_value}")
406 except (AttributeError, Exception):
407 # Skip fields that can't be accessed
408 continue
410 if field_summaries:
411 return ", ".join(field_summaries)
412 else:
413 # Fallback when no non-None fields are found
414 return f"{class_name} (default settings)"
417# MaterializationPathConfig is now LazyStepMaterializationConfig from lazy_config.py
418# Import moved to avoid circular dependency - use lazy import pattern
421@dataclass(frozen=True)
422class TilingKeybinding:
423 """Declarative mapping between key combination and window manager method."""
424 key: str
425 action: str # method name that already exists
426 description: str
429@dataclass(frozen=True)
430class TilingKeybindings:
431 """Declarative mapping of tiling keybindings to existing methods."""
433 # Focus controls
434 focus_next: TilingKeybinding = TilingKeybinding("ctrl+j", "focus_next_window", "Focus Next Window")
435 focus_prev: TilingKeybinding = TilingKeybinding("ctrl+k", "focus_previous_window", "Focus Previous Window")
437 # Layout controls - map to wrapper methods
438 horizontal_split: TilingKeybinding = TilingKeybinding("ctrl+shift+h", "set_horizontal_split", "Horizontal Split")
439 vertical_split: TilingKeybinding = TilingKeybinding("ctrl+shift+v", "set_vertical_split", "Vertical Split")
440 grid_layout: TilingKeybinding = TilingKeybinding("ctrl+shift+g", "set_grid_layout", "Grid Layout")
441 master_detail: TilingKeybinding = TilingKeybinding("ctrl+shift+m", "set_master_detail", "Master Detail")
442 toggle_floating: TilingKeybinding = TilingKeybinding("ctrl+shift+f", "toggle_floating", "Toggle Floating")
444 # Window movement - map to extracted window_manager methods
445 move_window_prev: TilingKeybinding = TilingKeybinding("ctrl+shift+left", "move_focused_window_prev", "Move Window Left")
446 move_window_next: TilingKeybinding = TilingKeybinding("ctrl+shift+right", "move_focused_window_next", "Move Window Right")
447 rotate_left: TilingKeybinding = TilingKeybinding("ctrl+alt+left", "rotate_window_order_left", "Rotate Windows Left")
448 rotate_right: TilingKeybinding = TilingKeybinding("ctrl+alt+right", "rotate_window_order_right", "Rotate Windows Right")
450 # Gap controls
451 gap_increase: TilingKeybinding = TilingKeybinding("ctrl+plus", "gap_increase", "Increase Gap")
452 gap_decrease: TilingKeybinding = TilingKeybinding("ctrl+minus", "gap_decrease", "Decrease Gap")
454 # Bulk operations
455 minimize_all: TilingKeybinding = TilingKeybinding("ctrl+shift+d", "minimize_all_windows", "Minimize All")
456 open_all: TilingKeybinding = TilingKeybinding("ctrl+shift+o", "open_all_windows", "Open All")
459@dataclass(frozen=True)
460class FunctionRegistryConfig:
461 """Configuration for function registry behavior across all libraries."""
462 enable_scalar_functions: bool = True
463 """
464 Whether to register functions that return scalars.
465 When True: Scalar-returning functions are wrapped as (array, scalar) tuples.
466 When False: Scalar-returning functions are filtered out entirely.
467 Applies uniformly to all libraries (CuPy, scikit-image, pyclesperanto).
468 """
471@dataclass(frozen=True)
472class TUIConfig:
473 """Configuration for OpenHCS Textual User Interface."""
474 default_tiling_layout: TilingLayout = TilingLayout.MASTER_DETAIL
475 """Default tiling layout for window manager on startup."""
477 default_window_gap: int = 1
478 """Default gap between windows in tiling mode (in characters)."""
480 enable_startup_notification: bool = True
481 """Whether to show notification about tiling mode on startup."""
483 keybindings: TilingKeybindings = field(default_factory=TilingKeybindings)
484 """Declarative mapping of all tiling keybindings."""
487@dataclass(frozen=True)
488class GlobalPipelineConfig:
489 """
490 Root configuration object for an OpenHCS pipeline session.
491 This object is intended to be instantiated at application startup and treated as immutable.
492 """
493 num_workers: int = field(default_factory=lambda: os.cpu_count() or 1)
494 """Number of worker processes/threads for parallelizable tasks."""
496 path_planning: PathPlanningConfig = field(default_factory=PathPlanningConfig)
497 """Configuration for path planning (directory suffixes)."""
499 vfs: VFSConfig = field(default_factory=VFSConfig)
500 """Configuration for Virtual File System behavior."""
502 zarr: ZarrConfig = field(default_factory=ZarrConfig)
503 """Configuration for Zarr storage backend."""
505 materialization_results_path: Path = Path("results")
506 """
507 Path for materialized analysis results (CSV, JSON files from special outputs).
509 This is a pipeline-wide setting that controls where all special output materialization
510 functions save their analysis results, regardless of which step produces them.
512 Can be relative to plate folder or absolute path.
513 Default: "results" creates a results/ folder in the plate directory.
514 Examples: "results", "./analysis", "/data/analysis_results", "../shared_results"
516 Note: This is separate from per-step image materialization, which is controlled
517 by the sub_dir field in each step's materialization_config.
518 """
520 analysis_consolidation: AnalysisConsolidationConfig = field(default_factory=AnalysisConsolidationConfig)
521 """Configuration for automatic analysis results consolidation."""
523 plate_metadata: PlateMetadataConfig = field(default_factory=PlateMetadataConfig)
524 """Configuration for plate metadata in consolidated outputs."""
526 function_registry: FunctionRegistryConfig = field(default_factory=FunctionRegistryConfig)
527 """Configuration for function registry behavior."""
529 materialization_defaults: StepMaterializationConfig = field(default_factory=StepMaterializationConfig)
530 """Default configuration for per-step materialization - configurable in UI."""
532 microscope: Microscope = Microscope.AUTO
533 """Default microscope type for auto-detection."""
535 use_threading: bool = field(default_factory=lambda: os.getenv('OPENHCS_USE_THREADING', 'false').lower() == 'true')
536 """Use ThreadPoolExecutor instead of ProcessPoolExecutor for debugging. Reads from OPENHCS_USE_THREADING environment variable."""
538 # Future extension point:
539 # logging_config: Optional[Dict[str, Any]] = None # For configuring logging levels, handlers
540 # plugin_settings: Dict[str, Any] = field(default_factory=dict) # For plugin-specific settings
542# --- Default Configuration Provider ---
544# Pre-instantiate default sub-configs for clarity if they have many fields or complex defaults.
545# For simple cases, direct instantiation in get_default_global_config is also fine.
546_DEFAULT_PATH_PLANNING_CONFIG = PathPlanningConfig()
547_DEFAULT_VFS_CONFIG = VFSConfig(
548 # Example: Set a default persistent_storage_root_path if desired for out-of-the-box behavior
549 # persistent_storage_root_path="./openhcs_output_data"
550)
551_DEFAULT_ZARR_CONFIG = ZarrConfig()
552_DEFAULT_ANALYSIS_CONSOLIDATION_CONFIG = AnalysisConsolidationConfig()
553_DEFAULT_PLATE_METADATA_CONFIG = PlateMetadataConfig()
554_DEFAULT_FUNCTION_REGISTRY_CONFIG = FunctionRegistryConfig()
555_DEFAULT_MATERIALIZATION_DEFAULTS = StepMaterializationConfig()
556_DEFAULT_TUI_CONFIG = TUIConfig()
558def get_default_global_config() -> GlobalPipelineConfig:
559 """
560 Provides a default instance of GlobalPipelineConfig.
562 This function is called if no specific configuration is provided to the
563 PipelineOrchestrator, ensuring the application can run with sensible defaults.
564 """
565 logger.info("Initializing with default GlobalPipelineConfig.")
566 return GlobalPipelineConfig(
567 # num_workers is already handled by field(default_factory) in GlobalPipelineConfig
568 path_planning=_DEFAULT_PATH_PLANNING_CONFIG,
569 vfs=_DEFAULT_VFS_CONFIG,
570 zarr=_DEFAULT_ZARR_CONFIG,
571 analysis_consolidation=_DEFAULT_ANALYSIS_CONSOLIDATION_CONFIG,
572 plate_metadata=_DEFAULT_PLATE_METADATA_CONFIG,
573 function_registry=_DEFAULT_FUNCTION_REGISTRY_CONFIG,
574 materialization_defaults=_DEFAULT_MATERIALIZATION_DEFAULTS
575 )
578# Import pipeline-specific classes - circular import solved by moving import to end
579from openhcs.core.pipeline_config import (
580 LazyStepMaterializationConfig as MaterializationPathConfig,
581 PipelineConfig,
582 set_current_pipeline_config,
583 ensure_pipeline_config_context,
584 create_pipeline_config_for_editing,
585 create_editing_config_from_existing_lazy_config
586)