171 lines
4.7 KiB
Python
Executable file
171 lines
4.7 KiB
Python
Executable file
#!/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 and ensure they are in chronological order."""
|
|
bookings = agenda.travel.parse_yaml("flights", data_dir)
|
|
|
|
prev_first_depart = None
|
|
for booking in bookings:
|
|
assert all(flight["airline"] in airlines for flight in booking["flights"])
|
|
check_currency(booking)
|
|
|
|
if prev_first_depart:
|
|
assert (
|
|
booking["flights"][0]["depart"] > prev_first_depart
|
|
), "Bookings are not in chronological order by first flight's departure."
|
|
prev_first_depart = booking["flights"][0]["depart"]
|
|
|
|
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()
|