Coverage for backend/django/idaes_factory/adapters/ml_adapters.py: 100%

1from __future__ import annotations 

2 

3from ahuora_builder_types.unit_model_schema import ValueArgSchema 

4from core.auxiliary.enums.generalEnums import PropertyType 

5from core.auxiliary.enums.unitOpGraphics import ConType 

6 

7from .port_adapter import PortListAdapter 

8from .property_info_adapter import serialise_property_info 

9from .stream_properties import serialise_stream 

10from ..queryset_lookup import get_all_ports 

11 

12 

13class MLPropertiesAdapter: 

14 """Serialise machine-learning block properties from ML column mappings.""" 

15 

16 def serialise(self, ctx, simulationObject): 

17 from core.auxiliary.models import MLModel 

18 

19 ml_model = MLModel.objects.get(simulationObject=simulationObject) 

20 return get_ml_properties(ctx, ml_model) 

21 

22 

23def get_ml_properties(ctx, ml_model): 

24 result = {} 

25 for column_mapping in ml_model.MLColumnMappings.all(): 

26 property_info = column_mapping.propertyInfo 

27 if property_info is not None: 

28 result[column_mapping.propertyKey] = serialise_property_info( 

29 ctx, 

30 property_info, 

31 is_tear=False, 

32 is_indexed=False, 

33 ) 

34 return result 

35 

36 

37class JSONModelAdapter: 

38 """Serialise the surrogate model JSON stored for a machine-learning block.""" 

39 

40 def serialise(self, ctx, simulationObject): 

41 from core.auxiliary.models import MLModel 

42 

43 return ValueArgSchema( 

44 value=MLModel.objects.get(simulationObject=simulationObject).surrogate_model 

45 ) 

46 

47 

48class IDAdapter: 

49 """Serialise ML input/output property value ids.""" 

50 

51 def serialise(self, ctx, simulationObject): 

52 from core.auxiliary.models import MLModel 

53 

54 ml_model = MLModel.objects.get(simulationObject=simulationObject) 

55 return get_id_mappings(ml_model) 

56 

57 

58def get_id_mappings(ml_model) -> ValueArgSchema: 

59 from core.auxiliary.models.MLColumnMapping import custom_property_port_index 

60 from flowsheetInternals.unitops.models.Port import Port 

61 

62 result = { 

63 "input": [], 

64 "output": [], 

65 } 

66 for mapping in ml_model.MLColumnMappings.all(): 

67 direction = ( 

68 ConType.Inlet 

69 if mapping.propertyType == PropertyType.InletProperty 

70 else ConType.Outlet 

71 ) 

72 if mapping.portIndex == custom_property_port_index: 

73 property_info = mapping.propertyInfo 

74 else: 

75 port = Port.objects.get( 

76 unitOp=ml_model.simulationObject, 

77 index=mapping.portIndex, 

78 direction=direction, 

79 ) 

80 property_info = port.stream.properties.get_property(mapping.propertyKey) 

81 if mapping.propertyType == PropertyType.InletProperty: 

82 result["input"].append(property_info.values.first().id) 

83 else: 

84 result["output"].append(property_info.values.first().id) 

85 return ValueArgSchema(value=result) 

86 

87 

88class UnitopNamesAdapter: 

89 """Serialise custom-property column names for machine-learning blocks.""" 

90 

91 def serialise(self, ctx, simulationObject): 

92 from core.auxiliary.models import MLModel 

93 

94 ml_model = MLModel.objects.get(simulationObject=simulationObject) 

95 return get_unitop_names(ml_model) 

96 

97 

98def get_unitop_names(ml_model) -> ValueArgSchema: 

99 from core.auxiliary.models.MLColumnMapping import custom_property_port_index 

100 

101 result = [] 

102 for item in ml_model.MLColumnMappings.filter(portIndex=custom_property_port_index): 

103 result.append(item.column) 

104 return ValueArgSchema(value=result) 

105 

106 

107class MLPortListAdapter(PortListAdapter): 

108 """Serialise variable machine-learning inlet and outlet port groups.""" 

109 

110 def __init__(self): 

111 pass 

112 

113 def serialise(self, ctx, unit_model) -> dict: 

114 inlets = get_all_ports(unit_model, "inlet") 

115 outlets = get_all_ports(unit_model, "outlet") 

116 

117 port_list = {} 

118 for index, inlet in enumerate(inlets, start=1): 

119 port_list[f"inlet_{index}"] = { 

120 "id": inlet.pk, 

121 "properties": serialise_stream(ctx, inlet.stream, is_inlet=True), 

122 } 

123 

124 for index, outlet in enumerate(outlets, start=1): 

125 port_list[f"outlet_{index}"] = { 

126 "id": outlet.pk, 

127 "properties": serialise_stream(ctx, outlet.stream, is_inlet=False), 

128 } 

129 

130 return port_list