2023-11-19 00:10:10 +01:00
|
|
|
from typing import Optional
|
2023-11-20 01:46:01 +01:00
|
|
|
|
|
|
|
from datetime import datetime
|
|
|
|
from dateutil.tz import tzutc
|
2023-11-19 00:10:10 +01:00
|
|
|
from marshmallow import fields, pre_dump
|
|
|
|
from marshmallow.schema import Schema
|
2023-11-20 01:46:01 +01:00
|
|
|
from marshmallow.validate import OneOf
|
2023-11-19 00:10:10 +01:00
|
|
|
|
|
|
|
from platypush.schemas import DateTime
|
|
|
|
|
|
|
|
|
|
|
|
def _get_precip_type(response: dict) -> Optional[str]:
|
|
|
|
if response.get('snow', {}).get('1h', 0) > 0:
|
|
|
|
return 'snow'
|
|
|
|
if response.get('rain', {}).get('1h', 0) > 0:
|
|
|
|
return 'rain'
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
|
|
class WeatherSchema(Schema):
|
|
|
|
"""
|
|
|
|
Schema for weather data.
|
|
|
|
"""
|
|
|
|
|
2023-11-23 01:06:38 +01:00
|
|
|
time = DateTime(
|
|
|
|
required=True,
|
|
|
|
attribute='dt',
|
|
|
|
metadata={
|
|
|
|
'description': 'Time of the weather condition',
|
|
|
|
'example': '2020-01-01T12:00:00+00:00',
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
2023-11-19 00:10:10 +01:00
|
|
|
summary = fields.Function(
|
|
|
|
lambda obj: obj.get('weather', [{'main': 'Unknown'}])[0].get('main', 'Unknown'),
|
|
|
|
metadata={
|
|
|
|
'description': 'Summary of the weather condition',
|
|
|
|
'example': 'Cloudy',
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
icon = fields.Function(
|
|
|
|
lambda obj: obj.get('weather', [{'icon': 'unknown'}])[0].get('icon', 'unknown'),
|
|
|
|
metadata={
|
|
|
|
'description': 'Icon representing the weather condition',
|
|
|
|
'example': 'cloudy',
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
precip_intensity = fields.Function(
|
|
|
|
lambda obj: obj.get('rain', obj.get('snow', {})).get('1h', 0),
|
|
|
|
metadata={
|
|
|
|
'description': 'Intensity of the precipitation',
|
|
|
|
'example': 0.0,
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
precip_type = fields.Function(
|
|
|
|
_get_precip_type,
|
|
|
|
metadata={
|
|
|
|
'description': 'Type of precipitation',
|
|
|
|
'example': 'rain',
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
temperature = fields.Function(
|
|
|
|
lambda obj: obj.get('main', {}).get('temp'),
|
|
|
|
metadata={
|
|
|
|
'description': 'Temperature in the configured unit of measure',
|
|
|
|
'example': 10.0,
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
apparent_temperature = fields.Function(
|
|
|
|
lambda obj: obj.get('main', {}).get('feels_like'),
|
|
|
|
metadata={
|
|
|
|
'description': 'Apparent temperature in the configured unit of measure',
|
|
|
|
'example': 9.0,
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
humidity = fields.Function(
|
|
|
|
lambda obj: obj.get('main', {}).get('humidity'),
|
|
|
|
metadata={
|
|
|
|
'description': 'Humidity percentage, between 0 and 100',
|
|
|
|
'example': 30,
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
pressure = fields.Function(
|
|
|
|
lambda obj: obj.get('main', {}).get('pressure'),
|
|
|
|
metadata={
|
|
|
|
'description': 'Pressure in hPa',
|
|
|
|
'example': 1000.0,
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
wind_speed = fields.Function(
|
|
|
|
lambda obj: obj.get('wind', {}).get('speed'),
|
|
|
|
metadata={
|
|
|
|
'description': 'Wind speed in the configured unit of measure',
|
|
|
|
'example': 10.0,
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
wind_direction = fields.Function(
|
|
|
|
lambda obj: obj.get('wind', {}).get('deg'),
|
|
|
|
metadata={
|
|
|
|
'description': 'Wind direction in degrees',
|
|
|
|
'example': 180.0,
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
wind_gust = fields.Function(
|
|
|
|
lambda obj: obj.get('wind', {}).get('gust'),
|
|
|
|
metadata={
|
|
|
|
'description': 'Wind gust in the configured unit of measure',
|
|
|
|
'example': 15.0,
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
cloud_cover = fields.Function(
|
|
|
|
lambda obj: obj.get('clouds', {}).get('all'),
|
|
|
|
metadata={
|
|
|
|
'description': 'Cloud cover percentage between 0 and 100',
|
|
|
|
'example': 0.5,
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
visibility = fields.Float(
|
|
|
|
metadata={
|
|
|
|
'description': 'Visibility in meters',
|
|
|
|
'example': 2000.0,
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
sunrise = DateTime(
|
|
|
|
metadata={
|
|
|
|
'description': 'Sunrise time',
|
|
|
|
'example': '2020-01-01T06:00:00+00:00',
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
sunset = DateTime(
|
|
|
|
metadata={
|
|
|
|
'description': 'Sunset time',
|
|
|
|
'example': '2020-01-01T18:00:00+00:00',
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
2023-11-20 01:46:01 +01:00
|
|
|
units = fields.String(
|
|
|
|
missing='metric',
|
|
|
|
validate=OneOf(['metric', 'imperial']),
|
|
|
|
metadata={
|
|
|
|
'description': 'Unit of measure',
|
|
|
|
'example': 'metric',
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
@staticmethod
|
|
|
|
def _timestamp_to_dt(timestamp: float) -> datetime:
|
|
|
|
return datetime.fromtimestamp(timestamp, tz=tzutc())
|
|
|
|
|
2023-11-19 00:10:10 +01:00
|
|
|
@pre_dump
|
|
|
|
def _pre_dump(self, data: dict, **_) -> dict:
|
|
|
|
sun_data = data.pop('sys', {})
|
|
|
|
sunrise = sun_data.get('sunrise')
|
|
|
|
sunset = sun_data.get('sunset')
|
|
|
|
|
|
|
|
if sunrise is not None:
|
2023-11-20 01:46:01 +01:00
|
|
|
data['sunrise'] = self._timestamp_to_dt(sunrise)
|
2023-11-19 00:10:10 +01:00
|
|
|
if sunset is not None:
|
2023-11-20 01:46:01 +01:00
|
|
|
data['sunset'] = self._timestamp_to_dt(sunset)
|
2023-11-19 00:10:10 +01:00
|
|
|
|
|
|
|
return data
|
2023-11-23 01:06:38 +01:00
|
|
|
|
|
|
|
|
|
|
|
class WeatherReportSchema(Schema):
|
|
|
|
"""
|
|
|
|
Schema for full weather reports.
|
|
|
|
"""
|
|
|
|
|
|
|
|
current = fields.Nested(WeatherSchema, required=True)
|
|
|
|
forecast = fields.List(fields.Nested(WeatherSchema), required=True)
|