Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Aggregation Pull Request #1 #51

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Aggregation Pull Request #1 #51

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4ea9a16
Inserted comments and made minor changes
Zy8712 Mar 17, 2021
a569f72
Aggregation Pull Request #1
Zy8712 Mar 17, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
117 changes: 77 additions & 40 deletions COVID19Py/covid19.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,14 @@
import requests
import json

class COVID19(object):
default_url = "https://covid-tracker-us.herokuapp.com"
#class acting as an aggregate for all operations related to setting up the user's data retrieval request
class UserRequestSetup(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_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'):
Expand Down Expand Up @@ -49,6 +48,36 @@ def __init__(self, url="https://covid-tracker-us.herokuapp.com", data_source='jh
raise ValueError("Invalid data source. Expected one of: %s" % self._valid_data_sources)
self.data_source = data_source


def _getSources(self):
response = requests.get(self.url + "/v2/sources")
response.raise_for_status()
return response.json()["sources"]


#originally _request, underscore removed as it is now referred to by class COVID19DataRetrieval and requires access. Underscore indicates protected status
def request(self, endpoint, params=None):
if params is None:
params = {}
response = requests.get(self.url + endpoint, {*params, "source": self.data_source}) #fixed a syntax error (originally **)
response.raise_for_status()
return response.json()



#class acting as an aggregate for all data retrieval operations
class COVID19DataRetrieval(object):
#originally variables from the class COVID19. Variables are only used in the operations contained within the COVIDDataRetrieval class, hence they were moved here
__previousData = None
__latestData = None


#"constructor" method that initializes data members of the class, where applicable
def __init__(self, userrequestsetup)
self.userrequestsetup = userrequestsetup #enables/allows us to access methods in the UserRequestSetup class, mainly the request method


#updates to latest data
def _update(self, timelines):
latest = self.getLatest()
locations = self.getLocations(timelines)
Expand All @@ -59,29 +88,34 @@ def _update(self, timelines):
"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.raise_for_status()
return response.json()
#############################
# Generalized data searches
#############################

#get all data at once
def getAll(self, timelines=False):
self._update(timelines)
return self.latestData


#gets the latest COVID-19 case data
def getLatest(self) -> List[Dict[str, int]]:
"""
:return: The latest amount of total confirmed cases, deaths, and recoveries.
"""
data = self.userrequestsetup.request("/v2/latest")
return data["latest"]


#get the latest changes (i.e.latest deltas)
def getLatestChanges(self):
changes = None
if self.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.latestData["latest"]["confirmed"] - self.previousData["latest"]["confirmed"], #changed from latestData to previousData
"deaths": self.latestData["latest"]["deaths"] - self.previousData["latest"]["deaths"], #changed from latestData to previousData
"recovered": self.latestData["latest"]["recovered"] - self.previousData["latest"]["recovered"], #changed from latestData to previousData
}
else:
changes = {
Expand All @@ -91,13 +125,8 @@ def getLatestChanges(self):
}
return changes

def getLatest(self) -> List[Dict[str, int]]:
"""
:return: The latest amount of total confirmed cases, deaths, and recoveries.
"""
data = self._request("/v2/latest")
return data["latest"]

#get case data relating all locations with COVID-19
def getLocations(self, timelines=False, rank_by: str = None) -> List[Dict]:
"""
Gets all locations affected by COVID-19, as well as latest case data.
Expand All @@ -107,52 +136,60 @@ def getLocations(self, timelines=False, rank_by: str = None) -> List[Dict]:
"""
data = None
if timelines:
data = self._request("/v2/locations", {"timelines": str(timelines).lower()})
data = self.userrequestsetup.request("/v2/locations", {"timelines": str(timelines).lower()})
else:
data = self._request("/v2/locations")
data = self.userrequestsetup.request("/v2/locations")

data = data["locations"]

ranking_criteria = ['confirmed', 'deaths', 'recovered']
if rank_by is not None:
if rank_by not in ranking_criteria:
raise ValueError("Invalid ranking criteria. Expected one of: %s" % ranking_criteria)

ranked = sorted(data, key=lambda i: i['latest'][rank_by], reverse=True)
data = ranked

return data

def getLocationByCountryCode(self, country_code, timelines=False) -> List[Dict]:

###############################
# Country-based data searches
###############################

#get case data by method of country name
def getLocationByCountry(self, country, timelines=False) -> List[Dict]:
"""
: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 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_code. If the country_code is invalid, it returns an empty list.
: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_code": country_code, "timelines": str(timelines).lower()})
data = self.userrequestsetup.request("/v2/locations", {"country": country, "timelines": str(timelines).lower()})
else:
data = self._request("/v2/locations", {"country_code": country_code})
data = self.userrequestsetup.request("/v2/locations", {"country": country})
return data["locations"]

def getLocationByCountry(self, country, timelines=False) -> List[Dict]:


#get case data by method of country code
def getLocationByCountryCode(self, country_code, timelines=False) -> List[Dict]:
"""
:param country: String denoting name of the country
: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 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.
:return: A list of areas that correspond to the country_code. If the country_code is invalid, it returns an empty list.
"""
data = None
if timelines:
data = self._request("/v2/locations", {"country": country, "timelines": str(timelines).lower()})
data = self.userrequestsetup.request("/v2/locations", {"country_code": country_code, "timelines": str(timelines).lower()})
else:
data = self._request("/v2/locations", {"country": country})
data = self.userrequestsetup.request("/v2/locations", {"country_code": country_code})
return data["locations"]


#get case data by method of country id
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))
data = self.userrequestsetup.request("/v2/locations/" + str(country_id))
return data["location"]