diff --git a/app/location/__init__.py b/app/location/__init__.py index 1da5e9e5..b4620292 100644 --- a/app/location/__init__.py +++ b/app/location/__init__.py @@ -5,120 +5,121 @@ # pylint: disable=redefined-builtin,invalid-name -class Location: # pylint: disable=too-many-instance-attributes - """ - A location in the world affected by the coronavirus. - """ - - def __init__( - self, id, country, province, coordinates, last_updated, confirmed, deaths, recovered, - ): # pylint: disable=too-many-arguments - # General info. - self.id = id - self.country = country.strip() - self.province = province.strip() - self.coordinates = coordinates - - # Last update. - self.last_updated = last_updated - - # Statistics. - self.confirmed = confirmed - self.deaths = deaths - self.recovered = recovered - - @property - def country_code(self): - """ - Gets the alpha-2 code represention of the country. Returns 'XX' if none is found. - - :returns: The country code. - :rtype: str - """ - return (countries.country_code(self.country) or countries.DEFAULT_COUNTRY_CODE).upper() - - @property - def country_population(self): - """ - Gets the population of this location. - - :returns: The population. - :rtype: int - """ - return country_population(self.country_code) - - def serialize(self): - """ - Serializes the location into a dict. - - :returns: The serialized location. - :rtype: dict - """ - return { - # General info. - "id": self.id, - "country": self.country, - "country_code": self.country_code, - "country_population": self.country_population, - "province": self.province, - # Coordinates. - "coordinates": self.coordinates.serialize(), - # Last updated. - "last_updated": self.last_updated, - # Latest data (statistics). - "latest": { - "confirmed": self.confirmed, - "deaths": self.deaths, - "recovered": self.recovered, - }, - } - - -class TimelinedLocation(Location): - """ - A location with timelines. - """ - - # pylint: disable=too-many-arguments - def __init__(self, id, country, province, coordinates, last_updated, timelines): - super().__init__( - # General info. - id, - country, - province, - coordinates, - last_updated, - # Statistics (retrieve latest from timelines). - confirmed=timelines.get("confirmed").latest or 0, - deaths=timelines.get("deaths").latest or 0, - recovered=timelines.get("recovered").latest or 0, - ) - - # Set timelines. - self.timelines = timelines - - # pylint: disable=arguments-differ - def serialize(self, timelines=False): - """ - Serializes the location into a dict. - - :param timelines: Whether to include the timelines. - :returns: The serialized location. - :rtype: dict - """ - serialized = super().serialize() - - # Whether to include the timelines or not. - if timelines: - serialized.update( - { - "timelines": { - # Serialize all the timelines. - key: value.serialize() - for (key, value) in self.timelines.items() - } - } - ) - - # Return the serialized location. - return serialized +class Locations: + class Location: # pylint: disable=too-many-instance-attributes + """ + A location in the world affected by the coronavirus. + """ + + def __init__( + self, id, country, province, coordinates, last_updated, confirmed, deaths, recovered, + ): # pylint: disable=too-many-arguments + # General info. + self.id = id + self.country = country.strip() + self.province = province.strip() + self.coordinates = coordinates + + # Last update. + self.last_updated = last_updated + + # Statistics. + self.confirmed = confirmed + self.deaths = deaths + self.recovered = recovered + + @property + def country_code(self): + """ + Gets the alpha-2 code represention of the country. Returns 'XX' if none is found. + + :returns: The country code. + :rtype: str + """ + return (countries.country_code(self.country) or countries.DEFAULT_COUNTRY_CODE).upper() + + @property + def country_population(self): + """ + Gets the population of this location. + + :returns: The population. + :rtype: int + """ + return country_population(self.country_code) + + def serialize(self): + """ + Serializes the location into a dict. + + :returns: The serialized location. + :rtype: dict + """ + return { + # General info. + "id": self.id, + "country": self.country, + "country_code": self.country_code, + "country_population": self.country_population, + "province": self.province, + # Coordinates. + "coordinates": self.coordinates.serialize(), + # Last updated. + "last_updated": self.last_updated, + # Latest data (statistics). + "latest": { + "confirmed": self.confirmed, + "deaths": self.deaths, + "recovered": self.recovered, + }, + } + + + class TimelinedLocation(Location): + """ + A location with timelines. + """ + + # pylint: disable=too-many-arguments + def __init__(self, id, country, province, coordinates, last_updated, timelines): + super().__init__( + # General info. + id, + country, + province, + coordinates, + last_updated, + # Statistics (retrieve latest from timelines). + confirmed=timelines.get("confirmed").latest or 0, + deaths=timelines.get("deaths").latest or 0, + recovered=timelines.get("recovered").latest or 0, + ) + + # Set timelines. + self.timelines = timelines + + # pylint: disable=arguments-differ + def serialize(self, timelines=False): + """ + Serializes the location into a dict. + + :param timelines: Whether to include the timelines. + :returns: The serialized location. + :rtype: dict + """ + serialized = super().serialize() + + # Whether to include the timelines or not. + if timelines: + serialized.update( + { + "timelines": { + # Serialize all the timelines. + key: value.serialize() + for (key, value) in self.timelines.items() + } + } + ) + + # Return the serialized location. + return serialized diff --git a/app/services/location/jhu.py b/app/services/location/jhu.py index ebed3960..1cc90252 100644 --- a/app/services/location/jhu.py +++ b/app/services/location/jhu.py @@ -10,7 +10,7 @@ from ...caches import check_cache, load_cache from ...coordinates import Coordinates -from ...location import TimelinedLocation +from ...location import Locations from ...models import Timeline from ...utils import countries from ...utils import date as date_util @@ -169,10 +169,11 @@ async def get_locations(): # Grab coordinates. coordinates = location["coordinates"] + locationTimeLined = Locations.TimelinedLocation # Create location (supporting timelines) and append. locations.append( - TimelinedLocation( + locationTimeLined( # General info. index, location["country"], diff --git a/app/utils/populations.py b/app/utils/populations.py index c02f15a9..01bc97ee 100644 --- a/app/utils/populations.py +++ b/app/utils/populations.py @@ -10,51 +10,52 @@ GEONAMES_URL = "http://api.geonames.org/countryInfoJSON" GEONAMES_BACKUP_PATH = "geonames_population_mappings.json" +class Population: # Fetching of the populations. -def fetch_populations(save=False): - """ - Returns a dictionary containing the population of each country fetched from the GeoNames. - https://www.geonames.org/ + def fetch_populations(save=False): + """ + Returns a dictionary containing the population of each country fetched from the GeoNames. + https://www.geonames.org/ - TODO: only skip writing to the filesystem when deployed with gunicorn, or handle concurent access, or use DB. + TODO: only skip writing to the filesystem when deployed with gunicorn, or handle concurent access, or use DB. - :returns: The mapping of populations. - :rtype: dict - """ - LOGGER.info("Fetching populations...") + :returns: The mapping of populations. + :rtype: dict + """ + LOGGER.info("Fetching populations...") - # Mapping of populations - mappings = {} + # Mapping of populations + mappings = {} - # Fetch the countries. - try: - countries = requests.get(GEONAMES_URL, params={"username": "dperic"}, timeout=1.25).json()[ + # Fetch the countries. + try: + countries = requests.get(GEONAMES_URL, params={"username": "dperic"}, timeout=1.25).json()[ "geonames" - ] - # Go through all the countries and perform the mapping. - for country in countries: - mappings.update({country["countryCode"]: int(country["population"]) or None}) - - if mappings and save: - LOGGER.info(f"Saving population data to {app.io.save(GEONAMES_BACKUP_PATH, mappings)}") - except (json.JSONDecodeError, KeyError, requests.exceptions.Timeout) as err: - LOGGER.warning(f"Error pulling population data. {err.__class__.__name__}: {err}") - mappings = app.io.load(GEONAMES_BACKUP_PATH) - LOGGER.info(f"Using backup data from {GEONAMES_BACKUP_PATH}") - # Finally, return the mappings. - LOGGER.info("Fetched populations") - return mappings - - -# Mapping of alpha-2 codes country codes to population. -POPULATIONS = fetch_populations() - -# Retrieving. -def country_population(country_code, default=None): - """ - Fetches the population of the country with the provided country code. - - :returns: The population. - :rtype: int - """ - return POPULATIONS.get(country_code, default) + ] + # Go through all the countries and perform the mapping. + for country in countries: + mappings.update({country["countryCode"]: int(country["population"]) or None}) + + if mappings and save: + LOGGER.info(f"Saving population data to {app.io.save(GEONAMES_BACKUP_PATH, mappings)}") + except (json.JSONDecodeError, KeyError, requests.exceptions.Timeout) as err: + LOGGER.warning(f"Error pulling population data. {err.__class__.__name__}: {err}") + mappings = app.io.load(GEONAMES_BACKUP_PATH) + LOGGER.info(f"Using backup data from {GEONAMES_BACKUP_PATH}") + # Finally, return the mappings. + LOGGER.info("Fetched populations") + return mappings + + + # Mapping of alpha-2 codes country codes to population. + POPULATIONS = fetch_populations() + + # Retrieving. + def country_population(country_code, default=None): + """ + Fetches the population of the country with the provided country code. + + :returns: The population. + :rtype: int + """ + return Population.POPULATIONS.get(country_code, default)