405 lines
12 KiB
Python
405 lines
12 KiB
Python
import asyncio
|
|
import configparser
|
|
import json
|
|
import operator
|
|
import os
|
|
import typing
|
|
import warnings
|
|
from datetime import date, datetime, time, timedelta, timezone
|
|
from time import time as unixtime
|
|
|
|
import dateutil
|
|
import dateutil.parser
|
|
import dateutil.tz
|
|
import exchange_calendars
|
|
import holidays
|
|
import httpx
|
|
import lxml
|
|
import pandas
|
|
import pytz
|
|
import yaml
|
|
from dateutil.easter import easter
|
|
from dateutil.relativedelta import FR, relativedelta
|
|
|
|
from . import (
|
|
birthday,
|
|
calendar,
|
|
conference,
|
|
economist,
|
|
fx,
|
|
gwr,
|
|
markets,
|
|
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(input_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)
|
|
|
|
year_later = input_date + timedelta(days=365)
|
|
|
|
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 < input_date:
|
|
continue
|
|
if event_date > year_later:
|
|
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 timedelta_display(delta: timedelta) -> str:
|
|
"""Format timedelta as a human readable string."""
|
|
total_seconds = int(delta.total_seconds())
|
|
days, remainder = divmod(total_seconds, 24 * 60 * 60)
|
|
hours, remainder = divmod(remainder, 60 * 60)
|
|
mins, secs = divmod(remainder, 60)
|
|
|
|
return " ".join(
|
|
f"{v:>3} {label}"
|
|
for v, label in ((days, "days"), (hours, "hrs"), (mins, "mins"))
|
|
if v
|
|
)
|
|
|
|
|
|
def stock_markets() -> list[str]:
|
|
"""Stock markets open and close times."""
|
|
# The trading calendars code is slow, maybe there is a faster way to do this
|
|
# Or we could cache the result
|
|
now = pandas.Timestamp.now(timezone.utc)
|
|
now_local = pandas.Timestamp.now(here)
|
|
markets = [
|
|
("XLON", "London"),
|
|
("XNYS", "US"),
|
|
]
|
|
reply = []
|
|
for code, label in markets:
|
|
cal = exchange_calendars.get_calendar(code)
|
|
|
|
if cal.is_open_on_minute(now_local):
|
|
next_close = cal.next_close(now).tz_convert(here)
|
|
next_close = next_close.replace(minute=round(next_close.minute, -1))
|
|
delta_close = timedelta_display(next_close - now_local)
|
|
|
|
prev_open = cal.previous_open(now).tz_convert(here)
|
|
prev_open = prev_open.replace(minute=round(prev_open.minute, -1))
|
|
delta_open = timedelta_display(now_local - prev_open)
|
|
|
|
msg = (
|
|
f"{label:>6} market opened {delta_open} ago, "
|
|
+ f"closes in {delta_close} ({next_close:%H:%M})"
|
|
)
|
|
else:
|
|
ts = cal.next_open(now)
|
|
ts = ts.replace(minute=round(ts.minute, -1))
|
|
ts = ts.tz_convert(here)
|
|
delta = timedelta_display(ts - now_local)
|
|
msg = f"{label:>6} market opens in {delta}" + (
|
|
f" ({ts:%H:%M})" if (ts - now_local) < timedelta(days=1) else ""
|
|
)
|
|
|
|
reply.append(msg)
|
|
return reply
|
|
|
|
|
|
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 birthdays 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)
|
|
]
|
|
|
|
|
|
def waste_collection_events() -> list[Event]:
|
|
"""Waste colllection events."""
|
|
postcode = "BS48 3HG"
|
|
uprn = "24071046"
|
|
|
|
html = waste_schedule.get_html(data_dir, postcode, uprn)
|
|
root = lxml.html.fromstring(html)
|
|
events = waste_schedule.parse(root)
|
|
return events
|
|
|
|
|
|
def bristol_waste_collection_events(start_date: date) -> list[Event]:
|
|
"""Waste colllection events."""
|
|
uprn = "358335"
|
|
|
|
return 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()
|
|
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,
|
|
) = await asyncio.gather(
|
|
fx.get_gbpusd(config),
|
|
gwr.advance_ticket_date(data_dir),
|
|
get_next_bank_holiday(today),
|
|
thespacedevs.get_launches(rocket_dir, limit=40),
|
|
)
|
|
|
|
reply = {
|
|
"now": now,
|
|
"gbpusd": gbpusd,
|
|
"next_economist": economist.next_pub_date(today),
|
|
"bank_holiday": bank_holiday,
|
|
"us_holiday": get_us_holidays(last_year, next_year),
|
|
"next_us_presidential_election": next_us_presidential_election,
|
|
"stock_markets": stock_markets(),
|
|
"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))
|
|
|
|
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 += waste_collection_events() + bristol_waste_collection_events(today)
|
|
|
|
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["fullcalendar_events"] = calendar.build_events(events)
|
|
|
|
return reply
|