Coverage for backend/django/core/auxiliary/unit_op_object_types.py: 90%

1from __future__ import annotations 

2 

3import json 

4from dataclasses import dataclass 

5from pathlib import Path 

6 

7 

8@dataclass(frozen=True, slots=True) 

9class UnitOpObjectType: 

10 """Object-type metadata loaded from one unit-operation JSON config.""" 

11 

12 key: str 

13 object_type: str 

14 enum_member: str 

15 display_type: str 

16 is_stream: bool 

17 

18 

19class UnitOpObjectTypes: 

20 """Attribute-style access to JSON-backed unit-operation object types.""" 

21 

22 def __init__(self, entries: list[UnitOpObjectType]) -> None: 

23 self._entries = tuple(entries) 

24 self._by_member = {entry.enum_member: entry for entry in entries} 

25 self._by_object_type = {entry.object_type: entry for entry in entries} 

26 

27 duplicate_members = _duplicates(entry.enum_member for entry in entries) 

28 if duplicate_members: 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true

29 raise ValueError( 

30 "Duplicate unit-op enumMember values in JSON config: " 

31 f"{sorted(duplicate_members)}" 

32 ) 

33 duplicate_object_types = _duplicates(entry.object_type for entry in entries) 

34 if duplicate_object_types: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true

35 raise ValueError( 

36 "Duplicate unit-op objectType values in JSON config: " 

37 f"{sorted(duplicate_object_types)}" 

38 ) 

39 

40 def __getattr__(self, name: str) -> str: 

41 try: 

42 return self._by_member[name].object_type 

43 except KeyError as exc: 

44 raise AttributeError(name) from exc 

45 

46 @property 

47 def choices(self) -> tuple[tuple[str, str], ...]: 

48 return tuple( 

49 (entry.object_type, entry.display_type) 

50 for entry in self._entries 

51 ) 

52 

53 @property 

54 def values(self) -> tuple[str, ...]: 

55 return tuple(entry.object_type for entry in self._entries) 

56 

57 def is_known(self, object_type: str) -> bool: 

58 return object_type in self._by_object_type 

59 

60 

61def load_unit_op_object_type_entries_from_json_dir( 

62 path: Path, 

63) -> list[UnitOpObjectType]: 

64 """Load object type metadata from per-unit JSON config files.""" 

65 

66 entries: list[UnitOpObjectType] = [] 

67 for config_path in sorted(path.glob("*.json")): 

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

69 entries.append( 

70 UnitOpObjectType( 

71 key=data["key"], 

72 object_type=data["objectType"], 

73 enum_member=data["enumMember"], 

74 display_type=data["displayType"], 

75 is_stream=bool(data.get("isStream", False)), 

76 ) 

77 ) 

78 

79 order = {object_type: index for index, object_type in enumerate(_LEGACY_ORDER)} 

80 return sorted( 

81 entries, 

82 key=lambda entry: (order.get(entry.object_type, len(order)), entry.object_type), 

83 ) 

84 

85 

86def load_unit_op_object_types_from_json_dir(path: Path) -> UnitOpObjectTypes: 

87 """Load attribute-style object type constants from JSON config files.""" 

88 

89 return UnitOpObjectTypes(load_unit_op_object_type_entries_from_json_dir(path)) 

90 

91 

92def load_unit_op_object_types() -> UnitOpObjectTypes: 

93 """Load object type constants from the conventional JSON config directory.""" 

94 

95 return load_unit_op_object_types_from_json_dir(_unit_op_config_dir()) 

96 

97 

98def load_heat_exchange_object_types() -> list[str]: 

99 """Return object types used by pinch heat-exchange grouping logic.""" 

100 

101 object_types = load_unit_op_object_types() 

102 return [ 

103 object_type 

104 for object_type in ( 

105 getattr(object_types, "Heater", None), 

106 getattr(object_types, "Cooler", None), 

107 getattr(object_types, "HeatExchanger", None), 

108 ) 

109 if object_type is not None 

110 ] 

111 

112 

113def _duplicates(values) -> set[str]: 

114 seen: set[str] = set() 

115 duplicates: set[str] = set() 

116 for value in values: 

117 if value in seen: 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true

118 duplicates.add(value) 

119 seen.add(value) 

120 return duplicates 

121 

122 

123def _unit_op_config_dir() -> Path: 

124 return ( 

125 Path(__file__).resolve().parents[2] 

126 / "flowsheetInternals/unitops/config/objects" 

127 ) 

128 

129 

130_LEGACY_ORDER = ( 

131 "decisionNode", 

132 "stream", 

133 "humid_air_stream", 

134 "pump", 

135 "compressor", 

136 "cooler", 

137 "valve", 

138 "turbine", 

139 "willans_turbine", 

140 "pl_turbine", 

141 "dts_turbine", 

142 "bs_turbine", 

143 "cs_turbine", 

144 "heater", 

145 "Tank", 

146 "recycle", 

147 "mixer", 

148 "splitter", 

149 "header", 

150 "phaseSeparator", 

151 "energy_stream", 

152 "ac_stream", 

153 "transformer_stream", 

154 "compoundSeparator", 

155 "heatExchanger", 

156 "heatPump", 

157 "group", 

158 "machineLearningBlock", 

159 "link", 

160 "grid", 

161 "boiler", 

162 "solar", 

163 "hydro", 

164 "wind", 

165 "energy_mixer", 

166 "bus", 

167 "storage", 

168 "acBus", 

169 "transmissionLine", 

170 "load", 

171 "transformer", 

172 "convertor", 

173 "direct_steam_injection", 

174 "translator", 

175 "RCT_CSTR", 

176 "stoich_hda", 

177 "pid_controller", 

178 "custom_variable", 

179 "integration", 

180 "energy_splitter", 

181 "mdb", 

182 "simple_header", 

183 "heat_user", 

184 "steam_user", 

185 "desuperheater", 

186 "heat_exchanger_ntu", 

187 "duty_heat_exchanger", 

188 "waterpipe", 

189 "heat_exchanger_lc", 

190 "heat_exchanger_1d", 

191 "plate_heat_exchanger", 

192 "reverse_osmosis_0d", 

193 "pressure_exchanger", 

194 "crystallizer", 

195)