Coverage for backend/django/diagnostics/views.py: 86%

1""" 

2Diagnostics API views. 

3 

4These endpoints exist to support the diagnostics UI: 

5- Run light-weight "what if?" validations used by the properties panel 

6 

7The goal is to keep views fairly thin: parse/validate the request, fetch the 

8minimal DB state needed, then delegate to the diagnostics rules engine. 

9""" 

10 

11from __future__ import annotations 

12 

13from django.db.models import Prefetch 

14from diagnostics.methods.get_solver_rule_findings import get_flowsheet_rule_findings, get_object_rule_findings 

15from drf_spectacular.types import OpenApiTypes 

16from drf_spectacular.utils import OpenApiParameter, extend_schema 

17from rest_framework import status 

18from rest_framework.decorators import api_view 

19from rest_framework.response import Response 

20 

21from core.validation import api_view_validate 

22from diagnostics.rules.engine import RuleFinding, evaluate_property_rules 

23from diagnostics.serializers import ( 

24 FlowsheetRuleFindingsSerializer, 

25 PropertyRuleFindingsSerializer, 

26) 

27from core.auxiliary.models.PropertyInfo import PropertyInfo 

28from core.auxiliary.models.PropertyValue import PropertyValue 

29from flowsheetInternals.unitops.models.SimulationObject import SimulationObject 

30 

31def _get_simulation_object_for_rules(object_id: int) -> SimulationObject: 

32 """ 

33 Fetch a SimulationObject with properties eagerly loaded for rule evaluation. 

34 

35 I prefetch properties/values here to avoid N+1 queries when we walk through 

36 all properties during rule evaluation. 

37 """ 

38 qs = SimulationObject.objects.filter(id=object_id).select_related("properties") 

39 values_qs = PropertyValue.objects.select_related( 

40 "controlManipulated", 

41 "controlSetPoint", 

42 ) 

43 

44 return qs.prefetch_related( 

45 Prefetch( 

46 "properties__ContainedProperties", 

47 queryset=PropertyInfo.objects.select_related("recycleConnection").prefetch_related( 

48 Prefetch("values", queryset=values_qs) 

49 ), 

50 ) 

51 ).get() 

52 

53 

54def _get_simulation_objects_for_rules(flowsheet_id: int): 

55 """ 

56 Fetch all SimulationObjects in a flowsheet with properties eagerly loaded. 

57 """ 

58 values_qs = PropertyValue.objects.select_related( 

59 "controlManipulated", 

60 "controlSetPoint", 

61 ) 

62 return ( 

63 SimulationObject.objects.filter(flowsheet_id=flowsheet_id) 

64 .select_related("properties") 

65 .prefetch_related( 

66 Prefetch( 

67 "properties__ContainedProperties", 

68 queryset=PropertyInfo.objects.select_related("recycleConnection").prefetch_related( 

69 Prefetch("values", queryset=values_qs) 

70 ), 

71 ) 

72 ) 

73 ) 

74 

75 

76@extend_schema( 

77 operation_id="diagnostics_evaluate_simulation_object_property_rules", 

78 parameters=[ 

79 OpenApiParameter(name="simulation_object_id", type=OpenApiTypes.INT, location=OpenApiParameter.PATH), 

80 OpenApiParameter( 

81 name="flowsheet", 

82 type=OpenApiTypes.INT, 

83 location=OpenApiParameter.QUERY, 

84 required=True, 

85 ), 

86 ], 

87 responses={200: PropertyRuleFindingsSerializer}, 

88) 

89@api_view_validate 

90@api_view(["GET"]) 

91def evaluate_object_property_rules(request, simulation_object_id: int) -> Response: 

92 """ 

93 Evaluate deterministic diagnostics rules against a unit op's current properties. 

94 Intended for inline properties-panel warnings/suggestions/errors. 

95 """ 

96 try: 

97 obj = _get_simulation_object_for_rules(simulation_object_id) 

98 except SimulationObject.DoesNotExist: 

99 return Response(status=status.HTTP_404_NOT_FOUND) 

100 

101 findings = evaluate_property_rules(obj) 

102 findings.extend(get_object_rule_findings(simulation_object_id)) 

103 response_serializer = PropertyRuleFindingsSerializer( 

104 {"object_id": obj.id, "findings": [f.to_dict() for f in findings]} 

105 ) 

106 return Response(response_serializer.data) 

107 

108 

109@extend_schema( 

110 operation_id="diagnostics_evaluate_flowsheet_property_rules", 

111 parameters=[ 

112 OpenApiParameter(name="flowsheet_id", type=OpenApiTypes.INT, location=OpenApiParameter.PATH), 

113 OpenApiParameter( 

114 name="flowsheet", 

115 type=OpenApiTypes.INT, 

116 location=OpenApiParameter.QUERY, 

117 required=True, 

118 ), 

119 ], 

120 responses={200: FlowsheetRuleFindingsSerializer}, 

121) 

122@api_view_validate 

123@api_view(["GET"]) 

124def evaluate_flowsheet_property_rules(request, flowsheet_id: int) -> Response: 

125 """ 

126 Evaluate deterministic diagnostics rules across all flowsheet objects. 

127 """ 

128 try: 

129 query_flowsheet_id = int(request.GET.get("flowsheet")) 

130 except (TypeError, ValueError): 

131 return Response({"detail": "Invalid flowsheet id"}, status=status.HTTP_400_BAD_REQUEST) 

132 

133 if query_flowsheet_id != flowsheet_id: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true

134 return Response( 

135 {"detail": "Flowsheet ID mismatch between path and query parameter"}, 

136 status=status.HTTP_400_BAD_REQUEST, 

137 ) 

138 

139 objects = _get_simulation_objects_for_rules(flowsheet_id) 

140 findings: list[RuleFinding] = [] 

141 for obj in objects: 

142 try: 

143 findings.extend(evaluate_property_rules(obj)) 

144 except Exception: 

145 # Keep the endpoint resilient: one malformed object path should not 

146 # prevent diagnostics from being returned for the rest of the flowsheet. 

147 continue 

148 

149 findings.extend(get_flowsheet_rule_findings(flowsheet_id)) 

150 

151 response_serializer = FlowsheetRuleFindingsSerializer( 

152 {"flowsheet_id": flowsheet_id, "findings": [f.to_dict() for f in findings]} 

153 ) 

154 return Response(response_serializer.data)