No need to maintain two different pieces of logic - a `utcnow()` for Python < 3.11 and `now(datetime.UTC)` for Python >= 3.11. `datetime.timezone.utc` existed long before datetime.UTC and that's what the `utcnow` facade should use. This means that all the `utcnow()` will always have `tzinfo=UTC` regardless of the Python version. There's still a problem with the `utcnow()`-generated timestamps that have been generated by previous versions of Python and stored on the db. Therefore, when the code performs comparisons with timestamps fetched from the db, it should always explicitly do a `.replace(tzinfo=utc)` to ensure that we always compare offset-aware datetime representations. See blog post for technical details: https://manganiello.blog/wheres-my-time-again
113 lines
3.4 KiB
Python
113 lines
3.4 KiB
Python
import datetime
|
|
from typing import Dict, Optional
|
|
|
|
import requests
|
|
from dateutil.tz import gettz
|
|
|
|
from platypush.message.event.sun import SunriseEvent, SunsetEvent
|
|
from platypush.plugins import RunnablePlugin, action
|
|
from platypush.schemas.sun import SunEventsSchema
|
|
from platypush.utils import utcnow
|
|
|
|
|
|
class SunPlugin(RunnablePlugin):
|
|
"""
|
|
Plugin to get sunset/sunrise events and info for a certain location.
|
|
"""
|
|
|
|
_base_url = 'https://api.sunrise-sunset.org/json'
|
|
_schema = SunEventsSchema()
|
|
_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():
|
|
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,
|
|
)
|
|
),
|
|
None,
|
|
)
|
|
|
|
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)
|
|
|
|
if not self.should_stop():
|
|
self._bus.post(next_event)
|
|
self.wait_stop(2)
|
|
|
|
@staticmethod
|
|
def _convert_time(t: str) -> datetime.datetime:
|
|
now = utcnow().replace(microsecond=0)
|
|
dt = datetime.datetime.strptime(
|
|
f'{now.year}-{now.month:02d}-{now.day:02d} {t}',
|
|
'%Y-%m-%d %I:%M:%S %p',
|
|
).replace(tzinfo=datetime.timezone.utc)
|
|
|
|
if dt < now:
|
|
dt += datetime.timedelta(days=1)
|
|
return dt
|
|
|
|
def _get_events(
|
|
self, latitude: Optional[float] = None, longitude: Optional[float] = None
|
|
) -> Dict[str, datetime.datetime]:
|
|
response = (
|
|
requests.get(
|
|
self._base_url,
|
|
timeout=10,
|
|
params={
|
|
'lat': latitude or self.latitude,
|
|
'lng': longitude or self.longitude,
|
|
},
|
|
)
|
|
.json()
|
|
.get('results', {})
|
|
)
|
|
|
|
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
|
|
"""
|
|
schema = SunEventsSchema()
|
|
return dict(
|
|
schema.dump(self._get_events(latitude=latitude, longitude=longitude))
|
|
)
|