Coverage for openhcs/core/pipeline/function_contracts.py: 84.2%

1""" 

2Function-level contract decorators for the pipeline compiler. 

3 

4This module provides decorators for declaring special input and output contracts 

5at the function level, enabling compile-time validation of dependencies between 

6processing functions in the pipeline. 

7 

8These decorators complement the class-level @special_in and @special_out decorators 

9by allowing more granular contract declarations at the function level. 

10 

11Doctrinal Clauses: 

12- Clause 3 — Declarative Primacy 

13- Clause 66 — Immutability After Construction 

14- Clause 88 — No Inferred Capabilities 

15- Clause 245 — Declarative Enforcement 

16- Clause 246 — Statelessness Mandate 

17- Clause 251 — Special Output Contract 

18""" 

19 

20from typing import Callable, Any, TypeVar, Set, Dict 

21 

22F = TypeVar('F', bound=Callable[..., Any]) 

23 

24 

25# Old special_output and special_input decorators are removed. 

26 

27def special_outputs(*output_specs) -> Callable[[F], F]: 

28 """ 

29 Decorator that marks a function as producing special outputs. 

30 

31 Args: 

32 *output_specs: Either strings or (string, materialization_function) tuples 

33 - String only: "positions" - no materialization function 

34 - Tuple: ("cell_counts", materialize_cell_counts) - with materialization 

35 

36 Examples: 

37 @special_outputs("positions", "metadata") # String only 

38 def process_image(image): 

39 return processed_image, positions, metadata 

40 

41 @special_outputs(("cell_counts", materialize_cell_counts)) # With materialization 

42 def count_cells(image): 

43 return processed_image, cell_count_results 

44 

45 @special_outputs("positions", ("cell_counts", materialize_cell_counts)) # Mixed 

46 def analyze_image(image): 

47 return processed_image, positions, cell_count_results 

48 """ 

49 def decorator(func: F) -> F: 

50 special_outputs_info = {} 

51 output_keys = set() 

52 

53 for spec in output_specs: 

54 if isinstance(spec, str): 

55 # String only - no materialization function 

56 output_keys.add(spec) 

57 special_outputs_info[spec] = None 

58 elif isinstance(spec, tuple) and len(spec) == 2: 58 ↛ 68line 58 didn't jump to line 68 because the condition on line 58 was always true

59 # (key, materialization_function) tuple 

60 key, mat_func = spec 

61 if not isinstance(key, str): 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true

62 raise ValueError(f"Special output key must be string, got {type(key)}: {key}") 

63 if not callable(mat_func): 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true

64 raise ValueError(f"Materialization function must be callable, got {type(mat_func)}: {mat_func}") 

65 output_keys.add(key) 

66 special_outputs_info[key] = mat_func 

67 else: 

68 raise ValueError(f"Invalid special output spec: {spec}. Must be string or (string, function) tuple.") 

69 

70 # Set both attributes for backward compatibility and new functionality 

71 func.__special_outputs__ = output_keys # For path planner (backward compatibility) 

72 func.__materialization_functions__ = special_outputs_info # For materialization system 

73 return func 

74 return decorator 

75 

76 

77def special_inputs(*input_names: str) -> Callable[[F], F]: 

78 """ 

79 Decorator that marks a function as requiring special inputs. 

80 

81 Args: 

82 *input_names: Names of the additional input parameters (excluding the first) 

83 that must be produced by other functions 

84 

85 Example: 

86 @special_inputs("positions", "metadata") 

87 def stitch_images(image_stack, positions, metadata): 

88 # First parameter is always the input image (3D array) 

89 # Additional parameters are special inputs from other functions 

90 return stitched_image 

91 """ 

92 def decorator(func: F) -> F: 

93 # For special_inputs, we store them as a dictionary with True as the value, 

94 # similar to the old special_input decorator, for compatibility with 

95 # existing logic in PathPlanner that expects a dict. 

96 # The 'required' flag is implicitly True for all named inputs here. 

97 # If optional special inputs are needed later, this structure can be extended. 

98 func.__special_inputs__ = {name: True for name in input_names} 

99 return func 

100 return decorator 

101 

102 

103 

104 

105