Coverage for src/hieraconf/cache_warming.py: 32%
59 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-02 21:44 +0000
1"""
2Generic cache warming for configuration analysis.
4This module provides a generic function to pre-warm analysis caches for any
5configuration hierarchy, eliminating first-load penalties in UI forms.
6"""
8import dataclasses
9import logging
10from typing import Type, Set, Optional, get_args, get_origin, Callable
12# Optional introspection - install openhcs for full functionality
13try:
14 from openhcs.introspection.signature_analyzer import SignatureAnalyzer
15 from openhcs.introspection.unified_parameter_analyzer import UnifiedParameterAnalyzer
16 from openhcs.ui.shared.parameter_form_service import ParameterFormService
17except ImportError:
18 SignatureAnalyzer = None
19 UnifiedParameterAnalyzer = None
20 ParameterFormService = None
22logger = logging.getLogger(__name__)
25def _extract_all_dataclass_types(base_type: Type, visited: Optional[Set[Type]] = None) -> Set[Type]:
26 """
27 Recursively extract all dataclass types from a configuration hierarchy.
29 Uses type introspection to discover all nested dataclass fields automatically.
30 This is fully generic and works for any dataclass hierarchy.
32 Args:
33 base_type: Root dataclass type to analyze
34 visited: Set of already-visited types (for cycle detection)
36 Returns:
37 Set of all dataclass types found in the hierarchy
38 """
39 if visited is None:
40 visited = set()
42 # Avoid infinite recursion on circular references
43 if base_type in visited:
44 return visited
46 # Only process dataclasses
47 if not dataclasses.is_dataclass(base_type):
48 return visited
50 visited.add(base_type)
52 # Introspect all fields to find nested dataclasses
53 for field in dataclasses.fields(base_type):
54 field_type = field.type
56 # Handle Optional[T] -> extract T
57 origin = get_origin(field_type)
58 if origin is not None:
59 # For Union types (including Optional), check all args
60 args = get_args(field_type)
61 for arg in args:
62 if arg is type(None):
63 continue
64 if dataclasses.is_dataclass(arg):
65 _extract_all_dataclass_types(arg, visited)
66 elif dataclasses.is_dataclass(field_type):
67 # Direct dataclass field
68 _extract_all_dataclass_types(field_type, visited)
70 return visited
73def prewarm_callable_analysis_cache(*callables: Callable) -> None:
74 """
75 Pre-warm analysis caches for callable signatures (functions, methods, constructors).
77 This is useful for warming caches for step editors, function pattern editors, etc.
78 that analyze function signatures rather than dataclass hierarchies.
80 Args:
81 *callables: One or more callables to analyze (functions, methods, __init__, etc.)
83 Example:
84 >>> from openhcs.core.steps.abstract import AbstractStep
85 >>> from hieraconf import prewarm_callable_analysis_cache
86 >>> prewarm_callable_analysis_cache(AbstractStep.__init__)
87 """
88 for callable_obj in callables:
89 SignatureAnalyzer.analyze(callable_obj)
90 UnifiedParameterAnalyzer.analyze(callable_obj)
92 logger.debug(f"Pre-warmed analysis cache for {len(callables)} callables")
95def prewarm_config_analysis_cache(base_config_type: Type) -> None:
96 """
97 Pre-warm analysis caches for all config types in a hierarchy.
99 This is a fully generic function that:
100 1. Uses type introspection to discover all dataclass types in the hierarchy
101 2. Pre-analyzes each type to populate analysis caches
102 3. Also analyzes the lazy version of the base config (if it exists)
103 4. Eliminates 1000ms+ first-load penalty when opening config windows
105 After this runs, first load is as fast as second load (~170ms instead of ~1000ms).
107 Args:
108 base_config_type: Root configuration type (e.g., GlobalPipelineConfig)
110 Example:
111 >>> from myapp.config import GlobalConfig
112 >>> from hieraconf import prewarm_config_analysis_cache
113 >>> prewarm_config_analysis_cache(GlobalConfig)
114 """
115 # Discover all dataclass types in the hierarchy using introspection
116 config_types = _extract_all_dataclass_types(base_config_type)
118 # Also add the lazy version of the base config if it exists
119 # (e.g., GlobalPipelineConfig -> PipelineConfig)
120 lazy_name = base_config_type.__name__.replace("Global", "")
121 if lazy_name != base_config_type.__name__:
122 import importlib
123 module = importlib.import_module(base_config_type.__module__)
124 lazy_type = getattr(module, lazy_name, None)
126 if lazy_type is not None and dataclasses.is_dataclass(lazy_type):
127 config_types.add(lazy_type)
129 # Create a single service instance to warm the class-level cache
130 service = ParameterFormService()
132 # Pre-analyze all config types to populate caches
133 for config_type in config_types:
134 # Warm SignatureAnalyzer cache (dataclass field analysis)
135 SignatureAnalyzer._analyze_dataclass(config_type)
137 # Warm UnifiedParameterAnalyzer cache (parameter info with descriptions)
138 param_info = UnifiedParameterAnalyzer.analyze(config_type)
140 # Warm ParameterFormService cache (form structure analysis)
141 # This is the expensive part that builds the recursive FormStructure
142 if dataclasses.is_dataclass(config_type):
143 # Extract parameters from the dataclass
144 params = {}
145 param_types = {}
146 for field in dataclasses.fields(config_type):
147 params[field.name] = None # Dummy value
148 param_types[field.name] = field.type
150 # Analyze to warm the cache
151 service.analyze_parameters(
152 params, param_types,
153 field_id='cache_warming',
154 parameter_info=param_info,
155 parent_dataclass_type=config_type
156 )
158 logger.debug(f"Pre-warmed analysis cache for {len(config_types)} config types")