Coverage for openhcs/config_framework/cache_warming.py: 19.0%

1""" 

2Generic cache warming for configuration analysis. 

3 

4This module provides a generic function to pre-warm analysis caches for any 

5configuration hierarchy, eliminating first-load penalties in UI forms. 

6""" 

7 

8import dataclasses 

9import logging 

10from typing import Type, Set, Optional, get_args, get_origin, Callable 

11 

12from openhcs.introspection.signature_analyzer import SignatureAnalyzer 

13from openhcs.introspection.unified_parameter_analyzer import UnifiedParameterAnalyzer 

14from openhcs.ui.shared.parameter_form_service import ParameterFormService 

15 

16logger = logging.getLogger(__name__) 

17 

18 

19def _extract_all_dataclass_types(base_type: Type, visited: Optional[Set[Type]] = None) -> Set[Type]: 

20 """ 

21 Recursively extract all dataclass types from a configuration hierarchy. 

22  

23 Uses type introspection to discover all nested dataclass fields automatically. 

24 This is fully generic and works for any dataclass hierarchy. 

25  

26 Args: 

27 base_type: Root dataclass type to analyze 

28 visited: Set of already-visited types (for cycle detection) 

29  

30 Returns: 

31 Set of all dataclass types found in the hierarchy 

32 """ 

33 if visited is None: 

34 visited = set() 

35 

36 # Avoid infinite recursion on circular references 

37 if base_type in visited: 

38 return visited 

39 

40 # Only process dataclasses 

41 if not dataclasses.is_dataclass(base_type): 

42 return visited 

43 

44 visited.add(base_type) 

45 

46 # Introspect all fields to find nested dataclasses 

47 for field in dataclasses.fields(base_type): 

48 field_type = field.type 

49 

50 # Handle Optional[T] -> extract T 

51 origin = get_origin(field_type) 

52 if origin is not None: 

53 # For Union types (including Optional), check all args 

54 args = get_args(field_type) 

55 for arg in args: 

56 if arg is type(None): 

57 continue 

58 if dataclasses.is_dataclass(arg): 

59 _extract_all_dataclass_types(arg, visited) 

60 elif dataclasses.is_dataclass(field_type): 

61 # Direct dataclass field 

62 _extract_all_dataclass_types(field_type, visited) 

63 

64 return visited 

65 

66 

67def prewarm_callable_analysis_cache(*callables: Callable) -> None: 

68 """ 

69 Pre-warm analysis caches for callable signatures (functions, methods, constructors). 

70 

71 This is useful for warming caches for step editors, function pattern editors, etc. 

72 that analyze function signatures rather than dataclass hierarchies. 

73 

74 Args: 

75 *callables: One or more callables to analyze (functions, methods, __init__, etc.) 

76 

77 Example: 

78 >>> from openhcs.core.steps.abstract import AbstractStep 

79 >>> from openhcs.config_framework import prewarm_callable_analysis_cache 

80 >>> prewarm_callable_analysis_cache(AbstractStep.__init__) 

81 """ 

82 for callable_obj in callables: 

83 SignatureAnalyzer.analyze(callable_obj) 

84 UnifiedParameterAnalyzer.analyze(callable_obj) 

85 

86 logger.debug(f"Pre-warmed analysis cache for {len(callables)} callables") 

87 

88 

89def prewarm_config_analysis_cache(base_config_type: Type) -> None: 

90 """ 

91 Pre-warm analysis caches for all config types in a hierarchy. 

92 

93 This is a fully generic function that: 

94 1. Uses type introspection to discover all dataclass types in the hierarchy 

95 2. Pre-analyzes each type to populate analysis caches 

96 3. Also analyzes the lazy version of the base config (if it exists) 

97 4. Eliminates 1000ms+ first-load penalty when opening config windows 

98 

99 After this runs, first load is as fast as second load (~170ms instead of ~1000ms). 

100 

101 Args: 

102 base_config_type: Root configuration type (e.g., GlobalPipelineConfig) 

103 

104 Example: 

105 >>> from myapp.config import GlobalConfig 

106 >>> from openhcs.config_framework import prewarm_config_analysis_cache 

107 >>> prewarm_config_analysis_cache(GlobalConfig) 

108 """ 

109 # Discover all dataclass types in the hierarchy using introspection 

110 config_types = _extract_all_dataclass_types(base_config_type) 

111 

112 # Also add the lazy version of the base config if it exists 

113 # (e.g., GlobalPipelineConfig -> PipelineConfig) 

114 lazy_name = base_config_type.__name__.replace("Global", "") 

115 if lazy_name != base_config_type.__name__: 

116 import importlib 

117 module = importlib.import_module(base_config_type.__module__) 

118 lazy_type = getattr(module, lazy_name, None) 

119 

120 if lazy_type is not None and dataclasses.is_dataclass(lazy_type): 

121 config_types.add(lazy_type) 

122 

123 # Create a single service instance to warm the class-level cache 

124 service = ParameterFormService() 

125 

126 # Pre-analyze all config types to populate caches 

127 for config_type in config_types: 

128 # Warm SignatureAnalyzer cache (dataclass field analysis) 

129 SignatureAnalyzer._analyze_dataclass(config_type) 

130 

131 # Warm UnifiedParameterAnalyzer cache (parameter info with descriptions) 

132 param_info = UnifiedParameterAnalyzer.analyze(config_type) 

133 

134 # Warm ParameterFormService cache (form structure analysis) 

135 # This is the expensive part that builds the recursive FormStructure 

136 if dataclasses.is_dataclass(config_type): 

137 # Extract parameters from the dataclass 

138 params = {} 

139 param_types = {} 

140 for field in dataclasses.fields(config_type): 

141 params[field.name] = None # Dummy value 

142 param_types[field.name] = field.type 

143 

144 # Analyze to warm the cache 

145 service.analyze_parameters( 

146 params, param_types, 

147 field_id='cache_warming', 

148 parameter_info=param_info, 

149 parent_dataclass_type=config_type 

150 ) 

151 

152 logger.debug(f"Pre-warmed analysis cache for {len(config_types)} config types") 

153