Coverage for backend/django/core/auxiliary/viewsets/ScenarioViewSet.py: 92%

1from core.viewset import ModelViewSet 

2 

3from django.db import transaction 

4from drf_spectacular.utils import extend_schema, OpenApiParameter, OpenApiTypes 

5from rest_framework import status 

6from rest_framework.decorators import action 

7from rest_framework.response import Response 

8from ..serializers.ScenarioSerializer import ScenarioSerializer, OptimizationDegreesOfFreedomSerializer 

9from ..models.Scenario import ( 

10 Scenario, 

11 OptimizationDegreesOfFreedom, 

12 ScenarioInputModeEnum, 

13) 

14from ..services import parameter_sweep, scenario_time_series 

15from core.managers import get_flowsheet_access 

16 

17 

18class ScenarioViewSet(ModelViewSet): 

19 serializer_class = ScenarioSerializer 

20 

21 

22 def get_queryset(self): 

23 queryset = Scenario.objects.all() 

24 simulationObjectId = self.request.query_params.get("simulationObjectId") 

25 if simulationObjectId is not None: 

26 queryset = queryset.filter(simulationObject=simulationObjectId) 

27 return queryset 

28 

29 @extend_schema( 

30 parameters=[ 

31 OpenApiParameter(name="simulationObjectId", required=True, type=OpenApiTypes.INT), 

32 ] 

33 ) 

34 def list(self, request): 

35 return super().list(request) 

36 

37 def _validate_sweep_mode_data(self, instance: Scenario, validated_data: dict): 

38 """Reject MSS mode changes that general scenario validation cannot assess. 

39 

40 The serializer only validates the submitted fields. Parameter sweep mode 

41 changes also depend on existing sweep definitions and optimisation state, 

42 so this guard runs before update side effects can clear CSV or sweep rows. 

43 """ 

44 requested_mode = validated_data.get("mss_input_mode", instance.mss_input_mode) 

45 requested_optimization = validated_data.get( 

46 "enable_optimization", 

47 instance.enable_optimization, 

48 ) 

49 sweep_definition_exists = hasattr(instance, "parameterSweepDefinition") 

50 clearing_sweep_mode = ( 

51 requested_mode == ScenarioInputModeEnum.Csv 

52 and ( 

53 instance.mss_input_mode == ScenarioInputModeEnum.ParameterSweep 

54 or sweep_definition_exists 

55 ) 

56 ) 

57 

58 if ( 

59 requested_optimization is True 

60 and ( 

61 requested_mode == ScenarioInputModeEnum.ParameterSweep 

62 or (sweep_definition_exists and not clearing_sweep_mode) 

63 ) 

64 ): 

65 return Response( 

66 { 

67 "enable_optimization": ( 

68 "Optimisation cannot be enabled while parameter sweep data " 

69 "or a sweep definition is active." 

70 ) 

71 }, 

72 status=status.HTTP_400_BAD_REQUEST, 

73 ) 

74 

75 if ( 75 ↛ 79line 75 didn't jump to line 79 because the condition on line 75 was never true

76 requested_mode == ScenarioInputModeEnum.ParameterSweep 

77 and requested_optimization is True 

78 ): 

79 return Response( 

80 { 

81 "mss_input_mode": ( 

82 "Parameter sweep mode cannot be enabled while optimisation is active." 

83 ) 

84 }, 

85 status=status.HTTP_400_BAD_REQUEST, 

86 ) 

87 

88 return None 

89 

90 def _update_scenario(self, request, *, partial: bool): 

91 instance = self.get_object() 

92 serializer = self.get_serializer(instance, data=request.data, partial=partial) 

93 serializer.is_valid(raise_exception=True) 

94 requested_mode = serializer.validated_data.get("mss_input_mode") 

95 previous_mode = instance.mss_input_mode 

96 had_parameter_sweep_definition = hasattr(instance, "parameterSweepDefinition") 

97 previous_time_series_values = scenario_time_series.mss_time_series_settings_snapshot(instance) 

98 

