304 lines
8.9 KiB
Python
304 lines
8.9 KiB
Python
"""Agenda data."""
|
|
|
|
import asyncio
|
|
import os
|
|
import typing
|
|
from datetime import date, datetime, timedelta
|
|
from time import time
|
|
|
|
import dateutil.rrule
|
|
import dateutil.tz
|
|
import flask
|
|
import lxml
|
|
import pytz
|
|
|
|
from . import (
|
|
accommodation,
|
|
birthday,
|
|
bristol_waste,
|
|
busy,
|
|
carnival,
|
|
conference,
|
|
domains,
|
|
economist,
|
|
events_yaml,
|
|
gandi,
|
|
gwr,
|
|
hn,
|
|
holidays,
|
|
meetup,
|
|
n_somerset_waste,
|
|
stock_market,
|
|
subscription,
|
|
sun,
|
|
thespacedevs,
|
|
travel,
|
|
uk_holiday,
|
|
)
|
|
from .event import Event
|
|
from .types import StrDict
|
|
from .utils import time_function
|
|
|
|
here = dateutil.tz.tzlocal()
|
|
|
|
# deadline to file tax return
|
|
# credit card expiry dates
|
|
# morzine ski lifts
|
|
# chalet availability calendar
|
|
|
|
# starlink visible
|
|
|
|
|
|
def timezone_transition(
|
|
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 <= t <= end
|
|
]
|
|
|
|
|
|
async def n_somerset_waste_collection_events(
|
|
data_dir: str, postcode: str, uprn: str
|
|
) -> list[Event]:
|
|
"""Waste colllection events."""
|
|
html = await n_somerset_waste.get_html(data_dir, postcode, uprn)
|
|
root = lxml.html.fromstring(html)
|
|
events = n_somerset_waste.parse(root)
|
|
return events
|
|
|
|
|
|
async def bristol_waste_collection_events(
|
|
data_dir: str, start_date: date, uprn: str
|
|
) -> list[Event]:
|
|
"""Waste colllection events."""
|
|
return await bristol_waste.get(start_date, data_dir, uprn)
|
|
|
|
|
|
def find_events_during_stay(
|
|
accommodation_events: list[Event], markets: list[Event]
|
|
) -> list[Event]:
|
|
"""Market events that happen during accommodation stays."""
|
|
overlapping_markets = []
|
|
for market in markets:
|
|
market_date = market.as_date
|
|
assert isinstance(market_date, date)
|
|
for e in accommodation_events:
|
|
start, end = e.as_date, e.end_as_date
|
|
assert start and end and all(isinstance(i, date) for i in (start, end))
|
|
# Check if the market date is within the accommodation dates.
|
|
if start <= market_date <= end:
|
|
overlapping_markets.append(market)
|
|
break # Breaks the inner loop if overlap is found.
|
|
return overlapping_markets
|
|
|
|
|
|
def hide_markets_while_away(
|
|
events: list[Event], accommodation_events: list[Event]
|
|
) -> None:
|
|
"""Hide markets that happen while away."""
|
|
optional = [
|
|
e
|
|
for e in events
|
|
if e.name == "market" or (e.title and "LHG Run Club" in e.title)
|
|
]
|
|
going = [e for e in events if e.going]
|
|
|
|
overlapping_markets = find_events_during_stay(
|
|
accommodation_events + going, optional
|
|
)
|
|
for market in overlapping_markets:
|
|
events.remove(market)
|
|
|
|
|
|
class AgendaData(typing.TypedDict, total=False):
|
|
"""Agenda Data."""
|
|
|
|
now: datetime
|
|
stock_markets: list[str]
|
|
rockets: list[thespacedevs.Summary]
|
|
gwr_advance_tickets: date | None
|
|
data_gather_seconds: float
|
|
stock_market_times_seconds: float
|
|
timings: list[tuple[str, float]]
|
|
events: list[Event]
|
|
accommodation_events: list[Event]
|
|
gaps: list[StrDict]
|
|
sunrise: datetime
|
|
sunset: datetime
|
|
last_week: date
|
|
two_weeks_ago: date
|
|
errors: list[tuple[str, Exception]]
|
|
|
|
|
|
def rocket_launch_events(rockets: list[thespacedevs.Summary]) -> list[Event]:
|
|
"""Rocket launch events."""
|
|
events: list[Event] = []
|
|
for launch in rockets:
|
|
dt = None
|
|
|
|
net_precision = launch["net_precision"]
|
|
skip = {"Year", "Month", "Quarter", "Fiscal Year"}
|
|
if net_precision == "Day":
|
|
dt = datetime.strptime(launch["net"], "%Y-%m-%dT%H:%M:%SZ").date()
|
|
elif (
|
|
net_precision
|
|
and net_precision not in skip
|
|
and "Year" not in net_precision
|
|
and launch["t0_time"]
|
|
):
|
|
dt = pytz.utc.localize(
|
|
datetime.strptime(launch["net"], "%Y-%m-%dT%H:%M:%SZ")
|
|
)
|
|
|
|
if not dt:
|
|
continue
|
|
|
|
rocket_name = (
|
|
f'{launch["rocket"]["full_name"]}: '
|
|
+ f'{launch["mission_name"] or "[no mission]"}'
|
|
)
|
|
e = Event(name="rocket", date=dt, title=rocket_name)
|
|
events.append(e)
|
|
|
|
return events
|
|
|
|
|
|
async def get_data(now: datetime, config: flask.config.Config) -> AgendaData:
|
|
"""Get data to display on agenda dashboard."""
|
|
data_dir = config["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=2 * 365)
|
|
|
|
minus_365 = now - timedelta(days=365)
|
|
plus_365 = now + timedelta(days=365)
|
|
|
|
t0 = time()
|
|
offline_mode = bool(config.get("OFFLINE_MODE"))
|
|
result_list = await asyncio.gather(
|
|
time_function(
|
|
"gwr_advance_tickets", gwr.advance_ticket_date, data_dir, offline_mode
|
|
),
|
|
time_function(
|
|
"backwell_bins",
|
|
n_somerset_waste_collection_events,
|
|
data_dir,
|
|
config["BACKWELL_POSTCODE"],
|
|
config["BACKWELL_UPRN"],
|
|
),
|
|
time_function(
|
|
"bristol_bins",
|
|
bristol_waste_collection_events,
|
|
data_dir,
|
|
today,
|
|
config["BRISTOL_UPRN"],
|
|
),
|
|
)
|
|
rockets = thespacedevs.read_cached_launches(rocket_dir)
|
|
|
|
results = {call[0]: call[1] for call in result_list}
|
|
|
|
errors = [(call[0], call[3]) for call in result_list if call[3]]
|
|
|
|
gwr_advance_tickets = results["gwr_advance_tickets"]
|
|
|
|
data_gather_seconds = time() - t0
|
|
t0 = time()
|
|
|
|
stock_market_times = stock_market.open_and_close()
|
|
stock_market_times_seconds = time() - t0
|
|
|
|
reply: AgendaData = {
|
|
"now": now,
|
|
"stock_markets": stock_market_times,
|
|
"rockets": rockets,
|
|
"gwr_advance_tickets": gwr_advance_tickets,
|
|
"data_gather_seconds": data_gather_seconds,
|
|
"stock_market_times_seconds": stock_market_times_seconds,
|
|
"timings": [(call[0], call[2]) for call in result_list],
|
|
}
|
|
|
|
my_data = config["PERSONAL_DATA"]
|
|
events = (
|
|
[
|
|
Event(name="mothers_day", date=uk_holiday.get_mothers_day(today)),
|
|
]
|
|
+ timezone_transition(minus_365, plus_365, "uk_clock_change", "Europe/London")
|
|
+ timezone_transition(
|
|
minus_365, plus_365, "us_clock_change", "America/New_York"
|
|
)
|
|
)
|
|
|
|
if gwr_advance_tickets:
|
|
events.append(Event(name="gwr_advance_tickets", date=gwr_advance_tickets))
|
|
|
|
us_hols = holidays.us_holidays(last_year, next_year)
|
|
events += holidays.get_nyse_holidays(last_year, next_year, us_hols)
|
|
|
|
accommodation_events = accommodation.get_events(
|
|
os.path.join(my_data, "accommodation.yaml")
|
|
)
|
|
|
|
holiday_list = holidays.get_all(last_year, next_year, data_dir)
|
|
events += holidays.combine_holidays(holiday_list)
|
|
if flask.g.user.is_authenticated:
|
|
events += birthday.get_birthdays(
|
|
last_year, os.path.join(my_data, "entities.yaml")
|
|
)
|
|
events += domains.renewal_dates(my_data)
|
|
events += accommodation_events
|
|
events += travel.all_events(my_data)
|
|
events += conference.get_list(os.path.join(my_data, "conferences.yaml"))
|
|
for key in "backwell_bins", "bristol_bins":
|
|
if results[key]:
|
|
events += results[key]
|
|
events += events_yaml.read(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 += rocket_launch_events(rockets)
|
|
events += [Event(name="today", date=today)]
|
|
|
|
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)
|
|
]
|
|
|
|
gaps = busy.find_gaps(busy_events)
|
|
|
|
events += [
|
|
Event(name="gap", date=gap["start"], end_date=gap["end"]) for gap in gaps
|
|
]
|
|
|
|
# Sort events by their datetime; the "today" event is prioritised
|
|
# at the top of the list for today. This is achieved by sorting first by
|
|
# the datetime attribute, and then ensuring that events with the name
|
|
# "today" are ordered before others on the same date.
|
|
events.sort(key=lambda e: (e.as_datetime, e.name != "today"))
|
|
|
|
reply["gaps"] = gaps
|
|
|
|
observer = sun.bristol()
|
|
reply["sunrise"] = sun.sunrise(observer)
|
|
reply["sunset"] = sun.sunset(observer)
|
|
reply["events"] = events
|
|
reply["accommodation_events"] = accommodation_events
|
|
reply["last_week"] = last_week
|
|
reply["two_weeks_ago"] = two_weeks_ago
|
|
|
|
reply["errors"] = errors
|
|
|
|
return reply
|