Coverage for openhcs/core/pipeline/pipeline_utils.py: 52.2%

1# openhcs/core/pipeline/pipeline_utils.py 

2"""Utility functions for the OpenHCS pipeline system.""" 

3from typing import Any, Callable, List, Optional, Tuple 

4 

5def get_core_callable(func_pattern: Any) -> Optional[Callable[..., Any]]: 

6 """ 

7 Extracts the first effective Python callable from a func_pattern. 

8 A func_pattern can be a direct callable, a (callable, kwargs) tuple, 

9 a list (chain) where the first element is one of these types, 

10 or a dict pattern where we extract from the first value. 

11 """ 

12 # Check for FunctionReference first 

13 try: 

14 from openhcs.core.pipeline.compiler import FunctionReference 

15 if isinstance(func_pattern, FunctionReference): 

16 return func_pattern.resolve() 

17 except ImportError: 

18 pass 

19 

20 if callable(func_pattern) and not isinstance(func_pattern, type): 

21 # It's a direct callable (and not an uninstantiated class) 

22 return func_pattern 

23 elif isinstance(func_pattern, tuple) and func_pattern: 23 ↛ 38line 23 didn't jump to line 38 because the condition on line 23 was always true

24 # It's a (callable, kwargs) tuple, check first element 

25 first_element = func_pattern[0] 

26 

27 # Check if first element is a FunctionReference 

28 try: 

29 from openhcs.core.pipeline.compiler import FunctionReference 

30 if isinstance(first_element, FunctionReference): 

31 return first_element.resolve() 

32 except ImportError: 

33 pass 

34 

35 # Check if first element is a regular callable 

36 if callable(first_element) and not isinstance(first_element, type): 36 ↛ 48line 36 didn't jump to line 48 because the condition on line 36 was always true

37 return first_element 

38 elif isinstance(func_pattern, list) and func_pattern: 

39 # It's a list (chain), recursively call for the first item 

40 return get_core_callable(func_pattern[0]) 

41 elif isinstance(func_pattern, dict) and func_pattern: 

42 # It's a dict pattern, extract from the first value 

43 # For transformed dict patterns, we want the transformed function with namespaced outputs 

44 for key, value in func_pattern.items(): 

45 core_callable = get_core_callable(value) 

46 if core_callable: 

47 return core_callable 

48 return None 

49