agenda/agenda/__init__.py

363 lines
11 KiB
Python

import asyncio
import configparser
import json
import operator
import os
import typing
import warnings
from datetime import date, datetime, time, timedelta
from time import time as unixtime
import dateutil
import dateutil.parser
import dateutil.tz
import holidays
import httpx
import lxml
import pytz
import yaml
from dateutil.easter import easter
from dateutil.relativedelta import FR, relativedelta
from . import (
birthday,
calendar,
conference,
economist,
fx,
gwr,
markets,
stock_market,
subscription,
sun,
thespacedevs,
travel,
waste_schedule,
)
from .types import Event
warnings.simplefilter(action="ignore", category=FutureWarning)
# deadline to file tax return
# credit card expiry dates
# morzine ski lifts
# chalet availablity calendar
# starlink visible
here = dateutil.tz.tzlocal()
next_us_presidential_election = date(2024, 11, 5)
xmas_last_posting_dates = {"first": date(2023, 12, 20), "second": date(2023, 12, 18)}
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")
def next_uk_mothers_day(input_date: date) -> date:
"""Calculate the date of the next UK Mother's Day from the current date."""
current_year = input_date.year
easter_date = easter(current_year)
# Calculate the date of Mother's Day, which is the fourth Sunday of Lent
mothers_day = easter_date + timedelta(weeks=3)
# Check if Mother's Day has already passed this year
if input_date > mothers_day:
# If it has passed, calculate for the next year
easter_date = easter(current_year + 1)
mothers_day = easter_date + timedelta(weeks=3)
return mothers_day
def next_uk_fathers_day(input_date: date) -> date:
"""Calculate the date of the next UK Father's Day from the current date."""
# Get the current date
# Calculate the day of the week for the current date (0 = Monday, 6 = Sunday)
current_day_of_week = input_date.weekday()
# Calculate the number of days until the next Sunday
days_until_sunday = (6 - current_day_of_week) % 7
# Calculate the date of the next Sunday
next_sunday = input_date + timedelta(days=days_until_sunday)
# Calculate the date of Father's Day, which is the third Sunday of June
fathers_day = date(next_sunday.year, 6, 1) + timedelta(
weeks=2, days=next_sunday.weekday()
)
# Check if Father's Day has already passed this year
if input_date > fathers_day:
# If it has passed, calculate for the next year
fathers_day = date(fathers_day.year + 1, 6, 1) + timedelta(
weeks=2, days=next_sunday.weekday()
)
return fathers_day
def timezone_transition(
start_dt: datetime, end_dt: 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_dt <= t <= end_dt
]
async def get_next_bank_holiday(start_date: date, end_date: date) -> list[Event]:
"""Date and name of the next UK bank holiday."""
url = "https://www.gov.uk/bank-holidays.json"
filename = os.path.join(data_dir, "bank-holidays.json")
mtime = os.path.getmtime(filename)
if (unixtime() - mtime) > 60 * 60 * 6: # six hours
async with httpx.AsyncClient() as client:
r = await client.get(url)
open(filename, "w").write(r.text)
events = json.load(open(filename))["england-and-wales"]["events"]
hols: list[Event] = []
for event in events:
event_date = datetime.strptime(event["date"], "%Y-%m-%d").date()
if event_date < start_date:
continue
if event_date > end_date:
break
hols.append(Event(name="bank_holiday", date=event_date, title=event["title"]))
return hols
def uk_financial_year_end(input_date: date) -> date:
"""Next date of the end of the UK financial year, April 5th."""
# Determine the year of the input date
year = input_date.year
# Calculate the UK financial year end date (April 5th)
uk_financial_year_end = date(year, 4, 5)
# If the input date is on or after the UK financial year end date,
# move to the next year
if input_date > uk_financial_year_end:
uk_financial_year_end = date(year + 1, 4, 5)
return uk_financial_year_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)
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 critical_mass(start_date: date, limit: int = 12) -> list[Event]:
"""Future dates for Critical Mass."""
events: list[Event] = []
current_date = start_date
tz = pytz.timezone("Europe/London")
t = time(18, 0)
for _ in range(limit):
# Calculate the last Friday of the current month
last_friday = current_date + relativedelta(day=31, weekday=FR(-1))
# Include it in the list only if it's on or after the start_date
if last_friday >= start_date:
events.append(
Event(
name="critical_mass",
date=tz.localize(datetime.combine(last_friday, t)),
)
)
# Move to the next month
current_date += relativedelta(months=1)
return events
def as_date(d: date | datetime) -> date:
"""Return date for given date or datetime."""
return d.date() if isinstance(d, datetime) else d
def get_accommodation(filepath: str) -> list[Event]:
"""Get accomodation from config."""
with open(filepath) as f:
return [
Event(
date=item["from"],
end_date=item["to"],
name="accommodation",
title=(
f'{item["location"]} Airbnb'
if item["operator"] == "airbnb"
else item["name"]
),
url=item.get("url"),
)
for item in yaml.safe_load(f)
]
async def waste_collection_events() -> 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(start_date: date) -> list[Event]:
"""Waste colllection events."""
uprn = "358335"
return await waste_schedule.get_bristol_gov_uk(start_date, data_dir, uprn)
async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
"""Get data to display on agenda dashboard."""
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),
get_next_bank_holiday(last_year, next_year),
thespacedevs.get_launches(rocket_dir, limit=40),
waste_collection_events(),
bristol_waste_collection_events(today),
)
reply = {
"now": now,
"gbpusd": gbpusd,
"economist": economist.publication_dates(last_year, next_year),
"bank_holiday": bank_holiday,
"us_holiday": get_us_holidays(last_year, next_year),
"next_us_presidential_election": next_us_presidential_election,
"stock_markets": stock_market.open_and_close(),
"uk_clock_change": timezone_transition(
minus_365, plus_365, "uk_clock_change", "Europe/London"
),
"us_clock_change": timezone_transition(
minus_365, plus_365, "us_clock_change", "America/New_York"
),
"mothers_day": next_uk_mothers_day(today),
"fathers_day": next_uk_fathers_day(today),
"uk_financial_year_end": uk_financial_year_end(today),
"xmas_last_posting_dates": xmas_last_posting_dates,
"gwr_advance_tickets": gwr_advance_tickets,
"critical_mass": critical_mass(today),
"market": (
markets.windmill_hill(last_year, next_year)
+ markets.tobacco_factory(last_year, next_year)
+ markets.nailsea_farmers(last_year, next_year)
),
"rockets": rockets,
}
skip = {"now", "gbpusd", "rockets", "stock_markets", "xmas_last_posting_dates"}
events: list[Event] = []
for key, value in reply.items():
if key in skip:
continue
if isinstance(value, list):
events += value
else:
assert isinstance(value, date)
event = Event(name=key, date=value)
events.append(event)
observer = sun.bristol()
reply["sunrise"] = sun.sunrise(observer)
reply["sunset"] = sun.sunset(observer)
for key, value in xmas_last_posting_dates.items():
events.append(
Event(
name=f"xmas_last_{key}",
date=value,
url="https://www.postoffice.co.uk/last-posting-dates",
)
)
my_data = config["data"]["personal-data"]
events += birthday.get_birthdays(last_year, os.path.join(my_data, "entities.yaml"))
events += get_accommodation(os.path.join(my_data, "accommodation.yaml"))
events += travel.all_events(today, config["data"]["personal-data"])
events += conference.get_list(os.path.join(my_data, "conferences.yaml"))
events += backwell_bins + bristol_bins
events += subscription.get_events(os.path.join(my_data, "subscriptions.yaml"))
next_up_series = Event(
date=date(2026, 6, 1),
title="70 Up",
name="next_up_series",
)
events.append(next_up_series)
events.sort(key=operator.attrgetter("as_datetime"))
reply["events"] = events
reply["last_week"] = last_week
reply["two_weeks_ago"] = two_weeks_ago
reply["fullcalendar_events"] = calendar.build_events(events)
return reply