2021-07-22 01:02:15 +02:00
|
|
|
import datetime
|
2024-05-20 02:01:40 +02:00
|
|
|
from typing import Dict, Optional
|
2021-07-22 01:02:15 +02:00
|
|
|
|
|
|
|
import requests
|
2024-05-20 02:01:40 +02:00
|
|
|
from dateutil.tz import gettz
|
2021-07-22 01:02:15 +02:00
|
|
|
|
|
|
|
from platypush.message.event.sun import SunriseEvent, SunsetEvent
|
|
|
|
from platypush.plugins import RunnablePlugin, action
|
|
|
|
from platypush.schemas.sun import SunEventsSchema
|
2024-05-31 19:52:32 +02:00
|
|
|
from platypush.utils import utcnow
|
2021-07-22 01:02:15 +02:00
|
|
|
|
|
|
|
|
|
|
|
class SunPlugin(RunnablePlugin):
|
|
|
|
"""
|
|
|
|
Plugin to get sunset/sunrise events and info for a certain location.
|
|
|
|
"""
|
2023-09-24 16:54:43 +02:00
|
|
|
|
2021-07-22 01:02:15 +02:00
|
|
|
_base_url = 'https://api.sunrise-sunset.org/json'
|
2024-05-20 02:01:40 +02:00
|
|
|
_schema = SunEventsSchema()
|
2021-07-22 01:02:15 +02:00
|
|
|
_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():
|
2024-05-20 02:01:40 +02:00
|
|
|
next_events = self._get_events()
|
|
|
|
next_event = next(
|
|
|
|
iter(
|
|
|
|
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,
|
2023-09-24 16:54:43 +02:00
|
|
|
)
|
2024-05-20 02:01:40 +02:00
|
|
|
),
|
|
|
|
None,
|
2023-09-24 16:54:43 +02:00
|
|
|
)
|
2021-07-22 01:02:15 +02:00
|
|
|
|
2024-05-20 02:01:40 +02:00
|
|
|
assert next_event is not None, 'No next event found'
|
|
|
|
wait_secs = max(
|
|
|
|
0, (next_event.time - datetime.datetime.now(tz=gettz())).seconds
|
|
|
|
)
|
|
|
|
self.wait_stop(wait_secs)
|
2021-07-22 01:02:15 +02:00
|
|
|
|
2024-05-20 02:01:40 +02:00
|
|
|
if not self.should_stop():
|
|
|
|
self._bus.post(next_event)
|
|
|
|
self.wait_stop(2)
|
2021-07-22 01:02:15 +02:00
|
|
|
|
|
|
|
@staticmethod
|
|
|
|
def _convert_time(t: str) -> datetime.datetime:
|
2024-05-31 19:52:32 +02:00
|
|
|
now = utcnow().replace(microsecond=0)
|
2024-05-20 02:01:40 +02:00
|
|
|
dt = datetime.datetime.strptime(
|
|
|
|
f'{now.year}-{now.month:02d}-{now.day:02d} {t}',
|
|
|
|
'%Y-%m-%d %I:%M:%S %p',
|
2024-06-01 01:34:47 +02:00
|
|
|
).replace(tzinfo=datetime.timezone.utc)
|
2021-07-22 01:02:15 +02:00
|
|
|
|
|
|
|
if dt < now:
|
|
|
|
dt += datetime.timedelta(days=1)
|
2024-05-20 02:01:40 +02:00
|
|
|
return dt
|
2021-07-22 01:02:15 +02:00
|
|
|
|
2024-05-20 02:01:40 +02:00
|
|
|
def _get_events(
|
2023-09-24 16:54:43 +02:00
|
|
|
self, latitude: Optional[float] = None, longitude: Optional[float] = None
|
2024-05-20 02:01:40 +02:00
|
|
|
) -> Dict[str, datetime.datetime]:
|
2023-09-24 16:54:43 +02:00
|
|
|
response = (
|
|
|
|
requests.get(
|
|
|
|
self._base_url,
|
2024-05-20 02:01:40 +02:00
|
|
|
timeout=10,
|
2023-09-24 16:54:43 +02:00
|
|
|
params={
|
|
|
|
'lat': latitude or self.latitude,
|
|
|
|
'lng': longitude or self.longitude,
|
|
|
|
},
|
|
|
|
)
|
|
|
|
.json()
|
|
|
|
.get('results', {})
|
|
|
|
)
|
2021-07-22 01:02:15 +02:00
|
|
|
|
2024-05-20 02:01:40 +02:00
|
|
|
return {
|
|
|
|
attr: self._convert_time(t)
|
|
|
|
for attr, t in response.items()
|
|
|
|
if attr in self._schema.declared_fields
|
|
|
|
}
|
|
|
|
|
|
|
|
@action
|
|
|
|
def get_events(
|
|
|
|
self, latitude: Optional[float] = None, longitude: Optional[float] = None
|
|
|
|
) -> dict:
|
|
|
|
"""
|
|
|
|
Return the next sun events.
|
|
|
|
|
|
|
|
:param latitude: Override the default latitude.
|
|
|
|
:param longitude: Override the default longitude.
|
|
|
|
:return: .. schema:: sun.SunEventsSchema
|
|
|
|
"""
|
2021-07-22 01:02:15 +02:00
|
|
|
schema = SunEventsSchema()
|
2024-05-20 02:01:40 +02:00
|
|
|
return dict(
|
|
|
|
schema.dump(self._get_events(latitude=latitude, longitude=longitude))
|
2023-09-24 16:54:43 +02:00
|
|
|
)
|