From c5e88727ae201cb90de91e6d83db71c19167bca3 Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Tue, 22 Jul 2025 21:27:51 +0100 Subject: [PATCH] Command to get airport YAML Fixes #199 --- get_airport.py | 224 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 224 insertions(+) create mode 100755 get_airport.py diff --git a/get_airport.py b/get_airport.py new file mode 100755 index 0000000..3e6e29a --- /dev/null +++ b/get_airport.py @@ -0,0 +1,224 @@ +#!/usr/bin/python3 + +import sys +from typing import Any, Dict, List + +import requests +import yaml + +# Define the base URL for the Wikidata API +WIKIDATA_API_URL = "https://www.wikidata.org/w/api.php" + + +def get_entity_label(qid: str) -> str | None: + """ + Fetches the English label for a given Wikidata entity QID. + + Args: + qid (str): The Wikidata entity ID (e.g., "Q6106"). + + Returns: + Optional[str]: The English label of the entity, or None if not found. + """ + params: Dict[str, str] = { + "action": "wbgetentities", + "ids": qid, + "format": "json", + "props": "labels", + "languages": "en", + } + try: + response = requests.get(WIKIDATA_API_URL, params=params) + response.raise_for_status() + entity = response.json().get("entities", {}).get(qid, {}) + return entity.get("labels", {}).get("en", {}).get("value") + except requests.exceptions.RequestException as e: + print(f"Error fetching label for QID {qid}: {e}", file=sys.stderr) + return None + + +def get_entity_details(qid: str) -> dict[str, Any] | None: + """ + Fetches and processes detailed information for a given airport QID. + + Args: + qid (str): The QID of the airport Wikidata entity. + + Returns: + Optional[Dict[str, Any]]: A dictionary containing the detailed airport data. + """ + params: Dict[str, str] = { + "action": "wbgetentities", + "ids": qid, + "format": "json", + "props": "claims|labels", + } + try: + response = requests.get(WIKIDATA_API_URL, params=params) + response.raise_for_status() + entity = response.json().get("entities", {}).get(qid, {}) + if not entity: + return None + + claims = entity.get("claims", {}) + + # Helper to safely extract claim values + def get_simple_claim_value(prop_id: str) -> str | None: + claim = claims.get(prop_id) + if not claim: + return None + v = claim[0].get("mainsnak", {}).get("datavalue", {}).get("value") + assert isinstance(v, str) or v is None + return v + + # Get IATA code, name, and website + iata = get_simple_claim_value("P238") + name = entity.get("labels", {}).get("en", {}).get("value") + website = get_simple_claim_value("P856") + + # Get City Name by resolving its QID + city_qid_claim = claims.get("P131") + city_name = None + if city_qid_claim: + city_qid = ( + city_qid_claim[0] + .get("mainsnak", {}) + .get("datavalue", {}) + .get("value", {}) + .get("id") + ) + if city_qid: + city_name = get_entity_label(city_qid) + + # Get coordinates + coords_claim = claims.get("P625") + latitude, longitude = None, None + if coords_claim: + coords = ( + coords_claim[0] + .get("mainsnak", {}) + .get("datavalue", {}) + .get("value", {}) + ) + latitude = coords.get("latitude") + longitude = coords.get("longitude") + + # Get elevation + elevation_claim = claims.get("P2044") + elevation = None + if elevation_claim: + amount_str = ( + elevation_claim[0] + .get("mainsnak", {}) + .get("datavalue", {}) + .get("value", {}) + .get("amount") + ) + if amount_str: + elevation = float(amount_str) if "." in amount_str else int(amount_str) + + # Get Country Code + country_claim = claims.get("P17") + country_code = None + if country_claim: + country_qid = ( + country_claim[0] + .get("mainsnak", {}) + .get("datavalue", {}) + .get("value", {}) + .get("id") + ) + if country_qid: + # Fetch the ISO 3166-1 alpha-2 code (P297) for the country entity + country_code_params = { + "action": "wbgetclaims", + "entity": country_qid, + "property": "P297", + "format": "json", + } + country_res = requests.get(WIKIDATA_API_URL, params=country_code_params) + country_res.raise_for_status() + country_claims = country_res.json().get("claims", {}).get("P297") + if country_claims: + code = ( + country_claims[0] + .get("mainsnak", {}) + .get("datavalue", {}) + .get("value") + ) + if code: + country_code = code.lower() + + data = { + "iata": iata, + "name": name, + "city": city_name, + "qid": qid, + "latitude": latitude, + "longitude": longitude, + "elevation": elevation, + "website": website, + "country": country_code, + } + + # Return the final structure, filtering out null values for cleaner output + return {iata: {k: v for k, v in data.items() if v is not None}} + + except requests.exceptions.RequestException as e: + print(f"Error fetching entity details for QID {qid}: {e}", file=sys.stderr) + return None + + +def find_airport_by_iata(iata_code: str) -> dict[str, Any] | None: + """ + Finds an airport by its IATA code using Wikidata's search API. + + Args: + iata_code (str): The IATA code of the airport (e.g., "PDX"). + + Returns: + Optional[Dict[str, Any]]: A dictionary with the airport data or None. + """ + params: Dict[str, str] = { + "action": "query", + "list": "search", + "srsearch": f"haswbstatement:P238={iata_code.upper()}", + "format": "json", + } + try: + response = requests.get(WIKIDATA_API_URL, params=params) + response.raise_for_status() + search_results: List[Dict[str, Any]] = ( + response.json().get("query", {}).get("search", []) + ) + + if not search_results: + print(f"No airport found with IATA code: {iata_code}", file=sys.stderr) + return None + + qid = search_results[0]["title"] + return get_entity_details(qid) + + except requests.exceptions.RequestException as e: + print(f"Error searching on Wikidata API: {e}", file=sys.stderr) + return None + + +if __name__ == "__main__": + if len(sys.argv) != 2: + print("Usage: python airport_lookup.py ", file=sys.stderr) + sys.exit(1) + + iata_code_arg = sys.argv[1] + airport_data = find_airport_by_iata(iata_code_arg) + + if airport_data: + print( + yaml.safe_dump( + airport_data, + default_flow_style=False, + allow_unicode=True, + sort_keys=False, + ), + end="", + )