agenda/busy.py

@@ -1,159 +0,0 @@
-"""Identify busy events and gaps when nothing is scheduled."""
-
-import itertools
-import typing
-from datetime import date, datetime, timedelta
-
-import flask
-
-from . import events_yaml
-from .types import Event, StrDict, Trip
-
-
-def busy_event(e: Event) -> bool:
-    """Busy."""
-    if e.name not in {
-        "event",
-        "accommodation",
-        "conference",
-        "transport",
-        "meetup",
-        "party",
-        "trip",
-    }:
-        return False
-
-    if e.title in ("IA UK board meeting", "Mill Road Winter Fair"):
-        return False
-
-    if e.name == "conference" and not e.going:
-        return False
-    if not e.title:
-        return True
-    if e.title == "LHG Run Club" or "Third Thursday Social" in e.title:
-        return False
-
-    lc_title = e.title.lower()
-    return (
-        "rebels" not in lc_title
-        and "south west data social" not in lc_title
-        and "dorkbot" not in lc_title
-    )
-
-
-def get_busy_events(
-    today: date, config: flask.config.Config, trips: list[Trip]
-) -> list[Event]:
-    """Find busy events from a year ago to two years in the future."""
-    last_year = today - timedelta(days=365)
-    next_year = today + timedelta(days=2 * 365)
-
-    my_data = config["PERSONAL_DATA"]
-    events = events_yaml.read(my_data, last_year, next_year, skip_trips=True)
-
-    for trip in trips:
-        event_type = "trip"
-        if trip.events and not trip.conferences:
-            event_type = trip.events[0]["name"]
-        elif len(trip.conferences) == 1 and trip.conferences[0].get("hackathon"):
-            event_type = "hackathon"
-        events.append(
-            Event(
-                name=event_type,
-                title=trip.title + " " + trip.country_flags,
-                date=trip.start,
-                end_date=trip.end,
-                url=flask.url_for("trip_page", start=trip.start.isoformat()),
-            )
-        )
-
-    busy_events = [
-        e
-        for e in sorted(events, key=lambda e: e.as_date)
-        if (e.as_date >= today or (e.end_date and e.end_as_date >= today))
-        and e.as_date < next_year
-        and busy_event(e)
-    ]
-
-    return busy_events
-
-
-def weekends(busy_events: list[Event]) -> typing.Sequence[StrDict]:
-    """Next ten weekends."""
-    today = datetime.today()
-    weekday = today.weekday()
-
-    # Calculate the difference to the next or previous Saturday
-    if weekday == 6:  # Sunday
-        start_date = (today - timedelta(days=1)).date()
-    else:
-        start_date = (today + timedelta(days=(5 - weekday))).date()
-
-    weekends_info = []
-    for i in range(52):
-        saturday = start_date + timedelta(weeks=i)
-        sunday = saturday + timedelta(days=1)
-
-        saturday_events = [
-            event
-            for event in busy_events
-            if event.end_date and event.as_date <= saturday <= event.end_as_date
-        ]
-        sunday_events = [
-            event
-            for event in busy_events
-            if event.end_date and event.as_date <= sunday <= event.end_as_date
-        ]
-
-        weekends_info.append(
-            {"date": saturday, "saturday": saturday_events, "sunday": sunday_events}
-        )
-
-    return weekends_info
-
-
-def find_gaps(events: list[Event], min_gap_days: int = 3) -> list[StrDict]:
-    """Gaps of at least `min_gap_days` between events in a list of events."""
-    # Sort events by start date
-
-    gaps: list[tuple[date, date]] = []
-    previous_event_end = None
-
-    by_start_date = {
-        d: list(on_day)
-        for d, on_day in itertools.groupby(events, key=lambda e: e.as_date)
-    }
-
-    by_end_date = {
-        d: list(on_day)
-        for d, on_day in itertools.groupby(events, key=lambda e: e.end_as_date)
-    }
-
-    for event in events:
-        # Use start date for current event
-        start_date = event.as_date
-
-        # If previous event exists, calculate the gap
-        if previous_event_end:
-            gap_days = (start_date - previous_event_end).days
-            if gap_days >= (min_gap_days + 2):
-                start_end = (
-                    previous_event_end + timedelta(days=1),
-                    start_date - timedelta(days=1),
-                )
-                gaps.append(start_end)
-
-        # Update previous event end date
-        end = event.end_as_date
-        if not previous_event_end or end > previous_event_end:
-            previous_event_end = end
-
-    return [
-        {
-            "start": gap_start,
-            "end": gap_end,
-            "after": by_start_date[gap_end + timedelta(days=1)],
-            "before": by_end_date[gap_start - timedelta(days=1)],
-        }
-        for gap_start, gap_end in gaps
-    ]

