63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
import os
|
|
import requests
|
|
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"""
|
|
<b> Weather average stats for 30 mins ({now_formatted})</b>
|
|
|
|
<b>Humidity Air:</b> {format_value(averages['avg_humidity_air'])}
|
|
<b>Humidity Ground:</b> {format_value(averages['avg_humidity_ground'])}
|
|
<b>Humidity Temperature:</b> {format_value(averages['avg_temperature', '°C'])}
|
|
<b>Light:</b> {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)
|
|
|
|
|