Coverage for backend/django/PinchAnalysis/views/StreamDataEntryViewSet.py: 84%

1from core.serializer_base import StateOwnedModelSerializer 

2from requests import Response 

3from PinchAnalysis.models.InputModels import StreamDataEntry 

4# from rest_framework import status 

5# from rest_framework.response import Response 

6from PinchAnalysis.models.next_stream_data_entry import next_sdes_for 

7from core.viewset import ModelViewSet 

8from rest_framework import serializers 

9from drf_spectacular.utils import extend_schema, OpenApiParameter, OpenApiTypes 

10# from rest_framework.decorators import action 

11 

12 

13class StreamDataEntrySerializer(StateOwnedModelSerializer): 

14 inlet_outlet_streams = serializers.SerializerMethodField() 

15 unitop_type = serializers.SerializerMethodField() 

16 temperatures = serializers.SerializerMethodField() 

17 pressures = serializers.SerializerMethodField() 

18 enthalpies = serializers.SerializerMethodField() 

19 heat_flow = serializers.SerializerMethodField() 

20 nextStream = serializers.SerializerMethodField(read_only=True) 

21 

22 class Meta: 

23 model = StreamDataEntry 

24 fields = [ 

25 'id', 

26 'flowsheet', 

27 'custom', 

28 'group', 

29 'streamDataProject', 

30 'inlet_outlet_streams', 

31 'unitop', 

32 'unitop_type', 

33 'property_package_mapping', 

34 'nextStream', 

35 't_h_data', 

36 'states', 

37 'created_at', 

38 'temperatures', 

39 'pressures', 

40 'enthalpies', 

41 'heat_flow', 

42 'expanded', 

43 ] 

44 

45 def get_nextStream(self, obj): 

46 nxt = next_sdes_for(obj) 

47 return [s.id for s in nxt] 

48 

49 def get_inlet_outlet_streams(self, obj): 

50 try: 

51 print(obj.inlet_outlet_stream) 

52 return {"inlet": obj.inlet_outlet_stream[0].id, "outlet": obj.inlet_outlet_stream[1].id} 

53 except Exception: 

54 return {"inlet": None, "outlet": None} 

55 

56 def get_temperatures(self, obj): 

57 try: 

58 inlet, outlet = obj.temperatures 

59 return {"inlet": inlet.get_value(), "outlet": outlet.get_value()} 

60 except Exception: 

61 return {"inlet": None, "outlet": None} 

62 

63 def get_pressures(self, obj): 

64 try: 

65 inlet, outlet = obj.pressures 

66 return {"inlet": inlet.get_value(), "outlet": outlet.get_value()} 

67 except Exception: 

68 return {"inlet": None, "outlet": None} 

69 

70 def get_enthalpies(self, obj): 

71 try: 

72 inlet, outlet = obj.enthalpies 

73 return {"inlet": inlet.get_value(), "outlet": outlet.get_value()} 

74 except Exception: 

75 return {"inlet": None, "outlet": None} 

76 

77 def get_heat_flow(self, obj): 

78 try: 

79 return obj.heat_flow.get_value() 

80 except Exception: 

81 return None 

82 

83 def get_unitop_type(self, obj): 

84 if obj.unitop: 84 ↛ 86line 84 didn't jump to line 86 because the condition on line 84 was always true

85 return getattr(obj.unitop.schema, "displayType", None) 

86 return None 

87 

88 

89class StreamDataEntryViewSet(ModelViewSet): 

90 serializer_class = StreamDataEntrySerializer 

91 

92 def get_queryset(self): 

93 queryset = StreamDataEntry.objects.all() 

94 stream_data_project = self.request.query_params.get("streamDataProject") 

95 if stream_data_project: 

96 queryset = queryset.filter(streamDataProject=stream_data_project) 

97 return queryset 

98 

99 def get_serializer_context(self): 

100 ctx = super().get_serializer_context() 

101 ctx["all_sdes"] = list( 

102 self.get_queryset() 

103 .select_related("unitop") 

104 .prefetch_related("unitop__ports__stream") 

105 ) 

106 return ctx 

107 

108 @extend_schema( 

109 parameters=[ 

110 OpenApiParameter(name="streamDataProject", required=True, type=OpenApiTypes.INT), 

111 ], 

112 responses=StreamDataEntrySerializer(many=True) 

113 ) 

114 def list(self, request): 

115 return super().list(request)