from django.shortcuts import render from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import viewsets, renderers from rest_framework.permissions import IsAuthenticated, AllowAny from rest_framework.decorators import api_view, permission_classes, action from django.core.validators import ValidationError from src.core.pyros_django.user_manager import views as user_views from api.serializers import AgentSurveySerializer, AlbumSerializer, FullSequenceSerializer, PlanSerializer, SPPeriodSerializer, ScientificProgramSerializer, SequenceSerializer, UserSerializer, AgentCmdSerializer from common.models import PyrosUser, SP_Period, ScientificProgram, Sequence, Album, Plan, UserLevel, SP_Period_User, AgentSurvey, AgentCmd from routine_manager.functions import check_sequence_file_validity from rest_framework.request import Request from django.db.models import Q from datetime import datetime, timezone, timedelta from src.pyros_logger import log # Create your views here. @api_view(["GET"]) def user_logout(request): """ Log out user and delete authentification token """ request.user.auth_token.delete() user_views.logout(request) return Response('User Logged out successfully') class UserViewSet(viewsets.ModelViewSet): """ API endpoint that allows users to be viewed. """ queryset = PyrosUser.objects.all().order_by('-date_joined') serializer_class = UserSerializer permission_classes = [IsAuthenticated] http_method_names = ["get"] def list(self, request): serializer_context = { 'request': request, } queryset = None current_user = self.request.user user = self.request.user user_role = str(UserLevel.objects.get( priority=user.get_priority()).name) if user_role in ("Unit-PI", "Admin"): queryset = PyrosUser.objects.all().order_by("-created") else: sp_of_current_user = user.get_scientific_program() pyros_users_with_roles = [] for sp in sp_of_current_user: for sp_period in sp.SP_Periods.all(): for user in SP_Period_User.objects.filter(SP_Period=sp_period).exclude(user=current_user).values_list("user", flat=True): pyros_users_with_roles.append( PyrosUser.objects.get(id=user)) pyros_users_with_roles.append( sp_period.scientific_program.sp_pi) # Include unit_pi and unit_board users unit_users = PyrosUser.objects.filter( user_level__name__in=("Unit-PI", "Unit-board")).distinct() queryset = pyros_users_with_roles + list(unit_users) serializer = UserSerializer( queryset, context=serializer_context, many=True) return Response(serializer.data) class SequenceViewSet(viewsets.ModelViewSet): """ API endpoint that allows users to view their sequences. """ queryset = Sequence.objects.all().order_by("-updated") serializer_class = SequenceSerializer permission_classes = [IsAuthenticated] http_method_names = ["get"] def get_queryset(self): """ This view should return a list of all the sequences for the currently authenticated user. """ user = self.request.user user_role = str(UserLevel.objects.get( priority=user.get_priority()).name) if user_role in ("Unit-PI", "Admin"): return Sequence.objects.all().order_by("-updated") else: return Sequence.objects.filter(pyros_user=user).order_by("-updated") class ScientificProgramViewSet(viewsets.ModelViewSet): """ API endpoint that allows users to view their scientific programs. """ queryset = ScientificProgram.objects.all().order_by("-created") serializer_class = ScientificProgramSerializer permission_classes = [IsAuthenticated] http_method_names = ["get"] def get_queryset(self): """ This view should return a list of all the sequences for the currently authenticated user. """ user = self.request.user user_role = str(UserLevel.objects.get( priority=user.get_priority()).name) if user_role in ("Unit-PI", "Admin"): return ScientificProgram.objects.all().order_by("-updated") else: return user.get_scientific_program().order_by("-updated") class SPPeriodViewSet(viewsets.ModelViewSet): """ API endpoint that allows users to view their sp_periods """ queryset = SP_Period.objects.all() serializer_class = SPPeriodSerializer permission_classes = [IsAuthenticated] http_method_names = ["get"] def get_queryset(self): """ This view should return a list of all the sp_period for the currently authenticated user. """ user = self.request.user user_role = str(UserLevel.objects.get( priority=user.get_priority()).name) if user_role in ("Unit-PI", "Admin"): return SP_Period.objects.all().order_by("-scientific_program") else: user_scientific_programs = user.get_scientific_program() return SP_Period.objects.filter(scientific_program__in=user_scientific_programs).order_by("-scientific_program") class FullSequenceViewSet(viewsets.ModelViewSet): """ API endpoint that allows users to view their full sequences. """ queryset = Sequence.objects.all().order_by("-updated") serializer_class = FullSequenceSerializer permission_classes = [IsAuthenticated] http_method_names = ["get"] def get_queryset(self): """ This view should return a list of all the sequences for the currently authenticated user. """ user = self.request.user user_role = str(UserLevel.objects.get( priority=user.get_priority()).name) if user_role in ("Unit-PI", "Admin"): return Sequence.objects.all().order_by("-updated") else: return Sequence.objects.filter(pyros_user=user).order_by("-updated") @action(detail=False, methods=["get"]) def get_sequences_for_period(self, request): """ Return the list of sequences corresponding to the requested period """ params = request.query_params start_date = datetime.strptime(params.get("start_date"), "%d/%m/%Y") end_date = datetime.strptime(params.get("end_date"), "%d/%m/%Y") queryset = Sequence.objects.filter( Q(start_date__gte=start_date) | Q(start_date__lte=end_date)) serializer = self.get_serializer(queryset, many=True) response = Response(serializer.data) return response class AlbumViewSet(viewsets.ModelViewSet): """ API endpoint that allows users to view their Albums. """ queryset = Album.objects.all().order_by("-updated") serializer_class = AlbumSerializer permission_classes = [IsAuthenticated] http_method_names = ["get"] def get_queryset(self): """ This view should return a list of all the albums for the currently authenticated user. """ user = self.request.user user_role = str(UserLevel.objects.get( priority=user.get_priority()).name) if user_role in ("Unit-PI", "Admin"): sequences = Sequence.objects.all().order_by("-updated") else: sequences = Sequence.objects.filter( pyros_user=user).order_by("-updated") return Album.objects.filter(sequence__in=sequences).order_by("-updated") class PlanViewSet(viewsets.ModelViewSet): """ API endpoint that allows users to view their sequences. """ queryset = Plan.objects.all().order_by("-updated") serializer_class = PlanSerializer permission_classes = [IsAuthenticated] http_method_names = ["get"] def get_queryset(self): """ This view should return a list of all the plans for the currently authenticated user. """ user = self.request.user user_role = str(UserLevel.objects.get( priority=user.get_priority()).name) if user_role in ("Unit-PI", "Admin"): sequences = Sequence.objects.all().order_by("-updated") else: sequences = Sequence.objects.filter( pyros_user=user).order_by("-updated") albums = Album.objects.filter( sequence__in=sequences).order_by("-updated") return Plan.objects.filter(album__in=albums).order_by("-updated") @api_view(["PUT"]) def submit_sequence_with_json(request): sequence_json = request.data response = check_sequence_file_validity(sequence_json, request) if response.get("succeed") == True: seq = Sequence.objects.get(id=response.get("sequence_id")) log.info( f"User {request.user} did action submit sequence {seq} for period {seq.period} ") return Response(response) class AgentSurveyViewSet(viewsets.ModelViewSet): """ API endpoint that allows users to view their sequences. """ queryset = AgentSurvey.objects.all() serializer_class = AgentSurveySerializer permission_classes = [IsAuthenticated] http_method_names = ["get"] lookup_field = "name" def get_queryset(self): agents = AgentSurvey.objects.all() datetime_now = datetime.utcnow() date_minus_two_days = datetime_now - timedelta(days=2) date_minus_two_days = date_minus_two_days.replace(tzinfo=timezone.utc) agents = agents.exclude(updated__lt=date_minus_two_days) return agents class AgentCmdViewSet(viewsets.ModelViewSet): """ API endpoint that allows users to view their sequences. """ queryset = AgentCmd.objects.all() serializer_class = AgentCmdSerializer permission_classes = [IsAuthenticated] http_method_names = ["get"] def get_queryset(self): agent_name = self.request.query_params.get('agent_name') number = self.request.query_params.get('number') if agent_name is None: if "agent_name" in self.kwargs: agent_name = self.kwargs["agent_name"] else: agent_name = None if agent_name is not None: commands_sent_by_agent = AgentCmd.get_commands_sent_by_agent(agent_name) commands_recivied_by_agent = AgentCmd.get_commands_sent_to_agent(agent_name) agent_cmds = commands_sent_by_agent | commands_recivied_by_agent agent_cmds = agent_cmds.exclude(full_name="get_specific_cmds") agent_cmds = agent_cmds.exclude(full_name="get_all_cmds") agent_cmds = agent_cmds.order_by("-s_deposit_time") if number: number = int(number) agent_cmds = agent_cmds[:number] return agent_cmds return AgentCmd.objects.all().order_by("-id")