diff --git a/COVID19Py/covid19.py b/COVID19Py/covid19.py index a68faa7..bec33f3 100644 --- a/COVID19Py/covid19.py +++ b/COVID19Py/covid19.py @@ -3,85 +3,40 @@ import json class COVID19(object): - default_url = "https://covid-tracker-us.herokuapp.com" - url = "" - data_source = "" - previousData = None - latestData = None - _valid_data_sources = [] - - mirrors_source = "https://raw.github.com/Kamaropoulos/COVID19Py/master/mirrors.json" - mirrors = None - - def __init__(self, url="https://covid-tracker-us.herokuapp.com", data_source='jhu'): - # Skip mirror checking if custom url was passed - if url == self.default_url: - # Load mirrors - response = requests.get(self.mirrors_source) - response.raise_for_status() - self.mirrors = response.json() - - # Try to get sources as a test - for mirror in self.mirrors: - # Set URL of mirror - self.url = mirror["url"] - result = None - try: - result = self._getSources() - except Exception as e: - # URL did not work, reset it and move on - self.url = "" - continue - - # TODO: Should have a better health-check, this is way too hacky... - if "jhu" in result: - # We found a mirror that worked just fine, let's stick with it - break - - # None of the mirrors worked. Raise an error to inform the user. - raise RuntimeError("No available API mirror was found.") - - else: - self.url = url + data = None - self._valid_data_sources = self._getSources() - if data_source not in self._valid_data_sources: - raise ValueError("Invalid data source. Expected one of: %s" % self._valid_data_sources) - self.data_source = data_source + def __init__(self, data=None): + if data is not None: + self.data = data def _update(self, timelines): latest = self.getLatest() - locations = self.getLocations(timelines) - if self.latestData: - self.previousData = self.latestData - self.latestData = { + locations = self.getAllLocations(timelines) + if self.data.latestData: + self.data.previousData = self.data.latestData + self.data.latestData = { "latest": latest, "locations": locations } - def _getSources(self): - response = requests.get(self.url + "/v2/sources") - response.raise_for_status() - return response.json()["sources"] - def _request(self, endpoint, params=None): if params is None: params = {} - response = requests.get(self.url + endpoint, {**params, "source":self.data_source}) + response = requests.get(self.data.url + endpoint, {**params, "source":self.data.data_source}) response.raise_for_status() return response.json() def getAll(self, timelines=False): self._update(timelines) - return self.latestData + return self.data.latestData def getLatestChanges(self): changes = None - if self.previousData: + if self.data.previousData: changes = { - "confirmed": self.latestData["latest"]["confirmed"] - self.latestData["latest"]["confirmed"], - "deaths": self.latestData["latest"]["deaths"] - self.latestData["latest"]["deaths"], - "recovered": self.latestData["latest"]["recovered"] - self.latestData["latest"]["recovered"], + "confirmed": self.data.latestData["latest"]["confirmed"] - self.data.latestData["latest"]["confirmed"], + "deaths": self.data.latestData["latest"]["deaths"] - self.data.latestData["latest"]["deaths"], + "recovered": self.data.latestData["latest"]["recovered"] - self.data.latestData["latest"]["recovered"], } else: changes = { @@ -98,7 +53,7 @@ def getLatest(self) -> List[Dict[str, int]]: data = self._request("/v2/latest") return data["latest"] - def getLocations(self, timelines=False, rank_by: str = None) -> List[Dict]: + def getAllLocations(self, timelines=False, rank_by: str = None) -> List[Dict]: """ Gets all locations affected by COVID-19, as well as latest case data. :param timelines: Whether timeline information should be returned as well. @@ -123,36 +78,168 @@ def getLocations(self, timelines=False, rank_by: str = None) -> List[Dict]: return data - def getLocationByCountryCode(self, country_code, timelines=False) -> List[Dict]: + # def getLocationData(self, location, timelines=False): + # """ + # :param location: Location containing any combination of strings denoting... + # - the name of the country + # - the ISO 3166-1 alpha-2 code (https://en.wikipedia.org/wiki/ISO_3166-1_alpha-2) of the country + # - the country id, an int + # :param timelines: Whether timeline information should be returned as well. + # :return: Either a list of areas that correspond to the location, or a dictionary with case information for the specified location. + # """ + # # retrieve by country name + # if location.country != "": + # data = None + # if timelines: + # data = self._request("/v2/locations", {"country": location.country, "timelines": str(timelines).lower()}) + # else: + # data = self._request("/v2/locations", {"country": location.country}) + # return data["locations"] + + # # retrieve by country code + # if location.country_code != "": + # data = None + # if timelines: + # data = self._request("/v2/locations", {"country_code": location.country_code, "timelines": str(timelines).lower()}) + # else: + # data = self._request("/v2/locations", {"country_code": location.country_code}) + # return data["locations"] + + # # retrieve by country id + # if location.country_id != "": + # data = self._request("/v2/locations/" + str(location.country_id)) + # return data["location"] + + def getLocationData(self, locationGroup, timelines=False): """ - :param country_code: String denoting the ISO 3166-1 alpha-2 code (https://en.wikipedia.org/wiki/ISO_3166-1_alpha-2) of the country + :param location: Location containing any combination of strings denoting... + - the name of the country + - the ISO 3166-1 alpha-2 code (https://en.wikipedia.org/wiki/ISO_3166-1_alpha-2) of the country + - the country id, an int :param timelines: Whether timeline information should be returned as well. - :return: A list of areas that correspond to the country_code. If the country_code is invalid, it returns an empty list. + :return: Either a list of areas that correspond to the location, or a dictionary with case information for the specified location. """ - data = None - if timelines: - data = self._request("/v2/locations", {"country_code": country_code, "timelines": str(timelines).lower()}) - else: - data = self._request("/v2/locations", {"country_code": country_code}) - return data["locations"] - - def getLocationByCountry(self, country, timelines=False) -> List[Dict]: - """ - :param country: String denoting name of the country - :param timelines: Whether timeline information should be returned as well. - :return: A list of areas that correspond to the country name. If the country is invalid, it returns an empty list. - """ - data = None - if timelines: - data = self._request("/v2/locations", {"country": country, "timelines": str(timelines).lower()}) + + output = [] + + for location in locationGroup.locList: + # retrieve by country name + if location.country != "": + data = None + if timelines: + data = self._request("/v2/locations", {"country": location.country, "timelines": str(timelines).lower()}) + else: + data = self._request("/v2/locations", {"country": location.country}) + output += data["locations"] + + # retrieve by country code + elif location.country_code != "": + data = None + if timelines: + data = self._request("/v2/locations", {"country_code": location.country_code, "timelines": str(timelines).lower()}) + else: + data = self._request("/v2/locations", {"country_code": location.country_code}) + output += data["locations"] + + # retrieve by country id + elif location.country_id != "": + data = self._request("/v2/locations/" + str(location.country_id)) + output += data["location"] + + return output + +class Data: + __instance = None + + default_url = "https://covid-tracker-us.herokuapp.com" + url = "" + data_source = "" + previousData = None + latestData = None + _valid_data_sources = [] + + mirrors_source = "https://raw.github.com/Kamaropoulos/COVID19Py/master/mirrors.json" + mirrors = None + + def __init__(self, url="https://covid-tracker-us.herokuapp.com", data_source="jhu"): + if Data.__instance != None: + raise Exception("Singleton!") else: - data = self._request("/v2/locations", {"country": country}) - return data["locations"] + Data.__instance = self + + # Skip mirror checking if custom url was passed + if url == Data.__instance.default_url: + # Load mirrors + response = requests.get(Data.__instance.mirrors_source) + response.raise_for_status() + Data.__instance.mirrors = response.json() + + # Try to get sources as a test + for mirror in Data.__instance.mirrors: + # Set URL of mirror + Data.__instance.url = mirror["url"] + result = None + try: + result = Data._getSources() + except Exception as e: + # URL did not work, reset it and move on + Data.__instance.url = "" + continue + + # TODO: Should have a better health-check, this is way too hacky... + if "jhu" in result: + # We found a mirror that worked just fine, let's stick with it + break + + # None of the mirrors worked. Raise an error to inform the user. + raise RuntimeError("No available API mirror was found.") + + else: + Data.__instance.url = url + + Data.__instance._valid_data_sources = Data._getSources() + if data_source not in Data.__instance._valid_data_sources: + raise ValueError("Invalid data source. Expected one of: %s" % Data.__instance._valid_data_sources) + Data.__instance.data_source = data_source + + def Instance(): + if Data.__instance == None: + Data() + return Data.__instance + + @staticmethod + def _getSources(): + response = requests.get(Data.__instance.url + "/v2/sources") + response.raise_for_status() + return response.json()["sources"] - def getLocationById(self, country_id: int): - """ - :param country_id: Country Id, an int - :return: A dictionary with case information for the specified location. - """ - data = self._request("/v2/locations/" + str(country_id)) - return data["location"] +class Location(object): + country = "" + country_code = "" + country_id = "" + + def __init__(self, country=None, country_code=None, country_id=None): + if country is not None: + self.country = country + if country_code is not None: + self.country_code = country_code + if country_id is not None: + self.country_id = country_id + + def __str__(self): + if self.country == "" and self.country_code == "" and self.country_id == "": + return "Warning! Invalid location provided." + + return "{: <25} ({}) \tid:{}".format(self.country, self.country_code, self.country_id) + +class LocationGroup(object): + locList = [] + + def __init__(self, locList=None): + if locList is not None: + self.locList = locList + + +# Testing +if __name__ == '__main__': + print(COVID19(Data()).getLocationData(LocationGroup([Location(country_code="US"), Location(country_code="CA")]))) \ No newline at end of file