agenda/carnival.py

@@ -1,33 +0,0 @@
-"""Calcuate the date for carnival."""
-
-from datetime import date, timedelta
-
-from dateutil.easter import easter
-
-from .types import Event
-
-
-def rio_carnival_events(start_date: date, end_date: date) -> list[Event]:
-    """List of events for Rio Carnival for each year between start_date and end_date."""
-    events = []
-    for year in range(start_date.year, end_date.year + 1):
-        easter_date = easter(year)
-        carnival_start = easter_date - timedelta(days=51)
-        carnival_end = easter_date - timedelta(days=46)
-
-        # Only include the carnival if it falls within the specified date range
-        if (
-            start_date <= carnival_start <= end_date
-            or start_date <= carnival_end <= end_date
-        ):
-            events.append(
-                Event(
-                    name="carnival",
-                    title="Rio Carnival",
-                    date=carnival_start,
-                    end_date=carnival_end,
-                    url="https://en.wikipedia.org/wiki/Rio_Carnival",
-                )
-            )
-
-    return events

agenda/data.py

@@ -1,6 +1,7 @@
 """Agenda data."""
 
 import asyncio
+import itertools
 import os
 import typing
 from datetime import date, datetime, timedelta
@@ -9,19 +10,19 @@ from time import time
 import dateutil.rrule
 import dateutil.tz
 import flask
+import isodate  # type: ignore
 import lxml
 import pytz
