Psion-fanatic/platypush
1
0
Fork 0
forked from platypush/platypush
Code Pull requests Activity

- Added Buienradar integration

Browse source 
- Refactored weather plugin/backend as Darksky plugin/backend
This commit is contained in:
Fabio Manganiello 2019-12-31 08:51:19 +01:00
parent 9eed1014eb
commit 49a7ee643e
8 changed files with 311 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
platypush/backend/http/static/js/widgets/date-time-weather/index.js
View file

@ -13,7 +13,7 @@ Vue.component('date-time-weather', {
methods: {
refresh: async function() {
let weather = (await request('weather.forecast.get_hourly_forecast')).data[0];
let weather = (await request('weather.darksky.get_hourly_forecast')).data[0];
this.onWeatherChange(weather);
},

52
platypush/backend/weather/buienradar.py Normal file
View file

@ -0,0 +1,52 @@
import time
from platypush.backend import Backend
from platypush.context import get_plugin
from platypush.message.event.weather import NewWeatherConditionEvent, NewPrecipitationForecastEvent
from platypush.plugins.weather.buienradar import WeatherBuienradarPlugin
class WeatherBuienradarBackend(Backend):
"""
Buienradar weather forecast backend. Listens for new weather or precipitation updates.
Triggers:
* :class:`platypush.message.event.weather.NewWeatherConditionEvent` when there is a weather condition update
Requires:
* The :mod:`platypush.plugins.weather.buienradar` plugin configured
"""
def __init__(self, poll_seconds=300, **kwargs):
super().__init__(**kwargs)
self.poll_seconds = poll_seconds
self.last_weather = None
self.last_precip = None
def run(self):
super().run()
plugin: WeatherBuienradarPlugin = get_plugin('weather.buienradar')
self.logger.info('Initialized weather forecast backend')
while not self.should_stop():
weather = plugin.get_weather().output
precip = plugin.get_precipitation().output
del weather['measured']
if precip != self.last_precip:
self.bus.post(NewPrecipitationForecastEvent(average=precip.get('average'),
total=precip.get('total'),
time_frame=precip.get('time_frame')))
if weather != self.last_weather:
self.bus.post(NewWeatherConditionEvent(**weather))
self.last_weather = weather
self.last_precip = precip
time.sleep(self.poll_seconds)
# vim:sw=4:ts=4:et:

11
platypush/backend/weather/forecast.py → platypush/backend/weather/darksky.py
View file

@ -2,11 +2,10 @@ import time
from platypush.backend import Backend
from platypush.context import get_plugin
from platypush.plugins.weather.forecast import WeatherForecastPlugin
from platypush.message.event.weather import NewWeatherConditionEvent
class WeatherForecastBackend(Backend):
class WeatherDarkskyBackend(Backend):
"""
Weather forecast backend - listens and propagates new weather events.
@ -16,7 +15,7 @@ class WeatherForecastBackend(Backend):
Requires:
* The :mod:`platypush.plugins.weather.forecast` plugin configured
* The :class:`platypush.plugins.weather.darksky.WeatherDarkskyPlugin` plugin configured
"""
def __init__(self, poll_seconds, **kwargs):
@ -24,12 +23,9 @@ class WeatherForecastBackend(Backend):
self.poll_seconds = poll_seconds
self.latest_update = {}
def send_message(self, msg):
pass
def run(self):
super().run()
weather = get_plugin('weather.forecast')
weather = get_plugin('weather.darksky')
self.logger.info('Initialized weather forecast backend')
while not self.should_stop():
@ -44,4 +40,3 @@ class WeatherForecastBackend(Backend):
# vim:sw=4:ts=4:et:

11
platypush/message/event/weather.py
View file

@ -1,5 +1,6 @@
from platypush.message.event import Event
class NewWeatherConditionEvent(Event):
"""
Event triggered when the weather condition changes
@ -8,5 +9,13 @@ class NewWeatherConditionEvent(Event):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# vim:sw=4:ts=4:et:
class NewPrecipitationForecastEvent(Event):
"""
Event triggered when the precipitation forecast changes
"""
def __init__(self, *args, average: float, total: float, time_frame: int, **kwargs):
super().__init__(*args, average=average, total=total, time_frame=time_frame, **kwargs)
# vim:sw=4:ts=4:et:

0
platypush/message/response/weather/__init__.py Normal file
View file

113
platypush/message/response/weather/buienradar.py Normal file
View file

@ -0,0 +1,113 @@
import datetime
from typing import List
from platypush.message import Mapping
from platypush.message.response import Response
class BuienradarWeatherResponse(Response):
def __init__(self,
barometer_fc: str,
condition_name: str,
condition_name_long: str,
condition_image: str,
feel_temperature: float,
ground_temperature: float,
humidity: int,
irradiance: int,
measured: datetime.datetime,
precipitation: float,
pressure: float,
rain_last_24_hours: float,
rain_last_hour: float,
station_name: str,
temperature: float,
visibility: int,
wind_azimuth: int,
wind_direction: str,
wind_force: int,
wind_gust: float,
wind_speed: float,
*args, **kwargs):
super().__init__(*args, output={
'barometer_fc': barometer_fc,
'condition_name': condition_name,
'condition_name_long': condition_name_long,
'condition_image': condition_image,
'feel_temperature': feel_temperature,
'ground_temperature': ground_temperature,
'humidity': humidity,
'irradiance': irradiance,
'measured': measured,
'precipitation': precipitation,
'pressure': pressure,
'rain_last_24_hours': rain_last_24_hours,
'rain_last_hour': rain_last_hour,
'station_name': station_name,
'temperature': temperature,
'visibility': visibility,
'wind_azimuth': wind_azimuth,
'wind_direction': wind_direction,
'wind_force': wind_force,
'wind_gust': wind_gust,
'wind_speed': wind_speed,
}, **kwargs)
class BuienradarPrecipitationResponse(Response):
def __init__(self,
average: float,
total: float,
time_frame: int,
*args, **kwargs):
super().__init__(*args, output={
'average': average,
'total': total,
'time_frame': time_frame,
}, **kwargs)
class BuienradarForecast(Mapping):
def __init__(self,
condition_name: str,
condition_name_long: str,
condition_image: str,
date_time: datetime.datetime,
rain: float,
min_rain: float,
max_rain: float,
rain_chance: float,
snow: int,
temperature: float,
wind_azimuth: int,
wind_direction: str,
wind_force: int,
wind_speed: float,
*args, **kwargs):
super().__init__(*args, output={
'condition_name': condition_name,
'condition_name_long': condition_name_long,
'condition_image': condition_image,
'date_time': date_time,
'rain': rain,
'min_rain': min_rain,
'max_rain': max_rain,
'rain_chance': rain_chance,
'snow': snow,
'temperature': temperature,
'wind_azimuth': wind_azimuth,
'wind_direction': wind_direction,
'wind_force': wind_force,
'wind_speed': wind_speed,
}, **kwargs)
class BuienradarForecastResponse(Response):
def __init__(self,
forecast=List[BuienradarForecast],
*args, **kwargs):
super().__init__(*args, output=forecast, **kwargs)
# vim:sw=4:ts=4:et:

131
platypush/plugins/weather/buienradar.py Normal file
View file

@ -0,0 +1,131 @@
from typing import Optional, Dict, Any
from platypush.plugins import Plugin, action
from platypush.message.response.weather.buienradar import BuienradarWeatherResponse, BuienradarPrecipitationResponse, \
BuienradarForecastResponse, BuienradarForecast
class WeatherBuienradarPlugin(Plugin):
"""
Plugin for getting weather updates through Buienradar - a Dutch weather app.
Requires:
* **buienradar** (``pip install buienradar``)
"""
def __init__(self, lat: float, long: float, time_frame: int = 120, **kwargs):
"""
:param lat: Default latitude
:param long: Default longitude
:param time_frame: Default number of minutes to look ahead for precipitation forecast
"""
super().__init__(**kwargs)
self.lat = lat
self.long = long
self.time_frame = time_frame
self.latest_bulletin = {}
def get_data(self, lat: Optional[float] = None, long: Optional[float] = None, time_frame: Optional[int] = None) \
-> Dict[str, Any]:
# noinspection PyPackageRequirements
from buienradar.buienradar import get_data, parse_data
# noinspection PyPackageRequirements
from buienradar.constants import SUCCESS, CONTENT, RAINCONTENT, DATA
lat = lat or self.lat
long = long or self.long
time_frame = time_frame or self.time_frame
result = get_data(latitude=lat, longitude=long)
if not result.get(SUCCESS):
raise RuntimeError('Error while retrieving data')
data = result.get(CONTENT)
rain_data = result.get(RAINCONTENT)
result = parse_data(data, rain_data, lat, long, time_frame)
return result.get(DATA, {})
@action
def get_weather(self, lat: Optional[float] = None, long: Optional[float] = None) -> BuienradarWeatherResponse:
"""
Get the current weather conditions.
:param lat: Weather latitude (default: configured latitude)
:param long: Weather longitude (default: configured longitude)
"""
data = self.get_data(lat, long, 60)
return BuienradarWeatherResponse(
barometer_fc=data.get('barometerfcname'),
condition_name=data.get('condition', {}).get('condition'),
condition_name_long=data.get('condition', {}).get('exact'),
condition_image=data.get('condition', {}).get('image'),
feel_temperature=data.get('feeltemperature'),
ground_temperature=data.get('groundtemperature'),
humidity=data.get('humidity'),
irradiance=data.get('irradiance'),
measured=data.get('measured'),
precipitation=data.get('precipitation'),
pressure=data.get('pressure'),
rain_last_24_hours=data.get('rainlast24hour'),
rain_last_hour=data.get('rainlasthour'),
station_name=data.get('stationname'),
temperature=data.get('temperature'),
visibility=data.get('visibility'),
wind_azimuth=data.get('windazimuth'),
wind_direction=data.get('wind_irection'),
wind_force=data.get('windforce'),
wind_gust=data.get('windgust'),
wind_speed=data.get('windspeed')
)
@action
def get_forecast(self, lat: Optional[float] = None, long: Optional[float] = None) -> BuienradarForecastResponse:
"""
Get the weather forecast for the next days.
:param lat: Weather latitude (default: configured latitude)
:param long: Weather longitude (default: configured longitude)
"""
data = self.get_data(lat, long, 60).get('forecast', [])
return BuienradarForecastResponse([
BuienradarForecast(
condition_name=d.get('condition', {}).get('condition'),
condition_name_long=d.get('condition', {}).get('exact'),
condition_image=d.get('condition', {}).get('image'),
date_time=d.get('datetime'),
rain=d.get('rain'),
min_rain=d.get('minrain'),
max_rain=d.get('maxrain'),
rain_chance=d.get('rainchance'),
snow=d.get('snow'),
temperature=d.get('temperature'),
wind_azimuth=d.get('windazimuth'),
wind_direction=d.get('winddirection'),
wind_force=d.get('windforce'),
wind_speed=d.get('windspeed'),
)
for d in data
])
@action
def get_precipitation(self, lat: Optional[float] = None, long: Optional[float] = None,
time_frame: Optional[int] = None) -> BuienradarPrecipitationResponse:
"""
Get the precipitation forecast for the specified time frame.
:param lat: Weather latitude (default: configured latitude)
:param long: Weather longitude (default: configured longitude)
:param time_frame: Time frame for the forecast in minutes (default: configured time_frame)
"""
data = self.get_data(lat, long, time_frame).get('precipitation_forecast', {})
return BuienradarPrecipitationResponse(
average=data.get('average'),
total=data.get('total'),
time_frame=data.get('timeframe'),
)
# vim:sw=4:ts=4:et:

2
platypush/plugins/weather/forecast.py → platypush/plugins/weather/darksky.py
View file

@ -2,7 +2,7 @@ from platypush.plugins import action
from platypush.plugins.http.request import HttpRequestPlugin
class WeatherForecastPlugin(HttpRequestPlugin):
class WeatherDarkskyPlugin(HttpRequestPlugin):
"""
Plugin for getting weather updates through Darksky API