Coverage for skema/gromet/execution_engine/execute.py: 67%

1from typing import Any, List, Dict 

2import importlib 

3import builtins # Used to call builtin functions 

4 

5module_imports = {} 

6module_imports["__builtins__"] = builtins 

7 

8def execute_primitive(primitive: str, inputs: List[Any]) -> Any: 

9 # If the module path has no . then we assume it is a builtin function  

10 # Builtin functions belong to the __builtins__ module 

11 module_path = primitive.rsplit(".", 1) 

12 if len(module_path) == 1: 

13 module = "__builtins__" 

14 primitive = primitive 

15 else: 

16 module = module_path[0] 

17 primitive = module_path[1] 

18 

19 # Next, we check if the primitive can be imported from another installed library like operators 

20 # or Numpy if it is installed 

21 if module not in module_imports: 

22 try: 

23 module_imports[module] = importlib.import_module(module) 

24 except: 

25 print(f"Could not find module to import for: {primitive}") 

26 return None 

27 

28 # Finally, we attempt to execute the primitive 

29 try: 

30 f = getattr(module_imports[module], primitive) 

31 return f(*inputs) 

32 except: 

33 print(f"Could not execute primitive: {primitive}") 

34 return None 

35 

36 

37#print(execute_primitive("operator.add", [10, 20])) 

38#print(execute_primitive("operator.sub", [1,-2])) 

39#print(execute_primitive("numpy.array", [[1,2,3]])) 

40#print(execute_primitive("list", [[1,2,3]])) 

41#print(module_imports)