+import yaml
+from dateutil.easter import easter
 
 from . import (
     accommodation,
     birthday,
-    busy,
     calendar,
-    carnival,
     conference,
     domains,
     economist,
-    events_yaml,
     gandi,
     gwr,
     hn,
@@ -33,9 +34,10 @@ from . import (
     thespacedevs,
     travel,
     uk_holiday,
+    uk_tz,
     waste_schedule,
 )
-from .types import Event
+from .types import Event, StrDict, Trip
 
 here = dateutil.tz.tzlocal()
 
@@ -59,6 +61,51 @@ def timezone_transition(
     ]
 
 
+def midnight(d: date) -> datetime:
+    """Convert from date to midnight on that day."""
+    return datetime.combine(d, datetime.min.time())
+
+
+def rio_carnival_events(start_date: date, end_date: date) -> list[Event]:
+    """List of events for Rio Carnival for each year between start_date and end_date."""
+    events = []
+    for year in range(start_date.year, end_date.year + 1):
+        easter_date = easter(year)
+        carnival_start = easter_date - timedelta(days=51)
+        carnival_end = easter_date - timedelta(days=46)
+
+        # Only include the carnival if it falls within the specified date range
+        if (
+            start_date <= carnival_start <= end_date
+            or start_date <= carnival_end <= end_date
+        ):
+            events.append(
+                Event(
+                    name="carnival",
+                    title="Rio Carnival",
+                    date=carnival_start,
+                    end_date=carnival_end,
+                    url="https://en.wikipedia.org/wiki/Rio_Carnival",
+                )
+            )
+
+    return events
+
+
+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"])
+
+    return [
+        i.date() if all_day else uk_tz.localize(i)
+        for i in dateutil.rrule.rrulestr(rrule, dtstart=midnight(start)).between(
+            midnight(start), midnight(end)
+        )
+    ]
+
+
 async def waste_collection_events(
     data_dir: str, postcode: str, uprn: str
 ) -> list[Event]:
@@ -76,6 +123,60 @@ async def bristol_waste_collection_events(
     return await waste_schedule.get_bristol_gov_uk(start_date, data_dir, uprn)
 
 
+def get_yaml_event_date_field(item: dict[str, str]) -> str:
+    """Event date field name."""
+    return (
+        "end_date"
+        if item["name"] == "travel_insurance"
+        else ("start_date" if "start_date" in item else "date")
+    )
+
+
+def get_yaml_event_end_date_field(item: dict[str, str]) -> str:
+    """Event date field name."""
+    return (
+        "end_date"
+        if item["name"] == "travel_insurance"
+        else ("start_date" if "start_date" in item else "date")
+    )
+
+
+def read_events_yaml(
+    data_dir: str, start: date, end: date, skip_trips: bool = False
+) -> list[Event]:
+    """Read eventes from YAML file."""
+    events: list[Event] = []
+    for item in yaml.safe_load(open(os.path.join(data_dir, "events.yaml"))):
+        if "trip" in item and skip_trips:
+            continue
+        duration = (
+            isodate.parse_duration(item["duration"]) if "duration" in item else None
+        )
+        dates = (
+            dates_from_rrule(item["rrule"], start, end)
+            if "rrule" in item
+            else [item[get_yaml_event_date_field(item)]]
+        )
+        for dt in dates:
+            e = Event(
+                name=item["name"],
+                date=dt,
+                end_date=(
+                    dt + duration
+                    if duration
+                    else (
+                        item.get("end_date")
+                        if item["name"] != "travel_insurance"
+                        else None
+                    )
+                ),
+                title=item.get("title"),
+                url=item.get("url"),
+            )
+            events.append(e)
+    return events
+
+
 def find_events_during_stay(
     accommodation_events: list[Event], markets: list[Event]
 ) -> list[Event]:
@@ -94,6 +195,84 @@ def find_events_during_stay(
     return overlapping_markets
 
 
+def find_gaps(events: list[Event], min_gap_days: int = 3) -> list[StrDict]:
+    """Gaps of at least `min_gap_days` between events in a list of events."""
+    # Sort events by start date
+
+    gaps: list[tuple[date, date]] = []
+    previous_event_end = None
+
+    by_start_date = {
+        d: list(on_day)
+        for d, on_day in itertools.groupby(events, key=lambda e: e.as_date)
+    }
+
+    by_end_date = {
+        d: list(on_day)
+        for d, on_day in itertools.groupby(events, key=lambda e: e.end_as_date)
+    }
+
+    for event in events:
+        # Use start date for current event
+        start_date = event.as_date
+
+        # If previous event exists, calculate the gap
+        if previous_event_end:
+            gap_days = (start_date - previous_event_end).days
+            if gap_days >= (min_gap_days + 2):
+                start_end = (
+                    previous_event_end + timedelta(days=1),
+                    start_date - timedelta(days=1),
+                )
+                gaps.append(start_end)
+
+        # Update previous event end date
+        end = event.end_as_date
+        if not previous_event_end or end > previous_event_end:
+            previous_event_end = end
+
+    return [
+        {
+            "start": gap_start,
+            "end": gap_end,
+            "after": by_start_date[gap_end + timedelta(days=1)],
+            "before": by_end_date[gap_start - timedelta(days=1)],
+        }
+        for gap_start, gap_end in gaps
+    ]
+
+
+def busy_event(e: Event) -> bool:
+    """Busy."""
+    if e.name not in {
+        "event",
+        "accommodation",
+        "conference",
+        "transport",
+        "meetup",
+        "party",
+        "trip",
+    }:
+        return False
+
+    if e.title in ("IA UK board meeting", "Mill Road Winter Fair"):
+        return False
+
+    if e.name == "conference" and not e.going:
+        return False
+    if not e.title:
+        return True
+    if e.title == "LHG Run Club" or "Third Thursday Social" in e.title:
+        return False
+
+    lc_title = e.title.lower()
+    return (
+        "rebels" not in lc_title
+        and "south west data social" not in lc_title
+        and "dorkbot" not in lc_title
+    )
+
+
 async def time_function(
     name: str,
     func: typing.Callable[..., typing.Coroutine[typing.Any, typing.Any, typing.Any]],
@@ -110,6 +289,77 @@ async def time_function(
     return name, result, end_time - start_time, exception
 
 
+def get_busy_events(
+    today: date, config: flask.config.Config, trips: list[Trip]
+) -> list[Event]:
+    """Find busy events from a year ago to two years in the future."""
+    last_year = today - timedelta(days=365)
+    next_year = today + timedelta(days=2 * 365)
+
+    my_data = config["PERSONAL_DATA"]
+    events = read_events_yaml(my_data, last_year, next_year, skip_trips=True)
+
+    for trip in trips:
+        event_type = "trip"
+        if trip.events and not trip.conferences:
+            event_type = trip.events[0]["name"]
+        elif len(trip.conferences) == 1 and trip.conferences[0].get("hackathon"):
+            event_type = "hackathon"
+        events.append(
+            Event(
+                name=event_type,
+                title=trip.title + " " + trip.country_flags,
+                date=trip.start,
+                end_date=trip.end,
+                url=flask.url_for("trip_page", start=trip.start.isoformat()),
+            )
+        )
+
+    busy_events = [
+        e
+        for e in sorted(events, key=lambda e: e.as_date)
+        if (e.as_date >= today or (e.end_date and e.end_as_date >= today))
+        and e.as_date < next_year
+        and busy_event(e)
+    ]
+
+    return busy_events
+
+
+def weekends(busy_events: list[Event]) -> typing.Sequence[StrDict]:
+    """Next ten weekends."""
+    today = datetime.today()
+    weekday = today.weekday()
+
+    # Calculate the difference to the next or previous Saturday
+    if weekday == 6:  # Sunday
+        start_date = (today - timedelta(days=1)).date()
+    else:
+        start_date = (today + timedelta(days=(5 - weekday))).date()
+
+    weekends_info = []
+    for i in range(52):
+        saturday = start_date + timedelta(weeks=i)
+        sunday = saturday + timedelta(days=1)
+
+        saturday_events = [
+            event
+            for event in busy_events
+            if event.end_date and event.as_date <= saturday <= event.end_as_date
+        ]
+        sunday_events = [
+            event
+            for event in busy_events
+            if event.end_date and event.as_date <= sunday <= event.end_as_date
+        ]
+
+        weekends_info.append(
+            {"date": saturday, "saturday": saturday_events, "sunday": sunday_events}
+        )
+
+    return weekends_info
+
+
 async def get_data(
     now: datetime, config: flask.config.Config
 ) -> typing.Mapping[str, str | object]:
@@ -202,13 +452,13 @@ async def get_data(
     for key in "backwell_bins", "bristol_bins":
         if results[key]:
             events += results[key]
-    events += events_yaml.read(my_data, last_year, next_year)
+    events += read_events_yaml(my_data, last_year, next_year)
     events += subscription.get_events(os.path.join(my_data, "subscriptions.yaml"))
     events += gandi.get_events(data_dir)
     events += economist.publication_dates(last_week, next_year)
     events += meetup.get_events(my_data)
     events += hn.whoishiring(last_year, next_year)
-    events += carnival.rio_carnival_events(last_year, next_year)
+    events += rio_carnival_events(last_year, next_year)
 
     # hide markets that happen while away
     optional = [
@@ -253,10 +503,10 @@ async def get_data(
     busy_events = [
         e
         for e in sorted(events, key=lambda e: e.as_date)
-        if e.as_date > today and e.as_date < next_year and busy.busy_event(e)
+        if e.as_date > today and e.as_date < next_year and busy_event(e)
     ]
 
-    gaps = busy.find_gaps(busy_events)
+    gaps = find_gaps(busy_events)
 
     events += [
         Event(name="gap", date=gap["start"], end_date=gap["end"]) for gap in gaps

agenda/events_yaml.py

@@ -1,85 +0,0 @@
-"""Read events from YAML."""
-
-import os
-import typing
-from datetime import date, datetime
-
-import dateutil.rrule
-import isodate  # type: ignore
-import yaml
-
-from . import uk_tz
-from .types import Event
-
-
-def midnight(d: date) -> datetime:
-    """Convert from date to midnight on that day."""
-    return datetime.combine(d, datetime.min.time())
-
-
-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"])
-
-    return [
-        i.date() if all_day else uk_tz.localize(i)
-        for i in dateutil.rrule.rrulestr(rrule, dtstart=midnight(start)).between(
-            midnight(start), midnight(end)
-        )
-    ]
-
-
-def get_yaml_event_date_field(item: dict[str, str]) -> str:
-    """Event date field name."""
-    return (
-        "end_date"
-        if item["name"] == "travel_insurance"
-        else ("start_date" if "start_date" in item else "date")
-    )
-
-
-def get_yaml_event_end_date_field(item: dict[str, str]) -> str:
-    """Event date field name."""
-    return (
-        "end_date"
-        if item["name"] == "travel_insurance"
-        else ("start_date" if "start_date" in item else "date")
-    )
-
-
-def read(
-    data_dir: str, start: date, end: date, skip_trips: bool = False
-) -> list[Event]:
-    """Read eventes from YAML file."""
-    events: list[Event] = []
-    for item in yaml.safe_load(open(os.path.join(data_dir, "events.yaml"))):
-        if "trip" in item and skip_trips:
-            continue
-        duration = (
-            isodate.parse_duration(item["duration"]) if "duration" in item else None
-        )
-        dates = (
-            dates_from_rrule(item["rrule"], start, end)
-            if "rrule" in item
-            else [item[get_yaml_event_date_field(item)]]
-        )
-        for dt in dates:
-            e = Event(
-                name=item["name"],
-                date=dt,
-                end_date=(
-                    dt + duration
-                    if duration
-                    else (
-                        item.get("end_date")
-                        if item["name"] != "travel_insurance"
-                        else None
-                    )
-                ),
-                title=item.get("title"),
-                url=item.get("url"),
-            )
-            events.append(e)
-    return events

agenda/types.py

@@ -268,26 +268,6 @@ class Event:
             else None
         )
 
-    def display_duration(self) -> str | None:
-        """Duration for display."""
-        if self.end_as_date != self.as_date or not self.has_time:
-            return None
-
-        assert isinstance(self.date, datetime.datetime)
-        assert isinstance(self.end_date, datetime.datetime)
-
-        secs: int = int((self.end_date - self.date).total_seconds())
-
-        hours: int = secs // 3600
-        mins: int = (secs % 3600) // 60
-
-        if mins == 0:
-            return f"{hours:d}h"
-        if hours == 0:
-            return f"{mins:d} mins"
-
-        return f"{hours:d}h {mins:02d} mins"
-
     def delta_days(self, today: datetime.date) -> str:
         """Return number of days from today as a string."""
         delta = (self.as_date - today).days

templates/index.html

@@ -174,10 +174,9 @@
 
       <div class="col-md-2{{ cell_bg }}">
         {% if event.end_date %}
-          {% set duration = event.display_duration() %}
-          {% if duration %}
+          {% if event.end_as_date == event.as_date and event.has_time %}
             end: {{event.end_date.strftime("%H:%M") }}
-            (duration: {{duration}})
+            (duration: {{event.end_date - event.date}})
           {% elif event.end_date != event.date %}
             {{event.end_date}}
           {% endif %}

web_view.py

@@ -92,8 +92,8 @@ async def gaps_page() -> str:
     """List of available gaps."""
     now = datetime.now()
     trip_list = agenda.trip.build_trip_list()
-    busy_events = agenda.busy.get_busy_events(now.date(), app.config, trip_list)
-    gaps = agenda.busy.find_gaps(busy_events)
+    busy_events = agenda.data.get_busy_events(now.date(), app.config, trip_list)
+    gaps = agenda.data.find_gaps(busy_events)
     return flask.render_template("gaps.html", today=now.date(), gaps=gaps)
 
 
@@ -102,8 +102,8 @@ async def weekends() -> str:
     """List of available gaps."""
     now = datetime.now()
     trip_list = agenda.trip.build_trip_list()
-    busy_events = agenda.busy.get_busy_events(now.date(), app.config, trip_list)
-    weekends = agenda.busy.weekends(busy_events)
+    busy_events = agenda.data.get_busy_events(now.date(), app.config, trip_list)
+    weekends = agenda.data.weekends(busy_events)
     return flask.render_template("weekends.html", today=now.date(), items=weekends)