edward/agenda
1
0
Fork 0
Code Issues 27 Pull requests Packages Projects Releases Wiki Activity

Merge branch 'main' of https://git.4angle.com/edward/agenda

Browse source
This commit is contained in:
Edward Betts 2023-11-06 18:57:05 +00:00
parent 3760b6ac9e 13a959711d
commit 943d45bd27
5 changed files with 184 additions and 138 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

153
agenda/__init__.py
View file

@ -1,24 +1,18 @@
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
from datetime import date, datetime, time, timedelta
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 (
@ -29,9 +23,12 @@ from . import (
fx,
gwr,
markets,
stock_market,
subscription,
sun,
thespacedevs,
travel,
uk_holiday,
waste_schedule,
)
from .types import Event
@ -62,51 +59,6 @@ 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]:
@ -119,29 +71,6 @@ def timezone_transition(
]
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
@ -158,60 +87,6 @@ def uk_financial_year_end(input_date: date) -> date:
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] = []
@ -272,7 +147,7 @@ def as_date(d: date | datetime) -> date:
def get_accommodation(filepath: str) -> list[Event]:
"""Get birthdays from config."""
"""Get accomodation from config."""
with open(filepath) as f:
return [
Event(
@ -330,7 +205,7 @@ async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
) = await asyncio.gather(
fx.get_gbpusd(config),
gwr.advance_ticket_date(data_dir),
get_next_bank_holiday(last_year, next_year),
uk_holiday.bank_holiday_list(last_year, next_year, data_dir),
thespacedevs.get_launches(rocket_dir, limit=40),
waste_collection_events(),
bristol_waste_collection_events(today),
@ -343,15 +218,15 @@ async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
"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(),
"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),
"mothers_day": uk_holiday.get_mothers_day(today),
"fathers_day": uk_holiday.get_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,
@ -381,7 +256,13 @@ async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
reply["sunset"] = sun.sunset(observer)
for key, value in xmas_last_posting_dates.items():
events.append(Event(name=f"xmas_last_{key}", date=value))
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"))
@ -390,6 +271,8 @@ async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
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",

63
agenda/stock_market.py Normal file
View file

@ -0,0 +1,63 @@
"""Stock market open and close times."""
from datetime import timedelta, timezone
import dateutil.tz
import exchange_calendars
import pandas
here = dateutil.tz.tzlocal()
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 open_and_close() -> 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

21
agenda/travel.py
View file

@ -38,6 +38,25 @@ def get(
]
def get_trains(from_date: date, filepath: str) -> list[Event]:
"""Get train events."""
events: list[Event] = []
for item in yaml.safe_load(open(filepath)):
if item["depart"].date() < from_date:
continue
events += [
Event(
date=leg["depart"],
end_date=leg["arrive"],
name="transport",
title=f'train from {leg["from"]} to {leg["to"]}',
url=item.get("url"),
)
for leg in item["legs"]
]
return events
def flight_number(flight: Leg) -> str:
"""Flight number."""
airline_code = flight["airline"]
@ -49,7 +68,7 @@ def flight_number(flight: Leg) -> str:
def all_events(from_date: date, data_dir: str) -> list[Event]:
"""Get all flights and rail journeys."""
trains = get(from_date, "train", os.path.join(data_dir, "trains.yaml"))
trains = get_trains(from_date, os.path.join(data_dir, "trains.yaml"))
flights = get(
from_date, "flight", os.path.join(data_dir, "flights.yaml"), extra=flight_number
)

81
agenda/uk_holiday.py Normal file
View file

@ -0,0 +1,81 @@
"""UK holidays."""
import json
import os
from datetime import date, datetime, timedelta
from time import time
import httpx
from dateutil.easter import easter
from .types import Event
async def bank_holiday_list(
start_date: date, end_date: date, data_dir: str
) -> 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 (time() - 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 get_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 get_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

4
agenda/waste_schedule.py
View file

@ -13,7 +13,7 @@ import pytz
from .types import Event
ttl_hours = 6
ttl_hours = 12
uk_tz = pytz.timezone("Europe/London")
@ -155,7 +155,7 @@ async def get_bristol_gov_uk_data(uprn: str) -> httpx.Response:
_uprn = str(uprn).zfill(12)
async with httpx.AsyncClient(timeout=10) as client:
async with httpx.AsyncClient(timeout=20) as client:
# Initialise form
payload = {"servicetypeid": "7dce896c-b3ba-ea11-a812-000d3a7f1cdc"}
response = await client.get(