Coverage for backend/ahuora-builder/src/ahuora_builder/methods/adapter_library.py: 83%

1from __future__ import annotations 

2 

3import importlib 

4import json 

5from collections.abc import Iterator, Mapping 

6from pathlib import Path 

7 

8from collections.abc import Callable 

9 

10type ModelConstructor = Callable 

11 

12 

13class JsonBackedAdapterLibrary(Mapping[str, ModelConstructor]): 

14 """Resolve unit model constructors from generated unit-operation JSON files.""" 

15 

16 def __init__(self, config_dir: Path | None = None) -> None: 

17 self.config_dir = config_dir or _default_config_dir() 

18 self._constructor_paths: dict[str, str] | None = None 

19 self._constructors: dict[str, ModelConstructor] = {} 

20 

21 def __getitem__(self, object_type: str) -> ModelConstructor: 

22 if object_type not in self._paths: 22 ↛ 23line 22 didn't jump to line 23 because the condition on line 22 was never true

23 raise KeyError(object_type) 

24 if object_type not in self._constructors: 

25 self._constructors[object_type] = _import_constructor( 

26 self._paths[object_type] 

27 ) 

28 return self._constructors[object_type] 

29 

30 def __iter__(self) -> Iterator[str]: 

31 return iter(self._paths) 

32 

33 def __len__(self) -> int: 

34 return len(self._paths) 

35 

36 @property 

37 def _paths(self) -> dict[str, str]: 

38 if self._constructor_paths is None: 

39 self._constructor_paths = _load_constructor_paths(self.config_dir) 

40 return self._constructor_paths 

41 

42 

43def _load_constructor_paths(config_dir: Path) -> dict[str, str]: 

44 paths: dict[str, str] = {} 

45 for config_path in sorted(config_dir.glob("*.json")): 

46 data = json.loads(config_path.read_text(encoding="utf-8")) 

47 object_type = data.get("objectType") 

48 adapter = data.get("idaesAdapter") 

49 constructor = adapter.get("constructor") if isinstance(adapter, dict) else None 

50 if not isinstance(object_type, str) or not isinstance(constructor, str): 

51 continue 

52 if object_type in paths: 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true

53 raise ValueError( 

54 f"Duplicate unit-op objectType {object_type!r} in {config_dir}" 

55 ) 

56 paths[object_type] = constructor 

57 return paths 

58 

59 

60def _import_constructor(path: str) -> ModelConstructor: 

61 module_path, _, object_name = path.rpartition(".") 

62 if not module_path or not object_name: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true

63 raise ValueError(f"Invalid constructor path in unit-op JSON: {path!r}") 

64 module = importlib.import_module(module_path) 

65 constructor = getattr(module, object_name) 

66 if not callable(constructor): 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 raise TypeError(f"Unit-op constructor is not callable: {path!r}") 

68 return constructor 

69 

70 

71def _default_config_dir() -> Path: 

72 current_path = Path(__file__).resolve() 

73 relative_config_path = Path("django/flowsheetInternals/unitops/config/objects") 

74 for parent in current_path.parents: 74 ↛ 83line 74 didn't jump to line 83 because the loop on line 74 didn't complete

75 config_dir = parent / relative_config_path 

76 if config_dir.is_dir(): 

77 return config_dir 

78 

79 config_dir = parent / "backend" / relative_config_path 

80 if config_dir.is_dir(): 80 ↛ 81line 80 didn't jump to line 81 because the condition on line 80 was never true

81 return config_dir 

82 

83 return current_path.parents[5] / "backend" / relative_config_path 

84 

85 

86""" 

87A mapping of IDAES model constructors keyed by platform unit-operation object 

88type. Entries are discovered from the generated unit-operation JSON files 

89instead of being generated into this source file. 

90""" 

91AdapterLibrary: Mapping[str, ModelConstructor] = JsonBackedAdapterLibrary()