diff --git a/docs/source/architecture/index.rst b/docs/source/architecture/index.rst index a89320c47..0542ddf01 100644 --- a/docs/source/architecture/index.rst +++ b/docs/source/architecture/index.rst @@ -143,6 +143,7 @@ TUI architecture, UI development patterns, and form management systems. service-layer-architecture gui_performance_patterns cross_window_update_optimization + scope_visual_feedback_system Development Tools ================= diff --git a/docs/source/architecture/scope_visual_feedback_system.rst b/docs/source/architecture/scope_visual_feedback_system.rst new file mode 100644 index 000000000..2c6e67b02 --- /dev/null +++ b/docs/source/architecture/scope_visual_feedback_system.rst @@ -0,0 +1,123 @@ +Scope-Based Visual Feedback System +=================================== + +Overview +-------- + +The scope visual feedback system provides visual cues to help users understand +which scope (orchestrator or step) a window belongs to. It uses MD5-based color +generation to create consistent, WCAG AA compliant colors for each scope. + +Architecture +------------ + +The system consists of several components: + +**ScopeColorService** (``services/scope_color_service.py``) + Singleton service that manages scope-to-color mappings. Generates colors + using MD5 hashing of scope_id strings to ensure consistency across sessions. + +**ScopeColorScheme** (``scope_visual_config.py``) + Dataclass holding all derived colors for a scope: + + - ``base_rgb``: The primary scope color + - ``orchestrator_border_rgb``: Border color for orchestrator windows + - ``orchestrator_item_border_rgb``: Border color for list items + - ``step_background_rgb``: Background tint for step windows + - ``flash_color_rgb``: Color used for flash animations + +**ScopedBorderMixin** (``scoped_border_mixin.py``) + Mixin for QDialog/QWidget that renders scope-based borders: + + - Simple solid borders for orchestrator-level scopes + - Layered patterned borders for step-level scopes + - Reactive updates via signal connections + +**Flash Animation System** + Smooth flash animations for visual feedback when values change: + + - ``SmoothFlashAnimatorBase``: Base class using QVariantAnimation + - ``WidgetFlashAnimator``: Flashes QWidget backgrounds + - ``TreeItemFlashAnimator``: Flashes QTreeWidgetItem backgrounds + - ``ListItemFlashAnimator``: Flashes QListWidgetItem backgrounds + +**TreeFormFlashMixin** (``tree_form_flash_mixin.py``) + Mixin for windows with tree + form that provides: + + - GroupBox flashing when navigating to sections + - Tree item flashing when placeholder values change cross-window + +Usage +----- + +Windows declare their scope via ``scope_id`` attribute: + +.. code-block:: python + + class MyWindow(ScopedBorderMixin, TreeFormFlashMixin, QDialog): + def __init__(self, scope_id: str): + super().__init__() + self.scope_id = scope_id + + # Initialize form_manager, tree_widget, etc. + self.setup_ui() + + # Initialize scope border rendering + self._init_scope_border() + + # Register flash hooks for placeholder changes + self._register_placeholder_flash_hook() + +Color Generation +---------------- + +Colors are generated using MD5 hashing to ensure: + +1. **Consistency**: Same scope_id always produces same color +2. **Uniqueness**: Different scopes get visually distinct colors +3. **Accessibility**: Colors meet WCAG AA contrast requirements + +The algorithm: + +1. Hash scope_id with MD5 +2. Extract hue from first 2 bytes (0-360°) +3. Use fixed saturation (0.65) and lightness (0.45) for WCAG compliance +4. Convert HSL to RGB + +Flash Animation Timing +---------------------- + +The smooth flash animation uses QVariantAnimation for 60fps transitions: + +- **Fade-in**: 100ms with OutQuad easing (quick snap to flash color) +- **Hold**: 150ms at max intensity while rapid updates continue +- **Fade-out**: 350ms with InOutCubic easing (smooth return to original) + +This creates a natural "pulse" effect that's noticeable but not jarring. + +Integration Points +------------------ + +The system integrates with: + +- ``ParameterFormManager``: Provides ``_on_placeholder_changed_callbacks`` hook +- ``ParameterOpsService``: Fires callbacks when placeholder values actually change +- ``AbstractManagerWidget``: Applies scope colors to list items +- ``ListItemDelegate``: Renders scope-colored borders on list items + +Files +----- + +New files added: + +- ``scope_color_utils.py``: Utility functions for scope color access +- ``scope_color_strategy.py``: Strategy pattern for color scheme derivation +- ``scope_visual_config.py``: Configuration dataclasses +- ``services/scope_color_service.py``: Core service implementation +- ``scoped_border_mixin.py``: Window border rendering mixin +- ``smooth_flash_base.py``: Base class for flash animations +- ``widget_flash_animation.py``: Widget flash animator +- ``tree_item_flash_animation.py``: Tree item flash animator +- ``list_item_flash_animation.py``: List item flash animator +- ``tree_form_flash_mixin.py``: Combined tree+form flash mixin + diff --git a/openhcs/config_framework/__init__.py b/openhcs/config_framework/__init__.py index e483d6cd6..a699a9f1f 100644 --- a/openhcs/config_framework/__init__.py +++ b/openhcs/config_framework/__init__.py @@ -97,6 +97,9 @@ build_context_stack, ) +# Context Stack Registry (single source of truth for resolved values) +from openhcs.config_framework.context_stack_registry import ContextStackRegistry + # Placeholder from openhcs.config_framework.placeholder import LazyDefaultPlaceholderService diff --git a/openhcs/config_framework/context_stack_registry.py b/openhcs/config_framework/context_stack_registry.py new file mode 100644 index 000000000..89c085367 --- /dev/null +++ b/openhcs/config_framework/context_stack_registry.py @@ -0,0 +1,406 @@ +""" +Central Context Stack Registry - Single Source of Truth for Resolved Config Values. + +This service eliminates the architectural smell where forms build their own context +stacks internally. Instead, the registry maintains context stacks for all scopes and +serves as the single source of truth for resolved values. + +Architecture: +- Singleton service managing all resolved configuration values +- Immutability pattern: stores concrete values separately, never mutates context_obj +- Materialization API: creates new instances with resolved values applied +- Signal-based reactivity: emits value_changed when resolved values change +- Lazy resolution: creates fresh lazy instances within context for proper inheritance + +Usage: + registry = ContextStackRegistry.instance() + + # Register scope + registry.register_scope(scope_id, context_obj, dataclass_type) + + # Live edits (immutable) - from forms with nested types + registry.set(scope_id, field_name, value, dataclass_type) + + # Read resolved value (for forms with nested types) + resolved = registry.resolve(scope_id, dataclass_type, field_name) + + # Read resolved value (for preview labels using stored type) + resolved = registry.resolve(scope_id, field_name=field_name) + + # Materialize for compilation + materialized = registry.get_materialized_object(scope_id) +""" + +from typing import Any, Dict, Optional, Set +from dataclasses import fields, is_dataclass +import logging +from contextlib import ExitStack + +from PyQt6.QtCore import QObject, pyqtSignal + +from openhcs.config_framework.context_manager import ( + build_context_stack, + config_context, + get_root_from_scope_key, +) +from openhcs.config_framework.lazy_factory import get_base_type_for_lazy + +logger = logging.getLogger(__name__) + + +class ContextStackRegistry(QObject): + """Central registry for context stacks and resolved configuration values. + + Maintains single source of truth for all resolved config values across the application. + Forms write to registry, UI reads from registry, no duplicate resolution paths. + """ + + # Signal: (scope_id, field_path, old_value, new_value) + value_changed = pyqtSignal(str, str, object, object) + + _instance: Optional['ContextStackRegistry'] = None + + def __init__(self): + super().__init__() + + # Registered scopes: scope_id -> (context_obj, dataclass_type) + self._scopes: Dict[str, tuple[Any, type]] = {} + + # Concrete value overrides: scope_id -> {field_path: value} + self._concrete_values: Dict[str, Dict[str, Any]] = {} + + # Resolved value cache: scope_id -> {field_path: value} + self._resolved_cache: Dict[str, Dict[str, Any]] = {} + + # LiveContextService token tracking for cache invalidation + self._last_live_context_token: int = 0 + + @classmethod + def instance(cls) -> 'ContextStackRegistry': + """Get singleton instance.""" + if cls._instance is None: + cls._instance = cls() + return cls._instance + + @classmethod + def reset_instance(cls) -> None: + """Reset singleton (for testing).""" + cls._instance = None + + def register_scope(self, scope_id: str, context_obj: Any, dataclass_type: type) -> None: + """Register a scope for resolution. + + If scope_id already exists, REPLACES context_obj and clears concrete values (code mode). + + Args: + scope_id: Unique scope identifier (plate path or plate::step_token) + context_obj: The object being edited (step, pipeline_config, etc.) + dataclass_type: The dataclass type being edited + """ + is_replacement = scope_id in self._scopes + + self._scopes[scope_id] = (context_obj, dataclass_type) + + if is_replacement: + # Code mode replacement - clear concrete values and cache + self._concrete_values[scope_id] = {} + self._resolved_cache[scope_id] = {} + logger.info(f"📦 REPLACED scope: {scope_id} (code mode)") + else: + # New registration + self._concrete_values[scope_id] = {} + self._resolved_cache[scope_id] = {} + logger.info(f"📦 REGISTERED scope: {scope_id}") + + def unregister_scope(self, scope_id: str) -> None: + """Unregister scope. Called when form closes or item deleted.""" + self._scopes.pop(scope_id, None) + self._concrete_values.pop(scope_id, None) + self._resolved_cache.pop(scope_id, None) + logger.info(f"📦 UNREGISTERED scope: {scope_id}") + + def set( + self, scope_id: str, field_name: str, value: Any, dataclass_type: Optional[type] = None + ) -> None: + """Set a concrete value override. Does NOT mutate context_obj (immutable). + + Emits value_changed if resolved value changes. + + Args: + scope_id: Scope identifier + field_name: Field name (e.g., "well_filter") + value: New value to set + dataclass_type: The dataclass type for this field. If None, uses stored root type. + """ + if scope_id not in self._scopes: + logger.warning(f"Cannot set value for unregistered scope: {scope_id}") + return + + _, stored_dataclass_type = self._scopes[scope_id] + if dataclass_type is None: + dataclass_type = stored_dataclass_type + + # Build cache key matching resolve() + cache_key = f"{dataclass_type.__name__}.{field_name}" + + # Get old resolved value before change + old_value = self.resolve(scope_id, dataclass_type, field_name) + + # Store concrete value with cache_key (immutable - don't mutate context_obj) + self._concrete_values[scope_id][cache_key] = value + + # Invalidate cache for this scope and descendants + self._invalidate_cache(scope_id, cache_key) + + # Get new resolved value after change + new_value = self.resolve(scope_id, dataclass_type, field_name) + + # Emit signal if resolved value actually changed + if old_value != new_value: + logger.info(f"⚡ VALUE CHANGED: {scope_id}.{field_path}: {old_value} -> {new_value}") + self.value_changed.emit(scope_id, field_path, old_value, new_value) + + def resolve( + self, scope_id: str, dataclass_type: Optional[type] = None, field_name: Optional[str] = None + ) -> Any: + """Resolve field value through context hierarchy using lazy dataclass resolution. + + Resolution order: + 1. Check concrete values (live edits) + 2. Check cache + 3. Build context stack and resolve via fresh lazy instance + + Args: + scope_id: Scope identifier + dataclass_type: The dataclass type to resolve for. If None, uses stored root type. + field_name: Name of the field to resolve (e.g., "well_filter") + + Returns: + Resolved value (raw, not formatted as placeholder text) + """ + if scope_id not in self._scopes: + logger.warning(f"Cannot resolve for unregistered scope: {scope_id}") + return None + + context_obj, stored_dataclass_type = self._scopes[scope_id] + + # Use stored type if not provided + if dataclass_type is None: + dataclass_type = stored_dataclass_type + + if field_name is None: + logger.warning(f"resolve() called without field_name for {scope_id}") + return None + + # Cache key combines type and field for proper namespacing + cache_key = f"{dataclass_type.__name__}.{field_name}" + logger.debug(f"🔍 Resolving {scope_id}.{cache_key}") + + # Check concrete values first (live edits) + if cache_key in self._concrete_values.get(scope_id, {}): + value = self._concrete_values[scope_id][cache_key] + logger.debug(f" ✅ Found in concrete values: {repr(value)[:50]}") + return value + + # Check cache + if cache_key in self._resolved_cache.get(scope_id, {}): + value = self._resolved_cache[scope_id][cache_key] + logger.debug(f" ✅ Found in cache: {repr(value)[:50]}") + return value + + # Build live context from all registered scopes + live_context = self._build_live_context(scope_id) + + # Get lazy version of the dataclass type for resolution + from openhcs.config_framework.placeholder import LazyDefaultPlaceholderService + lazy_type = dataclass_type + if not LazyDefaultPlaceholderService.has_lazy_resolution(dataclass_type): + lazy_type = LazyDefaultPlaceholderService._get_lazy_type_for_base(dataclass_type) + if not lazy_type: + lazy_type = dataclass_type # Fall back to original + + # Build context stack + stack = build_context_stack( + context_obj=context_obj, + overlay=self._concrete_values.get(scope_id, {}), + dataclass_type=lazy_type, + live_context=live_context, + ) + + # Resolve by creating fresh lazy instance within context + # This is the same approach used by get_lazy_resolved_placeholder() + with stack: + try: + instance = lazy_type() + resolved_value = getattr(instance, field_name) + logger.debug(f" ✅ Resolved via lazy instance: {repr(resolved_value)[:50]}") + except Exception as e: + logger.debug(f" ⚠️ Failed to resolve: {e}") + resolved_value = None + + # Cache result + self._resolved_cache[scope_id][cache_key] = resolved_value + + return resolved_value + + def get_materialized_object(self, scope_id: str) -> Any: + """Get context_obj with all resolved values applied. Returns NEW instance. + + This is used for compilation and code generation - creates a concrete instance + with all lazy fields resolved and concrete values applied. + + Args: + scope_id: Scope identifier + + Returns: + New instance with resolved values applied + """ + if scope_id not in self._scopes: + logger.warning(f"Cannot materialize unregistered scope: {scope_id}") + return None + + context_obj, dataclass_type = self._scopes[scope_id] + + # Resolve all fields using the new resolution API + materialized_kwargs = {} + + if is_dataclass(context_obj): + for field in fields(context_obj): + field_name = field.name + # Use resolve() which properly handles lazy dataclass resolution + resolved_value = self.resolve(scope_id, dataclass_type, field_name) + materialized_kwargs[field_name] = resolved_value + + # Create new instance + try: + return dataclass_type(**materialized_kwargs) + except Exception as e: + logger.error(f"Failed to materialize {scope_id}: {e}") + return context_obj # Fallback to original + + def get_resolved_state(self, scope_id: str) -> Dict[str, Any]: + """Get complete resolved state for dirty tracking. + + Returns dict of all field paths to their resolved values. + + Args: + scope_id: Scope identifier + + Returns: + Dict mapping field_path -> resolved_value + """ + if scope_id not in self._scopes: + return {} + + context_obj, dataclass_type = self._scopes[scope_id] + resolved_state = {} + + # Get all field paths and resolve each + if is_dataclass(context_obj): + for field in fields(context_obj): + field_name = field.name + resolved_state[field_name] = self.resolve(scope_id, dataclass_type, field_name) + + return resolved_state + + def _build_live_context(self, scope_id: str) -> Dict[type, Dict[str, Any]]: + """Build live context dict from all registered scopes. + + This replaces LiveContextService.collect() for registry-based resolution. + + Args: + scope_id: Current scope being resolved + + Returns: + Dict mapping dataclass_type -> field_values + """ + live_context = {} + + for other_scope_id, (other_obj, other_type) in self._scopes.items(): + if other_scope_id == scope_id: + continue # Skip self + + # Get base type for lazy dataclasses + base_type = get_base_type_for_lazy(other_type) + + # Merge concrete values with object's fields + field_values = {} + + if is_dataclass(other_obj): + for field in fields(other_obj): + field_name = field.name + # Prefer concrete values over object's values + if field_name in self._concrete_values.get(other_scope_id, {}): + field_values[field_name] = self._concrete_values[other_scope_id][field_name] + else: + field_values[field_name] = getattr(other_obj, field_name, None) + + live_context[base_type] = field_values + + return live_context + + def _invalidate_cache(self, scope_id: str, field_path: str) -> None: + """Invalidate resolved cache for scope and all visible descendants. + + When a value changes, we need to invalidate cache for: + 1. The scope itself + 2. All scopes where this scope is visible (descendants) + + Args: + scope_id: Scope that changed + field_path: Field that changed + """ + # Invalidate self + self._resolved_cache[scope_id] = {} + + # Invalidate all scopes where this scope is visible + editing_root = get_root_from_scope_key(scope_id) + + for other_scope_id in self._scopes.keys(): + if other_scope_id == scope_id: + continue + + # Check if editing_scope_id is visible to other_scope_id + if self._is_scope_visible(scope_id, other_scope_id): + self._resolved_cache[other_scope_id] = {} + + def _is_scope_visible(self, editing_scope_id: Optional[str], target_scope_id: Optional[str]) -> bool: + """Check if edits from editing_scope_id should affect target_scope_id. + + Uses same visibility rules as ParameterFormManager and ResolvedItemStateService. + + Args: + editing_scope_id: Scope where edit happened + target_scope_id: Scope to check visibility for + + Returns: + True if editing_scope affects target_scope + """ + # Guard against None scope IDs + if editing_scope_id is None or target_scope_id is None: + return False + + # Global config edits affect everything + if 'Global' in editing_scope_id: + return True + + # Same scope + if editing_scope_id == target_scope_id: + return True + + # Same root (same plate) + editing_root = get_root_from_scope_key(editing_scope_id) + target_root = get_root_from_scope_key(target_scope_id) + + if editing_root and target_root: + return editing_root == target_root + + return False + + def clear_all(self) -> None: + """Clear all state. Called on pipeline close/reset.""" + self._scopes.clear() + self._concrete_values.clear() + self._resolved_cache.clear() + logger.info("Cleared all registry state") + diff --git a/openhcs/pyqt_gui/shared/style_generator.py b/openhcs/pyqt_gui/shared/style_generator.py index a5cc5eb1d..8b2131130 100644 --- a/openhcs/pyqt_gui/shared/style_generator.py +++ b/openhcs/pyqt_gui/shared/style_generator.py @@ -103,14 +103,18 @@ def generate_tree_widget_style(self) -> str: QTreeWidget::item, QListWidget::item {{ padding: 4px; border-bottom: 1px solid {cs.to_hex(cs.border_color)}; + background-color: transparent; }} - QTreeWidget::item:hover, QListWidget::item:hover {{ + QTreeWidget::item:hover {{ background-color: {cs.to_hex(cs.hover_bg)}; }} - QTreeWidget::item:selected, QListWidget::item:selected {{ + QTreeWidget::item:selected {{ background-color: {cs.to_hex(cs.selection_bg)}; color: {cs.to_hex(cs.selection_text)}; }} + QListWidget::item:hover, QListWidget::item:selected {{ + background-color: transparent; + }} QHeaderView::section {{ background-color: {cs.to_hex(cs.panel_bg)}; color: {cs.to_hex(cs.text_primary)}; @@ -495,13 +499,13 @@ def generate_plate_manager_style(self) -> str: QListWidget::item {{ padding: 5px; border: none; + background-color: transparent; }} QListWidget::item:selected {{ - background-color: {cs.to_hex(cs.selection_bg)}; - color: {cs.to_hex(cs.selection_text)}; + background-color: transparent; }} QListWidget::item:hover {{ - background-color: {cs.to_hex(cs.hover_bg)}; + background-color: transparent; }} QFrame {{ background-color: {cs.to_hex(cs.window_bg)}; diff --git a/openhcs/pyqt_gui/widgets/pipeline_editor.py b/openhcs/pyqt_gui/widgets/pipeline_editor.py index ccd3d7667..74640615d 100644 --- a/openhcs/pyqt_gui/widgets/pipeline_editor.py +++ b/openhcs/pyqt_gui/widgets/pipeline_editor.py @@ -36,6 +36,7 @@ # Import ABC base class (Phase 4 migration) from openhcs.pyqt_gui.widgets.shared.abstract_manager_widget import AbstractManagerWidget +from openhcs.pyqt_gui.widgets.shared.scope_visual_config import ListItemType from openhcs.utils.performance_monitor import timer @@ -81,19 +82,14 @@ class PipelineEditorWidget(AbstractManagerWidget): 'items_changed_signal': 'pipeline_changed', # emit on changes 'preserve_selection_pred': lambda self: bool(self.pipeline_steps), 'list_item_data': 'index', # store index, not item - } - - # Declarative item hooks (replaces 9 trivial method overrides) - ITEM_HOOKS = { - 'id_accessor': ('attr', 'name'), # getattr(item, 'name', '') - 'backing_attr': 'pipeline_steps', # self.pipeline_steps - 'selection_attr': 'selected_step', # self.selected_step = ... - 'selection_signal': 'step_selected', # self.step_selected.emit(...) - 'selection_emit_id': False, # emit the full step object - 'selection_clear_value': None, # emit None when cleared - 'items_changed_signal': 'pipeline_changed', # self.pipeline_changed.emit(...) - 'preserve_selection_pred': lambda self: bool(self.pipeline_steps), - 'list_item_data': 'index', # store the step index + # Scope coloring - builder for composite scope_id + # REGISTRY REFACTOR: Removed @idx suffix - scope_id identifies STEP, not position + # Visual feedback (border patterns) falls back to hashing token if no @index + 'scope_item_type': ListItemType.STEP, + 'scope_id_builder': lambda item, idx, w: ( + f"{orch.plate_path}::{getattr(item, '_pipeline_scope_token', f'step_{idx}')}" + if (orch := w._get_current_orchestrator()) else None + ), } # Declarative preview field configuration (processed automatically in ABC.__init__) @@ -390,6 +386,7 @@ def handle_save(edited_step): on_save_callback=handle_save, orchestrator=orchestrator, gui_config=self.gui_config, + step_position=len(self.pipeline_steps), # New step will be at end parent=self ) # Set original step for change detection @@ -436,6 +433,9 @@ def action_auto_load_pipeline(self): self.update_item_list() self.pipeline_changed.emit(self.pipeline_steps) self.status_message.emit(f"Auto-loaded {len(new_pipeline_steps)} steps from basic_pipeline.py") + + # Set baselines for all steps (marks them as "saved") + self._set_all_step_baselines() else: raise ValueError("No 'pipeline_steps = [...]' assignment found in basic_pipeline.py") @@ -520,6 +520,9 @@ def _apply_executed_code(self, namespace: dict) -> bool: # Broadcast to global event bus for ALL windows to receive self._broadcast_to_event_bus('pipeline', new_pipeline_steps) + + # Set baselines for all steps (marks them as "saved") + self._set_all_step_baselines() return True def _get_code_missing_error_message(self) -> str: @@ -554,6 +557,9 @@ def load_pipeline_from_file(self, file_path: Path): self.update_item_list() self.pipeline_changed.emit(self.pipeline_steps) self.status_message.emit(f"Loaded {len(steps)} steps from {file_path.name}") + + # Set baselines for all steps (marks them as "saved") + self._set_all_step_baselines() else: self.status_message.emit(f"Invalid pipeline format in {file_path.name}") @@ -564,7 +570,7 @@ def load_pipeline_from_file(self, file_path: Path): def save_pipeline_to_file(self, file_path: Path): """ Save pipeline to file (extracted from Textual version). - + Args: file_path: Path to save pipeline """ @@ -573,7 +579,10 @@ def save_pipeline_to_file(self, file_path: Path): with open(file_path, 'wb') as f: pickle.dump(list(self.pipeline_steps), f) self.status_message.emit(f"Saved pipeline to {file_path.name}") - + + # Set baselines for all steps (marks them as "saved") + self._set_all_step_baselines() + except Exception as e: logger.error(f"Failed to save pipeline: {e}") self.service_adapter.show_error_dialog(f"Failed to save pipeline: {e}") @@ -588,7 +597,26 @@ def save_pipeline_for_plate(self, plate_path: str, pipeline: List[FunctionStep]) """ self.plate_pipelines[plate_path] = pipeline logger.debug(f"Saved pipeline for plate: {plate_path}") - + + def _set_all_step_baselines(self) -> None: + """Set baselines for all steps in ResolvedItemStateService. + + Called after save/load to mark current resolved state as "saved". + + REGISTRY REFACTOR: Uses registry.get_resolved_state() for dirty tracking. + """ + from openhcs.pyqt_gui.widgets.shared.services.resolved_item_state_service import ( + ResolvedItemStateService + ) + + service = ResolvedItemStateService.instance() + + for step in self.pipeline_steps: + scope_id = self._build_step_scope_id(step) + if scope_id: + service.set_baseline(scope_id) + logger.debug(f"Set baseline for step: {scope_id}") + def set_current_plate(self, plate_path: str): """ Set current plate and load its pipeline (extracted from Textual version). @@ -867,6 +895,13 @@ def _show_item_editor(self, item: Any) -> None: from openhcs.pyqt_gui.windows.dual_editor_window import DualEditorWindow + # Find step position for scope-based border styling + step_position = None + for i, step in enumerate(self.pipeline_steps): + if step is step_to_edit: + step_position = i + break + def handle_save(edited_step): """Handle step save from editor.""" # Find and replace the step in the pipeline @@ -889,6 +924,7 @@ def handle_save(edited_step): on_save_callback=handle_save, orchestrator=orchestrator, gui_config=self.gui_config, + step_position=step_position, parent=self ) # Set original step for change detection @@ -953,6 +989,21 @@ def _get_context_stack_for_resolution(self, item: Any) -> List[Any]: item # step ] + def _get_config_source_for_item(self, item: Any) -> Any: + """Step is its own config source (required abstract method).""" + return item # FunctionStep contains its own configs + + def _get_scope_id_for_item(self, item: Any) -> Optional[str]: + """Get scope_id for a step item (required abstract method). + + REGISTRY REFACTOR: Returns step's scope_id. + + Item is a FunctionStep - build scope_id from current_plate and step token. + """ + if isinstance(item, FunctionStep): + return self._build_step_scope_id(item) + return None + # === CrossWindowPreviewMixin Hook === # _get_current_orchestrator() is implemented above (line ~795) - does actual lookup from plate manager # _configure_preview_fields() REMOVED - now uses declarative PREVIEW_FIELD_CONFIGS (line ~99) diff --git a/openhcs/pyqt_gui/widgets/plate_manager.py b/openhcs/pyqt_gui/widgets/plate_manager.py index 660438a56..a9112740a 100644 --- a/openhcs/pyqt_gui/widgets/plate_manager.py +++ b/openhcs/pyqt_gui/widgets/plate_manager.py @@ -44,6 +44,7 @@ from openhcs.pyqt_gui.widgets.shared.services.zmq_execution_service import ZMQExecutionService from openhcs.pyqt_gui.widgets.shared.services.compilation_service import CompilationService from openhcs.pyqt_gui.widgets.shared.services.live_context_service import LiveContextService +from openhcs.pyqt_gui.widgets.shared.scope_visual_config import ListItemType logger = logging.getLogger(__name__) @@ -86,6 +87,9 @@ class PlateManagerWidget(AbstractManagerWidget): 'selection_emit_id': True, 'selection_clear_value': '', 'items_changed_signal': None, 'list_item_data': 'item', 'preserve_selection_pred': lambda self: bool(self.orchestrators), + # Scope coloring - simple attribute lookup + 'scope_item_type': ListItemType.ORCHESTRATOR, + 'scope_id_attr': 'path', } # Signals @@ -505,6 +509,24 @@ def _open_config_window(self, config_class, current_config, on_save_callback, or config_window.raise_() config_window.activateWindow() + def _set_all_orchestrator_baselines(self) -> None: + """Set baselines for all orchestrators in ResolvedItemStateService. + + Called after config save to mark current resolved state as "saved". + + REGISTRY REFACTOR: Uses registry.get_resolved_state() for dirty tracking. + """ + from openhcs.pyqt_gui.widgets.shared.services.resolved_item_state_service import ( + ResolvedItemStateService + ) + + service = ResolvedItemStateService.instance() + + for plate_path in self.orchestrators.keys(): + scope_id = str(plate_path) + service.set_baseline(scope_id) + logger.debug(f"Set baseline for orchestrator: {scope_id}") + def action_edit_global_config(self): """Handle global configuration editing - affects all orchestrators.""" current_global_config = self.service_adapter.get_global_config() or GlobalPipelineConfig() @@ -517,6 +539,9 @@ def handle_global_config_save(new_config: GlobalPipelineConfig) -> None: self._update_orchestrator_global_config(orchestrator, new_config) self.service_adapter.show_info_dialog("Global configuration applied to all orchestrators") + # Set baselines for all orchestrators (marks them as "saved") + self._set_all_orchestrator_baselines() + self._open_config_window( config_class=GlobalPipelineConfig, current_config=current_global_config, @@ -1167,6 +1192,35 @@ def _get_context_stack_for_resolution(self, item: Any) -> List[Any]: # Return raw objects - LiveContextResolver handles merging live values internally return [get_current_global_config(GlobalPipelineConfig), pipeline_config] + def _get_config_source_for_item(self, item: Any) -> Any: + """Get pipeline_config for a plate item (required abstract method). + + Item is a plate dict {'name': ..., 'path': ...}, so we look up + the orchestrator by path to get pipeline_config. + """ + if isinstance(item, dict) and 'path' in item: + orchestrator = self.orchestrators.get(item['path']) + if orchestrator: + return orchestrator.pipeline_config + return None # No config source if no orchestrator + + def _get_scope_id_for_item(self, item: Any) -> Optional[str]: + """Get scope_id for a plate item (required abstract method). + + REGISTRY REFACTOR: Returns plate path as scope_id. + + Item can be: + - PipelineOrchestrator: return orchestrator.plate_path + - dict with 'path': return item['path'] + """ + from openhcs.core.orchestrator.orchestrator import PipelineOrchestrator + + if isinstance(item, PipelineOrchestrator): + return str(item.plate_path) + elif isinstance(item, dict) and 'path' in item: + return str(item['path']) + return None + # === CrossWindowPreviewMixin Hook === def _get_current_orchestrator(self): diff --git a/openhcs/pyqt_gui/widgets/shared/abstract_manager_widget.py b/openhcs/pyqt_gui/widgets/shared/abstract_manager_widget.py index 5e9f6b8ab..abdbdffde 100644 --- a/openhcs/pyqt_gui/widgets/shared/abstract_manager_widget.py +++ b/openhcs/pyqt_gui/widgets/shared/abstract_manager_widget.py @@ -101,6 +101,11 @@ class AbstractManagerWidget(QWidget, CrossWindowPreviewMixin, ABC, metaclass=_Co # 'preserve_selection_pred': Callable[[self], bool] - Predicate for selection preservation # 'list_item_data': 'item' | 'index' - What to store in UserRole (default: 'item') # + # # Scope coloring (optional - for visual feedback): + # 'scope_item_type': ListItemType | None - Enum value for item type (ORCHESTRATOR, STEP) + # 'scope_id_attr': str | None - Simple: scope_id = getattr(item, attr) + # 'scope_id_builder': Callable | None - Complex: scope_id = builder(item, index, widget) + # # Example (PlateManager): # ITEM_HOOKS = { # 'id_accessor': 'path', @@ -112,6 +117,8 @@ class AbstractManagerWidget(QWidget, CrossWindowPreviewMixin, ABC, metaclass=_Co # 'items_changed_signal': None, # 'preserve_selection_pred': lambda self: bool(self.orchestrators), # 'list_item_data': 'item', + # 'scope_item_type': ListItemType.ORCHESTRATOR, + # 'scope_id_attr': 'path', # } ITEM_HOOKS: Dict[str, Any] = {} @@ -156,7 +163,8 @@ def __init__(self, service_adapter, color_scheme=None, gui_config=None, parent=N self.status_label: Optional[QLabel] = None self.item_list: Optional[ReorderableListWidget] = None - # Live context resolver for config attribute resolution + # REGISTRY REFACTOR: Use ContextStackRegistry instead of LiveContextResolver + # (LiveContextResolver still used for _merge_with_live_values until that's refactored) self._live_context_resolver = LiveContextResolver() # Initialize CrossWindowPreviewMixin @@ -165,6 +173,9 @@ def __init__(self, service_adapter, color_scheme=None, gui_config=None, parent=N # Process declarative preview field configs (AFTER mixin init) self._process_preview_field_configs() + # Connect to ResolvedItemStateService for reactive updates + self._connect_resolved_item_state_service() + def _get_default_gui_config(self): """Get default GUI config fallback.""" from openhcs.pyqt_gui.config import get_default_pyqt_gui_config @@ -190,6 +201,120 @@ def _process_preview_field_configs(self) -> None: else: logger.warning(f"Invalid PREVIEW_FIELD_CONFIGS entry: {config}") + def _connect_resolved_item_state_service(self) -> None: + """Connect to ResolvedItemStateService for reactive flash and dirty updates.""" + from openhcs.pyqt_gui.widgets.shared.services.resolved_item_state_service import ( + ResolvedItemStateService + ) + service = ResolvedItemStateService.instance() + service.item_resolved_changed.connect(self._on_item_resolved_changed) + service.item_dirty_changed.connect(self._on_item_dirty_changed) + logger.info(f"🔗 Connected to ResolvedItemStateService: {type(self).__name__}") + + def _on_item_resolved_changed(self, scope_id: str, field_name: str, new_value) -> None: + """Handle resolved value change - trigger flash animation. + + Called by ResolvedItemStateService when a field's resolved value changes. + """ + logger.info(f"⚡ _on_item_resolved_changed: scope_id={scope_id}, field={field_name}") + if self.item_list is None: + logger.info(f" ⚠️ item_list is None") + return + + # Find the row for this scope_id + row = self._find_row_for_scope_id(scope_id) + if row is None: + logger.info(f" ⚠️ No row found for scope_id={scope_id}") + return + + # Get scope info for flash animation + item = self._get_backing_items()[row] + scope_info = self._get_list_item_scope(item, row) + if scope_info is None: + logger.info(f" ⚠️ No scope_info for row={row}") + return + + scope_id_actual, item_type = scope_info + logger.info(f" 🎯 Triggering flash for row={row}, scope_id={scope_id_actual}") + + # Trigger flash animation + from openhcs.pyqt_gui.widgets.shared.list_item_flash_animation import flash_list_item + flash_list_item(self.item_list, row, scope_id_actual, item_type) + logger.info(f" ✅ Flash triggered for {scope_id}.{field_name}") + + def _on_item_dirty_changed(self, scope_id: str, is_dirty: bool) -> None: + """Handle dirty status change - update display with unsaved marker. + + Called by ResolvedItemStateService when item dirty status changes. + """ + # Find the row for this scope_id + row = self._find_row_for_scope_id(scope_id) + if row is None: + return + + # Update the list item text to show/hide dirty marker + list_item = self.item_list.item(row) + if list_item is None: + return + + current_text = list_item.text() + has_marker = current_text.startswith("* ") + + if is_dirty and not has_marker: + list_item.setText(f"* {current_text}") + elif not is_dirty and has_marker: + list_item.setText(current_text[2:]) + + logger.debug(f"Dirty marker updated for {scope_id}: {is_dirty}") + + def _find_row_for_scope_id(self, scope_id: str) -> int | None: + """Find list row for a given scope_id.""" + if self.item_list is None: + return None + + backing_items = self._get_backing_items() + for row, item in enumerate(backing_items): + scope_info = self._get_list_item_scope(item, row) + if scope_info and scope_info[0] == scope_id: + return row + return None + + def _register_items_with_state_service(self, items: list, _update_context) -> None: + """Register items with ResolvedItemStateService for flash/dirty tracking. + + Called during update_item_list to ensure all items are tracked. + No field list needed - lazydataclass inheritance determines propagation. + + REGISTRY REFACTOR: Also registers scopes with ContextStackRegistry and sets baseline. + """ + from openhcs.config_framework import ContextStackRegistry + from openhcs.pyqt_gui.widgets.shared.services.resolved_item_state_service import ( + ResolvedItemStateService + ) + + registry = ContextStackRegistry.instance() + service = ResolvedItemStateService.instance() + + for index, item in enumerate(items): + scope_info = self._get_list_item_scope(item, index) + if scope_info is None: + continue + + scope_id, _ = scope_info + + # Register scope with ContextStackRegistry (for preview label resolution) + # This ensures scopes are registered BEFORE preview labels are built + # Use _get_config_source_for_item to get the context_obj (pipeline_config or step) + context_obj = self._get_config_source_for_item(item) + if context_obj: + dataclass_type = type(context_obj) + registry.register_scope(scope_id, context_obj, dataclass_type) + + # Register with state service for dirty tracking + service.register_item(scope_id) + # Set baseline for dirty tracking (captures current resolved state) + service.set_baseline(scope_id) + # ========== UI Infrastructure (Concrete) ========== def setup_ui(self) -> None: @@ -830,6 +955,61 @@ def _patch_lazy_constructors(self): from openhcs.introspection import patch_lazy_constructors return patch_lazy_constructors() + # ========== Scope Coloring (Declarative via ITEM_HOOKS) ========== + + def _get_list_item_scope(self, item: Any, index: int) -> Optional[Tuple[str, Any]]: + """Get scope info for item. Pure data-driven - no branching on item type. + + Returns: + Tuple of (scope_id, item_type) or None if scope coloring not configured. + """ + hooks = self.ITEM_HOOKS + item_type = hooks.get('scope_item_type') + if not item_type: + return None + + # Two declarative options - builder takes precedence + builder = hooks.get('scope_id_builder') + if builder: + scope_id = builder(item, index, self) + else: + scope_id_attr = hooks.get('scope_id_attr') + if scope_id_attr: + # Use same pattern as _get_item_id for attribute resolution + if isinstance(item, dict): + scope_id = item.get(scope_id_attr) + else: + scope_id = getattr(item, scope_id_attr, None) + else: + scope_id = None + + return (scope_id, item_type) if scope_id else None + + # Custom data role for scope border color (UserRole+10 to avoid conflicts) + SCOPE_BORDER_ROLE = Qt.ItemDataRole.UserRole + 10 + + def _apply_list_item_scope_color(self, list_item: QListWidgetItem, item: Any, index: int) -> None: + """Apply scope-based background and border colors. Called by update_item_list().""" + scope_info = self._get_list_item_scope(item, index) + if not scope_info: + logger.info(f"🎨 No scope info for item {index}") + return + + scope_id, item_type = scope_info + from openhcs.pyqt_gui.widgets.shared.scope_color_utils import get_scope_color_scheme + scheme = get_scope_color_scheme(scope_id) + + # Set background color + bg_color = item_type.get_background_color(scheme) + if bg_color: + list_item.setBackground(bg_color) + logger.info(f"🎨 Set background for item {index}: {bg_color.name()}") + + # Set border color (stored in custom role, drawn by delegate) + border_color = scheme.to_qcolor_orchestrator_border() + list_item.setData(self.SCOPE_BORDER_ROLE, border_color) + logger.info(f"🎨 Set border for item {index}: {border_color.name()} (role={self.SCOPE_BORDER_ROLE})") + # ========== List Update Template ========== def update_item_list(self) -> None: @@ -862,6 +1042,10 @@ def update_item_list(self) -> None: def update_func(): """Update with in-place optimization when structure unchanged.""" backing_items = self._get_backing_items() + + # REGISTRY REFACTOR: Register scopes BEFORE formatting (so preview labels can resolve) + self._register_items_with_state_service(backing_items, update_context) + current_count = self.item_list.count() expected_count = len(backing_items) @@ -882,6 +1066,9 @@ def update_func(): # Extra data (e.g., enabled flag) for role_offset, value in self._get_list_item_extra_data(item_obj, index).items(): list_item.setData(Qt.ItemDataRole.UserRole + role_offset, value) + + # Apply scope-based background color (if configured via ITEM_HOOKS) + self._apply_list_item_scope_color(list_item, item_obj, index) else: # Structure changed - rebuild list self.item_list.clear() @@ -894,6 +1081,9 @@ def update_func(): for role_offset, value in self._get_list_item_extra_data(item_obj, index).items(): list_item.setData(Qt.ItemDataRole.UserRole + role_offset, value) + # Apply scope-based background color (if configured via ITEM_HOOKS) + self._apply_list_item_scope_color(list_item, item_obj, index) + self.item_list.addItem(list_item) # Post-update (e.g., auto-select first) @@ -1072,57 +1262,43 @@ def _resolve_preview_field_value( fallback_context: Optional[Dict[str, Any]] = None, ) -> Any: """ - Resolve a preview field path using the live context resolver. + Resolve a preview field path using the registry. - For dotted paths like 'path_planning_config.well_filter': - 1. Use getattr to navigate to the config object (preserves lazy type) - 2. Use _resolve_config_attr only for the final attribute (triggers MRO resolution) + REGISTRY REFACTOR: Uses registry.resolve() instead of LiveContextResolver. Args: item: Semantic item for context stack (orchestrator/plate dict or step) config_source: Root config object to resolve from (pipeline_config or step) field_path: Dot-separated field path (e.g., 'napari_streaming_config' or 'vfs_config.backend') - live_context_snapshot: Optional live context for resolving lazy values + live_context_snapshot: DEPRECATED - not used (registry is single source of truth) fallback_context: Optional context dict for fallback resolver Returns: Resolved value or None """ - parts = field_path.split('.') + from openhcs.config_framework import ContextStackRegistry - if len(parts) == 1: - # Simple field - resolve directly - return self._resolve_config_attr( - item, - config_source, - parts[0], - live_context_snapshot - ) - - # Dotted path: navigate to parent config using getattr, then resolve final attr - # This preserves the lazy type (e.g., LazyPathPlanningConfig) so MRO resolution works - current_obj = config_source - for part in parts[:-1]: - if current_obj is None: - return self._apply_preview_field_fallback(field_path, fallback_context) - current_obj = getattr(current_obj, part, None) - - if current_obj is None: + # Get scope_id from item + scope_id = self._get_scope_id_for_item(item) + logger.debug(f"🔎 _resolve_preview_field_value: field_path={field_path}, scope_id={scope_id}") + if not scope_id: + logger.debug(f" ❌ No scope_id, using fallback") return self._apply_preview_field_fallback(field_path, fallback_context) - # Resolve final attribute using live context resolver (triggers MRO inheritance) - resolved_value = self._resolve_config_attr( - item, - current_obj, - parts[-1], - live_context_snapshot - ) - - if resolved_value is None: + # Resolve via registry - uses stored dataclass_type for this scope + # field_path is the field_name (e.g., 'napari_streaming_config') + registry = ContextStackRegistry.instance() + try: + resolved_value = registry.resolve(scope_id, field_name=field_path) + logger.debug(f" ✅ Resolved: {repr(resolved_value)[:50]}") + if resolved_value is None: + logger.debug(f" ⚠️ Value is None, using fallback") + return self._apply_preview_field_fallback(field_path, fallback_context) + return resolved_value + except Exception as e: + logger.warning(f"Failed to resolve {field_path} via registry: {e}") return self._apply_preview_field_fallback(field_path, fallback_context) - return resolved_value - # === Config Preview Building (shared by both widgets) === def _build_preview_labels( @@ -1150,9 +1326,12 @@ def _build_preview_labels( """ from openhcs.pyqt_gui.widgets.config_preview_formatters import format_config_indicator + enabled_fields = self.get_enabled_preview_fields() + logger.debug(f"📋 _build_preview_labels: item={type(item).__name__}, enabled_fields={enabled_fields}") + labels = [] - for field_path in self.get_enabled_preview_fields(): + for field_path in enabled_fields: value = self._resolve_preview_field_value( item=item, config_source=config_source, @@ -1162,6 +1341,7 @@ def _build_preview_labels( ) if value is None: + logger.debug(f" ⏭️ Skipping {field_path} (value is None)") continue # Check if value is a dataclass config object @@ -1279,13 +1459,36 @@ def _emit_items_changed(self) -> None: signal = getattr(self, signal_name) signal.emit(self._get_backing_items()) - # === Config Resolution Hook (subclass must implement) === + # === Config Resolution Hooks (subclass must implement) === @abstractmethod def _get_context_stack_for_resolution(self, item: Any) -> List[Any]: """Build context stack for config resolution. Subclass must implement.""" ... + @abstractmethod + def _get_config_source_for_item(self, item: Any) -> Any: + """Get config source for an item. Used for reactive state tracking. + + PipelineEditor: return item (step is its own config source) + PlateManager: return orchestrator.pipeline_config + """ + ... + + @abstractmethod + def _get_scope_id_for_item(self, item: Any) -> Optional[str]: + """Get scope_id for an item for registry resolution. + + REGISTRY REFACTOR: Required for preview field resolution via registry. + + PipelineEditor: return step's scope_id (e.g., '/path/to/plate::step_token@index') + PlateManager: return plate path (e.g., '/path/to/plate') + + Returns: + Scope ID string or None if item is invalid + """ + ... + # === CrossWindowPreviewMixin Hook (declarative default) === def _handle_full_preview_refresh(self) -> None: diff --git a/openhcs/pyqt_gui/widgets/shared/list_item_delegate.py b/openhcs/pyqt_gui/widgets/shared/list_item_delegate.py index 6776ddb5f..eea5fcabf 100644 --- a/openhcs/pyqt_gui/widgets/shared/list_item_delegate.py +++ b/openhcs/pyqt_gui/widgets/shared/list_item_delegate.py @@ -5,10 +5,13 @@ and other widgets that display items with preview labels. """ +import logging from PyQt6.QtWidgets import QStyledItemDelegate, QStyleOptionViewItem, QStyle -from PyQt6.QtGui import QPainter, QColor, QFontMetrics +from PyQt6.QtGui import QPainter, QColor, QFontMetrics, QPen from PyQt6.QtCore import Qt, QRect +logger = logging.getLogger(__name__) + class MultilinePreviewItemDelegate(QStyledItemDelegate): """Custom delegate to render multiline items with grey preview text. @@ -41,6 +44,8 @@ def __init__(self, name_color: QColor, preview_color: QColor, selected_text_colo def paint(self, painter: QPainter, option: QStyleOptionViewItem, index) -> None: """Paint the item with multiline support and grey preview text.""" + from PyQt6.QtGui import QBrush + # Prepare a copy to let style draw backgrounds, hover, selection, borders, etc. opt = QStyleOptionViewItem(option) self.initStyleOption(opt, index) @@ -49,7 +54,36 @@ def paint(self, painter: QPainter, option: QStyleOptionViewItem, index) -> None: text = opt.text or "" opt.text = "" - # Let the style draw background, selection, hover, borders + # Check selection/hover state + is_selected = option.state & QStyle.StateFlag.State_Selected + is_hover = option.state & QStyle.StateFlag.State_MouseOver + + # Draw scope-based background color from item's BackgroundRole + # This is set via QListWidgetItem.setBackground() in _apply_list_item_scope_color + item_bg = index.data(Qt.ItemDataRole.BackgroundRole) + if item_bg: + # Get the color from the brush + bg_color = item_bg.color() + + # Invert opacity for selection: normal=30-40%, selected=70-80% + if is_selected: + # Invert: if alpha was 40% (102), make it 85% (217) + inverted_alpha = min(255, 255 - bg_color.alpha() + 50) + bg_color.setAlpha(inverted_alpha) + elif is_hover: + # Slight boost for hover + boosted_alpha = min(255, bg_color.alpha() + 30) + bg_color.setAlpha(boosted_alpha) + + painter.fillRect(option.rect, bg_color) + + # Clear style's background and selection drawing - we handle it ourselves + opt.backgroundBrush = QBrush() + if is_selected: + # Remove selected state so style doesn't draw blue + opt.state = opt.state & ~QStyle.StateFlag.State_Selected + + # Let the style draw hover effects and borders (but not selection background) self.parent().style().drawControl(QStyle.ControlElement.CE_ItemViewItem, opt, painter, self.parent()) # Now draw text manually with custom colors @@ -130,6 +164,22 @@ def paint(self, painter: QPainter, option: QStyleOptionViewItem, index) -> None: y_offset += line_height painter.restore() + + # Draw scope border (stored in SCOPE_BORDER_ROLE = UserRole+10) + border_data = index.data(Qt.ItemDataRole.UserRole + 10) + if border_data is not None and isinstance(border_data, QColor): + rect = option.rect + # Check if there's any clipping set + clip_region = painter.clipRegion() + clip_rect = painter.clipBoundingRect() + logger.info(f"🎨 DELEGATE row={index.row()} rect=({rect.left()},{rect.top()},{rect.width()},{rect.height()}) " + f"clip_empty={clip_region.isEmpty()} clip_rect={clip_rect} color={border_data.name()}") + painter.save() + painter.setClipping(False) # Disable any clipping + # Draw left border (5px wide) as scope indicator - use fillRect for solid bar + border_rect = QRect(rect.left(), rect.top(), 5, rect.height()) + painter.fillRect(border_rect, border_data) + painter.restore() def sizeHint(self, option: QStyleOptionViewItem, index) -> 'QSize': """Calculate size hint based on number of lines in text.""" diff --git a/openhcs/pyqt_gui/widgets/shared/list_item_flash_animation.py b/openhcs/pyqt_gui/widgets/shared/list_item_flash_animation.py new file mode 100644 index 000000000..e4e280088 --- /dev/null +++ b/openhcs/pyqt_gui/widgets/shared/list_item_flash_animation.py @@ -0,0 +1,168 @@ +"""Flash animation for QListWidgetItem updates.""" + +import logging +from typing import Optional +from PyQt6.QtCore import QTimer +from PyQt6.QtWidgets import QListWidget +from PyQt6.QtGui import QColor + +from .scope_visual_config import ScopeVisualConfig, ListItemType + +logger = logging.getLogger(__name__) + + +class ListItemFlashAnimator: + """Manages flash animation for QListWidgetItem background color changes. + + Design: + - Does NOT store item references (items can be destroyed during flash) + - Stores (list_widget, row, scope_id, item_type) for color recomputation + - Gracefully handles item destruction (checks if item exists before restoring) + """ + + def __init__( + self, + list_widget: QListWidget, + row: int, + scope_id: str, + item_type: ListItemType + ): + """Initialize animator. + + Args: + list_widget: Parent list widget + row: Row index of item + scope_id: Scope identifier for color recomputation + item_type: Type of list item (orchestrator or step) + """ + self.list_widget = list_widget + self.row = row + self.scope_id = scope_id + self.item_type = item_type + self.config = ScopeVisualConfig() + self._flash_timer: Optional[QTimer] = None + self._is_flashing: bool = False + + def flash_update(self) -> None: + """Trigger flash animation on item background by increasing opacity.""" + item = self.list_widget.item(self.row) + if item is None: # Item was destroyed + logger.debug(f"Flash skipped - item at row {self.row} no longer exists") + return + + # Get the correct background color from scope + from .scope_color_utils import get_scope_color_scheme + color_scheme = get_scope_color_scheme(self.scope_id) + correct_color = self.item_type.get_background_color(color_scheme) + + if correct_color is not None: + # Increase opacity to 100% for flash (from 15% for orchestrator, 5% for steps) + flash_color = QColor(correct_color) + flash_color.setAlpha(255) # Full opacity + item.setBackground(flash_color) + + if self._is_flashing: + # Already flashing - restart timer (flash color already re-applied above) + if self._flash_timer: + self._flash_timer.stop() + self._flash_timer.start(self.config.FLASH_DURATION_MS) + return + + self._is_flashing = True + + # Setup timer to restore correct background + self._flash_timer = QTimer(self.list_widget) + self._flash_timer.setSingleShot(True) + self._flash_timer.timeout.connect(self._restore_background) + self._flash_timer.start(self.config.FLASH_DURATION_MS) + + def _restore_background(self) -> None: + """Restore correct background color by recomputing from scope.""" + item = self.list_widget.item(self.row) + if item is None: # Item was destroyed during flash + logger.debug(f"Flash restore skipped - item at row {self.row} was destroyed") + self._is_flashing = False + return + + # Recompute correct color from scope_id (handles list rebuilds during flash) + from PyQt6.QtGui import QBrush + from .scope_color_utils import get_scope_color_scheme + color_scheme = get_scope_color_scheme(self.scope_id) + + # Use enum-based polymorphic dispatch to get correct color + correct_color = self.item_type.get_background_color(color_scheme) + + # Handle None (transparent) background + if correct_color is None: + item.setBackground(QBrush()) # Empty brush = transparent + else: + item.setBackground(correct_color) + + self._is_flashing = False + + +# Global registry of animators (keyed by (list_widget_id, item_row)) +_list_item_animators: dict[tuple[int, int], ListItemFlashAnimator] = {} + + +def flash_list_item( + list_widget: QListWidget, + row: int, + scope_id: str, + item_type: ListItemType +) -> None: + """Flash a list item to indicate update. + + Args: + list_widget: List widget containing the item + row: Row index of item to flash + scope_id: Scope identifier for color recomputation + item_type: Type of list item (orchestrator or step) + """ + config = ScopeVisualConfig() + if not config.LIST_ITEM_FLASH_ENABLED: + return + + item = list_widget.item(row) + if item is None: + return + + key = (id(list_widget), row) + + # Get or create animator + if key not in _list_item_animators: + _list_item_animators[key] = ListItemFlashAnimator( + list_widget, row, scope_id, item_type + ) + else: + # Update scope_id and item_type in case item was recreated + animator = _list_item_animators[key] + animator.scope_id = scope_id + animator.item_type = item_type + + animator = _list_item_animators[key] + animator.flash_update() + + +def clear_all_animators(list_widget: QListWidget) -> None: + """Clear all animators for a specific list widget. + + Call this before clearing/rebuilding the list to prevent + flash timers from accessing destroyed items. + + Args: + list_widget: List widget whose animators should be cleared + """ + widget_id = id(list_widget) + keys_to_remove = [k for k in _list_item_animators.keys() if k[0] == widget_id] + + for key in keys_to_remove: + animator = _list_item_animators[key] + # Stop any active flash timers + if animator._flash_timer and animator._flash_timer.isActive(): + animator._flash_timer.stop() + del _list_item_animators[key] + + if keys_to_remove: + logger.debug(f"Cleared {len(keys_to_remove)} flash animators for list widget") + diff --git a/openhcs/pyqt_gui/widgets/shared/parameter_form_manager.py b/openhcs/pyqt_gui/widgets/shared/parameter_form_manager.py index 0e8a9b597..46bdb6c82 100644 --- a/openhcs/pyqt_gui/widgets/shared/parameter_form_manager.py +++ b/openhcs/pyqt_gui/widgets/shared/parameter_form_manager.py @@ -8,7 +8,7 @@ import dataclasses from dataclasses import dataclass, field, is_dataclass, fields as dataclass_fields import logging -from typing import Any, Dict, Type, Optional, Tuple, List +from typing import Any, Dict, Type, Optional, Tuple, List, Callable from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QScrollArea, QLabel, QPushButton, QLineEdit, QCheckBox, QComboBox, QGroupBox, QSpinBox, QDoubleSpinBox @@ -357,6 +357,10 @@ def __init__(self, object_instance: Any, field_id: str, config: Optional[FormMan # Debounce timer for cross-window placeholder refresh self._cross_window_refresh_timer = None + # Hook: callbacks invoked when a placeholder value changes + # Signature: callback(config_name: str, field_name: str) + self._on_placeholder_changed_callbacks: list[Callable[[str, str], None]] = [] + # STEP 8: _user_set_fields starts empty and is populated only when user edits widgets # (via _emit_parameter_change). Do NOT populate during initialization, as that would # include inherited values that weren't explicitly set by the user. @@ -369,6 +373,17 @@ def __init__(self, object_instance: Any, field_id: str, config: Optional[FormMan lambda name, manager: setattr(manager, '_initial_load_complete', True) ) + # STEP 9.5: Register with ContextStackRegistry (ROOT MANAGERS ONLY) + # Nested managers share parent's scope_id and should NOT register + # (they would REPLACE the root's registration with their nested config) + is_nested = self._parent_manager is not None + if not is_nested: + with timer(" Register with registry", threshold_ms=1.0): + from openhcs.config_framework import ContextStackRegistry + registry = ContextStackRegistry.instance() + registry.register_scope(self.scope_id, self.object_instance, self.dataclass_type) + logger.info(f"📦 Registered scope with registry: {self.scope_id}") + # STEP 10: Execute initial refresh strategy (enum dispatch) with timer(" Initial refresh", threshold_ms=10.0): InitialRefreshStrategy.execute(self) @@ -1101,6 +1116,8 @@ def unregister_from_cross_window_updates(self): SIMPLIFIED: Just unregister from LiveContextService. The token increment in unregister() notifies all listeners to refresh. + + REGISTRY REFACTOR: Also unregister from ContextStackRegistry. """ logger.info(f"🔍 UNREGISTER: {self.field_id}") @@ -1116,6 +1133,13 @@ def unregister_from_cross_window_updates(self): # Remove from registry (triggers token increment → notifies listeners) LiveContextService.unregister(self) + # REGISTRY REFACTOR: Unregister from ContextStackRegistry (ROOT MANAGERS ONLY) + if not self._parent_manager: + from openhcs.config_framework import ContextStackRegistry + registry = ContextStackRegistry.instance() + registry.unregister_scope(self.scope_id) + logger.info(f"📦 Unregistered scope from registry: {self.scope_id}") + except Exception as e: logger.warning(f"🔍 UNREGISTER: Error: {e}") @@ -1278,6 +1302,13 @@ def _schedule_cross_window_refresh(self, changed_field: Optional[str] = None, em Set to False when refresh is triggered by another window's context_refreshed to prevent infinite ping-pong loops. """ + # CRITICAL: Only ROOT managers should handle cross-window refresh. + # Nested managers don't have the flash callback set, and letting them + # handle signals would cause duplicate refreshes and broken flash animation. + if self._parent_manager is not None: + logger.debug(f"⏭️ SKIP_SCHEDULE [{self.field_id}]: nested manager, root will handle") + return + logger.info(f"⏰ SCHEDULE_REFRESH [{self.field_id}]: field={changed_field}, emit_signal={emit_signal}, scope={self.scope_id}") # Cancel existing timer if any @@ -1318,19 +1349,11 @@ def _refresh_field_in_tree(self, field_name: str): """Refresh a specific field's placeholder in this manager and all nested managers. The field might be in this manager directly, or in any nested manager. - We refresh it wherever it exists. - - Args: - field_name: The leaf field name to refresh (e.g., "well_filter") + We refresh it wherever it exists. Flash animation is handled via + _on_placeholder_changed_callbacks hook in ParameterOpsService. """ - has_widget = field_name in self.widgets - nested_names = list(self.nested_managers.keys()) - logger.info(f" 🌳 TREE [{self.field_id}]: field={field_name}, has_widget={has_widget}, nested={nested_names}") - - # Try to refresh in this manager - if has_widget: + if field_name in self.widgets: self._parameter_ops_service.refresh_single_placeholder(self, field_name) - # Also try in all nested managers (the field might be nested) for nested_manager in self.nested_managers.values(): nested_manager._refresh_field_in_tree(field_name) diff --git a/openhcs/pyqt_gui/widgets/shared/scope_color_strategy.py b/openhcs/pyqt_gui/widgets/shared/scope_color_strategy.py new file mode 100644 index 000000000..e29896a26 --- /dev/null +++ b/openhcs/pyqt_gui/widgets/shared/scope_color_strategy.py @@ -0,0 +1,128 @@ +"""Pluggable color generation strategies for scope-based styling. + +Follows Strategy Pattern to allow different color generation algorithms: +- MD5HashStrategy: Deterministic color from scope_id hash (default) +- ManualColorStrategy: User-selected colors with fallback to generated +- Future: SequentialStrategy, ThemeAwareStrategy, etc. +""" + +from abc import ABC, abstractmethod +from typing import Tuple, Dict, Optional +from enum import Enum, auto +import hashlib +import colorsys +import logging + +logger = logging.getLogger(__name__) + + +class ColorStrategyType(Enum): + """Available color generation strategies.""" + MD5_HASH = auto() # Deterministic from scope_id hash + MANUAL = auto() # User-selected color with fallback + + +class ScopeColorStrategy(ABC): + """Abstract base for color generation strategies.""" + + # Subclasses must set this class attribute + strategy_type: ColorStrategyType + + @abstractmethod + def generate_color(self, scope_id: str) -> Tuple[int, int, int]: + """Generate RGB color for scope. Returns (r, g, b) in 0-255 range.""" + ... + + +class MD5HashStrategy(ScopeColorStrategy): + """Deterministic color from MD5 hash of scope_id. Default strategy. + + Uses distinctipy for perceptually distinct colors when available, + falls back to HSV-based generation otherwise. + """ + + strategy_type = ColorStrategyType.MD5_HASH + PALETTE_SIZE = 50 + + def __init__(self): + self._palette: Optional[list] = None + + def _get_palette(self) -> list: + """Get or generate distinct color palette.""" + if self._palette is None: + self._palette = self._generate_palette() + return self._palette + + def _generate_palette(self) -> list: + """Generate perceptually distinct colors.""" + try: + from distinctipy import distinctipy + colors = distinctipy.get_colors( + self.PALETTE_SIZE, + exclude_colors=[(0, 0, 0), (1, 1, 1)], + pastel_factor=0.5 + ) + # Convert from 0-1 to 0-255 range + return [tuple(int(c * 255) for c in color) for color in colors] + except ImportError: + logger.debug("distinctipy not installed, using HSV fallback") + return self._generate_hsv_palette() + + def _generate_hsv_palette(self) -> list: + """Fallback HSV-based palette generation.""" + palette = [] + for i in range(self.PALETTE_SIZE): + hue = 360 * i / self.PALETTE_SIZE + r, g, b = colorsys.hsv_to_rgb(hue / 360, 0.5, 0.8) + palette.append((int(r * 255), int(g * 255), int(b * 255))) + return palette + + def _hash_to_index(self, scope_id: str) -> int: + """Generate deterministic palette index from scope_id.""" + hash_bytes = hashlib.md5(scope_id.encode('utf-8')).digest() + hash_int = int.from_bytes(hash_bytes[:4], byteorder='big') + return hash_int % self.PALETTE_SIZE + + def generate_color(self, scope_id: str) -> Tuple[int, int, int]: + """Generate color from MD5 hash of scope_id.""" + palette = self._get_palette() + index = self._hash_to_index(scope_id) + return palette[index] + + +class ManualColorStrategy(ScopeColorStrategy): + """User-selected colors with persistence. + + Falls back to MD5HashStrategy for scopes without manual colors. + """ + + strategy_type = ColorStrategyType.MANUAL + + def __init__(self): + self._colors: Dict[str, Tuple[int, int, int]] = {} + self._fallback = MD5HashStrategy() + + def set_color(self, scope_id: str, rgb: Tuple[int, int, int]) -> None: + """Set manual color for scope.""" + self._colors[scope_id] = rgb + + def clear_color(self, scope_id: str) -> None: + """Clear manual color, revert to fallback.""" + self._colors.pop(scope_id, None) + + def has_manual_color(self, scope_id: str) -> bool: + """Check if scope has manual color override.""" + return scope_id in self._colors + + def get_all_manual_colors(self) -> Dict[str, Tuple[int, int, int]]: + """Get all manual color assignments (for persistence).""" + return dict(self._colors) + + def load_manual_colors(self, colors: Dict[str, Tuple[int, int, int]]) -> None: + """Load manual colors from persistence.""" + self._colors.update(colors) + + def generate_color(self, scope_id: str) -> Tuple[int, int, int]: + """Get manual color or fall back to generated.""" + return self._colors.get(scope_id) or self._fallback.generate_color(scope_id) + diff --git a/openhcs/pyqt_gui/widgets/shared/scope_color_utils.py b/openhcs/pyqt_gui/widgets/shared/scope_color_utils.py new file mode 100644 index 000000000..448a7bb85 --- /dev/null +++ b/openhcs/pyqt_gui/widgets/shared/scope_color_utils.py @@ -0,0 +1,258 @@ +"""Utilities for generating scope-based colors using perceptually distinct palettes. + +This module provides: +- Helper functions for color manipulation and extraction +- The main get_scope_color_scheme() function that delegates to ScopeColorService +- Internal _build_color_scheme_from_rgb() for constructing schemes from base colors + +The actual color generation is handled by pluggable strategies via ScopeColorService. +""" + +import hashlib +import colorsys +import logging +from typing import Optional, Tuple + +from .scope_visual_config import ScopeColorScheme + +logger = logging.getLogger(__name__) + + +def _ensure_wcag_compliant( + color_rgb: tuple[int, int, int], + background: tuple[int, int, int] = (255, 255, 255), + min_ratio: float = 4.5 +) -> tuple[int, int, int]: + """Ensure color meets WCAG AA contrast requirements against background. + + Args: + color_rgb: RGB color tuple (0-255 range) + background: Background RGB color tuple (0-255 range), default white + min_ratio: Minimum contrast ratio (4.5 for WCAG AA normal text, 3.0 for large text) + + Returns: + Adjusted RGB color tuple that meets contrast requirements + """ + try: + from wcag_contrast_ratio.contrast import rgb as wcag_rgb + + # Convert to 0-1 range for wcag library + color_01 = tuple(c / 255.0 for c in color_rgb) + bg_01 = tuple(c / 255.0 for c in background) + + # Calculate current contrast ratio + current_ratio = wcag_rgb(color_01, bg_01) + + if current_ratio >= min_ratio: + return color_rgb # Already compliant + + # Darken color until it meets contrast requirements + # Convert to HSV for easier manipulation + h, s, v = colorsys.rgb_to_hsv(*color_01) + + # Reduce value (brightness) to increase contrast + while v > 0.1: # Don't go completely black + v *= 0.9 # Reduce by 10% each iteration + adjusted_rgb_01 = colorsys.hsv_to_rgb(h, s, v) + ratio = wcag_rgb(adjusted_rgb_01, bg_01) + + if ratio >= min_ratio: + # Convert back to 0-255 range + adjusted_rgb = tuple(int(c * 255) for c in adjusted_rgb_01) + logger.debug(f"Adjusted color from ratio {current_ratio:.2f} to {ratio:.2f}") + return adjusted_rgb + + # If we couldn't meet requirements by darkening, return darkest version + logger.warning(f"Could not meet WCAG contrast ratio {min_ratio} for color {color_rgb}") + return tuple(int(c * 255) for c in colorsys.hsv_to_rgb(h, s, 0.1)) + + except ImportError: + logger.warning("wcag-contrast-ratio not installed, skipping WCAG compliance check") + return color_rgb + except Exception as e: + logger.warning(f"WCAG compliance check failed: {e}") + return color_rgb + + +def extract_orchestrator_scope(scope_id: Optional[str]) -> Optional[str]: + """Extract orchestrator scope from a scope_id. + + Scope IDs follow the pattern: + - Orchestrator scope: "plate_path" (e.g., "/path/to/plate") + - Step scope: "plate_path::step_token" (e.g., "/path/to/plate::step_0") + + Args: + scope_id: Full scope identifier (can be orchestrator or step scope) + + Returns: + Orchestrator scope (plate_path) or None if scope_id is None + + Examples: + >>> extract_orchestrator_scope("/path/to/plate") + '/path/to/plate' + >>> extract_orchestrator_scope("/path/to/plate::step_0") + '/path/to/plate' + >>> extract_orchestrator_scope(None) + None + """ + if scope_id is None: + return None + + # Split on :: separator + if '::' in scope_id: + return scope_id.split('::', 1)[0] + else: + return scope_id + + + + + +def extract_step_index(scope_id: str) -> int: + """Extract per-orchestrator step index from step scope_id. + + The scope_id format is "plate_path::step_token@position" where position + is the step's index within its orchestrator's pipeline (0-based). + + This ensures each orchestrator has independent step indexing for visual styling. + + Args: + scope_id: Step scope in format "plate_path::step_token@position" + + Returns: + Step index (0-based) for visual styling, or 0 if not a step scope + """ + if '::' not in scope_id: + return 0 + + # Extract the part after :: + step_part = scope_id.split('::')[1] + + # Check if position is included (format: "step_token@position") + if '@' in step_part: + try: + position_str = step_part.split('@')[1] + return int(position_str) + except (IndexError, ValueError): + pass + + # Fallback for old format without @position: hash the step token + hash_bytes = hashlib.md5(step_part.encode()).digest() + return int.from_bytes(hash_bytes[:2], byteorder='big') % 27 + + +def hsv_to_rgb(hue: int, saturation: int, value: int) -> tuple[int, int, int]: + """Convert HSV color to RGB tuple. + + Args: + hue: Hue in range [0, 359] + saturation: Saturation in range [0, 100] + value: Value (brightness) in range [0, 100] + + Returns: + RGB tuple with values in range [0, 255] + """ + # Normalize to [0, 1] range for colorsys + h = hue / 360.0 + s = saturation / 100.0 + v = value / 100.0 + + # Convert to RGB + r, g, b = colorsys.hsv_to_rgb(h, s, v) + + # Scale to [0, 255] + return (int(r * 255), int(g * 255), int(b * 255)) + + +def get_scope_color_scheme(scope_id: Optional[str]) -> ScopeColorScheme: + """Get color scheme for scope via ScopeColorService. + + This is the main entry point for getting scope colors. Delegates to + ScopeColorService which handles caching, strategy selection, and reactive updates. + + Args: + scope_id: Scope identifier (can be orchestrator or step scope) + + Returns: + ScopeColorScheme with all derived colors and border info + """ + from .services.scope_color_service import ScopeColorService + return ScopeColorService.instance().get_color_scheme(scope_id) + + +def _build_color_scheme_from_rgb( + base_rgb: Tuple[int, int, int], + scope_id: str +) -> ScopeColorScheme: + """Build complete color scheme from base RGB color. + + Internal function used by ScopeColorService. Takes a base color (from any strategy) + and builds the full scheme with all derived colors and border layers. + + Args: + base_rgb: Base RGB color tuple (0-255 range) + scope_id: Scope identifier for step index extraction + + Returns: + ScopeColorScheme with all derived colors and border info + """ + # Extract orchestrator scope (removes step token if present) + orchestrator_scope = extract_orchestrator_scope(scope_id) + + # Orchestrator background is base color (transparency handled in to_qcolor methods) + orch_bg_rgb = base_rgb + + # Darker version for border (scale by ~0.78) + orch_border_rgb = tuple(int(c * 200 / 255) for c in base_rgb) + + # Get step index for border logic + step_index = extract_step_index(scope_id) if '::' in (scope_id or '') else 0 + + # Steps use same color as orchestrator + step_item_rgb = orch_bg_rgb + + # === BORDER LAYER CALCULATION === + # Cycle through 9 tint+pattern combinations before adding layers: + # - 3 tints (0=dark, 1=neutral, 2=bright) + # - 3 patterns (solid, dashed, dotted) + num_border_layers = (step_index // 9) + 1 + combo_index = step_index % 9 + step_pattern_index = combo_index // 3 + step_tint = combo_index % 3 + + border_patterns = ['solid', 'dashed', 'dotted'] + step_border_layers = [] + for layer in range(num_border_layers): + if layer == 0: + border_tint = step_tint + border_pattern = border_patterns[step_pattern_index] + else: + layer_combo = (combo_index + layer * 3) % 9 + border_tint = (layer_combo // 3) % 3 + border_pattern = border_patterns[layer_combo % 3] + step_border_layers.append((3, border_tint, border_pattern)) + + step_border_width = num_border_layers * 3 + + # === STEP WINDOW BORDER === + tint_index = step_index % 3 + tint_factors = [0.7, 1.0, 1.4] + tint_factor = tint_factors[tint_index] + step_window_rgb = tuple(min(255, int(c * tint_factor)) for c in base_rgb) + + # === WCAG COMPLIANCE === + orch_border_rgb = _ensure_wcag_compliant(orch_border_rgb, background=(255, 255, 255)) + step_window_rgb = _ensure_wcag_compliant(step_window_rgb, background=(255, 255, 255)) + + return ScopeColorScheme( + scope_id=orchestrator_scope, + hue=0, + orchestrator_item_bg_rgb=orch_bg_rgb, + orchestrator_item_border_rgb=orch_border_rgb, + step_window_border_rgb=step_window_rgb, + step_item_bg_rgb=step_item_rgb, + step_border_width=step_border_width, + step_border_layers=step_border_layers, + base_color_rgb=orch_bg_rgb, + ) + diff --git a/openhcs/pyqt_gui/widgets/shared/scope_visual_config.py b/openhcs/pyqt_gui/widgets/shared/scope_visual_config.py new file mode 100644 index 000000000..4c63338c3 --- /dev/null +++ b/openhcs/pyqt_gui/widgets/shared/scope_visual_config.py @@ -0,0 +1,148 @@ +"""Configuration for scope-based visual feedback (colors, flash animations).""" + +from dataclasses import dataclass +from enum import Enum +from typing import Optional + + +@dataclass +class ScopeVisualConfig: + """Configuration for scope-based visual feedback. + + Controls colors, flash animations, and styling for scope-aware UI elements. + All values are configurable for easy tuning. + """ + + # === Orchestrator List Item Colors (HSV) === + ORCHESTRATOR_ITEM_BG_SATURATION: int = 40 # Visible but not overwhelming + ORCHESTRATOR_ITEM_BG_VALUE: int = 85 # Medium-light background + ORCHESTRATOR_ITEM_BORDER_SATURATION: int = 30 + ORCHESTRATOR_ITEM_BORDER_VALUE: int = 80 + + # === Step List Item Colors (HSV) === + STEP_ITEM_BG_SATURATION: int = 35 # Slightly less saturated than orchestrator + STEP_ITEM_BG_VALUE: int = 88 # Slightly lighter than orchestrator + + # === Step Window Border Colors (HSV) === + STEP_WINDOW_BORDER_SATURATION: int = 60 # More saturated for visibility + STEP_WINDOW_BORDER_VALUE: int = 70 # Medium brightness + STEP_WINDOW_BORDER_WIDTH_PX: int = 4 # Thicker for visibility + STEP_WINDOW_BORDER_STYLE: str = "solid" + + # === Flash Animation === + FLASH_DURATION_MS: int = 300 # Duration of flash effect + FLASH_COLOR_RGB: tuple[int, int, int] = (144, 238, 144) # Light green + LIST_ITEM_FLASH_ENABLED: bool = True + WIDGET_FLASH_ENABLED: bool = True + + +@dataclass +class ScopeColorScheme: + """Color scheme for a specific scope.""" + scope_id: Optional[str] + hue: int + + # Orchestrator colors + orchestrator_item_bg_rgb: tuple[int, int, int] + orchestrator_item_border_rgb: tuple[int, int, int] + + # Step colors + step_window_border_rgb: tuple[int, int, int] + step_item_bg_rgb: Optional[tuple[int, int, int]] # None = transparent background + step_border_width: int = 0 # Total border width (for backward compat) + step_border_layers: list = None # List of (width, tint_index) for layered borders + base_color_rgb: tuple[int, int, int] = (128, 128, 128) # Base orchestrator color for tint calculation + + def __post_init__(self): + """Initialize mutable defaults.""" + if self.step_border_layers is None: + self.step_border_layers = [] + + def to_qcolor_orchestrator_bg(self) -> 'QColor': + """Get QColor for orchestrator list item background with alpha transparency.""" + from PyQt6.QtGui import QColor + r, g, b = self.orchestrator_item_bg_rgb + # 40% opacity for visible background tint + return QColor(r, g, b, int(255 * 0.40)) + + def to_qcolor_orchestrator_border(self) -> 'QColor': + """Get QColor for orchestrator list item border.""" + from PyQt6.QtGui import QColor + return QColor(*self.orchestrator_item_border_rgb) + + def to_qcolor_step_window_border(self) -> 'QColor': + """Get QColor for step window border.""" + from PyQt6.QtGui import QColor + return QColor(*self.step_window_border_rgb) + + def to_qcolor_step_item_bg(self) -> Optional['QColor']: + """Get QColor for step list item background with alpha transparency. + + Returns None for transparent background (no background color). + """ + if self.step_item_bg_rgb is None: + return None + from PyQt6.QtGui import QColor + r, g, b = self.step_item_bg_rgb + # 30% opacity for visible background tint + return QColor(r, g, b, int(255 * 0.30)) + + def to_stylesheet_step_window_border(self) -> str: + """Generate stylesheet for step window border with layered borders. + + Uses custom border painting via paintEvent override since Qt stylesheets + don't properly support multiple layered borders with patterns on QDialog. + + This method returns a simple placeholder border - actual layered rendering + happens in the window's paintEvent. + """ + if not self.step_border_layers or len(self.step_border_layers) == 0: + # No borders - use simple window border with step color + r, g, b = self.step_window_border_rgb + return f"border: 4px solid rgb({r}, {g}, {b});" + + # Calculate total border width for spacing purposes + total_width = sum(layer[0] for layer in self.step_border_layers) + + # Return empty border - actual painting happens in paintEvent + # We still need to reserve space for the border + return f"border: {total_width}px solid transparent;" + + +class ListItemType(Enum): + """Type of list item for scope-based coloring. + + Uses enum-driven polymorphic dispatch to select correct background color + from ScopeColorScheme without if/else conditionals. + + Pattern follows OpenHCS ProcessingContract enum design: + - Enum value stores method name + - Enum method uses getattr() for polymorphic dispatch + - Extensible: add new item types without modifying existing code + """ + ORCHESTRATOR = "to_qcolor_orchestrator_bg" + STEP = "to_qcolor_step_item_bg" + + def get_background_color(self, color_scheme: ScopeColorScheme) -> 'QColor': + """Get background color for this item type via polymorphic dispatch. + + Args: + color_scheme: ScopeColorScheme containing all color variants + + Returns: + QColor for this item type's background + """ + method = getattr(color_scheme, self.value) + return method() + + +def get_scope_visual_config() -> ScopeVisualConfig: + """Get singleton instance of ScopeVisualConfig.""" + global _config_instance + if _config_instance is None: + _config_instance = ScopeVisualConfig() + return _config_instance + + +_config_instance: Optional[ScopeVisualConfig] = None + diff --git a/openhcs/pyqt_gui/widgets/shared/scoped_border_mixin.py b/openhcs/pyqt_gui/widgets/shared/scoped_border_mixin.py new file mode 100644 index 000000000..d8c87bd07 --- /dev/null +++ b/openhcs/pyqt_gui/widgets/shared/scoped_border_mixin.py @@ -0,0 +1,134 @@ +"""Mixin for scope-based window border rendering. + +Provides automatic border styling based on scope_id: +- Simple solid borders for orchestrator-level scopes +- Layered patterned borders for step-level scopes (with tints + patterns) +- Reactive updates via ScopeColorService signals +""" + +from typing import Optional, Tuple, List +from PyQt6.QtGui import QPainter, QPen, QColor +from PyQt6.QtCore import Qt +import logging + +logger = logging.getLogger(__name__) + + +class ScopedBorderMixin: + """Mixin that renders scope-based borders on QDialog/QWidget subclasses. + + Subclasses declare scope via: + self.scope_id: str # Required - determines color scheme + self.step_position: Optional[int] # Optional - for step windows (layered borders) + + Then call self._init_scope_border() after scope_id is set. + + The mixin automatically: + - Gets color scheme from ScopeColorService + - Reserves border space via stylesheet + - Renders layered borders in paintEvent (if step scope with layers) + - Subscribes to color change signals for reactive updates + """ + + # Class-level config (override in subclass if needed) + BORDER_TINT_FACTORS: Tuple[float, ...] = (0.7, 1.0, 1.4) + BORDER_PATTERNS = { + 'solid': (Qt.PenStyle.SolidLine, None), + 'dashed': (Qt.PenStyle.DashLine, [8, 6]), + 'dotted': (Qt.PenStyle.DotLine, [2, 6]), + } + + _scope_color_scheme = None # Set by _init_scope_border + + def _init_scope_border(self) -> None: + """Initialize scope-based border. Call after scope_id is set.""" + scope_id = getattr(self, 'scope_id', None) + if not scope_id: + return + + from openhcs.pyqt_gui.widgets.shared.scope_color_utils import get_scope_color_scheme + self._scope_color_scheme = get_scope_color_scheme(scope_id) + + # Reserve border space via stylesheet + border_style = self._scope_color_scheme.to_stylesheet_step_window_border() + current_style = self.styleSheet() if hasattr(self, 'styleSheet') else '' + # Apply to QDialog (for BaseFormDialog subclasses) + self.setStyleSheet(f"{current_style}\nQDialog {{ {border_style} }}") + + # Subscribe to color change signals for reactive updates + self._subscribe_to_color_changes() + + if hasattr(self, 'update'): + self.update() # Trigger repaint + + def _subscribe_to_color_changes(self) -> None: + """Subscribe to ScopeColorService signals for reactive updates.""" + from openhcs.pyqt_gui.widgets.shared.services.scope_color_service import ScopeColorService + service = ScopeColorService.instance() + + # Connect signals (check if not already connected) + scope_id = getattr(self, 'scope_id', None) + if scope_id: + service.color_changed.connect(self._on_scope_color_changed) + service.all_colors_reset.connect(self._on_all_colors_reset) + + def _on_scope_color_changed(self, changed_scope_id: str) -> None: + """Handle color change for specific scope.""" + scope_id = getattr(self, 'scope_id', None) + if scope_id and (scope_id == changed_scope_id or scope_id.startswith(f"{changed_scope_id}::")): + self._refresh_scope_border() + + def _on_all_colors_reset(self) -> None: + """Handle bulk color reset (strategy change).""" + self._refresh_scope_border() + + def _refresh_scope_border(self) -> None: + """Refresh border with current color scheme.""" + self._scope_color_scheme = None # Clear cached scheme + self._init_scope_border() + + def paintEvent(self, event) -> None: + """Render layered scope borders. Calls super().paintEvent first.""" + super().paintEvent(event) + + if not self._scope_color_scheme: + return + + layers = getattr(self._scope_color_scheme, 'step_border_layers', None) + if not layers: + return + + self._paint_border_layers(layers) + + def _paint_border_layers(self, layers: List[Tuple]) -> None: + """Paint multi-layer borders with tints and patterns.""" + painter = QPainter(self) + painter.setRenderHint(QPainter.RenderHint.Antialiasing) + + rect = self.rect() + inset = 0 + base_rgb = self._scope_color_scheme.base_color_rgb + + for layer in layers: + # Unpack layer tuple: (width, tint_idx, pattern) + width, tint_idx, pattern = (layer + ('solid',))[:3] + + # Tinted color + tint = self.BORDER_TINT_FACTORS[tint_idx] + color = QColor(*(min(255, int(c * tint)) for c in base_rgb)).darker(120) + + # Create pen with pattern + pen = QPen(color, width) + style, dash_pattern = self.BORDER_PATTERNS.get(pattern, self.BORDER_PATTERNS['solid']) + pen.setStyle(style) + if dash_pattern: + pen.setDashPattern(dash_pattern) + + # Draw layer + offset = int(inset + width / 2) + painter.setPen(pen) + painter.drawRect(rect.adjusted(offset, offset, -offset - 1, -offset - 1)) + inset += width + + painter.end() + diff --git a/openhcs/pyqt_gui/widgets/shared/scrollable_form_mixin.py b/openhcs/pyqt_gui/widgets/shared/scrollable_form_mixin.py index 228140e2e..b81cee86e 100644 --- a/openhcs/pyqt_gui/widgets/shared/scrollable_form_mixin.py +++ b/openhcs/pyqt_gui/widgets/shared/scrollable_form_mixin.py @@ -15,16 +15,19 @@ class ScrollableFormMixin: Mixin for widgets that have: - self.scroll_area: QScrollArea containing the form - self.form_manager: ParameterFormManager with nested_managers - + Provides scroll-to-section functionality. + + If the class also inherits from TreeFormFlashMixin, the GroupBox will be + flashed after scrolling to provide visual feedback. """ - + # Type hints for attributes that must be provided by the implementing class scroll_area: QScrollArea form_manager: 'ParameterFormManager' # Forward reference - + def _scroll_to_section(self, field_name: str): - """Scroll to a specific section in the form.""" + """Scroll to a specific section in the form and flash it for visibility.""" logger.info(f"🔍 Scrolling to section: {field_name}") if not hasattr(self, 'scroll_area') or self.scroll_area is None: @@ -49,10 +52,14 @@ def _scroll_to_section(self, field_name: str): # Map widget position to scroll area coordinates and scroll to it widget_pos = first_widget.mapTo(self.scroll_area.widget(), first_widget.rect().topLeft()) v_scroll_bar = self.scroll_area.verticalScrollBar() - + # Scroll to widget position with offset to show context above target_scroll = max(0, widget_pos.y() - 50) v_scroll_bar.setValue(target_scroll) - + logger.info(f"✅ Scrolled to {field_name}") + # Flash the GroupBox if TreeFormFlashMixin is available + if hasattr(self, '_flash_groupbox_for_field'): + self._flash_groupbox_for_field(field_name) + diff --git a/openhcs/pyqt_gui/widgets/shared/services/field_change_dispatcher.py b/openhcs/pyqt_gui/widgets/shared/services/field_change_dispatcher.py index c2f340519..5eb4a7cd5 100644 --- a/openhcs/pyqt_gui/widgets/shared/services/field_change_dispatcher.py +++ b/openhcs/pyqt_gui/widgets/shared/services/field_change_dispatcher.py @@ -69,7 +69,7 @@ def _dispatch_impl(self, event: FieldChangeEvent) -> None: logger.warning(f"🚫 DISPATCH BLOCKED: {source.field_id} has _in_reset=True") return - # 1. Update source's data model + # 1. Update source's data model AND registry source.parameters[event.field_name] = event.value # CRITICAL: Always add to _user_set_fields, even for reset # This ensures get_user_modified_values() includes None for reset fields, @@ -77,9 +77,18 @@ def _dispatch_impl(self, event: FieldChangeEvent) -> None: # Previously we discarded on reset, but that caused preview labels to show # the saved-on-disk value instead of the reset (None) value. source._user_set_fields.add(event.field_name) + + # REGISTRY REFACTOR: Update registry (single source of truth) + # Skip if scope_id is None (can happen for unregistered managers) + if source.scope_id is not None: + from openhcs.config_framework import ContextStackRegistry + registry = ContextStackRegistry.instance() + registry.set(source.scope_id, event.field_name, event.value, source.dataclass_type) + if DEBUG_DISPATCHER: reset_note = " (reset to None)" if event.is_reset else "" logger.info(f" ✅ Updated source.parameters[{event.field_name}], ADDED to _user_set_fields{reset_note}") + logger.info(f" ✅ Updated registry: {source.scope_id}.{event.field_name}") # PERFORMANCE OPTIMIZATION: Invalidate cache but DON'T notify listeners yet # This allows sibling refreshes to share the cached live context @@ -90,9 +99,11 @@ def _dispatch_impl(self, event: FieldChangeEvent) -> None: root = root._parent_manager from openhcs.pyqt_gui.widgets.shared.services.live_context_service import LiveContextService - LiveContextService.increment_token(notify=False) # Invalidate cache only + # Build full field path for reactive updates (e.g., "GlobalConfig.output_dir") + full_field_path = f"{source.field_id}.{event.field_name}" + LiveContextService.increment_token(notify=False, changed_field=full_field_path) if DEBUG_DISPATCHER: - logger.info(f" 🔄 Incremented live context token to {LiveContextService.get_token()} (notify deferred)") + logger.info(f" 🔄 Incremented live context token to {LiveContextService.get_token()} (field={full_field_path}, notify deferred)") # 2. Mark parent chain as modified BEFORE refreshing siblings # This ensures root.get_user_modified_values() includes this field on first keystroke diff --git a/openhcs/pyqt_gui/widgets/shared/services/live_context_service.py b/openhcs/pyqt_gui/widgets/shared/services/live_context_service.py index d62228d37..c796d545c 100644 --- a/openhcs/pyqt_gui/widgets/shared/services/live_context_service.py +++ b/openhcs/pyqt_gui/widgets/shared/services/live_context_service.py @@ -54,6 +54,9 @@ class LiveContextService: _live_context_token_counter: int = 0 _live_context_cache: Optional['TokenCache'] = None # Initialized on first use + # Track which field changed for reactive updates + _last_changed_field: Optional[str] = None + # ========== TOKEN MANAGEMENT ========== @classmethod @@ -62,15 +65,27 @@ def get_token(cls) -> int: return cls._live_context_token_counter @classmethod - def increment_token(cls, notify: bool = True) -> None: + def get_last_changed_field(cls) -> Optional[str]: + """Get the field path that triggered the last token increment. + + Returns: + Field path (e.g., "GlobalConfig.output_dir") or None if unknown. + """ + return cls._last_changed_field + + @classmethod + def increment_token(cls, notify: bool = True, changed_field: Optional[str] = None) -> None: """Increment token to invalidate all caches. Args: notify: If True (default), notify all listeners of the change. Set to False when you need to invalidate caches but will notify listeners later (e.g., after sibling refresh completes). + changed_field: Optional field path that triggered the change. + Used by ResolvedItemStateService for targeted updates. """ cls._live_context_token_counter += 1 + cls._last_changed_field = changed_field if notify: cls._notify_change() diff --git a/openhcs/pyqt_gui/widgets/shared/services/parameter_ops_service.py b/openhcs/pyqt_gui/widgets/shared/services/parameter_ops_service.py index 8d9d5deac..f040722c2 100644 --- a/openhcs/pyqt_gui/widgets/shared/services/parameter_ops_service.py +++ b/openhcs/pyqt_gui/widgets/shared/services/parameter_ops_service.py @@ -164,10 +164,13 @@ def _update_reset_tracking(manager, param_name: str, reset_value: Any) -> None: # ========== PLACEHOLDER REFRESH (from PlaceholderRefreshService) ========== # DELETED: refresh_affected_siblings - moved to FieldChangeDispatcher + # DELETED: _build_full_field_path - registry now uses (dataclass_type, field_name) instead def refresh_single_placeholder(self, manager, field_name: str) -> None: """Refresh placeholder for a single field in a manager. + REGISTRY REFACTOR: Uses registry.resolve() instead of building context stack. + Only updates if: 1. The field exists as a widget in the manager 2. The current value is None (needs placeholder) @@ -192,81 +195,53 @@ def refresh_single_placeholder(self, manager, field_name: str) -> None: logger.info(f" ✅ {field_name} value is None, computing placeholder...") from openhcs.pyqt_gui.widgets.shared.widget_strategies import PyQt6WidgetEnhancer - from openhcs.pyqt_gui.widgets.shared.parameter_form_manager import ParameterFormManager - from openhcs.config_framework.context_manager import build_context_stack - - # Find root manager to get complete form values (enables sibling inheritance) - # Root form (GlobalPipelineConfig/PipelineConfig/Step) contains all nested configs - root_manager = manager - while getattr(root_manager, '_parent_manager', None) is not None: - root_manager = root_manager._parent_manager - - # Build context stack for resolution (use ROOT type for cache sharing) - live_context_snapshot = ParameterFormManager.collect_live_context( - scope_filter=manager.scope_id, - for_type=root_manager.dataclass_type - ) - live_context = live_context_snapshot.values if live_context_snapshot else None - - # Use root manager's values and type for context (not just this nested manager's) - # PERFORMANCE OPTIMIZATION: Get root_values from live_context instead of calling - # get_user_modified_values() again (which calls get_current_values()) - root_type = root_manager.dataclass_type - is_nested = root_manager != manager - root_values = live_context.get(root_type) if live_context and is_nested else None - if root_values: - value_types = {k: type(v).__name__ for k, v in root_values.items()} - logger.info(f" 🔍 ROOT: field_id={root_manager.field_id}, type={root_type}, values={value_types}") - if root_type: - from openhcs.core.lazy_placeholder_simplified import LazyDefaultPlaceholderService - lazy_root_type = LazyDefaultPlaceholderService._get_lazy_type_for_base(root_type) - if lazy_root_type: - root_type = lazy_root_type - - # CRITICAL: Exclude the field being resolved from the overlay. - # If we include it, the overlay's None value shadows the inherited value - # from parent configs (e.g., streaming_defaults.well_filter=None would - # shadow well_filter_config.well_filter=2). - overlay_without_field = {k: v for k, v in manager.parameters.items() if k != field_name} - - stack = build_context_stack( - context_obj=manager.context_obj, - overlay=overlay_without_field, - dataclass_type=manager.dataclass_type, - live_context=live_context, - is_global_config_editing=getattr(manager.config, 'is_global_config_editing', False), - global_config_type=getattr(manager.config, 'global_config_type', None), - root_form_values=root_values, - root_form_type=root_type, - ) - - with stack: - from openhcs.core.lazy_placeholder_simplified import LazyDefaultPlaceholderService - dataclass_type_for_resolution = manager.dataclass_type - if dataclass_type_for_resolution: - lazy_type = LazyDefaultPlaceholderService._get_lazy_type_for_base(dataclass_type_for_resolution) - if lazy_type: - dataclass_type_for_resolution = lazy_type - - placeholder_text = manager.service.get_placeholder_text(field_name, dataclass_type_for_resolution) + from openhcs.config_framework import ContextStackRegistry + from openhcs.config_framework.placeholder import LazyDefaultPlaceholderService + + # REGISTRY REFACTOR: Use registry.resolve() with dataclass_type and field_name + # The registry handles lazy instance creation and context resolution + registry = ContextStackRegistry.instance() + resolved_value = registry.resolve(manager.scope_id, manager.dataclass_type, field_name) + + logger.info(f" Registry resolved: {manager.dataclass_type.__name__}.{field_name} = {repr(resolved_value)[:50]}") + + # Apply placeholder to widget + # NOTE: resolved_value can be None - that's a valid value meaning "no default" + # Use LazyDefaultPlaceholderService for consistent formatting (None -> "(none)") + widget = manager.widgets.get(field_name) + if widget: + # Format placeholder text using the same logic as the old code + placeholder_text = LazyDefaultPlaceholderService._format_placeholder_text( + resolved_value, LazyDefaultPlaceholderService.PLACEHOLDER_PREFIX + ) logger.info(f" 📝 Computed placeholder: {repr(placeholder_text)[:50]}") + # Get old placeholder to detect actual changes + old_placeholder = getattr(widget, 'placeholderText', lambda: None)() + if placeholder_text: - widget = manager.widgets[field_name] PyQt6WidgetEnhancer.apply_placeholder_text(widget, placeholder_text) logger.info(f" ✅ Applied placeholder to widget") - # Keep enabled-field styling in sync when placeholder changes the visual state - if field_name == 'enabled': + # Keep enabled-field styling in sync when placeholder changes the visual state + if field_name == 'enabled': + try: + resolved_value = manager._widget_ops.get_value(widget) + manager._enabled_field_styling_service.on_enabled_field_changed( + manager, 'enabled', resolved_value + ) + except Exception: + logger.exception("Failed to apply enabled styling after placeholder refresh") + + # Hook: notify listeners ONLY if placeholder actually changed + if old_placeholder != placeholder_text: + for callback in getattr(manager, '_on_placeholder_changed_callbacks', []): try: - resolved_value = manager._widget_ops.get_value(widget) - manager._enabled_field_styling_service.on_enabled_field_changed( - manager, 'enabled', resolved_value - ) + callback(manager.field_id, field_name, manager.dataclass_type) except Exception: - logger.exception("Failed to apply enabled styling after placeholder refresh") - else: - logger.warning(f" ⚠️ No placeholder text computed") + logger.exception("Failed to call placeholder changed callback") + else: + logger.warning(f" ⚠️ No resolved value from registry") def refresh_with_live_context(self, manager, use_user_modified_only: bool = False) -> None: """Refresh placeholders using live values from tree registry.""" @@ -277,77 +252,36 @@ def refresh_with_live_context(self, manager, use_user_modified_only: bool = Fals ) def refresh_all_placeholders(self, manager, use_user_modified_only: bool = False) -> None: - """Refresh placeholder text for all widgets in a form.""" + """Refresh placeholder text for all widgets in a form. + + REGISTRY REFACTOR: Uses registry.resolve() instead of building context stack. + """ with timer(f"_refresh_all_placeholders ({manager.field_id})", threshold_ms=5.0): if not manager.dataclass_type: logger.debug(f"[PLACEHOLDER] {manager.field_id}: No dataclass_type, skipping") return from openhcs.pyqt_gui.widgets.shared.widget_strategies import PyQt6WidgetEnhancer - from openhcs.pyqt_gui.widgets.shared.parameter_form_manager import ParameterFormManager - from openhcs.config_framework.context_manager import build_context_stack - - logger.debug(f"[PLACEHOLDER] {manager.field_id}: Building context stack") - # Find root manager to get complete form values (enables sibling inheritance) - root_manager = manager - while getattr(root_manager, '_parent_manager', None) is not None: - root_manager = root_manager._parent_manager - - live_context_snapshot = ParameterFormManager.collect_live_context( - scope_filter=manager.scope_id, - for_type=root_manager.dataclass_type - ) - # Extract .values dict from LiveContextSnapshot for build_context_stack - live_context = live_context_snapshot.values if live_context_snapshot else None - overlay = manager.get_user_modified_values() if use_user_modified_only else manager.parameters - - # Handle excluded params in overlay - if overlay: - overlay_dict = overlay.copy() - for excluded_param in getattr(manager, 'exclude_params', []): - if excluded_param not in overlay_dict and hasattr(manager.object_instance, excluded_param): - overlay_dict[excluded_param] = getattr(manager.object_instance, excluded_param) - else: - overlay_dict = None - - # PERFORMANCE OPTIMIZATION: Get root_values from live_context instead of calling - # get_user_modified_values() again (which calls get_current_values()) - root_type = root_manager.dataclass_type - is_nested = root_manager != manager - root_values = live_context.get(root_type) if live_context and is_nested else None - if root_type: - from openhcs.core.lazy_placeholder_simplified import LazyDefaultPlaceholderService - lazy_root_type = LazyDefaultPlaceholderService._get_lazy_type_for_base(root_type) - if lazy_root_type: - root_type = lazy_root_type - - # Use framework-agnostic context stack building from config_framework - stack = build_context_stack( - context_obj=manager.context_obj, - overlay=overlay_dict, - dataclass_type=manager.dataclass_type, - live_context=live_context, - is_global_config_editing=getattr(manager.config, 'is_global_config_editing', False), - global_config_type=getattr(manager.config, 'global_config_type', None), - root_form_values=root_values, - root_form_type=root_type, - ) - - with stack: - monitor = get_monitor("Placeholder resolution per field") - from openhcs.core.lazy_placeholder_simplified import LazyDefaultPlaceholderService - dataclass_type_for_resolution = manager.dataclass_type - if dataclass_type_for_resolution: - lazy_type = LazyDefaultPlaceholderService._get_lazy_type_for_base(dataclass_type_for_resolution) - if lazy_type: - dataclass_type_for_resolution = lazy_type - - for param_name, widget in manager.widgets.items(): - current_value = manager.parameters.get(param_name) - should_apply_placeholder = (current_value is None) - - if should_apply_placeholder: - with monitor.measure(): - placeholder_text = manager.service.get_placeholder_text(param_name, dataclass_type_for_resolution) - if placeholder_text: - PyQt6WidgetEnhancer.apply_placeholder_text(widget, placeholder_text) + from openhcs.config_framework import ContextStackRegistry + from openhcs.config_framework.placeholder import LazyDefaultPlaceholderService + + logger.debug(f"[PLACEHOLDER] {manager.field_id}: Resolving via registry") + + registry = ContextStackRegistry.instance() + monitor = get_monitor("Placeholder resolution per field") + + for param_name, widget in manager.widgets.items(): + current_value = manager.parameters.get(param_name) + should_apply_placeholder = (current_value is None) + + if should_apply_placeholder: + with monitor.measure(): + # Resolve via registry using dataclass_type and field_name + resolved_value = registry.resolve(manager.scope_id, manager.dataclass_type, param_name) + # NOTE: resolved_value can be None - that's a valid value + # Use LazyDefaultPlaceholderService for consistent formatting + placeholder_text = LazyDefaultPlaceholderService._format_placeholder_text( + resolved_value, LazyDefaultPlaceholderService.PLACEHOLDER_PREFIX + ) + if placeholder_text: + PyQt6WidgetEnhancer.apply_placeholder_text(widget, placeholder_text) diff --git a/openhcs/pyqt_gui/widgets/shared/services/resolved_item_state_service.py b/openhcs/pyqt_gui/widgets/shared/services/resolved_item_state_service.py new file mode 100644 index 000000000..e40ae718a --- /dev/null +++ b/openhcs/pyqt_gui/widgets/shared/services/resolved_item_state_service.py @@ -0,0 +1,165 @@ +""" +Reactive flash/dirty service for list items. + +REGISTRY REFACTOR: Listens to ContextStackRegistry.value_changed instead of +ParameterFormManager.context_value_changed. + +Architecture: +- Listens to registry.value_changed (scope_id, field_path, old_value, new_value) +- Uses scope visibility rules to filter which items are affected +- ANY change in visible scope triggers flash - no field tracking needed +- Lazydataclass inheritance handles field propagation automatically +- Dirty tracking via registry.get_resolved_state() comparison to baseline + +Usage: + service = ResolvedItemStateService.instance() + + # Register item scope (no field list - automatic via lazydataclass) + service.register_item(scope_id) + + # Set baseline for dirty tracking + service.set_baseline(scope_id) + + # Values come in via registry.value_changed - flash if scope visible +""" + +from typing import Any, Optional, Set, Dict +import logging + +from PyQt6.QtCore import QObject, pyqtSignal + +logger = logging.getLogger(__name__) + + +class ResolvedItemStateService(QObject): + """Reactive service that flashes items when visible scope changes. + + REGISTRY REFACTOR: Uses ContextStackRegistry as single source of truth. + NO field tracking - lazydataclass inheritance handles propagation. + """ + + _instance: Optional['ResolvedItemStateService'] = None + + # Signals + item_resolved_changed = pyqtSignal(str, str, object) # scope_id, field_name, new_value + item_dirty_changed = pyqtSignal(str, bool) # scope_id, is_dirty + + def __init__(self): + super().__init__() + # Registered scope_ids + self._registrations: Set[str] = set() + + # Baselines for dirty tracking (scope_id -> resolved state dict) + self._baselines: Dict[str, Dict[str, Any]] = {} + + # REGISTRY REFACTOR: Connect to registry.value_changed + from openhcs.config_framework import ContextStackRegistry + registry = ContextStackRegistry.instance() + registry.value_changed.connect(self._on_registry_value_changed) + logger.info("🔗 Connected to ContextStackRegistry.value_changed") + + @classmethod + def instance(cls) -> 'ResolvedItemStateService': + if cls._instance is None: + cls._instance = cls() + return cls._instance + + @classmethod + def reset_instance(cls) -> None: + """Reset singleton for testing.""" + cls._instance = None + + # --- Registration --- + + def register_item(self, scope_id: str) -> None: + """Register a list item scope for flash tracking.""" + self._registrations.add(scope_id) + logger.debug(f"📦 Registered scope: {scope_id}") + + def unregister_item(self, scope_id: str) -> None: + """Unregister item when removed from list.""" + self._registrations.discard(scope_id) + self._baselines.pop(scope_id, None) + + def set_baseline(self, scope_id: str) -> None: + """Set baseline for dirty tracking. + + REGISTRY REFACTOR: Captures current resolved state from registry. + Call this after save, load, or when item is first registered. + """ + from openhcs.config_framework import ContextStackRegistry + + registry = ContextStackRegistry.instance() + baseline = registry.get_resolved_state(scope_id) + if baseline is not None: + self._baselines[scope_id] = baseline + logger.debug(f"� Set baseline for {scope_id}: {len(baseline)} fields") + # Check if dirty status changed + self._check_dirty_status(scope_id) + + def _on_registry_value_changed(self, scope_id: str, field_path: str, + old_value: Any, new_value: Any) -> None: + """Handle registry value change - flash and check dirty status. + + REGISTRY REFACTOR: Listens to ContextStackRegistry.value_changed. + """ + # Extract leaf field name + field_name = field_path.split('.')[-1] + + # Flash all registered items where scope is visible + for registered_scope_id in self._registrations: + if self._is_scope_visible(scope_id, registered_scope_id): + logger.info(f"⚡ Flash {registered_scope_id} (field={field_name})") + self.item_resolved_changed.emit(registered_scope_id, field_name, new_value) + + # Check dirty status for this item + self._check_dirty_status(registered_scope_id) + + def _check_dirty_status(self, scope_id: str) -> None: + """Check if scope is dirty compared to baseline and emit signal if changed.""" + from openhcs.config_framework import ContextStackRegistry + + if scope_id not in self._baselines: + return # No baseline set yet + + registry = ContextStackRegistry.instance() + current_state = registry.get_resolved_state(scope_id) + if current_state is None: + return + + baseline = self._baselines[scope_id] + is_dirty = current_state != baseline + + # Emit signal (AbstractManagerWidget will update display) + self.item_dirty_changed.emit(scope_id, is_dirty) + logger.debug(f"Dirty check {scope_id}: {is_dirty}") + + def _is_scope_visible(self, editing_scope_id: str, target_scope_id: str) -> bool: + """Check if edits from editing_scope_id should affect target_scope_id. + + Uses the same visibility rules as ParameterFormManager. + """ + from openhcs.config_framework.context_manager import get_root_from_scope_key + + # Global config edits affect everything + if 'Global' in editing_scope_id: + return True + + # Same scope + if editing_scope_id == target_scope_id: + return True + + # Same root (same plate) + editing_root = get_root_from_scope_key(editing_scope_id) + target_root = get_root_from_scope_key(target_scope_id) + + if editing_root and target_root: + return editing_root == target_root + + return False + + def clear_all(self) -> None: + """Clear all state. Called on pipeline close/reset.""" + self._registrations.clear() + logger.debug("Cleared all registrations") + diff --git a/openhcs/pyqt_gui/widgets/shared/services/scope_color_service.py b/openhcs/pyqt_gui/widgets/shared/services/scope_color_service.py new file mode 100644 index 000000000..79852f5c7 --- /dev/null +++ b/openhcs/pyqt_gui/widgets/shared/services/scope_color_service.py @@ -0,0 +1,151 @@ +"""Reactive service for scope-based colors with strategy support. + +Singleton service that: +- Manages color generation strategies (pluggable) +- Caches color schemes for performance +- Emits signals when colors change (reactive updates) +- Supports manual color overrides for future right-click menu +""" + +from typing import Optional, Dict, Tuple +from PyQt6.QtCore import QObject, pyqtSignal +import logging + +from openhcs.pyqt_gui.widgets.shared.scope_color_strategy import ( + ScopeColorStrategy, MD5HashStrategy, ManualColorStrategy, ColorStrategyType +) + +logger = logging.getLogger(__name__) + + +class ScopeColorService(QObject): + """Singleton service managing scope colors with reactive updates. + + Emits signals when colors change, allowing widgets to update reactively. + Supports pluggable strategies and manual overrides. + + Usage: + service = ScopeColorService.instance() + scheme = service.get_color_scheme(scope_id) + + # Subscribe to changes + service.color_changed.connect(my_widget.on_color_changed) + + # Manual color override (future right-click menu) + service.set_manual_color(scope_id, (255, 0, 0)) + """ + + # Signals for reactive updates + color_changed = pyqtSignal(str) # scope_id changed + all_colors_reset = pyqtSignal() # bulk reset (strategy change) + + _instance: Optional['ScopeColorService'] = None + + @classmethod + def instance(cls) -> 'ScopeColorService': + """Get singleton instance.""" + if cls._instance is None: + cls._instance = cls() + return cls._instance + + @classmethod + def reset_instance(cls) -> None: + """Reset singleton (for testing).""" + cls._instance = None + + def __init__(self): + super().__init__() + self._strategies: Dict[ColorStrategyType, ScopeColorStrategy] = { + ColorStrategyType.MD5_HASH: MD5HashStrategy(), + ColorStrategyType.MANUAL: ManualColorStrategy(), + } + self._active_strategy_type = ColorStrategyType.MD5_HASH + self._scheme_cache: Dict[str, 'ScopeColorScheme'] = {} + + # === Strategy Management === + + @property + def active_strategy(self) -> ScopeColorStrategy: + """Get currently active strategy.""" + return self._strategies[self._active_strategy_type] + + def set_strategy(self, strategy_type: ColorStrategyType) -> None: + """Switch active color generation strategy.""" + if strategy_type != self._active_strategy_type: + self._active_strategy_type = strategy_type + self._invalidate_all() + logger.info(f"Color strategy changed to {strategy_type.name}") + + def register_strategy(self, strategy: ScopeColorStrategy) -> None: + """Register custom strategy.""" + self._strategies[strategy.strategy_type] = strategy + logger.info(f"Registered color strategy: {strategy.strategy_type.name}") + + # === Color Access (cached) === + + def get_color_scheme(self, scope_id: Optional[str]) -> 'ScopeColorScheme': + """Get color scheme for scope. Cached and reactive.""" + # Import here to avoid circular dependency + from openhcs.pyqt_gui.widgets.shared.scope_color_utils import ( + _build_color_scheme_from_rgb, + extract_orchestrator_scope, + ) + + if scope_id is None: + return self._get_neutral_scheme() + + if scope_id not in self._scheme_cache: + # Get base color from strategy + orchestrator_scope = extract_orchestrator_scope(scope_id) + rgb = self.active_strategy.generate_color(orchestrator_scope) + # Build full scheme with all derived colors + self._scheme_cache[scope_id] = _build_color_scheme_from_rgb(rgb, scope_id) + + return self._scheme_cache[scope_id] + + def _get_neutral_scheme(self) -> 'ScopeColorScheme': + """Get neutral gray scheme for None scope.""" + from openhcs.pyqt_gui.widgets.shared.scope_visual_config import ScopeColorScheme + return ScopeColorScheme( + scope_id=None, + hue=0, + orchestrator_item_bg_rgb=(240, 240, 240), + orchestrator_item_border_rgb=(180, 180, 180), + step_window_border_rgb=(128, 128, 128), + step_item_bg_rgb=(245, 245, 245), + step_border_width=0, + ) + + # === Manual Color API === + + def set_manual_color(self, scope_id: str, rgb: Tuple[int, int, int]) -> None: + """Set manual color override for scope.""" + manual = self._strategies[ColorStrategyType.MANUAL] + if isinstance(manual, ManualColorStrategy): + manual.set_color(scope_id, rgb) + self._invalidate(scope_id) + logger.debug(f"Set manual color for {scope_id}: {rgb}") + + def clear_manual_color(self, scope_id: str) -> None: + """Clear manual override, revert to generated color.""" + manual = self._strategies[ColorStrategyType.MANUAL] + if isinstance(manual, ManualColorStrategy): + manual.clear_color(scope_id) + self._invalidate(scope_id) + logger.debug(f"Cleared manual color for {scope_id}") + + # === Cache Invalidation + Signals === + + def _invalidate(self, scope_id: str) -> None: + """Invalidate cache and emit change signal.""" + # Invalidate this scope and any step scopes derived from it + keys_to_remove = [k for k in self._scheme_cache if k == scope_id or k.startswith(f"{scope_id}::")] + for key in keys_to_remove: + self._scheme_cache.pop(key, None) + self.color_changed.emit(scope_id) + + def _invalidate_all(self) -> None: + """Invalidate all and emit reset signal.""" + self._scheme_cache.clear() + self.all_colors_reset.emit() + diff --git a/openhcs/pyqt_gui/widgets/shared/smooth_flash_base.py b/openhcs/pyqt_gui/widgets/shared/smooth_flash_base.py new file mode 100644 index 000000000..459153397 --- /dev/null +++ b/openhcs/pyqt_gui/widgets/shared/smooth_flash_base.py @@ -0,0 +1,128 @@ +"""Base class for smooth flash animations using QVariantAnimation. + +Uses QVariantAnimation for smooth 60fps color transitions: +- Rapid fade-in (~100ms) with OutQuad easing +- Hold at max flash while rapid updates continue +- Smooth fade-out (~350ms) with InOutCubic easing when updates stop +""" + +import logging +from abc import abstractmethod +from typing import Optional +from PyQt6.QtCore import QVariantAnimation, QEasingCurve, QTimer +from PyQt6.QtGui import QColor + +logger = logging.getLogger(__name__) + + +class SmoothFlashAnimatorBase: + """Base class for smooth flash animation with hold-at-max behavior. + + Uses QVariantAnimation for 60fps color interpolation with: + - Rapid fade-in: 100ms with OutQuad easing (quick snap to flash color) + - Hold at max: stays at flash color while rapid updates continue + - Smooth fade-out: 350ms with InOutCubic easing (when updates stop) + + Subclasses must implement: + - _apply_color(color: QColor) -> None: Apply interpolated color to target + - _on_animation_complete() -> None: Cleanup when animation finishes + """ + + # Animation timing constants (configurable per-class) + FADE_IN_DURATION_MS: int = 100 # Rapid fade-in + FADE_OUT_DURATION_MS: int = 350 # Smooth fade-out + HOLD_DURATION_MS: int = 150 # Hold at max flash before fade-out + + def __init__(self, flash_color: QColor, original_color: QColor): + """Initialize animator. + + Args: + flash_color: Target color at max flash intensity + original_color: Color to return to after flash + """ + self.flash_color = flash_color + self._original_color = original_color + self._is_flashing: bool = False + + # Create fade-in animation + self._fade_in_anim = QVariantAnimation() + self._fade_in_anim.setDuration(self.FADE_IN_DURATION_MS) + self._fade_in_anim.setEasingCurve(QEasingCurve.Type.OutQuad) + self._fade_in_anim.valueChanged.connect(self._apply_color) + self._fade_in_anim.finished.connect(self._on_fade_in_complete) + + # Create fade-out animation + self._fade_out_anim = QVariantAnimation() + self._fade_out_anim.setDuration(self.FADE_OUT_DURATION_MS) + self._fade_out_anim.setEasingCurve(QEasingCurve.Type.InOutCubic) + self._fade_out_anim.valueChanged.connect(self._apply_color) + self._fade_out_anim.finished.connect(self._on_fade_out_complete) + + # Hold timer - resets on each flash, starts fade-out when expires + self._hold_timer = QTimer() + self._hold_timer.setSingleShot(True) + self._hold_timer.timeout.connect(self._start_fade_out) + + def flash_update(self) -> None: + """Trigger smooth flash animation. Call this on each update.""" + if not self._can_flash(): + return + + # If already flashing, just reset the hold timer (stay at max flash) + if self._is_flashing: + self._hold_timer.stop() + self._fade_out_anim.stop() # Cancel fade-out if it started + self._apply_color(self.flash_color) # Ensure at max + self._hold_timer.start(self.HOLD_DURATION_MS) + return + + # First flash - prepare and start fade-in + self._prepare_flash() + self._is_flashing = True + + # Start fade-in: original -> flash color + self._fade_in_anim.setStartValue(self._original_color) + self._fade_in_anim.setEndValue(self.flash_color) + self._fade_in_anim.start() + + def _on_fade_in_complete(self) -> None: + """Called when fade-in completes. Start hold timer.""" + self._hold_timer.start(self.HOLD_DURATION_MS) + + def _start_fade_out(self) -> None: + """Called when hold timer expires. Start fade-out animation.""" + self._fade_out_anim.setStartValue(self.flash_color) + self._fade_out_anim.setEndValue(self._original_color) + self._fade_out_anim.start() + + def _on_fade_out_complete(self) -> None: + """Called when fade-out completes.""" + self._is_flashing = False + self._on_animation_complete() + + def stop(self) -> None: + """Stop all animations immediately.""" + self._fade_in_anim.stop() + self._fade_out_anim.stop() + self._hold_timer.stop() + self._is_flashing = False + + # --- Abstract methods for subclasses --- + + def _can_flash(self) -> bool: + """Return True if flash should proceed. Override to add visibility checks.""" + return True + + def _prepare_flash(self) -> None: + """Called before first flash starts. Override to capture original state.""" + pass + + @abstractmethod + def _apply_color(self, color: QColor) -> None: + """Apply interpolated color to target. Called ~60 times/sec during animation.""" + raise NotImplementedError + + def _on_animation_complete(self) -> None: + """Called when animation finishes. Override to restore original state.""" + pass + diff --git a/openhcs/pyqt_gui/widgets/shared/tree_form_flash_mixin.py b/openhcs/pyqt_gui/widgets/shared/tree_form_flash_mixin.py new file mode 100644 index 000000000..3a2f0ec79 --- /dev/null +++ b/openhcs/pyqt_gui/widgets/shared/tree_form_flash_mixin.py @@ -0,0 +1,165 @@ +"""Mixin for widgets that have both a tree and a form with flash animations. + +This mixin provides: +1. GroupBox flashing when scrolling to a section (double-click tree item) +2. Tree item flashing when nested config placeholders change (cross-window updates) + +Used by: +- ConfigWindow +- StepParameterEditorWidget +""" + +import logging +from typing import Optional +from PyQt6.QtWidgets import QTreeWidget, QTreeWidgetItem +from PyQt6.QtCore import Qt + +logger = logging.getLogger(__name__) + + +class TreeFormFlashMixin: + """Mixin for widgets with tree + form that need flash animations. + + Requirements: + - Must have `self.form_manager` (ParameterFormManager instance) + - Must have `self.hierarchy_tree` or `self.tree_widget` (QTreeWidget instance) + - Must have `self.scope_id` (str for scope color scheme) + + Usage: + class MyWidget(TreeFormFlashMixin, QWidget): + def __init__(self): + super().__init__() + # ... create form_manager, tree_widget, scope_id ... + self._register_placeholder_flash_hook() + """ + + def _register_placeholder_flash_hook(self) -> None: + """Register flash callback. Called after each nested manager is created.""" + # Register on root manager + if self._on_placeholder_changed not in self.form_manager._on_placeholder_changed_callbacks: + self.form_manager._on_placeholder_changed_callbacks.append(self._on_placeholder_changed) + + # Register on all existing nested managers + def register_on_manager(name: str, manager) -> None: + if self._on_placeholder_changed not in manager._on_placeholder_changed_callbacks: + manager._on_placeholder_changed_callbacks.append(self._on_placeholder_changed) + + self.form_manager._apply_to_nested_managers(register_on_manager) + + # CRITICAL: Also register a post-build callback to catch async-created nested managers + def register_on_new_nested(): + self.form_manager._apply_to_nested_managers(register_on_manager) + + self.form_manager._on_build_complete_callbacks.append(register_on_new_nested) + logger.info(f"🔗 Registered placeholder flash hook on {type(self).__name__}") + + def _on_placeholder_changed(self, config_name: str, field_name: str, dataclass_type: type) -> None: + """Called when any placeholder value changes. Flash the tree item and groupbox.""" + logger.info(f"🔥 Placeholder changed: {config_name}.{field_name} (type={dataclass_type.__name__ if dataclass_type else 'None'})") + self._flash_for_config(config_name, dataclass_type) + + def _flash_groupbox_for_field(self, field_name: str): + """Flash the GroupBox for a specific field. + + Args: + field_name: Name of the field whose GroupBox should flash + """ + # Get the GroupBox widget from root manager + group_box = self.form_manager.widgets.get(field_name) + + if not group_box: + logger.warning(f"No GroupBox widget found for {field_name}") + return + + # Flash the GroupBox using scope border color + from PyQt6.QtGui import QColor + from openhcs.pyqt_gui.widgets.shared.scope_color_utils import get_scope_color_scheme + from openhcs.pyqt_gui.widgets.shared.widget_flash_animation import flash_widget + + # Get scope color scheme + color_scheme = get_scope_color_scheme(self.scope_id) + + # Use orchestrator border color for flash (same as window border) + border_rgb = color_scheme.orchestrator_item_border_rgb + flash_color = QColor(*border_rgb, 180) # Border color with high opacity + + # Use global registry to prevent overlapping flashes + flash_widget(group_box, flash_color=flash_color) + logger.info(f"✅ Flashed GroupBox for {field_name}") + + def _flash_for_config(self, config_name: str, dataclass_type: type = None) -> None: + """Flash tree item AND groupbox for a config. + + Args: + config_name: Name of the config (e.g., 'well_filter_config') + dataclass_type: The dataclass type (for tree matching - more reliable than name) + """ + from PyQt6.QtGui import QColor + from openhcs.pyqt_gui.widgets.shared.scope_color_utils import get_scope_color_scheme + + scope_id = getattr(self, 'scope_id', None) + if not scope_id: + return + + color_scheme = get_scope_color_scheme(scope_id) + border_rgb = color_scheme.orchestrator_item_border_rgb + flash_color = QColor(*border_rgb, 200) + + # 1. Flash the tree item (if tree exists) + tree_widget = getattr(self, 'tree_widget', None) or getattr(self, 'hierarchy_tree', None) + if tree_widget is not None: + # Prefer matching by class type (tree stores "class" in item data) + item = self._find_tree_item_by_class(dataclass_type, tree_widget) if dataclass_type else None + # Fall back to field_name matching + if item is None: + item = self._find_tree_item_by_field_name(config_name, tree_widget) + if item: + from openhcs.pyqt_gui.widgets.shared.tree_item_flash_animation import flash_tree_item + flash_tree_item(tree_widget, item, flash_color) + + # 2. Flash the groupbox + self._flash_groupbox_for_field(config_name) + + def _find_tree_item_by_class(self, dataclass_type: type, tree_widget: QTreeWidget, parent_item: Optional[QTreeWidgetItem] = None) -> Optional[QTreeWidgetItem]: + """Recursively find tree item by dataclass type (handles Lazy variants).""" + if parent_item is None: + for i in range(tree_widget.topLevelItemCount()): + result = self._find_tree_item_by_class(dataclass_type, tree_widget, tree_widget.topLevelItem(i)) + if result: + return result + return None + + data = parent_item.data(0, Qt.ItemDataRole.UserRole) + if data: + tree_class = data.get('class') + # Match by: exact type, subclass, or base class (handles LazyX vs X) + if tree_class and (tree_class == dataclass_type or + issubclass(dataclass_type, tree_class) or + issubclass(tree_class, dataclass_type)): + return parent_item + + for i in range(parent_item.childCount()): + result = self._find_tree_item_by_class(dataclass_type, tree_widget, parent_item.child(i)) + if result: + return result + return None + + def _find_tree_item_by_field_name(self, field_name: str, tree_widget: QTreeWidget, parent_item: Optional[QTreeWidgetItem] = None) -> Optional[QTreeWidgetItem]: + """Recursively find tree item by field_name (fallback for non-dataclass items).""" + if parent_item is None: + for i in range(tree_widget.topLevelItemCount()): + result = self._find_tree_item_by_field_name(field_name, tree_widget, tree_widget.topLevelItem(i)) + if result: + return result + return None + + data = parent_item.data(0, Qt.ItemDataRole.UserRole) + if data and data.get('field_name') == field_name: + return parent_item + + for i in range(parent_item.childCount()): + result = self._find_tree_item_by_field_name(field_name, tree_widget, parent_item.child(i)) + if result: + return result + return None + diff --git a/openhcs/pyqt_gui/widgets/shared/tree_item_flash_animation.py b/openhcs/pyqt_gui/widgets/shared/tree_item_flash_animation.py new file mode 100644 index 000000000..19a697206 --- /dev/null +++ b/openhcs/pyqt_gui/widgets/shared/tree_item_flash_animation.py @@ -0,0 +1,166 @@ +"""Flash animation for QTreeWidgetItem updates.""" + +import logging +from typing import Optional +from PyQt6.QtWidgets import QTreeWidget, QTreeWidgetItem +from PyQt6.QtGui import QColor, QBrush, QFont + +from .smooth_flash_base import SmoothFlashAnimatorBase +from .scope_visual_config import ScopeVisualConfig + +logger = logging.getLogger(__name__) + + +class TreeItemFlashAnimator(SmoothFlashAnimatorBase): + """Smooth flash animation for QTreeWidgetItem background and font changes. + + Design: + - Does NOT store item references (items can be destroyed during flash) + - Stores (tree_widget, item_id) for item lookup + - Gracefully handles item destruction (checks if item exists before restoring) + - Flashes both background color AND font weight for visibility + """ + + def __init__( + self, + tree_widget: QTreeWidget, + item: QTreeWidgetItem, + flash_color: QColor + ): + """Initialize animator. + + Args: + tree_widget: Parent tree widget + item: Tree item to flash + flash_color: Color to flash with + """ + self.tree_widget = tree_widget + self.item_id = id(item) # Store ID, not reference + + # Store original state when animator is created + self.original_background = item.background(0) + self.original_font = item.font(0) + original_color = self.original_background.color() if self.original_background.style() else QColor(255, 255, 255, 0) + + super().__init__(flash_color, original_color) + + def _find_item(self) -> Optional[QTreeWidgetItem]: + """Find tree item by ID (handles item recreation).""" + def search_tree(parent_item=None): + if parent_item is None: + for i in range(self.tree_widget.topLevelItemCount()): + item = self.tree_widget.topLevelItem(i) + if id(item) == self.item_id: + return item + result = search_tree(item) + if result: + return result + else: + for i in range(parent_item.childCount()): + child = parent_item.child(i) + if id(child) == self.item_id: + return child + result = search_tree(child) + if result: + return result + return None + return search_tree() + + def _can_flash(self) -> bool: + """Check if item still exists.""" + return self._find_item() is not None + + def _prepare_flash(self) -> None: + """Apply bold font on first flash.""" + item = self._find_item() + if item: + flash_font = QFont(self.original_font) + flash_font.setBold(True) + item.setFont(0, flash_font) + + def _apply_color(self, color: QColor) -> None: + """Apply interpolated color to tree item.""" + item = self._find_item() + if item is None: + return + item.setBackground(0, QBrush(color)) + self.tree_widget.viewport().update() + + def _on_animation_complete(self) -> None: + """Restore original background and font.""" + item = self._find_item() + if item is not None: + item.setBackground(0, self.original_background) + item.setFont(0, self.original_font) + self.tree_widget.viewport().update() + logger.debug(f"✅ Smooth flash complete for tree item") + + +# Global registry of animators (keyed by (tree_widget_id, item_id)) +_tree_item_animators: dict[tuple[int, int], TreeItemFlashAnimator] = {} + + +def flash_tree_item( + tree_widget: QTreeWidget, + item: QTreeWidgetItem, + flash_color: QColor +) -> None: + """Flash a tree item to indicate update. + + Args: + tree_widget: Tree widget containing the item + item: Tree item to flash + flash_color: Color to flash with + """ + logger.info(f"🔥 flash_tree_item called for item: {item.text(0)}") + + config = ScopeVisualConfig() + if not config.LIST_ITEM_FLASH_ENABLED: # Reuse list item flash config + logger.info(f"🔥 Flash DISABLED in config") + return + + if item is None: + logger.info(f"🔥 Item is None") + return + + logger.info(f"🔥 Creating/getting animator for tree item") + + key = (id(tree_widget), id(item)) + + # Get or create animator + if key not in _tree_item_animators: + logger.info(f"🔥 Creating NEW animator for tree item") + _tree_item_animators[key] = TreeItemFlashAnimator( + tree_widget, item, flash_color + ) + else: + logger.info(f"🔥 Reusing existing animator for tree item") + # Update flash color in case it changed + animator = _tree_item_animators[key] + animator.flash_color = flash_color + + animator = _tree_item_animators[key] + logger.info(f"🔥 Calling animator.flash_update() for tree item") + animator.flash_update() + + +def clear_all_tree_animators(tree_widget: QTreeWidget) -> None: + """Clear all animators for a specific tree widget. + + Call this before clearing/rebuilding the tree to prevent + flash animations from accessing destroyed items. + + Args: + tree_widget: Tree widget whose animators should be cleared + """ + widget_id = id(tree_widget) + keys_to_remove = [k for k in _tree_item_animators.keys() if k[0] == widget_id] + + for key in keys_to_remove: + animator = _tree_item_animators[key] + animator.stop() # Stop all animations + del _tree_item_animators[key] + + if keys_to_remove: + logger.debug(f"Cleared {len(keys_to_remove)} flash animators for tree widget") + diff --git a/openhcs/pyqt_gui/widgets/shared/widget_flash_animation.py b/openhcs/pyqt_gui/widgets/shared/widget_flash_animation.py new file mode 100644 index 000000000..aa1be0852 --- /dev/null +++ b/openhcs/pyqt_gui/widgets/shared/widget_flash_animation.py @@ -0,0 +1,121 @@ +"""Flash animation for form widgets (QLineEdit, QComboBox, etc.).""" + +import logging +from typing import Optional +from PyQt6.QtWidgets import QWidget, QGroupBox +from PyQt6.QtGui import QColor, QPalette + +from .smooth_flash_base import SmoothFlashAnimatorBase +from .scope_visual_config import ScopeVisualConfig + +logger = logging.getLogger(__name__) + + +class WidgetFlashAnimator(SmoothFlashAnimatorBase): + """Smooth flash animation for form widget background color changes. + + Uses stylesheet manipulation for GroupBox (since stylesheets override palettes), + and palette manipulation for input widgets. + """ + + def __init__(self, widget: QWidget, flash_color: Optional[QColor] = None): + """Initialize animator. + + Args: + widget: Widget to animate + flash_color: Optional custom flash color (defaults to config FLASH_COLOR_RGB) + """ + self.widget = widget + self.config = ScopeVisualConfig() + resolved_flash_color = flash_color or QColor(*self.config.FLASH_COLOR_RGB, 180) + self._original_stylesheet: Optional[str] = None + self._use_stylesheet: bool = False + + # Get initial original color (will be re-captured on first flash) + palette = widget.palette() + original_color = palette.color(QPalette.ColorRole.Base) + + super().__init__(resolved_flash_color, original_color) + + def _can_flash(self) -> bool: + """Check if widget is visible.""" + return self.widget is not None and self.widget.isVisible() + + def _prepare_flash(self) -> None: + """Capture original state before flash starts.""" + self._use_stylesheet = isinstance(self.widget, QGroupBox) + if self._use_stylesheet: + self._original_stylesheet = self.widget.styleSheet() + palette = self.widget.palette() + self._original_color = palette.color(QPalette.ColorRole.Window) + else: + palette = self.widget.palette() + self._original_color = palette.color(QPalette.ColorRole.Base) + + def _apply_color(self, color: QColor) -> None: + """Apply interpolated color to widget.""" + if not self.widget: + return + + if self._use_stylesheet: + # GroupBox: Apply via stylesheet + r, g, b, a = color.red(), color.green(), color.blue(), color.alpha() + style = f"QGroupBox {{ background-color: rgba({r}, {g}, {b}, {a}); }}" + self.widget.setStyleSheet(style) + else: + # Other widgets: Apply via palette + palette = self.widget.palette() + palette.setColor(QPalette.ColorRole.Base, color) + self.widget.setPalette(palette) + + def _on_animation_complete(self) -> None: + """Restore original stylesheet for GroupBox.""" + if self._use_stylesheet and self._original_stylesheet is not None: + self.widget.setStyleSheet(self._original_stylesheet) + logger.debug(f"✅ Smooth flash complete for {type(self.widget).__name__}") + + +# Global registry of animators (keyed by widget id) +_widget_animators: dict[int, WidgetFlashAnimator] = {} + + +def flash_widget(widget: QWidget, flash_color: Optional[QColor] = None) -> None: + """Flash a widget to indicate update. + + Args: + widget: Widget to flash + flash_color: Optional custom flash color (defaults to config FLASH_COLOR_RGB) + """ + config = ScopeVisualConfig() + if not config.WIDGET_FLASH_ENABLED: + return + + if not widget or not widget.isVisible(): + return + + widget_id = id(widget) + + # Get or create animator + if widget_id not in _widget_animators: + _widget_animators[widget_id] = WidgetFlashAnimator(widget, flash_color=flash_color) + else: + # Update flash color if provided + if flash_color is not None: + _widget_animators[widget_id].flash_color = flash_color + + animator = _widget_animators[widget_id] + animator.flash_update() + + +def cleanup_widget_animator(widget: QWidget) -> None: + """Cleanup animator when widget is destroyed. + + Args: + widget: Widget being destroyed + """ + widget_id = id(widget) + if widget_id in _widget_animators: + animator = _widget_animators[widget_id] + animator.stop() # Stop all animations + del _widget_animators[widget_id] + diff --git a/openhcs/pyqt_gui/widgets/step_parameter_editor.py b/openhcs/pyqt_gui/widgets/step_parameter_editor.py index f202e483e..8a7f481c3 100644 --- a/openhcs/pyqt_gui/widgets/step_parameter_editor.py +++ b/openhcs/pyqt_gui/widgets/step_parameter_editor.py @@ -21,6 +21,7 @@ from openhcs.pyqt_gui.widgets.shared.config_hierarchy_tree import ConfigHierarchyTreeHelper from openhcs.pyqt_gui.widgets.shared.collapsible_splitter_helper import CollapsibleSplitterHelper from openhcs.pyqt_gui.widgets.shared.scrollable_form_mixin import ScrollableFormMixin +from openhcs.pyqt_gui.widgets.shared.tree_form_flash_mixin import TreeFormFlashMixin from openhcs.pyqt_gui.shared.color_scheme import PyQt6ColorScheme from openhcs.pyqt_gui.shared.style_generator import StyleSheetGenerator from openhcs.pyqt_gui.config import PyQtGUIConfig, get_default_pyqt_gui_config @@ -32,7 +33,7 @@ logger = logging.getLogger(__name__) -class StepParameterEditorWidget(ScrollableFormMixin, QWidget): +class StepParameterEditorWidget(TreeFormFlashMixin, ScrollableFormMixin, QWidget): """ Step parameter editor using dynamic form generation. @@ -40,6 +41,8 @@ class StepParameterEditorWidget(ScrollableFormMixin, QWidget): constructor signature with nested dataclass support. Inherits from ScrollableFormMixin to provide scroll-to-section functionality. + + Inherits from TreeFormFlashMixin for visual flash animations on cross-window updates. """ # Signals @@ -139,6 +142,9 @@ def __init__(self, step: FunctionStep, service_adapter=None, color_scheme: Optio self.setup_ui() self.setup_connections() + # Register placeholder flash hook (TreeFormFlashMixin) - AFTER setup_ui so tree exists + self._register_placeholder_flash_hook() + logger.debug(f"Step parameter editor initialized for step: {getattr(step, 'name', 'Unknown')}") def _is_optional_lazy_dataclass_in_pipeline(self, param_type, param_name): @@ -386,8 +392,11 @@ def _handle_parameter_change(self, param_name: str, value: Any): # Remove type name prefix path_parts = path_parts[1:] - # For nested fields, the form manager already updated self.step via _mark_parents_modified - # For top-level fields, we need to update self.step + # REGISTRY REFACTOR: Use registry.set() instead of setattr (immutable) + # Registry stores concrete values separately, never mutates context_obj + from openhcs.config_framework import ContextStackRegistry + registry = ContextStackRegistry.instance() + if len(path_parts) == 1: leaf_field = path_parts[0] @@ -404,12 +413,14 @@ def _handle_parameter_change(self, param_name: str, value: Any): except Exception: pass # Use original if refresh fails - # Update step attribute - setattr(self.step, leaf_field, final_value) - logger.debug(f"Updated step parameter {leaf_field}={final_value}") + # Store in registry (immutable - doesn't mutate self.step) + registry.set(self.form_manager.scope_id, leaf_field, final_value) + logger.debug(f"Registry set: {self.form_manager.scope_id}.{leaf_field}={final_value}") else: - # Nested field - already updated by _mark_parents_modified - logger.debug(f"Nested field {'.'.join(path_parts)} already updated by dispatcher") + # Nested field - store full path in registry + full_field_path = '.'.join(path_parts) + registry.set(self.form_manager.scope_id, full_field_path, value) + logger.debug(f"Registry set: {self.form_manager.scope_id}.{full_field_path}={value}") self.step_parameter_changed.emit() @@ -577,6 +588,16 @@ def _handle_edited_step_code(self, edited_code: str) -> None: # Update step object self.step = new_step + # REGISTRY REFACTOR: Re-register scope with new object (code mode replacement) + from openhcs.config_framework import ContextStackRegistry + registry = ContextStackRegistry.instance() + registry.register_scope( + self.form_manager.scope_id, + new_step, + self.form_manager.dataclass_type + ) + logger.info(f"📦 Re-registered scope (code mode): {self.form_manager.scope_id}") + # IMPORTANT: # Do NOT block cross-window updates here. We want code-mode edits # to behave like a sequence of normal widget edits so that diff --git a/openhcs/pyqt_gui/windows/base_form_dialog.py b/openhcs/pyqt_gui/windows/base_form_dialog.py index 2f8101912..28f988827 100644 --- a/openhcs/pyqt_gui/windows/base_form_dialog.py +++ b/openhcs/pyqt_gui/windows/base_form_dialog.py @@ -42,6 +42,8 @@ def __init__(self, ...): from typing import Callable from PyQt6.QtCore import Qt +from openhcs.pyqt_gui.widgets.shared.scoped_border_mixin import ScopedBorderMixin + logger = logging.getLogger(__name__) @@ -50,7 +52,12 @@ class HasUnregisterMethod(Protocol): def unregister_from_cross_window_updates(self) -> None: ... -class BaseFormDialog(QDialog): +class BaseFormDialog(ScopedBorderMixin, QDialog): + """Base class for dialogs with form managers and scope-based borders. + + Inherits from ScopedBorderMixin to provide automatic scope-based border rendering. + Subclasses should call self._init_scope_border() after setting self.scope_id. + """ def _setup_save_button(self, button: 'QPushButton', save_callback: Callable): """ diff --git a/openhcs/pyqt_gui/windows/config_window.py b/openhcs/pyqt_gui/windows/config_window.py index d40526b84..829d85809 100644 --- a/openhcs/pyqt_gui/windows/config_window.py +++ b/openhcs/pyqt_gui/windows/config_window.py @@ -23,6 +23,7 @@ from openhcs.pyqt_gui.widgets.shared.config_hierarchy_tree import ConfigHierarchyTreeHelper from openhcs.pyqt_gui.widgets.shared.scrollable_form_mixin import ScrollableFormMixin from openhcs.pyqt_gui.widgets.shared.collapsible_splitter_helper import CollapsibleSplitterHelper +from openhcs.pyqt_gui.widgets.shared.tree_form_flash_mixin import TreeFormFlashMixin from openhcs.pyqt_gui.shared.style_generator import StyleSheetGenerator from openhcs.pyqt_gui.shared.color_scheme import PyQt6ColorScheme from openhcs.pyqt_gui.windows.base_form_dialog import BaseFormDialog @@ -40,7 +41,7 @@ # Infrastructure classes removed - functionality migrated to ParameterFormManager service layer -class ConfigWindow(ScrollableFormMixin, BaseFormDialog): +class ConfigWindow(TreeFormFlashMixin, ScrollableFormMixin, BaseFormDialog): """ PyQt6 Configuration Window. @@ -51,6 +52,8 @@ class ConfigWindow(ScrollableFormMixin, BaseFormDialog): cross-window placeholder updates when the dialog closes. Inherits from ScrollableFormMixin to provide scroll-to-section functionality. + + Inherits from TreeFormFlashMixin for visual flash animations on cross-window updates. """ # Signals @@ -121,6 +124,9 @@ def __init__(self, config_class: Type, current_config: Any, scope_id=self.scope_id # Pass scope_id to limit cross-window updates to same orchestrator ) + # REGISTRY REFACTOR: ResolvedItemStateService now connects directly to registry + # No need to connect to individual managers + if is_global_config_type(self.config_class): self._original_global_config_snapshot = copy.deepcopy(current_config) self.form_manager.parameter_changed.connect(self._on_global_config_field_changed) @@ -131,6 +137,9 @@ def __init__(self, config_class: Type, current_config: Any, # Setup UI self.setup_ui() + # Register placeholder flash hook (TreeFormFlashMixin) - AFTER setup_ui so tree exists + self._register_placeholder_flash_hook() + logger.debug(f"Config window initialized for {config_class.__name__}") def setup_ui(self): @@ -233,6 +242,9 @@ def setup_ui(self): self.style_generator.generate_tree_widget_style() ) + # Initialize scope-based border styling (ScopedBorderMixin) + self._init_scope_border() + def _create_inheritance_tree(self) -> QTreeWidget: """Create tree widget showing inheritance hierarchy for navigation.""" tree = self.tree_helper.create_tree_widget() diff --git a/openhcs/pyqt_gui/windows/dual_editor_window.py b/openhcs/pyqt_gui/windows/dual_editor_window.py index 87d95fa0f..651597a6d 100644 --- a/openhcs/pyqt_gui/windows/dual_editor_window.py +++ b/openhcs/pyqt_gui/windows/dual_editor_window.py @@ -46,7 +46,7 @@ class DualEditorWindow(BaseFormDialog): def __init__(self, step_data: Optional[FunctionStep] = None, is_new: bool = False, on_save_callback: Optional[Callable] = None, color_scheme: Optional[PyQt6ColorScheme] = None, - orchestrator=None, gui_config=None, parent=None): + orchestrator=None, gui_config=None, step_position: Optional[int] = None, parent=None): """ Initialize the dual editor window. @@ -74,6 +74,7 @@ def __init__(self, step_data: Optional[FunctionStep] = None, is_new: bool = Fals self.is_new = is_new self.on_save_callback = on_save_callback self.orchestrator = orchestrator # Store orchestrator for context management + self.step_position = step_position # For scope-based border styling # Pattern management (extracted from Textual version) self.pattern_manager = PatternDataManager() @@ -225,6 +226,11 @@ def setup_ui(self): self._function_sync_timer.timeout.connect(self._flush_function_editor_sync) self._pending_function_editor_sync = False + # Set scope_id for border styling (ScopedBorderMixin via BaseFormDialog) + step_name = getattr(self.editing_step, 'name', 'unknown_step') + self.scope_id = self._build_step_scope_id(step_name) + self._init_scope_border() + def _update_window_title(self): title = "New Step" if getattr(self, 'is_new', False) else f"Edit Step: {getattr(self.editing_step, 'name', 'Unknown')}" self.setWindowTitle(title) @@ -238,11 +244,15 @@ def _update_save_button_text(self): self.save_button.setText(new_text) def _build_step_scope_id(self, fallback_name: str) -> str: + """Build scope_id for this step window. + + Format: plate_path::step_token (matches PipelineEditorWidget scope_id_builder) + + REGISTRY REFACTOR: Removed @position suffix - scope_id identifies STEP, not position. + """ plate_scope = getattr(self.orchestrator, 'plate_path', 'no_orchestrator') - token = getattr(self.editing_step, '_pipeline_scope_token', None) - if token: - return f"{plate_scope}::{token}" - return f"{plate_scope}::{fallback_name}" + token = getattr(self.editing_step, '_pipeline_scope_token', None) or fallback_name + return f"{plate_scope}::{token}" def create_step_tab(self): """Create the step settings tab (using dedicated widget).""" diff --git a/plans/context-registry/plan_01_context_stack_registry.md b/plans/context-registry/plan_01_context_stack_registry.md new file mode 100644 index 000000000..ad9653065 --- /dev/null +++ b/plans/context-registry/plan_01_context_stack_registry.md @@ -0,0 +1,167 @@ +# plan_01_context_stack_registry.md +## Component: Central Context Stack Registry + +### Objective + +Create a **ContextStackRegistry** service that serves as the single source of truth for all resolved configuration values across the application. This eliminates the current architecture where: +- ParameterFormManager builds its own context stacks +- Preview labels resolve values through a separate path +- Flash/dirty tracking would need yet another resolution path + +**Scope**: Works for BOTH plate/orchestrator list items (PlateManager) AND step list items (PipelineEditor). + +### Rollout and Testing + +- Plans 01-04 will be applied back-to-back with end-to-end testing after the full integration. No interim compatibility shims or partial regression fixes are planned between steps. + +### Problem Statement + +**Current Architecture (Broken)**: +1. Forms build context stacks internally via `build_context_stack()` in `refresh_single_placeholder()` +2. Preview labels resolve via `_resolve_config_attr()` + `LiveContextResolver` +3. No single source of truth for "what is the resolved value of field X for scope Y" +4. Flash/dirty tracking requires re-resolving values (expensive, complex) + +**User's Vision**: +> "why don't we just have the context stacks built for all objects at all time and have the parameter form manager access them rather than build its own and maintain it internally? this way there's a single source of truth for context stacks and ui elements can read from it" + +### Plan + +1. **Create ContextStackRegistry singleton service** + - Location: `openhcs/config_framework/context_stack_registry.py` + - Singleton pattern matching existing services (LiveContextService, ResolvedItemStateService) + +2. **Define core API** + ```python + class ContextStackRegistry(QObject): + # Signal: (scope_id, field_path, old_value, new_value) + value_changed = pyqtSignal(str, str, object, object) + + def register_scope(self, scope_id: str, context_obj: Any, dataclass_type: type) -> None: + """Register a scope for resolution. Called when form opens or item created. + If scope_id already exists, REPLACES context_obj and clears concrete values (code mode).""" + + def unregister_scope(self, scope_id: str) -> None: + """Unregister scope. Called when form closes or item deleted.""" + + def set(self, scope_id: str, field_path: str, value: Any) -> None: + """Set a concrete value override. Does NOT mutate context_obj (immutable). + Emits value_changed if resolved value changes.""" + + def resolve(self, scope_id: str, field_path: str) -> Any: + """Resolve field value through context hierarchy. Cached.""" + + def get_resolved_state(self, scope_id: str) -> Dict[str, Any]: + """Get all resolved values for a scope. For dirty comparison.""" + + def get_materialized_object(self, scope_id: str) -> Any: + """Get context_obj with all resolved values applied. For compilation/serialization. + Returns a NEW instance with all fields materialized from resolved state.""" + ``` + +3. **Internal implementation** + - Reuse existing `build_context_stack()` from `context_manager.py` + - Reuse existing `config_context()` for resolution + - Lazy resolution with caching (resolve on first access) + - Cache invalidation is field-aware: when `set()` fires or LiveContextService token changes, invalidate only the affected `field_path` for the scope and its descendants + - **Immutability**: `set()` stores concrete values but does NOT mutate context_obj + - **Materialization**: `get_materialized_object()` creates new instance with resolved values applied + +4. **Scope hierarchy handling** + - Use existing `get_root_from_scope_key()` for hierarchy + - When value changes at scope X, invalidate cache for X and all children (same plate root) + - No explicit parent/child map needed - iterate registered scopes and match roots + - Scope ID format: `/path/to/plate` (plate), `/path/to/plate::step_token@index` (step) + +5. **Integration with LiveContextService** + - Registry does NOT use LiveContextService.collect() (avoids circular dependency) + - Registry builds stacks from: registered context_obj + concrete values from registry.set() + - Forms still register with LiveContextService for cross-window coordination + - Resolution precedence: concrete values (registry.set) > context_obj values > inherited defaults + +### Findings + +**Existing Infrastructure to Reuse**: +- `build_context_stack()` in `context_manager.py` - builds complete context stack +- `config_context()` - context manager for resolution +- `get_root_from_scope_key()` - extracts plate root from scope_id +- `LiveContextService` - collects live values from open forms +- `LiveContextResolver` - resolves through context hierarchy with caching + +**Key Patterns from Codebase**: +- Singleton services: `ResolvedItemStateService.instance()`, `LiveContextService` (class methods) +- Signal-based reactivity: `context_value_changed`, `item_resolved_changed` +- Scope visibility: `_is_scope_visible()` in ResolvedItemStateService + +**State Mutation Paths**: + +There are exactly 2 ways configuration state changes: + +1. **ParameterFormManager edits** (field-level changes): + - User edits field in form + - Form calls `registry.set(scope_id, field_path, value)` + - Registry stores concrete value (does NOT mutate context_obj - immutable) + - Registry emits `value_changed` if resolved value changed + - **For Steps**: Currently uses `setattr(step, field, value)` - will be REMOVED in plan_02 + - **For PipelineConfig**: Creates new instance on save via callback + +2. **Code mode edits** (instance replacement): + - User edits code, saves + - Code is `exec()`'d, creates **NEW instances** (PipelineConfig or step list) + - **For PipelineConfig**: Calls `orchestrator.apply_pipeline_config(new_config)` - REPLACES instance + - **For Steps**: Replaces entire list `pipeline_editor.plate_pipelines[plate_path] = new_steps` + - Registry handles via re-registration: `register_scope(scope_id, new_obj, ...)` (replaces if exists) + +**Unified Mutation Path**: +- Live edits → `registry.set(scope_id, field_path, value)` +- Replacements → `registry.register_scope(scope_id, new_obj, ...)` (re-registration replaces) + +**No other mutation paths exist**. If new paths are added later (server pushes, reloads), they must explicitly call `register_scope()` with new objects. + +**Current Resolution Flow** (to be replaced): +1. `refresh_single_placeholder()` calls `build_context_stack()` +2. Stack includes: global layer, intermediate layers, context_obj, root form, overlay +3. Resolution happens inside `with stack:` context +4. Placeholder text computed via `service.get_placeholder_text()` + +**Materialization Example**: +```python +# Initial step from code mode: +step = FunctionStep( + func=my_function, + step_materialization_config=LazyStepMaterializationConfig( + sub_dir="checkpoints" # Override + # output_dir_suffix=None - will inherit from PipelineConfig.path_planning_config + ) +) + +# Registry internal state after registration: +# _context_objs[scope_id] = step (the original object) +# _concrete_values[scope_id] = {} # Empty initially + +# User edits output_dir_suffix to "_custom" in form: +registry.set(scope_id, "step_materialization_config.output_dir_suffix", "_custom") + +# Registry internal state after edit: +# _context_objs[scope_id] = step (UNCHANGED - immutable) +# _concrete_values[scope_id] = {"step_materialization_config.output_dir_suffix": "_custom"} + +# When compiling pipeline: +materialized_step = registry.get_materialized_object(scope_id) +# Returns NEW FunctionStep with: +# step_materialization_config.sub_dir = "checkpoints" (from original) +# step_materialization_config.output_dir_suffix = "_custom" (from concrete values) +# All other fields resolved through context hierarchy + +orchestrator.compile([materialized_step]) +``` + +**Why Immutability + Materialization**: +- ✅ Single Responsibility: Registry resolves, doesn't mutate +- ✅ No side effects: `set()` is pure +- ✅ Explicit boundary: Materialization makes snapshot clear +- ✅ Testable: Easy to test without object mutation concerns + +### Implementation Draft + +*To be written after smell loop approval* diff --git a/plans/context-registry/plan_02_form_manager_integration.md b/plans/context-registry/plan_02_form_manager_integration.md new file mode 100644 index 000000000..506c82742 --- /dev/null +++ b/plans/context-registry/plan_02_form_manager_integration.md @@ -0,0 +1,77 @@ +# plan_02_form_manager_integration.md +## Component: ParameterFormManager Integration with Registry + +### Objective + +Migrate ParameterFormManager to use ContextStackRegistry as the single source of truth for resolved values. Forms will: +- **Write** to registry when user edits a field +- **Read** from registry for placeholder resolution +- **No longer** build their own context stacks + +### Rollout and Testing + +- Applied immediately after plan_01 and before plans 03/04, with no interim compatibility shims; any downstream listeners are expected to be updated in later plans. End-to-end testing happens after plan_04. + +### Plan + +1. **On form initialization** + - Call `registry.register_scope(scope_id, context_obj, dataclass_type)` + - Store registry reference for later use + +2. **On user edit (parameter_changed signal) - UNIFIED MUTATION PATH** + - Call `registry.set(scope_id, field_path, value)` for ALL objects (steps, PipelineConfig, everything) + - **Remove `setattr(self.step, field_name, value)` from step_parameter_editor.py** (no more in-place mutation) + - **Remove all `setattr()` patterns that mutate context_obj** (objects are immutable from registry's perspective) + - Registry stores concrete values, does NOT mutate context_obj + - Registry handles cache invalidation and signal emission + - Remove direct `context_value_changed` emission (registry does it; downstream listeners are re-wired in plans 03/04) + +3. **On placeholder resolution** + - Replace `refresh_single_placeholder()` implementation + - Call `registry.resolve(scope_id, field_path)` instead of building stack + - Apply resolved value as placeholder text + +4. **On form close** + - Call `registry.unregister_scope(scope_id)` + +5. **Code mode integration** + - When code mode creates new objects, call `registry.register_scope(scope_id, new_obj, ...)` (re-registration) + - Registry detects re-registration (scope_id exists), replaces context_obj, clears concrete values + - Registry emits `value_changed` for all fields that changed between old and new object + +6. **Materialization for compilation/serialization** + - When compiling pipeline: `materialized_steps = [registry.get_materialized_object(scope_id) for scope_id in step_scopes]` + - When generating code: Use `registry.get_materialized_object()` to get concrete instances + - Objects in `pipeline_editor.pipeline_steps` remain as-is (original from code mode or creation) + +7. **Remove deprecated code** + - Remove `build_context_stack()` calls from ParameterOpsService + - Remove internal stack building in `refresh_all_placeholders()` + - Simplify `_schedule_cross_window_refresh()` to just trigger registry update + - Keep `parameters` dict in ParameterFormManager for LiveContextService compatibility + +### Findings + +**Current ParameterFormManager Flow**: +- `__init__`: Registers with LiveContextService, connects signals +- `_emit_parameter_change()`: Emits `context_value_changed` signal +- `refresh_single_placeholder()`: Builds context stack, resolves placeholder +- `refresh_all_placeholders()`: Iterates widgets, builds stack, resolves each + +**Assumptions/Notes**: +- Registry is the sole ingestion path for live edits: ParameterFormManager calls `registry.set()` +- Forms keep `parameters` dict so LiveContextService can read from it (no circular dependency) +- Registry does NOT read from LiveContextService (avoids loops) +- Objects are immutable from registry's perspective (no setattr on context_obj) + +**Files to Modify**: +- `openhcs/pyqt_gui/widgets/shared/parameter_form_manager.py` +- `openhcs/pyqt_gui/widgets/shared/services/parameter_ops_service.py` +- `openhcs/pyqt_gui/widgets/shared/services/field_change_dispatcher.py` +- `openhcs/pyqt_gui/widgets/step_parameter_editor.py` (remove setattr mutations) +- `openhcs/pyqt_gui/widgets/plate_manager.py` (code mode re-registration) +- `openhcs/pyqt_gui/widgets/pipeline_editor.py` (code mode re-registration, materialization for compilation) + +### Implementation Draft + +*To be written after plan_01 is implemented and smell loop approved* diff --git a/plans/context-registry/plan_03_preview_label_integration.md b/plans/context-registry/plan_03_preview_label_integration.md new file mode 100644 index 000000000..c65c3fbae --- /dev/null +++ b/plans/context-registry/plan_03_preview_label_integration.md @@ -0,0 +1,58 @@ +# plan_03_preview_label_integration.md +## Component: Preview Label Integration with Registry + +### Objective + +Migrate preview label resolution to use ContextStackRegistry. Preview labels will: +- **Read** from registry for resolved values +- **React** to registry.value_changed signal for updates +- **No longer** resolve through separate LiveContextResolver path + +### Plan + +1. **Replace `_resolve_preview_field_value()`** + - Call `registry.resolve(scope_id, field_path)` instead of building context + - Remove `_resolve_config_attr()` calls + - Remove `LiveContextResolver` usage in abstract_manager_widget + +2. **Replace `_build_preview_labels()`** + - Iterate enabled preview fields + - Call `registry.resolve()` for each field + - Format and return labels + +3. **React to registry changes** + - Connect to `registry.value_changed` signal + - On ANY value change for a scope, refresh ALL preview fields for that scope + - Preview labels don't cache - they just call `registry.resolve()` for each field when rendering + - This handles composite previews correctly (multiple fields contributing to label) + - Use scope visibility to filter which items need update + +4. **Remove deprecated code** + - Remove `_live_context_resolver` from AbstractManagerWidget + - Remove `CrossWindowPreviewMixin._on_live_context_changed()` (registry handles it) + - Simplify `_handle_full_preview_refresh()` to use registry + +### Rollout and Testing + +- Executed immediately after plan_02; downstream listeners are updated here. End-to-end testing waits until plan_04 completion; no interim compatibility shims. + +### Findings + +**Current Preview Label Flow**: +- `_build_preview_labels()` iterates enabled fields +- `_resolve_preview_field_value()` navigates dotted paths +- `_resolve_config_attr()` uses LiveContextResolver with context stack +- `CrossWindowPreviewMixin` listens to LiveContextService for changes + +**Assumptions/Notes**: +- Preview labels are assumed to depend on the matching `field_path`. If any labels compose multiple fields, add a small dependency map or occasional full refresh to keep those labels in sync. + +**Files to Modify**: +- `openhcs/pyqt_gui/widgets/shared/abstract_manager_widget.py` +- `openhcs/pyqt_gui/widgets/mixins/cross_window_preview_mixin.py` +- `openhcs/pyqt_gui/widgets/plate_manager.py` +- `openhcs/pyqt_gui/widgets/pipeline_editor.py` + +### Implementation Draft + +*To be written after plan_02 is implemented and smell loop approved* diff --git a/plans/context-registry/plan_04_flash_dirty_integration.md b/plans/context-registry/plan_04_flash_dirty_integration.md new file mode 100644 index 000000000..dc5086295 --- /dev/null +++ b/plans/context-registry/plan_04_flash_dirty_integration.md @@ -0,0 +1,58 @@ +# plan_04_flash_dirty_integration.md +## Component: Flash and Dirty Tracking via Registry + +### Objective + +Implement flash animations and dirty tracking using ContextStackRegistry signals. This becomes trivial because: +- **Flash**: Registry emits `(scope_id, field_path, old_value, new_value)` - flash if `old != new` +- **Dirty**: Compare `registry.get_resolved_state(scope_id)` to saved baseline + +### Rollout and Testing + +- Final step in the back-to-back rollout. End-to-end testing happens after this plan is completed; no interim verification is planned before this point. + +### Plan + +1. **Flash implementation** + - Listen to `registry.value_changed` signal + - If `old_value != new_value`, emit flash for affected scope_id + - Use scope visibility to determine which list items flash + - No field tracking needed - registry handles propagation + +2. **Dirty tracking implementation** + - **On register_scope()**: Capture initial baseline if none exists (newly opened scope) + - **On save**: Update baseline to current `registry.get_resolved_state(scope_id)` + - **On load**: Capture baseline from loaded state + - **On check**: Compare current resolved state to baseline + - Emit `item_dirty_changed(scope_id, is_dirty)` signal + +3. **Simplify ResolvedItemStateService** + - Remove field tracking (registry handles it) + - Remove `_on_context_value_changed()` (use registry signal) + - Keep scope registration for flash targeting + - Add baseline storage for dirty tracking + +4. **UI integration** + - AbstractManagerWidget connects to service signals + - `_on_item_resolved_changed()` triggers flash animation + - `_on_item_dirty_changed()` updates dirty indicator + +### Findings + +**User's Exact Specification**: +1. Flash: "only flash if the resolved state changes from before the signal and after the signal" +2. Preview text: "just reads the current resolved state always, simple as" +3. Dirty: "checks if the current resolved state is different from the saved resolved state" + +**Current ResolvedItemStateService**: +- Listens to `context_value_changed` from forms +- Uses scope visibility to filter +- Flashes on ANY change (wrong - should only flash if resolved value changes) + +**Files to Modify**: +- `openhcs/pyqt_gui/widgets/shared/services/resolved_item_state_service.py` +- `openhcs/pyqt_gui/widgets/shared/abstract_manager_widget.py` + +### Implementation Draft + +*To be written after plan_03 is implemented and smell loop approved* diff --git a/pyproject.toml b/pyproject.toml index 15a56a151..30f6bb32f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -90,6 +90,8 @@ gui = [ "PyQt6-QScintilla>=2.14.1", "pyqtgraph>=0.13.7", # Used for system monitor visualization "GPUtil>=1.4.0", + "distinctipy>=1.2.2", # Perceptually distinct colors for scope-based visual feedback + "wcag-contrast-ratio>=0.9", # WCAG AA compliance for scope colors # plotext removed - PyQt GUI now uses pyqtgraph instead # psutil moved to core dependencies (required by ui/shared/system_monitor_core.py) ] @@ -173,6 +175,8 @@ all = [ "PyQt6-QScintilla>=2.14.1", "pyqtgraph>=0.13.7", "GPUtil>=1.4.0", + "distinctipy>=1.2.2", # Perceptually distinct colors for scope-based visual feedback + "wcag-contrast-ratio>=0.9", # WCAG AA compliance for scope colors # plotext removed - PyQt GUI now uses pyqtgraph instead # psutil moved to core dependencies (required by ui/shared/system_monitor_core.py)