Coverage for backend/django/core/auxiliary/views/UploadMSSData.py: 88%

1from rest_framework.response import Response 

2from django.db import transaction 

3from drf_spectacular.utils import extend_schema 

4from core.auxiliary.models.DataColumn import DataColumn 

5from core.auxiliary.models.Scenario import Scenario, ScenarioInputModeEnum 

6from core.auxiliary.services.parameter_sweep import clear_mss_input_data 

7from core.auxiliary.services.mss_input_data import ( 

8 MssInputCell, 

9 MssInputRow, 

10 MssInputSource, 

11 replace_mss_input_rows, 

12) 

13from rest_framework.decorators import api_view 

14from rest_framework import serializers, status 

15from core.validation import api_view_validate 

16from core.managers import get_flowsheet_access 

17 

18class UploadDataSerializer(serializers.Serializer): 

19 # THe data format is e.g: 

20 # { 

21 # "data": { 

22 # "heater_enthalpy": [1, 2, 3, 4, 5], 

23 # "heater_temperature": [1, 2, 3, 4, 5] 

24 # } 

25 # "flowsheet": 1 

26 # } 

27 flowsheet = serializers.IntegerField() 

28 scenario=serializers.IntegerField() 

29 data = serializers.DictField( # column name 

30 child=serializers.ListField( # List of values 

31 child=serializers.FloatField() # Value 

32 ) 

33 ) 

34 

35 

36@api_view_validate 

37@extend_schema(request=UploadDataSerializer, responses=None) 

38@api_view(['POST']) 

39def upload_data(request) -> Response: 

40 flowsheet_id = request.GET.get("flowsheet") 

41 access_state = get_flowsheet_access(request.user, flowsheet_id) 

42 if access_state.has_read_access and not access_state.has_write_access: 

43 return Response( 

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

45 status=status.HTTP_403_FORBIDDEN, 

46 ) 

47 

48 try: 

49 serializer = UploadDataSerializer(data=request.data) 

50 serializer.is_valid(raise_exception=True) 

51 validated_data = serializer.validated_data 

52 data = validated_data.get('data') 

53 flowsheet_id = validated_data.get('flowsheet') 

54 scenario_id = validated_data.get('scenario') 

55 except Exception as e: 

56 return Response(status=400, data=f"Invalid csv data: {e}") 

57 

58 if not data: 58 ↛ 59line 58 didn't jump to line 59 because the condition on line 58 was never true

59 return Response(status=400, data="CSV data must include at least one column.") 

60 

61 num_rows = len(list(next(iter(data.values())))) 

62 if any(len(values) != num_rows for values in data.values()): 

63 return Response(status=400, data="CSV columns must all have the same number of rows.") 

64 

65 try: 

66 scenario = Scenario.objects.get( 

67 id=scenario_id, 

68 flowsheet_state__flowsheet_id=flowsheet_id, 

69 ) 

70 except Scenario.DoesNotExist: 

71 return Response(status=404, data="Scenario not found.") 

72 

73 with transaction.atomic(): 

74 replacing_parameter_sweep = ( 

75 scenario.mss_input_mode == ScenarioInputModeEnum.ParameterSweep 

76 or hasattr(scenario, "parameterSweepDefinition") 

77 ) 

78 if replacing_parameter_sweep: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true

79 clear_mss_input_data(scenario) 

80 

81 scenario.mss_input_mode = ScenarioInputModeEnum.Csv 

82 scenario.save(update_fields=["mss_input_mode"]) 

83 

84 # Step 1: Create any missing data column under the given optimization 

85 data_columns = [] 

86 for key in data: 

87 data_columns.append( 

88 DataColumn( 

89 name=key, 

90 scenario_id=scenario_id, 

91 flowsheet_state=scenario.flowsheet_state, 

92 ) 

93 ) 

94 

95 DataColumn.objects.bulk_create(data_columns, ignore_conflicts=True) 

96 

97 # Step 2: Get updated data columns 

98 data_columns = DataColumn.objects.filter(scenario_id=scenario_id) 

99 column_map = {column.name: column for column in data_columns} 

100 

101 # Step 3: Replace row/cell values through the MSS input mutation boundary 

102 replace_mss_input_rows( 

103 scenario=scenario, 

104 rows=[ 

105 MssInputRow( 

106 index=i, 

107 cells=[ 

108 MssInputCell( 

109 data_column_id=column_map[column_name].id, 

110 value=values[i], 

111 ) 

112 for column_name, values in data.items() 

113 ], 

114 ) 

115 for i in range(num_rows) 

116 ], 

117 source=MssInputSource.legacy_upload, 

118 ) 

119 

120 return Response(status=200, data="Data uploaded successfully")