Reorganise code

This commit is contained in:
Edward Betts 2023-11-07 16:55:05 +01:00
parent 943d45bd27
commit ac456ce5be
6 changed files with 302 additions and 308 deletions

View file

@ -1,291 +1,12 @@
import asyncio
import configparser
import operator
import os
import typing
import warnings
from datetime import date, datetime, time, timedelta
"""Agenda functions."""
from datetime import date, datetime, time
import dateutil
import dateutil.parser
import dateutil.tz
import holidays
import lxml
import pytz
import yaml
from dateutil.relativedelta import FR, relativedelta
from . import (
birthday,
calendar,
conference,
economist,
fx,
gwr,
markets,
stock_market,
subscription,
sun,
thespacedevs,
travel,
uk_holiday,
waste_schedule,
)
from .types import Event
warnings.simplefilter(action="ignore", category=FutureWarning)
uk_tz = pytz.timezone("Europe/London")
# 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 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
]
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),
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),
)
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": 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,
"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
def uk_time(d: date, t: time) -> datetime:
"""Combine time and date for UK timezone."""
return uk_tz.localize(datetime.combine(d, t))

281
agenda/data.py Normal file
View file

@ -0,0 +1,281 @@
"""Agenda data."""
import asyncio
import configparser
import operator
import os
import typing
from datetime import date, datetime, time, timedelta
import dateutil.tz
import holidays
import lxml
import pytz
from dateutil.relativedelta import FR, relativedelta
from . import (
accommodation,
birthday,
calendar,
conference,
economist,
fx,
gwr,
markets,
stock_market,
subscription,
sun,
thespacedevs,
travel,
uk_holiday,
uk_time,
waste_schedule,
)
from .types import Event
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)}
# deadline to file tax return
# credit card expiry dates
# morzine ski lifts
# chalet availablity calendar
# starlink visible
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
]
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 uk_financial_year_ends(start_date: date, end_date: date) -> list[Event]:
"""Generate a list of UK financial year end dates (April 5th) between two dates."""
# Initialize an empty list to store the financial year ends
financial_year_ends: list[date] = []
# Start from the year of the start date
year = start_date.year
# Calculate the first possible financial year end date
financial_year_end = date(year, 4, 5)
# Loop over the years until we reach the year of the end date
while financial_year_end < end_date:
# If the financial year end date is on or after the start date,
# add it to the list
if financial_year_end >= start_date:
financial_year_ends.append(financial_year_end)
year += 1 # Move to the next year
financial_year_end = date(year, 4, 5)
return [Event(name="uk_financial_year_end", date=d) for d in financial_year_ends]
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, end_date: date) -> list[Event]:
"""Future dates for Critical Mass."""
events: list[Event] = []
current_date = start_date
t = time(18, 0)
while current_date < end_date:
# Calculate the last Friday of the current month
last_friday = current_date + relativedelta(day=31, weekday=FR(-1))
events.append(Event(name="critical_mass", date=uk_time(last_friday, t)))
# Move to the next month
current_date += relativedelta(months=1)
return events
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)
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,
"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": 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,
"critical_mass": critical_mass(last_year, next_year),
"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 += accommodation.get_events(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

View file

@ -1,18 +1,10 @@
"""Next publication date of the Economist."""
from datetime import date, datetime, time, timedelta
import pytz
from datetime import date, time, timedelta
from . import uk_time
from .types import Event
uk_tz = pytz.timezone("Europe/London")
def add_pub_time(pub_date: date) -> datetime:
"""Publication time is 19:00."""
return uk_tz.localize(datetime.combine(pub_date, time(19, 0)))
def publication_dates(start_date: date, end_date: date) -> list[Event]:
"""List of Economist publication dates."""
@ -23,13 +15,14 @@ def publication_dates(start_date: date, end_date: date) -> list[Event]:
current_date = start_date
publication_dates = []
t = time(19, 0)
while current_date <= end_date:
if (
current_date.weekday() == publication_day
and current_date.isocalendar().week not in non_publication_weeks
):
publication_dates.append(add_pub_time(current_date))
publication_dates.append(uk_time(current_date, t))
current_date += timedelta(days=1)
return [Event(name="economist", date=pub_date) for pub_date in publication_dates]

View file

@ -1,10 +1,11 @@
"""Market days."""
from datetime import date, datetime, time, timedelta
from datetime import date, time, timedelta
import pytz
from dateutil.relativedelta import SA, relativedelta
from . import uk_time
from .types import Event
uk_tz = pytz.timezone("Europe/London")
@ -15,8 +16,8 @@ def event(title: str, d: date, start: time, end: time, url: str) -> Event:
return Event(
name="market",
title=title,
date=uk_tz.localize(datetime.combine(d, start)),
end_date=uk_tz.localize(datetime.combine(d, end)),
date=uk_time(d, start),
end_date=uk_time(d, end),
url=url,
)

View file

@ -5,18 +5,16 @@ import os
import re
import typing
from collections import defaultdict
from datetime import date, datetime, timedelta, time
from datetime import date, datetime, time, timedelta
import httpx
import lxml.html
import pytz
from . import uk_time
from .types import Event
ttl_hours = 12
uk_tz = pytz.timezone("Europe/London")
def make_waste_dir(data_dir: str) -> None:
"""Make waste dir if missing."""
@ -96,7 +94,7 @@ def parse(root: lxml.html.HtmlElement) -> list[Event]:
return [
Event(
name="waste_schedule",
date=uk_tz.localize(datetime.combine(d, time(6, 30))),
date=uk_time(d, time(6, 30)),
title="Backwell: " + ", ".join(services),
)
for d, services in by_date.items()

View file

@ -11,10 +11,10 @@ import flask
import werkzeug
import werkzeug.debug.tbtools
from agenda import get_data
import agenda.data
app = flask.Flask(__name__)
app.debug = True
app.debug = False
@app.errorhandler(werkzeug.exceptions.InternalServerError)
@ -48,7 +48,7 @@ def exception_handler(e: werkzeug.exceptions.InternalServerError) -> tuple[str,
async def index() -> str:
"""Index page."""
now = datetime.now()
data = await get_data(now)
data = await agenda.data.get_data(now)
return flask.render_template("index.html", today=now.date(), **data)