99 validation_response = self._validate_sweep_mode_data( 

100 instance, 

101 serializer.validated_data, 

102 ) 

103 if validation_response is not None: 

104 return validation_response 

105 

106 with transaction.atomic(): 

107 self.perform_update(serializer) 

108 if requested_mode is not None: 

109 parameter_sweep.clear_mss_input_data_after_mode_switch( 

110 serializer.instance, 

111 previous_mode=previous_mode, 

112 requested_mode=requested_mode, 

113 had_parameter_sweep_definition=had_parameter_sweep_definition, 

114 ) 

115 if scenario_time_series.mss_time_series_settings_changed( 

116 serializer.instance, 

117 previous_values=previous_time_series_values, 

118 ): 

119 scenario_time_series.handle_mss_time_series_configuration_changed(serializer.instance) 

120 

121 if getattr(instance, "_prefetched_objects_cache", None): 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true

122 instance._prefetched_objects_cache = {} 

123 

124 instance.refresh_from_db() 

125 return Response(self.get_serializer(instance).data) 

126 

127 def partial_update(self, request, *args, **kwargs): 

128 return self._update_scenario(request, partial=True) 

129 

130 def update(self, request, *args, **kwargs): 

131 return self._update_scenario(request, partial=False) 

132 

133 @extend_schema( 

134 operation_id="core_scenario_eligible_parameter_sweep_targets_list", 

135 responses=parameter_sweep.EligibleParameterSweepTargetsResponse, 

136 ) 

137 @action(detail=True, methods=["get"], url_path="eligible_parameter_sweep_targets") 

138 def eligible_parameter_sweep_targets(self, request, pk=None): 

139 scenario = self.get_object() 

140 targets = parameter_sweep.eligible_parameter_sweep_targets( 

141 scenario.flowsheet_state.flowsheet_id 

142 ) 

143 return Response( 

144 parameter_sweep.EligibleParameterSweepTargetsResponse(targets).model_dump( 

145 mode="json", 

146 ) 

147 ) 

148 

149 @extend_schema( 

150 methods=["POST"], 

151 request=parameter_sweep.ParameterSweepRequest, 

152 responses=parameter_sweep.ParameterSweepPreviewResponse, 

153 ) 

154 @action(detail=True, methods=["post"], url_path="parameter_sweep_preview") 

155 def parameter_sweep_preview(self, request, pk=None): 

156 scenario = self.get_object() 

157 preview = parameter_sweep.preview_parameter_sweep(scenario, request.data) 

158 return Response(preview.model_dump(mode="json")) 

159 

160 @extend_schema( 

161 request=parameter_sweep.ParameterSweepRequest, 

162 responses=parameter_sweep.ParameterSweepGenerateResponse, 

163 ) 

164 @action(detail=True, methods=["post"], url_path="generate_parameter_sweep") 

165 def generate_parameter_sweep(self, request, pk=None): 

166 scenario = self.get_object() 

167 access_state = get_flowsheet_access( 

168 request.user, 

169 scenario.flowsheet_state.flowsheet_id, 

170 ) 

171 if access_state.has_read_access and not access_state.has_write_access: 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true

172 return Response( 

173 {"error": "This flowsheet is shared with read-only access."}, 

174 status=status.HTTP_403_FORBIDDEN, 

175 ) 

176 if scenario.enable_optimization: 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true

177 return Response( 

178 { 

179 "enable_optimization": ( 

180 "Parameter sweep generation is not available while optimisation is enabled." 

181 ) 

182 }, 

183 status=status.HTTP_400_BAD_REQUEST, 

184 ) 

185 

186 result = parameter_sweep.generate_parameter_sweep( 

187 scenario, 

188 request.data, 

189 ) 

190 return Response(result.model_dump(mode="json")) 

191 

192 

193class OptimizationDegreesOfFreedomViewSet(ModelViewSet): 

194 serializer_class = OptimizationDegreesOfFreedomSerializer 

195 

196 def get_queryset(self): 

197 return OptimizationDegreesOfFreedom.objects.all()