84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from farm_data.models import SensorData
|
||
|
|
|
||
|
|
from .services import get_forecast_for_location
|
||
|
|
|
||
|
|
|
||
|
|
WMO_CONDITIONS = {
|
||
|
|
0: "صاف",
|
||
|
|
1: "عمدتاً صاف",
|
||
|
|
2: "نیمهابری",
|
||
|
|
3: "ابری",
|
||
|
|
45: "مه",
|
||
|
|
48: "مه یخزده",
|
||
|
|
51: "نمنم باران",
|
||
|
|
61: "بارش خفیف",
|
||
|
|
63: "بارش متوسط",
|
||
|
|
65: "بارش شدید",
|
||
|
|
71: "برف خفیف",
|
||
|
|
80: "رگبار",
|
||
|
|
95: "رعد و برق",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _safe_number(value, default=0):
|
||
|
|
return default if value is None else value
|
||
|
|
|
||
|
|
|
||
|
|
def _average(values, default=0):
|
||
|
|
clean_values = [value for value in values if value is not None]
|
||
|
|
if not clean_values:
|
||
|
|
return default
|
||
|
|
return sum(clean_values) / len(clean_values)
|
||
|
|
|
||
|
|
|
||
|
|
def _weather_condition(weather_code):
|
||
|
|
return WMO_CONDITIONS.get(weather_code, "نامشخص")
|
||
|
|
|
||
|
|
|
||
|
|
def _build_farm_weather_card(forecasts: list[Any]) -> dict[str, Any]:
|
||
|
|
if not forecasts:
|
||
|
|
return {
|
||
|
|
"condition": "نامشخص",
|
||
|
|
"temperature": 0,
|
||
|
|
"unit": "°C",
|
||
|
|
"humidity": 0,
|
||
|
|
"windSpeed": 0,
|
||
|
|
"windUnit": "km/h",
|
||
|
|
"chartData": {"labels": [], "series": [[]]},
|
||
|
|
}
|
||
|
|
|
||
|
|
current_forecast = forecasts[0]
|
||
|
|
labels = [str(forecast.forecast_date) for forecast in forecasts[:7]]
|
||
|
|
series = [[round(_safe_number(forecast.temperature_mean, 0)) for forecast in forecasts[:7]]]
|
||
|
|
|
||
|
|
return {
|
||
|
|
"condition": _weather_condition(current_forecast.weather_code),
|
||
|
|
"temperature": round(_safe_number(current_forecast.temperature_mean, current_forecast.temperature_max)),
|
||
|
|
"unit": "°C",
|
||
|
|
"humidity": round(_average([current_forecast.humidity_mean], default=0)),
|
||
|
|
"windSpeed": round(_safe_number(current_forecast.wind_speed_max, 0)),
|
||
|
|
"windUnit": "km/h",
|
||
|
|
"chartData": {
|
||
|
|
"labels": labels,
|
||
|
|
"series": series,
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
class FarmWeatherService:
|
||
|
|
def get_farm_weather_card(self, *, farm_uuid: str) -> dict[str, Any]:
|
||
|
|
sensor = (
|
||
|
|
SensorData.objects.select_related("center_location")
|
||
|
|
.filter(farm_uuid=farm_uuid)
|
||
|
|
.first()
|
||
|
|
)
|
||
|
|
if sensor is None:
|
||
|
|
raise ValueError("Farm not found.")
|
||
|
|
|
||
|
|
forecasts = get_forecast_for_location(sensor.center_location, days=7)
|
||
|
|
return _build_farm_weather_card(forecasts)
|