Adjust bristol waste function names

This commit is contained in:
Edward Betts 2024-07-14 23:10:00 +08:00
parent 0f3f596cb3
commit 5c4eac60ee
3 changed files with 13 additions and 15 deletions

View file

@ -17,9 +17,7 @@ ttl_hours = 12
BristolSchedule = list[dict[str, typing.Any]] BristolSchedule = list[dict[str, typing.Any]]
async def get_bristol_data( async def get_data(data_dir: str, uprn: str, refresh: bool = False) -> BristolSchedule:
data_dir: str, uprn: str, refresh: bool = False
) -> BristolSchedule:
"""Get Bristol Waste schedule, with cache.""" """Get Bristol Waste schedule, with cache."""
now = datetime.now() now = datetime.now()
waste_dir = os.path.join(data_dir, "waste") waste_dir = os.path.join(data_dir, "waste")
@ -41,7 +39,7 @@ async def get_bristol_data(
return get_from_recent() return get_from_recent()
try: try:
r = await get_bristol_gov_uk_data(uprn) r = await get_web_data(uprn)
except httpx.ReadTimeout: except httpx.ReadTimeout:
return get_from_recent() return get_from_recent()
@ -51,7 +49,7 @@ async def get_bristol_data(
return typing.cast(BristolSchedule, r.json()["data"]) return typing.cast(BristolSchedule, r.json()["data"])
async def get_bristol_gov_uk_data(uprn: str) -> httpx.Response: async def get_web_data(uprn: str) -> httpx.Response:
"""Get JSON from Bristol City Council.""" """Get JSON from Bristol City Council."""
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
HEADERS = { HEADERS = {
@ -101,38 +99,38 @@ async def get_bristol_gov_uk_data(uprn: str) -> httpx.Response:
return response return response
def bristol_service(item: dict[str, typing.Any]) -> str: def get_service(item: dict[str, typing.Any]) -> str:
"""Bristol waste service name.""" """Bristol waste service name."""
service: str = item["containerName"] service: str = item["containerName"]
return "Recycling" if "Recycling" in service else service.partition(" ")[2] return "Recycling" if "Recycling" in service else service.partition(" ")[2]
def bristol_collections(item: dict[str, typing.Any]) -> typing.Iterable[date]: def collections(item: dict[str, typing.Any]) -> typing.Iterable[date]:
"""Bristol dates from collections.""" """Bristol dates from collections."""
for collection in item["collection"]: for collection in item["collection"]:
for collection_date_key in ["nextCollectionDate", "lastCollectionDate"]: for collection_date_key in ["nextCollectionDate", "lastCollectionDate"]:
yield date.fromisoformat(collection[collection_date_key][:10]) yield date.fromisoformat(collection[collection_date_key][:10])
def bristol_by_date(data: BristolSchedule, start_date: date) -> dict[date, list[str]]: def get_by_date(data: BristolSchedule, start_date: date) -> dict[date, list[str]]:
"""Build list of Bristol services by date.""" """Build list of Bristol services by date."""
by_date: defaultdict[date, list[str]] = defaultdict(list) by_date: defaultdict[date, list[str]] = defaultdict(list)
for item in data: for item in data:
service = bristol_service(item) service = get_service(item)
for d in bristol_collections(item): for d in collections(item):
if d < start_date and service not in by_date[d]: if d < start_date and service not in by_date[d]:
by_date[d].append(service) by_date[d].append(service)
return by_date return by_date
async def get_bristol_gov_uk( async def get(
start_date: date, data_dir: str, uprn: str, refresh: bool = False start_date: date, data_dir: str, uprn: str, refresh: bool = False
) -> list[Event]: ) -> list[Event]:
"""Get waste collection schedule from Bristol City Council.""" """Get waste collection schedule from Bristol City Council."""
data = await get_bristol_data(data_dir, uprn, refresh) data = await get_data(data_dir, uprn, refresh)
by_date = bristol_by_date(data, start_date) by_date = get_by_date(data, start_date)
return [ return [
Event(name="waste_schedule", date=d, title="Bristol: " + ", ".join(services)) Event(name="waste_schedule", date=d, title="Bristol: " + ", ".join(services))
for d, services in by_date.items() for d, services in by_date.items()

View file

@ -73,7 +73,7 @@ async def bristol_waste_collection_events(
data_dir: str, start_date: date, uprn: str data_dir: str, start_date: date, uprn: str
) -> list[Event]: ) -> list[Event]:
"""Waste colllection events.""" """Waste colllection events."""
return await bristol_waste.get_bristol_gov_uk(start_date, data_dir, uprn) return await bristol_waste.get(start_date, data_dir, uprn)
def find_events_during_stay( def find_events_during_stay(

View file

@ -39,7 +39,7 @@ async def update_bank_holidays(config: flask.config.Config) -> None:
async def update_bristol_bins(config: flask.config.Config) -> None: async def update_bristol_bins(config: flask.config.Config) -> None:
"""Update waste schedule from Bristol City Council.""" """Update waste schedule from Bristol City Council."""
t0 = time() t0 = time()
events = await agenda.bristol_waste.get_bristol_gov_uk( events = await agenda.bristol_waste.get(
date.today(), config["DATA_DIR"], config["BRISTOL_UPRN"], refresh=True date.today(), config["DATA_DIR"], config["BRISTOL_UPRN"], refresh=True
) )
time_taken = time() - t0 time_taken = time() - t0