kerykeion.fetch_geonames

  1# -*- coding: utf-8 -*-
  2"""
  3    This is part of Kerykeion (C) 2024 Giacomo Battaglia
  4"""
  5
  6
  7import logging
  8from requests import Request
  9from requests_cache import CachedSession
 10from typing import Union
 11
 12
 13class FetchGeonames:
 14    """
 15    Class to handle requests to the GeoNames API
 16
 17    Args:
 18    city_name (str): Name of the city
 19    country_code (str): Two letters country code
 20    username (str, optional): GeoNames username, defaults to "century.boy".
 21    """
 22
 23    def __init__(
 24        self,
 25        city_name: str,
 26        country_code: str,
 27        username: str = "century.boy",
 28    ):
 29        self.session = CachedSession(
 30            cache_name="cache/kerykeion_geonames_cache",
 31            backend="sqlite",
 32            expire_after=86400,
 33        )
 34
 35        self.username = username
 36        self.city_name = city_name
 37        self.country_code = country_code
 38        self.base_url = "http://api.geonames.org/searchJSON"
 39        self.timezone_url = "http://api.geonames.org/timezoneJSON"
 40
 41    def __get_timezone(self, lat: Union[str, float, int], lon: Union[str, float, int]) -> dict[str, str]:
 42        """
 43        Get the timezone for a given latitude and longitude
 44        """
 45        # Dictionary that will be returned:
 46        timezone_data = {}
 47
 48        params = {"lat": lat, "lng": lon, "username": self.username}
 49
 50        prepared_request = Request("GET", self.timezone_url, params=params).prepare()
 51        logging.debug(f"Requesting data from GeoName timezones: {prepared_request.url}")
 52
 53        try:
 54            response = self.session.send(prepared_request)
 55            response_json = response.json()
 56
 57        except Exception as e:
 58            logging.error(f"Error fetching {self.timezone_url}: {e}")
 59            return {}
 60
 61        try:
 62            timezone_data["timezonestr"] = response_json["timezoneId"]
 63
 64        except Exception as e:
 65            logging.error(f"Error serializing data maybe wrong username? Details: {e}")
 66            return {}
 67
 68        if hasattr(response, "from_cache"):
 69            timezone_data["from_tz_cache"] = response.from_cache  # type: ignore
 70
 71        return timezone_data
 72
 73    def __get_contry_data(self, city_name: str, country_code: str) -> dict[str, str]:
 74        """
 75        Get the city data *whitout timezone* for a given city and country name
 76        """
 77        # Dictionary that will be returned:
 78        city_data_whitout_tz = {}
 79
 80        params = {
 81            "q": city_name,
 82            "contry": country_code,
 83            "username": self.username,
 84            "maxRows": 1,
 85            "style": "SHORT",
 86            "featureClass": "A",
 87            "featureClass": "P",
 88        }
 89
 90        prepared_request = Request("GET", self.base_url, params=params).prepare()
 91        logging.debug(f"Requesting data from geonames basic: {prepared_request.url}")
 92
 93        try:
 94            response = self.session.send(prepared_request)
 95            response_json = response.json()
 96            logging.debug(f"Response from GeoNames: {response_json}")
 97
 98        except Exception as e:
 99            logging.error(f"Error in fetching {self.base_url}: {e}")
100            return {}
101
102        try:
103            city_data_whitout_tz["name"] = response_json["geonames"][0]["name"]
104            city_data_whitout_tz["lat"] = response_json["geonames"][0]["lat"]
105            city_data_whitout_tz["lng"] = response_json["geonames"][0]["lng"]
106            city_data_whitout_tz["countryCode"] = response_json["geonames"][0]["countryCode"]
107
108        except Exception as e:
109            logging.error(f"Error serializing data maybe wrong username? Details: {e}")
110            return {}
111
112        if hasattr(response, "from_cache"):
113            city_data_whitout_tz["from_country_cache"] = response.from_cache  # type: ignore
114
115        return city_data_whitout_tz
116
117    def get_serialized_data(self) -> dict[str, str]:
118        """
119        Returns all the data necessary for the Kerykeion calculation.
120
121        Returns:
122            dict[str, str]: _description_
123        """
124        city_data_response = self.__get_contry_data(self.city_name, self.country_code)
125        try:
126            timezone_response = self.__get_timezone(city_data_response["lat"], city_data_response["lng"])
127
128        except Exception as e:
129            logging.error(f"Error in fetching timezone: {e}")
130            return {}
131
132        return {**timezone_response, **city_data_response}
133
134
135if __name__ == "__main__":
136    from kerykeion.utilities import setup_logging
137    setup_logging(level="debug")
138
139    geonames = FetchGeonames("Montichiari", "IT")
140    print(geonames.get_serialized_data())