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

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 drf_spectacular.types import OpenApiTypes 

15from drf_spectacular.utils import OpenApiParameter, extend_schema 

16from rest_framework import status 

17from rest_framework.decorators import api_view 

18from rest_framework.response import Response 

19 

20from core.validation import api_view_validate 

21from diagnostics.rules.engine import evaluate_property_rules 

22from diagnostics.serializers import ( 

23 FlowsheetRuleFindingsSerializer, 

24 PropertyRuleFindingsSerializer, 

25) 

26from core.auxiliary.models.PropertyInfo import PropertyInfo 

27from core.auxiliary.models.PropertyValue import PropertyValue 

28from flowsheetInternals.unitops.models.SimulationObject import SimulationObject 

29 

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

31 """ 

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

33 

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

35 all properties during rule evaluation. 

36 """ 

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

38 values_qs = PropertyValue.objects.select_related( 

39 "controlManipulated", 

40 "controlSetPoint", 

41 ) 

42 

43 return qs.prefetch_related( 

44 Prefetch( 

45 "properties__ContainedProperties", 

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

47 Prefetch("values", queryset=values_qs) 

48 ), 

49 ) 

50 ).get() 

51 

52 

53def _get_simulation_objects_for_rules(flowsheet_id: int): 

54 """ 

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

56 """ 

57 values_qs = PropertyValue.objects.select_related( 

58 "controlManipulated", 

59 "controlSetPoint", 

60 ) 

61 return ( 

62 SimulationObject.objects.filter(flowsheet_id=flowsheet_id) 

63 .select_related("properties") 

64 .prefetch_related( 

65 Prefetch( 

66 "properties__ContainedProperties", 

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

68 Prefetch("values", queryset=values_qs) 

69 ), 

70 ) 

71 ) 

72 ) 

73 

74 

75@extend_schema( 

76 operation_id="diagnostics_evaluate_simulation_object_property_rules", 

77 parameters=[ 

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

79 OpenApiParameter( 

80 name="flowsheet", 

81 type=OpenApiTypes.INT, 

82 location=OpenApiParameter.QUERY, 

83 required=True, 

84 ), 

85 ], 

86 responses={200: PropertyRuleFindingsSerializer}, 

87) 

88@api_view_validate 

89@api_view(["GET"]) 

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

91 """ 

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

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

94 """ 

95 try: 

96 obj = _get_simulation_object_for_rules(simulation_object_id) 

97 except SimulationObject.DoesNotExist: 

98 return Response(status=status.HTTP_404_NOT_FOUND) 

99 

100 findings = evaluate_property_rules(obj) 

101 response_serializer = PropertyRuleFindingsSerializer( 

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

103 ) 

104 return Response(response_serializer.data) 

105 

106 

107@extend_schema( 

108 operation_id="diagnostics_evaluate_flowsheet_property_rules", 

109 parameters=[ 

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

111 OpenApiParameter( 

112 name="flowsheet", 

113 type=OpenApiTypes.INT, 

114 location=OpenApiParameter.QUERY, 

115 required=True, 

116 ), 

117 ], 

118 responses={200: FlowsheetRuleFindingsSerializer}, 

119) 

120@api_view_validate 

121@api_view(["GET"]) 

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

123 """ 

124 Evaluate deterministic diagnostics rules across all flowsheet objects. 

125 """ 

126 try: 

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

128 except (TypeError, ValueError): 

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

130 

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

132 return Response( 

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

134 status=status.HTTP_400_BAD_REQUEST, 

135 ) 

136 

137 objects = _get_simulation_objects_for_rules(flowsheet_id) 

138 findings = [] 

139 for obj in objects: 

140 try: 

141 findings.extend(evaluate_property_rules(obj)) 

142 except Exception: 

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

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

145 continue 

146 

147 response_serializer = FlowsheetRuleFindingsSerializer( 

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

149 ) 

150 return Response(response_serializer.data)