Split out busy events code
This commit is contained in:
parent
7bb6110f45
commit
dd59c809e1
159
agenda/busy.py
Normal file
159
agenda/busy.py
Normal file
|
|
@ -0,0 +1,159 @@
|
|||
"""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
|
||||
]
|
||||
158
agenda/data.py
158
agenda/data.py
|
|
@ -1,7 +1,6 @@
|
|||
"""Agenda data."""
|
||||
|
||||
import asyncio
|
||||
import itertools
|
||||
import os
|
||||
import typing
|
||||
from datetime import date, datetime, timedelta
|
||||
|
|
@ -16,6 +15,7 @@ import pytz
|
|||
from . import (
|
||||
accommodation,
|
||||
birthday,
|
||||
busy,
|
||||
calendar,
|
||||
carnival,
|
||||
conference,
|
||||
|
|
@ -33,10 +33,9 @@ from . import (
|
|||
thespacedevs,
|
||||
travel,
|
||||
uk_holiday,
|
||||
uk_tz,
|
||||
waste_schedule,
|
||||
)
|
||||
from .types import Event, StrDict, Trip
|
||||
from .types import Event
|
||||
|
||||
here = dateutil.tz.tzlocal()
|
||||
|
||||
|
|
@ -95,84 +94,6 @@ 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]],
|
||||
|
|
@ -189,77 +110,6 @@ 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]:
|
||||
|
|
@ -403,10 +253,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_event(e)
|
||||
if e.as_date > today and e.as_date < next_year and busy.busy_event(e)
|
||||
]
|
||||
|
||||
gaps = find_gaps(busy_events)
|
||||
gaps = busy.find_gaps(busy_events)
|
||||
|
||||
events += [
|
||||
Event(name="gap", date=gap["start"], end_date=gap["end"]) for gap in gaps
|
||||
|
|
|
|||
|
|
@ -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.data.get_busy_events(now.date(), app.config, trip_list)
|
||||
gaps = agenda.data.find_gaps(busy_events)
|
||||
busy_events = agenda.busy.get_busy_events(now.date(), app.config, trip_list)
|
||||
gaps = agenda.busy.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.data.get_busy_events(now.date(), app.config, trip_list)
|
||||
weekends = agenda.data.weekends(busy_events)
|
||||
busy_events = agenda.busy.get_busy_events(now.date(), app.config, trip_list)
|
||||
weekends = agenda.busy.weekends(busy_events)
|
||||
return flask.render_template("weekends.html", today=now.date(), items=weekends)
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue