264 lines
7.6 KiB
Python
264 lines
7.6 KiB
Python
"""Agenda data."""
|
||
|
||
import asyncio
|
||
import configparser
|
||
import operator
|
||
import os
|
||
import typing
|
||
from datetime import date, datetime, timedelta
|
||
|
||
import dateutil.rrule
|
||
import dateutil.tz
|
||
import holidays # type: ignore
|
||
import isodate
|
||
import lxml
|
||
import pytz
|
||
import yaml
|
||
|
||
from . import (
|
||
accommodation,
|
||
birthday,
|
||
calendar,
|
||
conference,
|
||
economist,
|
||
fx,
|
||
gwr,
|
||
stock_market,
|
||
subscription,
|
||
sun,
|
||
thespacedevs,
|
||
travel,
|
||
uk_holiday,
|
||
uk_midnight,
|
||
waste_schedule,
|
||
)
|
||
from .types import Event
|
||
|
||
here = dateutil.tz.tzlocal()
|
||
|
||
# deadline to file tax return
|
||
# credit card expiry dates
|
||
# morzine ski lifts
|
||
# chalet availablity calendar
|
||
|
||
# starlink visible
|
||
|
||
|
||
def timezone_transition(
|
||
start: datetime, end: datetime, key: str, tz_name: str
|
||
) -> list[Event]:
|
||
"""Clocks changes."""
|
||
tz = pytz.timezone(tz_name)
|
||
return [
|
||
Event(name=key, date=pytz.utc.localize(t).astimezone(tz))
|
||
for t in tz._utc_transition_times # type: ignore
|
||
if start <= t <= end
|
||
]
|
||
|
||
|
||
def get_us_holidays(start_date: date, end_date: date) -> list[Event]:
|
||
"""Date and name of next US holiday."""
|
||
found: list[Event] = []
|
||
for year in range(start_date.year, end_date.year + 1):
|
||
hols = holidays.UnitedStates(years=year)
|
||
found += [
|
||
Event(name="us_holiday", date=hol_date, title=title.replace("'", "’"))
|
||
for hol_date, title in hols.items()
|
||
if start_date < hol_date < end_date
|
||
]
|
||
|
||
extra = []
|
||
for e in found:
|
||
if e.title != "Thanksgiving":
|
||
continue
|
||
extra += [
|
||
Event(
|
||
name="us_holiday", date=e.date + timedelta(days=1), title="Black Friday"
|
||
),
|
||
Event(
|
||
name="us_holiday", date=e.date + timedelta(days=4), title="Cyber Monday"
|
||
),
|
||
]
|
||
|
||
return found + extra
|
||
|
||
|
||
def dates_from_rrule(
|
||
rrule: str, start: date, end: date
|
||
) -> typing.Sequence[datetime | date]:
|
||
"""Generate events from an RRULE between start_date and end_date."""
|
||
all_day = not any(param in rrule for param in ["BYHOUR", "BYMINUTE", "BYSECOND"])
|
||
|
||
return [
|
||
i.date() if all_day else i
|
||
for i in dateutil.rrule.rrulestr(rrule, dtstart=uk_midnight(start)).between(
|
||
uk_midnight(start), uk_midnight(end)
|
||
)
|
||
]
|
||
|
||
|
||
async def waste_collection_events(data_dir: str) -> list[Event]:
|
||
"""Waste colllection events."""
|
||
postcode = "BS48 3HG"
|
||
uprn = "24071046"
|
||
|
||
html = await waste_schedule.get_html(data_dir, postcode, uprn)
|
||
root = lxml.html.fromstring(html)
|
||
events = waste_schedule.parse(root)
|
||
return events
|
||
|
||
|
||
async def bristol_waste_collection_events(
|
||
data_dir: str, start_date: date
|
||
) -> list[Event]:
|
||
"""Waste colllection events."""
|
||
uprn = "358335"
|
||
|
||
return await waste_schedule.get_bristol_gov_uk(start_date, data_dir, uprn)
|
||
|
||
|
||
def combine_holidays(events: list[Event]) -> list[Event]:
|
||
"""Combine UK and US holidays with the same date and title."""
|
||
combined: dict[tuple[date, str], Event] = {}
|
||
|
||
for e in events:
|
||
assert isinstance(e.title, str) and isinstance(e.date, date)
|
||
event_key = (e.date, e.title)
|
||
combined[event_key] = (
|
||
Event(name="bank_holiday", date=e.date, title=e.title + " (UK & US)")
|
||
if event_key in combined
|
||
else e
|
||
)
|
||
|
||
return list(combined.values())
|
||
|
||
|
||
def get_yaml_event_date_field(item: dict[str, str]) -> str:
|
||
"""Event date field name."""
|
||
return (
|
||
"end_date"
|
||
if item["name"] == "travel_insurance"
|
||
else ("start_date" if "start_date" in item else "date")
|
||
)
|
||
|
||
|
||
def get_yaml_event_end_date_field(item: dict[str, str]) -> str:
|
||
"""Event date field name."""
|
||
return (
|
||
"end_date"
|
||
if item["name"] == "travel_insurance"
|
||
else ("start_date" if "start_date" in item else "date")
|
||
)
|
||
|
||
|
||
def read_events_yaml(data_dir: str, start: date, end: date) -> list[Event]:
|
||
"""Read eventes from YAML file."""
|
||
events: list[Event] = []
|
||
for item in yaml.safe_load(open(os.path.join(data_dir, "events.yaml"))):
|
||
duration = (
|
||
isodate.parse_duration(item["duration"]) if "duration" in item else None
|
||
)
|
||
dates = (
|
||
dates_from_rrule(item["rrule"], start, end)
|
||
if "rrule" in item
|
||
else [item[get_yaml_event_date_field(item)]]
|
||
)
|
||
for dt in dates:
|
||
e = Event(
|
||
name=item["name"],
|
||
date=dt,
|
||
end_date=(
|
||
dt + duration
|
||
if duration
|
||
else (
|
||
item.get("end_date")
|
||
if item["name"] != "travel_insurance"
|
||
else None
|
||
)
|
||
),
|
||
title=item.get("title"),
|
||
url=item.get("url"),
|
||
)
|
||
events.append(e)
|
||
return events
|
||
|
||
|
||
async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
|
||
"""Get data to display on agenda dashboard."""
|
||
config_filename = os.path.join(os.path.dirname(__file__), "..", "config")
|
||
|
||
assert os.path.exists(config_filename)
|
||
|
||
config = configparser.ConfigParser()
|
||
config.read(config_filename)
|
||
|
||
data_dir = config.get("data", "dir")
|
||
|
||
rocket_dir = os.path.join(data_dir, "thespacedevs")
|
||
today = now.date()
|
||
two_weeks_ago = today - timedelta(weeks=2)
|
||
last_week = today - timedelta(weeks=1)
|
||
last_year = today - timedelta(days=365)
|
||
next_year = today + timedelta(days=365)
|
||
|
||
minus_365 = now - timedelta(days=365)
|
||
plus_365 = now + timedelta(days=365)
|
||
|
||
(
|
||
gbpusd,
|
||
gwr_advance_tickets,
|
||
bank_holiday,
|
||
rockets,
|
||
backwell_bins,
|
||
bristol_bins,
|
||
) = await asyncio.gather(
|
||
fx.get_gbpusd(config),
|
||
gwr.advance_ticket_date(data_dir),
|
||
uk_holiday.bank_holiday_list(last_year, next_year, data_dir),
|
||
thespacedevs.get_launches(rocket_dir, limit=40),
|
||
waste_collection_events(data_dir),
|
||
bristol_waste_collection_events(data_dir, today),
|
||
)
|
||
|
||
reply = {
|
||
"now": now,
|
||
"gbpusd": gbpusd,
|
||
"stock_markets": stock_market.open_and_close(),
|
||
"rockets": rockets,
|
||
}
|
||
|
||
my_data = config["data"]["personal-data"]
|
||
events = (
|
||
[
|
||
Event(name="gwr_advance_tickets", date=gwr_advance_tickets),
|
||
Event(name="mothers_day", date=uk_holiday.get_mothers_day(today)),
|
||
]
|
||
+ timezone_transition(minus_365, plus_365, "uk_clock_change", "Europe/London")
|
||
+ timezone_transition(
|
||
minus_365, plus_365, "us_clock_change", "America/New_York"
|
||
)
|
||
)
|
||
|
||
events += combine_holidays(bank_holiday + get_us_holidays(last_year, next_year))
|
||
events += birthday.get_birthdays(last_year, os.path.join(my_data, "entities.yaml"))
|
||
events += accommodation.get_events(os.path.join(my_data, "accommodation.yaml"))
|
||
events += travel.all_events(config["data"]["personal-data"])
|
||
events += conference.get_list(os.path.join(my_data, "conferences.yaml"))
|
||
events += backwell_bins + bristol_bins
|
||
events += read_events_yaml(my_data, last_year, next_year)
|
||
events += subscription.get_events(os.path.join(my_data, "subscriptions.yaml"))
|
||
events += economist.publication_dates(last_week, next_year)
|
||
|
||
events.sort(key=operator.attrgetter("as_datetime"))
|
||
|
||
observer = sun.bristol()
|
||
reply["sunrise"] = sun.sunrise(observer)
|
||
reply["sunset"] = sun.sunset(observer)
|
||
reply["events"] = events
|
||
reply["last_week"] = last_week
|
||
reply["two_weeks_ago"] = two_weeks_ago
|
||
|
||
reply["fullcalendar_events"] = calendar.build_events(events)
|
||
|
||
return reply
|