diff --git a/agenda/__init__.py b/agenda/__init__.py index 4e5e2b4..07698d8 100644 --- a/agenda/__init__.py +++ b/agenda/__init__.py @@ -5,7 +5,6 @@ import os import typing import warnings from datetime import date, datetime, time, timedelta, timezone -from decimal import Decimal from time import time as unixtime from typing import List @@ -21,11 +20,11 @@ import pytz import requests import yaml from dateutil.easter import easter -from dateutil.relativedelta import FR, SA, relativedelta +from dateutil.relativedelta import FR, relativedelta from agenda import thespacedevs -from . import gwr, waste_schedule +from . import fx, gwr, markets, waste_schedule from .types import Event warnings.simplefilter(action="ignore", category=FutureWarning) @@ -150,35 +149,6 @@ def get_next_bank_holiday(input_date: date) -> list[Event]: return hols -def get_gbpusd() -> Decimal: - """Get the current value for GBPUSD, with caching.""" - now = datetime.now() - now_str = now.strftime("%Y-%m-%d_%H:%M") - fx_dir = os.path.join(data_dir, "fx") - existing_data = os.listdir(fx_dir) - existing = [f for f in existing_data if f.endswith("_GBPUSD.json")] - if existing: - recent_filename = max(existing) - recent = datetime.strptime(recent_filename, "%Y-%m-%d_%H:%M_GBPUSD.json") - delta = now - recent - - if existing and delta < timedelta(hours=6): - full = os.path.join(fx_dir, recent_filename) - data = json.load(open(full), parse_float=Decimal) - if "quotes" in data and "USDGBP" in data["quotes"]: - return typing.cast(Decimal, 1 / data["quotes"]["USDGBP"]) - - url = "http://api.exchangerate.host/live" - params = {"currencies": "GBP,USD", "access_key": access_key} - - filename = f"{fx_dir}/{now_str}_GBPUSD.json" - r = requests.get(url, params=params) - open(filename, "w").write(r.text) - data = json.loads(r.text, parse_float=Decimal) - - return typing.cast(Decimal, 1 / data["quotes"]["USDGBP"]) - - def next_economist(input_date: date) -> date: """Next date that the Economist is published.""" # Define the publication day (Thursday) and the day of the week of the input date @@ -315,124 +285,6 @@ def critical_mass(start_date: date, limit: int = 12) -> list[Event]: return events -def windmill_hill_market_days(start_date: date) -> list[Event]: - """Windmill Hill Market days for the next 12 months from a given date.""" - events: list[Event] = [] - current_date = start_date - url = ( - "https://www.windmillhillcityfarm.org.uk" - + "/visit-us/shops-more/windmill-hill-market-bristol-market/" - ) - - # To keep count of how many market days have been calculated - count = 0 - - tz = pytz.timezone("Europe/London") - start = time(10, 0) - end = time(15, 0) - - while count < 12: - # Skip months outside of April to December - if current_date.month < 4 or current_date.month > 12: - current_date += relativedelta(months=1) - current_date = date(current_date.year, current_date.month, 1) - continue - - # Calculate the first Saturday of the current month - first_saturday = current_date + relativedelta(day=1, weekday=SA(+1)) - - # Include it in the list only if it's on or after the start_date - if first_saturday >= start_date: - events.append( - Event( - name="market", - title="Windmill Hill Market", - date=tz.localize(datetime.combine(first_saturday, start)), - end_date=tz.localize(datetime.combine(first_saturday, end)), - url=url, - ) - ) - count += 1 - - # Move to the next month - current_date += relativedelta(months=1) - current_date = date(current_date.year, current_date.month, 1) - - return events - - -def tobacco_factory_market_days(start_date: date) -> list[Event]: - """Tobacco Factory Market days for the next 12 months from a given date.""" - events: list[Event] = [] - current_date = start_date - count = 0 - - url = "https://tobaccofactory.com/whats-on/sunday-market/" - - tz = pytz.timezone("Europe/London") - start = time(10, 0) - end = time(14, 30) - - while count < 52: # 52 weeks in a year - # Calculate the next Sunday from the current date - next_sunday = current_date + relativedelta(weekday=6) # Sunday is 6 - - # Include it in the list only if it's on or after the start_date - if next_sunday >= start_date: - events.append( - Event( - name="market", - title="Tobacco Factory Sunday Market", - date=tz.localize(datetime.combine(next_sunday, start)), - end_date=tz.localize(datetime.combine(next_sunday, end)), - url=url, - ) - ) - count += 1 - - # Move to the next week - current_date += timedelta(weeks=1) - - return events - - -def nailsea_farmers_market_days(start_date: date) -> list[Event]: - """Nailsea Farmers Market days for the next 12 months from a given date.""" - events: list[Event] = [] - current_date = start_date - count = 0 - - tz = pytz.timezone("Europe/London") - t = time(9, 0) # The market starts at 9am - - while count < 12: - # Calculate the 3rd Saturday of the current month - third_saturday = current_date + relativedelta(day=1, weekday=SA(+3)) - - # Include it in the list only if it's on or after the start_date - if third_saturday >= start_date: - events.append( - Event( - name="market", - title="Nailsea Farmers Market", - date=tz.localize(datetime.combine(third_saturday, t)), - ) - ) - count += 1 - - # Move to the next month - current_date += relativedelta(months=1) - current_date = date(current_date.year, current_date.month, 1) - - return events - - -# Test the function -if __name__ == "__main__": - start_date = date(2023, 10, 29) - print(windmill_hill_market_days(start_date)) - - def as_date(d: date | datetime) -> date: """Return date for given date or datetime.""" return d.date() if isinstance(d, datetime) else d @@ -676,7 +528,7 @@ def get_data(now: datetime) -> typing.Mapping[str, str | object]: reply = { "now": now, - "gbpusd": get_gbpusd(), + "gbpusd": fx.get_gbpusd(config), "next_economist": next_economist(today), "bank_holiday": get_next_bank_holiday(today), "us_holiday": get_us_holidays(today), @@ -691,9 +543,9 @@ def get_data(now: datetime) -> typing.Mapping[str, str | object]: "gwr_advance_tickets": gwr.advance_ticket_date(data_dir), "critical_mass": critical_mass(today), "market": ( - windmill_hill_market_days(today) - + tobacco_factory_market_days(today) - + nailsea_farmers_market_days(today) + markets.windmill_hill(today) + + markets.tobacco_factory(today) + + markets.nailsea_farmers(today) ), "rockets": thespacedevs.get_launches(rocket_dir, limit=40), } diff --git a/agenda/fx.py b/agenda/fx.py new file mode 100644 index 0000000..5d397e1 --- /dev/null +++ b/agenda/fx.py @@ -0,0 +1,42 @@ +"""Currency exchange rates.""" + +import configparser +import json +import os +import typing +from datetime import datetime, timedelta +from decimal import Decimal + +import requests + + +def get_gbpusd(config: configparser.ConfigParser) -> Decimal: + """Get the current value for GBPUSD, with caching.""" + access_key = config.get("exchangerate", "access_key") + data_dir = config.get("data", "dir") + + now = datetime.now() + now_str = now.strftime("%Y-%m-%d_%H:%M") + fx_dir = os.path.join(data_dir, "fx") + existing_data = os.listdir(fx_dir) + existing = [f for f in existing_data if f.endswith("_GBPUSD.json")] + if existing: + recent_filename = max(existing) + recent = datetime.strptime(recent_filename, "%Y-%m-%d_%H:%M_GBPUSD.json") + delta = now - recent + + if existing and delta < timedelta(hours=6): + full = os.path.join(fx_dir, recent_filename) + data = json.load(open(full), parse_float=Decimal) + if "quotes" in data and "USDGBP" in data["quotes"]: + return typing.cast(Decimal, 1 / data["quotes"]["USDGBP"]) + + url = "http://api.exchangerate.host/live" + params = {"currencies": "GBP,USD", "access_key": access_key} + + filename = f"{fx_dir}/{now_str}_GBPUSD.json" + r = requests.get(url, params=params) + open(filename, "w").write(r.text) + data = json.loads(r.text, parse_float=Decimal) + + return typing.cast(Decimal, 1 / data["quotes"]["USDGBP"]) diff --git a/agenda/markets.py b/agenda/markets.py new file mode 100644 index 0000000..8bf042d --- /dev/null +++ b/agenda/markets.py @@ -0,0 +1,120 @@ +"""Market days.""" + +from datetime import date, datetime, time, timedelta + +import pytz +from dateutil.relativedelta import SA, relativedelta + +from .types import Event + + +def windmill_hill(start_date: date) -> list[Event]: + """Windmill Hill Market days for the next 12 months from a given date.""" + events: list[Event] = [] + current_date = start_date + url = ( + "https://www.windmillhillcityfarm.org.uk" + + "/visit-us/shops-more/windmill-hill-market-bristol-market/" + ) + + # To keep count of how many market days have been calculated + count = 0 + + tz = pytz.timezone("Europe/London") + start = time(10, 0) + end = time(15, 0) + + while count < 12: + # Skip months outside of April to December + if current_date.month < 4 or current_date.month > 12: + current_date += relativedelta(months=1) + current_date = date(current_date.year, current_date.month, 1) + continue + + # Calculate the first Saturday of the current month + first_saturday = current_date + relativedelta(day=1, weekday=SA(+1)) + + # Include it in the list only if it's on or after the start_date + if first_saturday >= start_date: + events.append( + Event( + name="market", + title="Windmill Hill Market", + date=tz.localize(datetime.combine(first_saturday, start)), + end_date=tz.localize(datetime.combine(first_saturday, end)), + url=url, + ) + ) + count += 1 + + # Move to the next month + current_date += relativedelta(months=1) + current_date = date(current_date.year, current_date.month, 1) + + return events + + +def tobacco_factory(start_date: date) -> list[Event]: + """Tobacco Factory Market days for the next 12 months from a given date.""" + events: list[Event] = [] + current_date = start_date + count = 0 + + url = "https://tobaccofactory.com/whats-on/sunday-market/" + + tz = pytz.timezone("Europe/London") + start = time(10, 0) + end = time(14, 30) + + while count < 52: # 52 weeks in a year + # Calculate the next Sunday from the current date + next_sunday = current_date + relativedelta(weekday=6) # Sunday is 6 + + # Include it in the list only if it's on or after the start_date + if next_sunday >= start_date: + events.append( + Event( + name="market", + title="Tobacco Factory Sunday Market", + date=tz.localize(datetime.combine(next_sunday, start)), + end_date=tz.localize(datetime.combine(next_sunday, end)), + url=url, + ) + ) + count += 1 + + # Move to the next week + current_date += timedelta(weeks=1) + + return events + + +def nailsea_farmers(start_date: date) -> list[Event]: + """Nailsea Farmers Market days for the next 12 months from a given date.""" + events: list[Event] = [] + current_date = start_date + count = 0 + + tz = pytz.timezone("Europe/London") + t = time(9, 0) # The market starts at 9am + + while count < 12: + # Calculate the 3rd Saturday of the current month + third_saturday = current_date + relativedelta(day=1, weekday=SA(+3)) + + # Include it in the list only if it's on or after the start_date + if third_saturday >= start_date: + events.append( + Event( + name="market", + title="Nailsea Farmers Market", + date=tz.localize(datetime.combine(third_saturday, t)), + ) + ) + count += 1 + + # Move to the next month + current_date += relativedelta(months=1) + current_date = date(current_date.year, current_date.month, 1) + + return events