# Standard imports from datetime import datetime, timezone, timedelta # Django imports 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 rest_framework.request import Request from django.db.models import Q # Project imports 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, EnvDataSerializer, SensorDataLastSerializer from user_manager.models import PyrosUser, UserLevel, SP_Period, ScientificProgram, SP_Period_User #from scientific_program.models import SP_Period, ScientificProgram, SP_Period_User from common.models import AgentSurvey, AgentCmd, Env_data, Sensors_data, Sensors_data_last_value from routine_manager.models import Sequence, Album, Plan from routine_manager.functions import check_sequence_file_validity_and_save, create_sequence_pickle from src.pyros_logger import log import yaml # 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_programs() 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: sp_of_user = user.get_scientific_programs() return Sequence.objects.filter(scientific_program__in=sp_of_user).order_by("-updated") #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_programs().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_programs() 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 (Sequence with their albums and plans). """ queryset = Sequence.objects.all().order_by("-updated") serializer_class = FullSequenceSerializer permission_classes = [IsAuthenticated] http_method_names = ["get"] def retrieve(self, request, pk=None): pk = pk.lstrip("0") 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 plans. """ 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_file(request): """ Submit a sequence file (yaml) """ file = request.FILES.get("sequence_file") yaml_content = None response = None status = 0 if file is None: status = -1 message = "File does not exist" elif file.size > 1000000: status = -1 message = "File is too big (more than 1 000 000 bytes)" else: yaml_content = yaml.safe_load(file.read()) response = check_sequence_file_validity_and_save(yaml_content, request) if response.get("succeed") == True: seq = Sequence.objects.get(id=response.get("sequence_id")) create_sequence_pickle(seq) 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 the status of agents. """ 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 the commands of agents. """ 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',20) 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") class EnvDataViewSet(viewsets.ModelViewSet): """ API endpoint to see environment data. """ queryset = Env_data.objects.all().order_by("monitoring_name") serializer_class = EnvDataSerializer permission_classes = [IsAuthenticated] http_method_names = ["get"] class SensorDataLastViewSet(viewsets.ModelViewSet): """ API endpoint to see sensor data. """ queryset = Sensors_data_last_value.objects.all().order_by("key") serializer_class = SensorDataLastSerializer permission_classes = [IsAuthenticated] http_method_names = ["get"]