Coverage for backend/django/Economics/formulas/native_properties/materialize.py: 87%

1from __future__ import annotations 

2 

3from django.core.exceptions import ObjectDoesNotExist 

4from django.db import transaction 

5 

6from core.auxiliary.models.PropertyInfo import PropertyInfo 

7from core.auxiliary.models.PropertySet import PropertySet 

8from core.auxiliary.models.PropertyValue import PropertyValue 

9from Economics.formulas.models import EconomicsMetricFormula 

10from Economics.formulas.property_state import apply_economics_property_state 

11from Economics.studies.models import EconomicsStudy 

12from Economics.formulas.native_properties.specs import EconomicsNativePropertySpec, native_property_specs 

13 

14 

15def materialize_economics_native_properties(study: EconomicsStudy) -> dict[str, PropertyValue]: 

16 """Create or update managed native-property rows for a flowsheet study.""" 

17 

18 with transaction.atomic(): 

19 property_set = _native_property_set(study) 

20 values: dict[str, PropertyValue] = {} 

21 for spec in native_property_specs(study): 

22 values[spec.field_key] = _materialize_spec(study=study, property_set=property_set, spec=spec) 

23 return values 

24 

25 

26def _materialize_spec( 

27 *, 

28 study: EconomicsStudy, 

29 property_set: PropertySet, 

30 spec: EconomicsNativePropertySpec, 

31) -> PropertyValue: 

32 property_info = _native_property_info(study=study, property_set=property_set, spec=spec) 

33 value = _single_scalar_value(property_info) 

34 _link_native_formula_owner(study=study, spec=spec, property_value=value) 

35 apply_economics_property_state( 

36 property_info, 

37 editable=spec.editable, 

38 formula_incomplete=None, 

39 ) 

40 return value 

41 

42 

43def _link_native_formula_owner( 

44 *, 

45 study: EconomicsStudy, 

46 spec: EconomicsNativePropertySpec, 

47 property_value: PropertyValue, 

48) -> None: 

49 EconomicsMetricFormula.objects.update_or_create( 

50 flowsheet=study.flowsheet, 

51 study=study, 

52 metric_key=_native_metric_key(spec), 

53 defaults={ 

54 "property_value": property_value, 

55 "formula_key": spec.field_key, 

56 "unit": property_value.property.unit, 

57 }, 

58 ) 

59 

60 

61def _native_property_info( 

62 *, 

63 study: EconomicsStudy, 

64 property_set: PropertySet, 

65 spec: EconomicsNativePropertySpec, 

66) -> PropertyInfo: 

67 property_info = _source_owned_native_property_info(study=study, spec=spec) 

68 defaults = { 

69 "set": property_set, 

70 "type": "numeric", 

71 "unitType": spec.unit_type, 

72 "unit": spec.unit, 

73 "displayName": spec.display_name, 

74 "index": 0, 

75 } 

76 if property_info is None: 

77 property_info = PropertyInfo.objects.create( 

78 flowsheet=study.flowsheet, 

79 key=spec.property_key, 

80 **defaults, 

81 ) 

82 return property_info 

83 changed_fields = [] 

84 for field_name, value in defaults.items(): 

85 if getattr(property_info, field_name) != value: 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true

86 setattr(property_info, field_name, value) 

87 changed_fields.append(field_name) 

88 if property_info.key != spec.property_key: 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true

89 property_info.key = spec.property_key 

90 changed_fields.append("key") 

91 if changed_fields: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true

92 property_info.save(update_fields=changed_fields) 

93 return property_info 

94 

95 

96def _source_owned_native_property_info( 

97 *, 

98 study: EconomicsStudy, 

99 spec: EconomicsNativePropertySpec, 

100) -> PropertyInfo | None: 

101 property_value = ( 

102 EconomicsMetricFormula.objects.filter( 

103 flowsheet=study.flowsheet, 

104 study=study, 

105 metric_key=_native_metric_key(spec), 

106 property_value__isnull=False, 

107 ) 

108 .select_related("property_value__property") 

109 .order_by("pk") 

110 .first() 

111 ) 

112 if property_value is None or property_value.property_value is None: 

113 return None 

114 return property_value.property_value.property 

115 

116 

117def _single_scalar_value(property_info: PropertyInfo) -> PropertyValue: 

118 values = list(property_info.values.order_by("pk")) 

119 value = values[0] if values else None 

120 for duplicate in values[1:]: 120 ↛ 121line 120 didn't jump to line 121 because the loop on line 120 never started

121 duplicate.delete() 

122 if value is None: 

123 value = PropertyValue.objects.create( 

124 flowsheet=property_info.flowsheet, 

125 property=property_info, 

126 value=None, 

127 displayValue=None, 

128 enabled=True, 

129 ) 

130 return value 

131 

132 

133def _native_property_set(study: EconomicsStudy) -> PropertySet: 

134 root_grouping = getattr(study.flowsheet, "rootGrouping", None) 

135 try: 

136 root_object = getattr(root_grouping, "simulationObject", None) 

137 except ObjectDoesNotExist: 

138 root_object = None 

139 if root_object is not None: 

140 property_set, _ = PropertySet.objects.get_or_create( 

141 flowsheet=study.flowsheet, 

142 simulationObject=root_object, 

143 ) 

144 return property_set 

145 property_set = PropertySet.objects.filter(flowsheet=study.flowsheet, simulationObject__isnull=True).order_by("pk").first() 

146 if property_set is not None: 

147 return property_set 

148 return PropertySet.objects.create(flowsheet=study.flowsheet, simulationObject=None) 

149 

150 

151def _native_metric_key(spec: EconomicsNativePropertySpec) -> str: 

152 return spec.result_metric_key or spec.field_key