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

1from requests import Response 

2from PinchAnalysis.models.InputModels import StreamDataEntry 

3# from rest_framework import status 

4# from rest_framework.response import Response 

5from PinchAnalysis.models.next_stream_data_entry import next_sdes_for 

6from core.viewset import ModelViewSet 

7from rest_framework import serializers 

8from drf_spectacular.utils import extend_schema, OpenApiParameter, OpenApiTypes 

9# from rest_framework.decorators import action 

10 

11 

12class StreamDataEntrySerializer(serializers.ModelSerializer): 

13 inlet_outlet_streams = serializers.SerializerMethodField() 

14 unitop_type = serializers.SerializerMethodField() 

15 temperatures = serializers.SerializerMethodField() 

16 pressures = serializers.SerializerMethodField() 

17 enthalpies = serializers.SerializerMethodField() 

18 heat_flow = serializers.SerializerMethodField() 

19 nextStream = serializers.SerializerMethodField(read_only=True) 

20 

21 class Meta: 

22 model = StreamDataEntry 

23 fields = [ 

24 'id', 

25 'flowsheet', 

26 'custom', 

27 'group', 

28 'streamDataProject', 

29 'inlet_outlet_streams', 

30 'unitop', 

31 'unitop_type', 

32 'property_package_mapping', 

33 'nextStream', 

34 't_h_data', 

35 'states', 

36 'created_at', 

37 'temperatures', 

38 'pressures', 

39 'enthalpies', 

40 'heat_flow', 

41 'expanded', 

42 ] 

43 

44 def get_nextStream(self, obj): 

45 nxt = next_sdes_for(obj) 

46 return [s.id for s in nxt] 

47 

48 def get_inlet_outlet_streams(self, obj): 

49 try: 

50 print(obj.inlet_outlet_stream) 

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

52 except Exception: 

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

54 

55 def get_temperatures(self, obj): 

56 try: 

57 inlet, outlet = obj.temperatures 

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

59 except Exception: 

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

61 

62 def get_pressures(self, obj): 

63 try: 

64 inlet, outlet = obj.pressures 

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

66 except Exception: 

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

68 

69 def get_enthalpies(self, obj): 

70 try: 

71 inlet, outlet = obj.enthalpies 

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

73 except Exception: 

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

75 

76 def get_heat_flow(self, obj): 

77 try: 

78 return obj.heat_flow.get_value() 

79 except Exception: 

80 return None 

81 

82 def get_unitop_type(self, obj): 

83 if obj.unitop: 

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

85 return None 

86 

87 

88class StreamDataEntryViewSet(ModelViewSet): 

89 serializer_class = StreamDataEntrySerializer 

90 

91 def get_queryset(self): 

92 queryset = StreamDataEntry.objects.all() 

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

94 if stream_data_project: 

95 queryset = queryset.filter(streamDataProject=stream_data_project) 

96 return queryset 

97 

98 def get_serializer_context(self): 

99 ctx = super().get_serializer_context() 

100 ctx["all_sdes"] = list( 

101 self.get_queryset() 

102 .select_related("unitop") 

103 .prefetch_related("unitop__ports__stream") 

104 ) 

105 return ctx 

106 

107 @extend_schema( 

108 parameters=[ 

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

110 ], 

111 responses=StreamDataEntrySerializer(many=True) 

112 ) 

113 def list(self, request): 

114 return super().list(request)