Coverage for backend/idaes_service/solver/methods/load_unit_model.py: 95%

1from typing import Any 

2 

3from pyomo.network import Port 

4from pyomo.environ import value as pyo_value, Var 

5from pyomo.core.base.constraint import ScalarConstraint 

6from pyomo.core.base.units_container import units 

7from idaes.core import UnitModelBlock 

8from idaes.core.util.tables import _get_state_from_port 

9from idaes.core.util.model_serializer import StoreSpec, from_json 

10 

11from idaes_service.solver.methods.BlockContext import BlockContext 

12from .adapter_library import UnitModelConstructor 

13from ..flowsheet_manager_type import FlowsheetManager 

14from common.models.idaes import UnitModelSchema 

15from .adapter import fix_block 

16from idaes_service.solver.methods.adapter import get_component 

17 

18def add_unit_model_to_flowsheet( 

19 unit_model_def: UnitModelSchema, 

20 adapter_constructor: UnitModelConstructor, 

21 flowsheet_manager: FlowsheetManager, 

22) -> UnitModelBlock: 

23 """ 

24 Add the unit model to the flowsheet. 

25 Args: 

26 unit_model_def: The definition of the unit model to be added to the flowsheet. 

27 adapter_schema: Methods used to create the model and parse the arguments. 

28 flowsheet_manager: Flowsheet manager to interact with the flowsheet. 

29 """ 

30 # Create the model 

31 idaes_model: UnitModelBlock =_create_model(unit_model_def, adapter_constructor, flowsheet_manager) 

32 

33 # Add the model to the flowsheet 

34 component_name = f"{unit_model_def.name}_{unit_model_def.id}" 

35 flowsheet_manager.model.fs.add_component(component_name, idaes_model) 

36 

37 # Import initial guesses 

38 initial_values = unit_model_def.initial_values 

39 

40 if initial_values: 

41 from_json(idaes_model, initial_values, wts=StoreSpec.value()) 

42 

43 # Fix properties 

44 block_context = BlockContext(flowsheet_manager.model.fs) 

45 _fix_properties(idaes_model,unit_model_def, flowsheet_manager,block_context) 

46 # Fix ports 

47 _fix_ports(idaes_model, unit_model_def, flowsheet_manager,block_context) 

48 # Apply degrees of freedom (add constraints for controlled vars) 

49 block_context.apply_elimination() 

50 return idaes_model 

51 

52def _create_model(unit_model_def : UnitModelSchema,adapter_constructor: UnitModelConstructor,flowsheet_manager : FlowsheetManager) -> UnitModelBlock: 

53 """ 

54 Create the kwargs for the model constructor from the unit model definition. 

55 """ 

56 arg_parsers = adapter_constructor.arg_parsers 

57 args = unit_model_def.args 

58 

59 kwargs : dict[str, Any] = {} 

60 for name in args: 

61 if not name in arg_parsers: 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true

62 raise ValueError( 

63 f"Argument {name} not found in model schema, available arguments are {[x for x in arg_parsers.keys()]}" 

64 ) 

65 for name in arg_parsers: 

66 register = arg_parsers[name] 

67 # try to get the argument from the schema passed in 

68 # however, in some cases the argument may not be required 

69 # i.e when specified by the model adapter itself 

70 # e.g methods.constant 

71 

72 # note that in the future we may have to support passing 

73 # optional arguments to the model constructor 

74 result = register.run(args.get(name, None),flowsheet_manager) 

75 kwargs[name] = result 

76 

77 idaes_model = adapter_constructor.model_constructor(**kwargs) 

78 return idaes_model 

79 

80def _fix_properties( unit_model: UnitModelBlock, unit_model_def: UnitModelSchema, flowsheet_manager : FlowsheetManager,block_context: BlockContext) -> None: 

81 """ 

82 Fix the properties of the unit model based on the properties in the unit model definition. 

83 """ 

84 # Loop through the properties in the unit model definition to fix the properties in the unit model 

85 properties = unit_model_def.properties 

86 fix_block( 

87 unit_model, properties, flowsheet_manager.model.fs, block_context 

88 ) 

89 

90 

91def _fix_ports(unit_model: UnitModelBlock, unit_model_def: UnitModelSchema, flowsheet_manager:FlowsheetManager,block_context: BlockContext) -> None: 

92 """ 

93 Fix the ports of the unit model based on the ports in the unit model definition. 

94 """ 

95 # Loop through the ports in the unit model definition to fix the ports in the unit model 

96 for port_name, port_schema in unit_model_def.ports.items(): 

97 # Get the port from the unit model 

98 port = get_component(unit_model, port_name) 

99 if not isinstance(port, Port): 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true

100 raise ValueError(f"Port {port_name} not found in model") 

101 # Register the port, so arcs can connect to it by id 

102 flowsheet_manager.ports.register_port(port_schema.id, port) 

103 # Set the port parameters 

104 sb = _get_state_from_port(port, 0) 

105 state_block = sb.parent_component() 

106 

107 # Prefer time-only properties_in/properties_out if available 

108 # decide by port name containing "inlet"/"outlet" and "hot"/"cold" 

109 pn = port_name.lower() 

110 if "inlet" in pn: 

111 if "hot" in pn and hasattr(unit_model, "hot_side") and hasattr(unit_model.hot_side, "properties_in"): 

112 state_block = unit_model.hot_side.properties_in 

113 elif "cold" in pn and hasattr(unit_model, "cold_side") and hasattr(unit_model.cold_side, "properties_in"): 

114 state_block = unit_model.cold_side.properties_in 

115 elif "outlet" in pn: 115 ↛ 121line 115 didn't jump to line 121 because the condition on line 115 was always true

116 if "hot" in pn and hasattr(unit_model, "hot_side") and hasattr(unit_model.hot_side, "properties_out"): 

117 state_block = unit_model.hot_side.properties_out 

118 elif "cold" in pn and hasattr(unit_model, "cold_side") and hasattr(unit_model.cold_side, "properties_out"): 

119 state_block = unit_model.cold_side.properties_out 

120 

121 if sb.config.defined_state: 

122 """ 

123 ie. Inlet state. 

124 

125 The inlet state needs a separate context, because its variables 

126 are all fixed during initialization, so applying elimination 

127 involving variables outside the inlet state would run into 

128 degrees of freedom issues. 

129 """ 

130 inlet_ctx = BlockContext(flowsheet_manager.model.fs) 

131 fix_block( 

132 state_block, 

133 port_schema.properties, 

134 flowsheet_manager.model.fs, 

135 inlet_ctx, 

136 ) 

137 inlet_ctx.apply_elimination() 

138 else: 

139 # The outlet state(s) can use the unit model context. 

140 fix_block( 

141 state_block, 

142 port_schema.properties, 

143 flowsheet_manager.model.fs, 

144 block_context, 

145 ) 

146 

147