#!/usr/bin/python3 """Load YAML data to ensure validity.""" import os import sys import typing from datetime import date, timedelta import yaml from rich.pretty import pprint import agenda import agenda.conference import agenda.data import agenda.travel import agenda.trip import agenda.types config = __import__("config.default", fromlist=[""]) data_dir = config.PERSONAL_DATA currencies = set(config.CURRENCIES + ["GBP"]) def check_currency(item: agenda.types.StrDict) -> None: """Throw error if currency is not in config.""" currency = item.get("currency") if not currency or currency in currencies: return None pprint(item) print(f"currency {currency!r} not in {currencies!r}") sys.exit(-1) def check_trips() -> None: """Check trips.""" trip_list = agenda.trip.build_trip_list(data_dir) print(len(trip_list), "trips") coords, routes = agenda.trip.get_coordinates_and_routes(trip_list, data_dir) print(len(coords), "coords") print(len(routes), "routes") def check_flights(airlines: set[str]) -> None: """Check flights.""" bookings = agenda.travel.parse_yaml("flights", data_dir) for booking in bookings: assert all(flight["airline"] in airlines for flight in booking["flights"]) for booking in bookings: check_currency(booking) print(len(bookings), "flights") def check_trains() -> None: """Check trains.""" trains = agenda.travel.parse_yaml("trains", data_dir) print(len(trains), "trains") def check_conferences() -> None: """Check conferences.""" filepath = os.path.join(data_dir, "conferences.yaml") conferences = [ agenda.conference.Conference(**conf) for conf in yaml.safe_load(open(filepath, "r")) ] for conf in conferences: if not conf.currency or conf.currency in currencies: continue pprint(conf) print(f"currency {conf.currency!r} not in {currencies!r}") sys.exit(-1) print(len(conferences), "conferences") def check_events() -> None: """Check events.""" today = date.today() last_year = today - timedelta(days=365) next_year = today + timedelta(days=2 * 365) events = agenda.events_yaml.read(data_dir, last_year, next_year) print(len(events), "events") def check_coordinates(item: agenda.types.StrDict) -> None: """Check coordinate are valid.""" if "latitude" not in item and "longitude" not in item: return assert "latitude" in item and "longitude" in item assert all(isinstance(item[key], (int, float)) for key in ("latitude", "longitude")) def check_accommodation() -> None: """Check accommodation.""" filepath = os.path.join(data_dir, "accommodation.yaml") accommodation_list = yaml.safe_load(open(filepath)) required_fields = ["type", "name", "country", "location", "trip", "from", "to"] for stay in accommodation_list: try: assert all(field in stay for field in required_fields) check_coordinates(stay) except AssertionError: pprint(stay) raise check_currency(stay) print(len(accommodation_list), "stays") def check_airports() -> None: """Check airports.""" airports = typing.cast( dict[str, agenda.types.StrDict], agenda.travel.parse_yaml("airports", data_dir) ) print(len(airports), "airports") for airport in airports.values(): assert "country" in airport assert agenda.get_country(airport["country"]) def check_stations() -> None: """Check stations.""" stations = agenda.travel.parse_yaml("stations", data_dir) print(len(stations), "stations") for station in stations: assert "country" in station assert agenda.get_country(station["country"]) def check_airlines() -> list[agenda.types.StrDict]: """Check airlines.""" airlines = agenda.travel.parse_yaml("airlines", data_dir) print(len(airlines), "airlines") for airline in airlines: assert airline.keys() == {"icao", "iata", "name"} assert len(airline["icao"]) == 3 assert len(airline["iata"]) == 2 return airlines def check() -> None: """Validate personal data YAML files.""" airlines = check_airlines() check_trips() check_flights({airline["iata"] for airline in airlines}) check_trains() check_conferences() check_events() check_accommodation() check_airports() check_stations() if __name__ == "__main__": check()