Coverage for backend/ahuora-builder/src/ahuora_builder/methods/load_unit_model.py: 97%

1from typing import Any 

2from pyomo.network import Port 

3from pyomo.environ import value as pyo_value, Var 

4from pyomo.core.base.constraint import ScalarConstraint 

5from pyomo.core.base.units_container import units 

6from idaes.core import UnitModelBlock 

7from idaes.core.util.tables import _get_state_from_port 

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

9from ahuora_builder.methods.BlockContext import BlockContext 

10from .adapter_library import ModelConstructor 

11from .adapter_methods import parse_args_dictionary 

12from ..flowsheet_manager_type import FlowsheetManager 

13from ahuora_builder_types import UnitModelSchema 

14from .adapter import fix_block 

15from ahuora_builder.methods.adapter import get_component 

16 

17def add_unit_model_to_flowsheet( 

18 unit_model_def: UnitModelSchema, 

19 model_constructor: ModelConstructor, 

20 flowsheet_manager: FlowsheetManager, 

21) -> UnitModelBlock: 

22 """ 

23 Add the unit model to the flowsheet. 

24 Args: 

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

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

27 flowsheet_manager: Flowsheet manager to interact with the flowsheet. 

28 """ 

29 # Create the model 

30 kwargs : dict[str, Any] = parse_args_dictionary(unit_model_def.args,flowsheet_manager) 

31 idaes_model : UnitModelBlock = model_constructor(**kwargs) 

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 

51 return idaes_model 

52 

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

54 """ 

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

56 """ 

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

58 properties = unit_model_def.properties 

59 fix_block( 

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

61 ) 

62 

63 

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

65 """ 

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

67 """ 

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

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

70 # Get the port from the unit model 

71 port = get_component(unit_model, port_name) 

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

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

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

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

76 # Set the port parameters 

77 sb = _get_state_from_port(port, 0) 

78 state_block = sb.parent_component() 

79 

80 # Prefer time-only properties_in/properties_out if available 

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

82 pn = port_name.lower() 

83 if "inlet" in pn: 

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

85 state_block = unit_model.hot_side.properties_in 

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

87 state_block = unit_model.cold_side.properties_in 

88 elif "outlet" in pn: 

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

90 state_block = unit_model.hot_side.properties_out 

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

92 state_block = unit_model.cold_side.properties_out 

93 

94 if sb.config.defined_state: 

95 """ 

96 ie. Inlet state. 

97 

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

99 are all fixed during initialization, so applying elimination 

100 involving variables outside the inlet state would run into 

101 degrees of freedom issues. 

102 """ 

103 inlet_ctx = BlockContext(flowsheet_manager.model.fs) 

104 fix_block( 

105 state_block, 

106 port_schema.properties, 

107 flowsheet_manager.model.fs, 

108 inlet_ctx, 

109 ) 

110 inlet_ctx.apply_elimination() 

111 else: 

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

113 fix_block( 

114 state_block, 

115 port_schema.properties, 

116 flowsheet_manager.model.fs, 

117 block_context, 

118 ) 

119 

120