import os
import requests
import openmeteo_requests
from datetime import datetime
from django.db.models import Avg
from django.utils import timezone
from datetime import timedelta
from rest_framework.pagination import PageNumberPagination
from rest_framework.request import Request
from rest_framework import permissions
from .models import WeatherStats
from project.settings_context import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID
class IsMikrokontroller(permissions.BasePermission):
def has_permission(self, request: Request, view):
mikro_key = os.environ['MIKRO_SECRET_KEY']
request_key = request.headers.get('X-Mikro-Key')
return request_key == mikro_key
class SmallPageNumberPagination(PageNumberPagination):
page_size = 200
page_size_query_param = 'page_size'
max_page_size = 500
def format_value(val, unit="%"):
return f"{val:.1f}{unit}" if val is not None else "—"
def send_telegram_stats():
now = timezone.now()
thirty_minutes_ago = now - timedelta(minutes=30)
now_formatted = now.strftime("%d.%m.%Y %H:%M")
# Collect data
averages = WeatherStats.objects.filter(
created_at__gte=thirty_minutes_ago
).aggregate(
avg_humidity_air = Avg('humidity_air'),
avg_humidity_ground = Avg('humidity_ground'),
avg_temperature = Avg('temperature'),
avg_light = Avg('light')
)
message = f"""
Weather average stats for 30 mins ({now_formatted})
Humidity Air: {format_value(averages['avg_humidity_air'])}
Humidity Ground: {format_value(averages['avg_humidity_ground'])}
Humidity Temperature: {format_value(averages['avg_temperature'])}°C
Light: {format_value(averages['avg_light'])}
"""
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
params = {
"chat_id": TELEGRAM_CHAT_ID,
"text": message,
"parse_mode": "HTML"
}
requests.get(url, params)
def get_weather_api_data(latitude, longitude):
now = timezone.now()
openmeteo = openmeteo_requests.Client()
start_date = now + timedelta(days=1)
end_date = start_date + timedelta(days=7)
start_date_formatted = start_date.strftime("%Y-%m-%d")
end_date_formatted = end_date.strftime("%Y-%m-%d")
url = "https://api.open-meteo.com/v1/forecast"
params = {
"latitude": latitude,
"longitude": longitude,
"daily": [
"temperature_2m_min",
"temperature_2m_max",
"wind_speed_10m_max",
"precipitation_probability_max",
"relative_humidity_2m_mean"
],
"timezone": "auto",
"start_date": start_date_formatted,
"end_date": end_date_formatted,
}
responses = openmeteo.weather_api(url, params=params)
response = responses[0]
days = response.Daily()
dates = days.Time()
temp_min = days.Variables(0).ValuesAsNumpy()
temp_max = days.Variables(1).ValuesAsNumpy()
wind_max = days.Variables(2).ValuesAsNumpy()
precip_prob_max = days.Variables(3).ValuesAsNumpy()
humidity_mean = days.Variables(4).ValuesAsNumpy()
result = []
count = len(days.Variables(0).ValuesAsNumpy())
base_time = days.Time()
for i in range(count):
timestamp = base_time + i * 86400
local_dt = datetime.fromtimestamp(timestamp)
day = {
"date": local_dt.strftime("%Y-%m-%d"),
"temp_night": float(temp_min[i]),
"temp_day": float(temp_max[i]),
"humidity": float(humidity_mean[i]),
"wind_speed": float(wind_max[i]),
"precip_probability": float(precip_prob_max[i])
}
result.append(day)
return result