Coverage for backend/django/PinchAnalysis/views/PinchAnalysisView.py: 87%

1from rest_framework.response import Response 

2from rest_framework.decorators import api_view 

3from drf_spectacular.utils import extend_schema 

4from rest_framework import serializers, status 

5from pinch_factory.pinch_factory import PinchFactory 

6from core.validation import api_view_validate 

7from core.managers import get_flowsheet_access 

8 

9class PinchRequestSerializer(serializers.Serializer): 

10 excludedSegments = serializers.ListField( 

11 child=serializers.IntegerField(), 

12 ) 

13 

14@api_view_validate 

15@extend_schema(request=PinchRequestSerializer) 

16@api_view(['POST']) 

17def pinch_analysis_view(request) -> Response: 

18 try: 

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

20 access_state = get_flowsheet_access(request.user, flowsheet_id) 

21 if access_state.has_read_access and not access_state.has_write_access: 

22 return Response( 

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

24 status=status.HTTP_403_FORBIDDEN, 

25 ) 

26 

27 # Validate the request 

28 serializer = PinchRequestSerializer(data=request.data) 

29 serializer.is_valid(raise_exception=True) 

30 validated_data = serializer.validated_data 

31 excluded_segments = validated_data.get("excludedSegments", []) 

32 

33 # Initialize PinchFactory and prepare data 

34 factory = PinchFactory(flowsheet_id) 

35 factory.run_calculate(excluded_segments=excluded_segments) 

36 

37 # Perform the calculation and return results 

38 return Response(data=None, status=200) 

39 

40 except ValueError as ve: 

41 return Response({"error": str(ve)}, status=400) 

42 

43 except Exception as e: 

44 return Response({"error": "An error occurred", "details": str(e)}, status=500)