forked from platypush/platypush
Fabio Manganiello
c3337ccc6c
Added an `add_dependencies` plugin to the Sphinx build process that parses the manifest files of the scanned backends and plugins and automatically generates the documentation for the required dependencies and triggered events. This means that those dependencies are no longer required to be listed in the docstring of the class itself. Also in this commit: - Black/LINT for some integrations that hadn't been touched in a long time. - Deleted some leftovers from previous refactors (deprecated `backend.mqtt`, `backend.zwave.mqtt`, `backend.http.request.rss`). - Deleted deprecated `inotify` backend - replaced by `file.monitor` (see #289).
111 lines
3.3 KiB
Python
111 lines
3.3 KiB
Python
import datetime
|
|
import time
|
|
from typing import Optional
|
|
|
|
import requests
|
|
from dateutil.tz import gettz, tzutc
|
|
|
|
from platypush.message.event.sun import SunriseEvent, SunsetEvent
|
|
from platypush.plugins import RunnablePlugin, action
|
|
from platypush.schemas.sun import SunEventsSchema
|
|
|
|
|
|
class SunPlugin(RunnablePlugin):
|
|
"""
|
|
Plugin to get sunset/sunrise events and info for a certain location.
|
|
"""
|
|
|
|
_base_url = 'https://api.sunrise-sunset.org/json'
|
|
_attr_to_event_class = {
|
|
'sunrise': SunriseEvent,
|
|
'sunset': SunsetEvent,
|
|
}
|
|
|
|
def __init__(self, latitude: float, longitude: float, **kwargs):
|
|
"""
|
|
:param latitude: Default latitude.
|
|
:param longitude: Default longitude.
|
|
"""
|
|
super().__init__(**kwargs)
|
|
self.latitude = latitude
|
|
self.longitude = longitude
|
|
|
|
def main(self):
|
|
while not self.should_stop():
|
|
# noinspection PyUnresolvedReferences
|
|
next_events = self.get_events().output
|
|
next_events = sorted(
|
|
[
|
|
event_class(
|
|
latitude=self.latitude,
|
|
longitude=self.longitude,
|
|
time=next_events[attr],
|
|
)
|
|
for attr, event_class in self._attr_to_event_class.items()
|
|
if next_events.get(attr)
|
|
],
|
|
key=lambda t: t.time,
|
|
)
|
|
|
|
for event in next_events:
|
|
# noinspection PyTypeChecker
|
|
dt = datetime.datetime.fromisoformat(event.time)
|
|
while (not self.should_stop()) and (
|
|
dt > datetime.datetime.now(tz=gettz())
|
|
):
|
|
time.sleep(1)
|
|
|
|
if dt <= datetime.datetime.now(tz=gettz()):
|
|
self.bus.post(event)
|
|
|
|
@staticmethod
|
|
def _convert_time(t: str) -> datetime.datetime:
|
|
now = datetime.datetime.now().replace(
|
|
tzinfo=gettz()
|
|
) # lgtm [py/call-to-non-callable]
|
|
dt = datetime.datetime.strptime(t, '%H:%M:%S %p')
|
|
dt = datetime.datetime(
|
|
year=now.year,
|
|
month=now.month,
|
|
day=now.day,
|
|
hour=dt.hour,
|
|
minute=dt.minute,
|
|
second=dt.second,
|
|
tzinfo=tzutc(),
|
|
)
|
|
|
|
if dt < now:
|
|
dt += datetime.timedelta(days=1)
|
|
return datetime.datetime.fromtimestamp(dt.timestamp(), tz=gettz())
|
|
|
|
@action
|
|
def get_events(
|
|
self, latitude: Optional[float] = None, longitude: Optional[float] = None
|
|
) -> dict:
|
|
"""
|
|
Return the next sun events.
|
|
|
|
:param latitude: Default latitude override.
|
|
:param longitude: Default longitude override.
|
|
:return: .. schema:: sun.SunEventsSchema
|
|
"""
|
|
response = (
|
|
requests.get(
|
|
self._base_url,
|
|
params={
|
|
'lat': latitude or self.latitude,
|
|
'lng': longitude or self.longitude,
|
|
},
|
|
)
|
|
.json()
|
|
.get('results', {})
|
|
)
|
|
|
|
schema = SunEventsSchema()
|
|
return schema.dump(
|
|
{
|
|
attr: self._convert_time(t)
|
|
for attr, t in response.items()
|
|
if attr in schema.declared_fields
|
|
}
|
|
)
|