diff --git a/agenda/__init__.py b/agenda/__init__.py index baf8813..ea2b6df 100644 --- a/agenda/__init__.py +++ b/agenda/__init__.py @@ -10,3 +10,8 @@ uk_tz = pytz.timezone("Europe/London") def uk_time(d: date, t: time) -> datetime: """Combine time and date for UK timezone.""" return uk_tz.localize(datetime.combine(d, t)) + + +def uk_midnight(d: date) -> datetime: + """Midnight UK time.""" + return uk_time(d, datetime.min.time()) diff --git a/agenda/data.py b/agenda/data.py index 99c0bbd..31151cc 100644 --- a/agenda/data.py +++ b/agenda/data.py @@ -5,14 +5,14 @@ import configparser import operator import os import typing -from datetime import date, datetime, time, timedelta +from datetime import date, datetime, timedelta +import dateutil.rrule import dateutil.tz -import holidays +import holidays # type: ignore import lxml import pytz import yaml -from dateutil.relativedelta import FR, relativedelta from . import ( accommodation, @@ -29,7 +29,7 @@ from . import ( thespacedevs, travel, uk_holiday, - uk_time, + uk_midnight, waste_schedule, ) from .types import Event @@ -47,41 +47,17 @@ next_us_presidential_election = date(2024, 11, 5) def timezone_transition( - start_dt: datetime, end_dt: datetime, key: str, tz_name: str + start: datetime, end: 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 + if start <= t <= end ] -def uk_financial_year_end(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] = [] @@ -109,22 +85,18 @@ def get_us_holidays(start_date: date, end_date: date) -> list[Event]: 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) +def dates_from_rrule( + rrule: str, start: date, end: date +) -> typing.Sequence[datetime | date]: + """Generate events from an RRULE between start_date and end_date.""" + all_day = not any(param in rrule for param in ["BYHOUR", "BYMINUTE", "BYSECOND"]) - 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 + return [ + i.date() if all_day else i + for i in dateutil.rrule.rrulestr(rrule, dtstart=uk_midnight(start)).between( + uk_midnight(start), uk_midnight(end) + ) + ] async def waste_collection_events(data_dir: str) -> list[Event]: @@ -163,17 +135,24 @@ def combine_holidays(events: list[Event]) -> list[Event]: return list(combined.values()) -def read_events_yaml(data_dir: str) -> list[Event]: +def read_events_yaml(data_dir: str, start: date, end: date) -> list[Event]: """Read eventes from YAML file.""" - return [ - Event( - name=item["name"], - date=item["end_date" if item["name"] == "travel_insurance" else "date"], - title=item.get("title"), - url=item.get("url"), + events: list[Event] = [] + for item in yaml.safe_load(open(os.path.join(data_dir, "events.yaml"))): + dates = ( + dates_from_rrule(item["rrule"], start, end) + if "rrule" in item + else [item["end_date" if item["name"] == "travel_insurance" else "date"]] ) - for item in yaml.safe_load(open(os.path.join(data_dir, "events.yaml"))) - ] + for dt in dates: + e = Event( + name=item["name"], + date=dt, + title=item.get("title"), + url=item.get("url"), + ) + events.append(e) + return events async def get_data(now: datetime) -> typing.Mapping[str, str | object]: @@ -226,10 +205,8 @@ async def get_data(now: datetime) -> typing.Mapping[str, str | object]: 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(last_year, next_year), + # "fathers_day": uk_holiday.get_fathers_day(today), "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) @@ -250,24 +227,21 @@ async def get_data(now: datetime) -> typing.Mapping[str, str | object]: event = Event(name=key, date=value) events.append(event) - observer = sun.bristol() - reply["sunrise"] = sun.sunrise(observer) - reply["sunset"] = sun.sunset(observer) - - events += combine_holidays(bank_holiday + get_us_holidays(last_year, next_year)) - my_data = config["data"]["personal-data"] + events += combine_holidays(bank_holiday + get_us_holidays(last_year, next_year)) 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(config["data"]["personal-data"]) events += conference.get_list(os.path.join(my_data, "conferences.yaml")) events += backwell_bins + bristol_bins - events += read_events_yaml(my_data) - + events += read_events_yaml(my_data, last_year, next_year) events += subscription.get_events(os.path.join(my_data, "subscriptions.yaml")) events.sort(key=operator.attrgetter("as_datetime")) + observer = sun.bristol() + reply["sunrise"] = sun.sunrise(observer) + reply["sunset"] = sun.sunset(observer) reply["events"] = events reply["last_week"] = last_week reply["two_weeks_ago"] = two_weeks_ago diff --git a/agenda/uk_holiday.py b/agenda/uk_holiday.py index 070d02d..817dc9d 100644 --- a/agenda/uk_holiday.py +++ b/agenda/uk_holiday.py @@ -52,30 +52,3 @@ def get_mothers_day(input_date: date) -> date: 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