import time from django.utils.decorators import method_decorator from django.views.decorators.cache import cache_page from django.core.cache import cache from rest_framework.generics import CreateAPIView, ListAPIView from rest_framework.views import APIView from rest_framework.request import Request from rest_framework.response import Response from rest_framework import status from drf_spectacular.utils import extend_schema, OpenApiParameter, OpenApiResponse from project.serializers import MessageSerializer from .models import WeatherStats from .serializers import WeatherStatSerializer, OpenWeatherAPISerializer from .utils import IsMikrokontroller, PageNumberPagination, get_weather_api_data CACHE_SAVE_CONTROL = 'stats_latest_saved_control' CACHE_KEY_STATS = 'stats_latest_data' CACHE_TIMEOUT = 30 # 30 seconds CACHE_KEY_WEATHER = 'weather_latest_data' CACHE_TIMEOUT_WEATHER = 60 * 5 # 5 minutes # Create your views here. @extend_schema(tags=['Weather'], description="Call method by mikrocontroller to set new data (required MIKRO_SECRET_KEY with header X-Mikro-Key) with delay 30 seconds", parameters=[ OpenApiParameter( name='X-Mikro-Key', type=str, location=OpenApiParameter.HEADER, description='Secret Key for microcontroller', required=True ) ] ) class CreateStatAPI(CreateAPIView): serializer_class = WeatherStatSerializer permission_classes = [ IsMikrokontroller ] queryset = WeatherStats.objects.all() def perform_create(self, serializer: WeatherStatSerializer): new_data = serializer.validated_data last_saved = cache.get(CACHE_SAVE_CONTROL) current_time = time.time() if last_saved is None or (current_time - last_saved) >= CACHE_TIMEOUT: serializer.save() cache.set(CACHE_SAVE_CONTROL, current_time, CACHE_TIMEOUT) cache.set(CACHE_KEY_STATS, new_data, CACHE_TIMEOUT) @extend_schema(tags=['Weather'], description="Get latest stat by microcontroller. Can be used to get actual data now", responses={ 200: WeatherStatSerializer, 404: MessageSerializer } ) class LastStatAPI(APIView): def get(self, request: Request, format=None): cached_data = cache.get(CACHE_KEY_STATS) if cached_data is None: return Response( {"message": "We don't have latest data from microcontroller. Maybe microcontroller is not connected"}, status=status.HTTP_404_NOT_FOUND ) return Response(cached_data) @extend_schema(tags=['Weather'], description="Get full history for graph") class StatsHistoryAPI(ListAPIView): serializer_class = WeatherStatSerializer queryset = WeatherStats.objects.order_by('-created_at') pagination_class = PageNumberPagination @method_decorator(cache_page(CACHE_TIMEOUT_WEATHER), name='dispatch') class OpenMeteoWeatherAPI(APIView): @extend_schema(tags=['Weather'], description="Weather prediction using OpenMeteo API data based on latitude and longitude for 7 days", parameters=[ OpenApiParameter( name='lat', type=float, location=OpenApiParameter.QUERY, description='Latitude', required=True ), OpenApiParameter( name='long', type=float, location=OpenApiParameter.QUERY, description='Longitude', required=True ) ], responses={ 200: OpenWeatherAPISerializer(many=True), 400: MessageSerializer } ) def get(self, request: Request, format=None): latitude = request.query_params.get('lat') longitude = request.query_params.get('long') if not latitude or not longitude: return Response("No latitude or longitude data", status=status.HTTP_400_BAD_REQUEST) weather_stats = get_weather_api_data(latitude, longitude) return Response(weather_stats)