78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
import datetime
|
|
import statistics
|
|
from typing import Literal
|
|
|
|
from .model import Cloudness, Precipitation, Sky, WeatherValue, WindDirectionDeg
|
|
|
|
MergeMode = Literal["interval", "average"]
|
|
|
|
|
|
def build_weather_value(date: datetime.datetime) -> WeatherValue:
|
|
return WeatherValue(
|
|
date=date,
|
|
sky=Sky(
|
|
cloudness=Cloudness.CLEAR,
|
|
precipitation=Precipitation.NO,
|
|
thunder=False,
|
|
fog=False,
|
|
),
|
|
temperature=[],
|
|
wind_speed=0,
|
|
wind_gust=0,
|
|
wind_direction=WindDirectionDeg(-1),
|
|
precipitation=0,
|
|
pressure=[],
|
|
humidity=-1,
|
|
)
|
|
|
|
|
|
def _merge_values(values: list[int], mode: MergeMode) -> list[int]:
|
|
if mode == "interval":
|
|
if (max_value := max(values)) == (min_value := min(values)):
|
|
return [max_value]
|
|
else:
|
|
return [max_value, min_value]
|
|
else:
|
|
return [round(sum(values) / len(values))]
|
|
|
|
|
|
def merge_weather_values(
|
|
date: datetime.datetime,
|
|
values: list[WeatherValue],
|
|
mode: MergeMode = "interval",
|
|
) -> WeatherValue:
|
|
result = build_weather_value(date)
|
|
temperatures = []
|
|
pressures = []
|
|
humidities = []
|
|
wind_speeds = []
|
|
wind_gusts = []
|
|
wind_directions = []
|
|
cloudnesses = []
|
|
precipitations = []
|
|
precipitation = 0
|
|
for value in values:
|
|
temperatures += value.temperature
|
|
pressures += value.pressure
|
|
humidities.append(value.humidity)
|
|
wind_speeds.append(value.wind_speed)
|
|
wind_gusts.append(value.wind_gust)
|
|
wind_directions.append(value.wind_direction)
|
|
cloudnesses.append(value.sky.cloudness)
|
|
precipitations.append(value.sky.precipitation)
|
|
precipitation += value.precipitation
|
|
result.temperature = _merge_values(temperatures, mode)
|
|
result.pressure = _merge_values(pressures, mode)
|
|
result.humidity = round(statistics.mean(humidities))
|
|
result.wind_speed = round(statistics.mean(wind_speeds))
|
|
result.wind_gust = round(statistics.mean(wind_gusts))
|
|
result.wind_direction = statistics.mean(wind_directions)
|
|
for item in cloudnesses:
|
|
if Cloudness.values.index(item) > Cloudness.values.index(result.sky.cloudness):
|
|
result.sky.cloudness = item
|
|
for item in precipitations:
|
|
if Precipitation.values.index(item) > Precipitation.values.index(result.sky.precipitation):
|
|
result.sky.precipitation = item
|
|
result.precipitation = precipitation
|
|
return result
|