diff --git a/agenda/__init__.py b/agenda/__init__.py index e23263c..baf8813 100644 --- a/agenda/__init__.py +++ b/agenda/__init__.py @@ -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)) diff --git a/agenda/data.py b/agenda/data.py new file mode 100644 index 0000000..de41eaf --- /dev/null +++ b/agenda/data.py @@ -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 diff --git a/agenda/economist.py b/agenda/economist.py index 99acdd8..d843303 100644 --- a/agenda/economist.py +++ b/agenda/economist.py @@ -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] diff --git a/agenda/markets.py b/agenda/markets.py index e8e48a7..3b05495 100644 --- a/agenda/markets.py +++ b/agenda/markets.py @@ -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, ) diff --git a/agenda/waste_schedule.py b/agenda/waste_schedule.py index 61603b0..7e5e9de 100644 --- a/agenda/waste_schedule.py +++ b/agenda/waste_schedule.py @@ -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() diff --git a/web_view.py b/web_view.py index 4d430b7..d7cbad3 100755 --- a/web_view.py +++ b/web_view.py @@ -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)