1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
|
# Import modules
import os
from pathlib import Path
import pandas as pd
import requests
import traceback
from datetime import datetime
from plotnine import *
import sys
import threading
import itertools
import time
class Spinner:
def __init__(self, message, delay=0.1):
self.spinner = itertools.cycle(['-', '/', '|', '\\'])
self.delay = delay
self.busy = False
self.spinner_visible = False
sys.stdout.write(message)
def write_next(self):
with self._screen_lock:
if not self.spinner_visible:
sys.stdout.write(next(self.spinner))
self.spinner_visible = True
sys.stdout.flush()
def remove_spinner(self, cleanup=False):
with self._screen_lock:
if self.spinner_visible:
sys.stdout.write('\b')
self.spinner_visible = False
if cleanup:
sys.stdout.write(' ') # overwrite spinner with blank
sys.stdout.write('\r') # move to next line
sys.stdout.flush()
def spinner_task(self):
while self.busy:
self.write_next()
time.sleep(self.delay)
self.remove_spinner()
def __enter__(self):
if sys.stdout.isatty():
self._screen_lock = threading.Lock()
self.busy = True
self.thread = threading.Thread(target=self.spinner_task)
self.thread.start()
def __exit__(self, exception, value, tb):
if sys.stdout.isatty():
self.busy = False
self.remove_spinner(cleanup=True)
else:
sys.stdout.write('\r')
# VARIABLES AND INITIALISATION
url_cases = 'https://coronavirus.data.gov.uk/downloads/csv/coronavirus-cases_latest.csv'
url_deaths = 'https://coronavirus.data.gov.uk/downloads/csv/coronavirus-deaths_latest.csv'
def getStats():
cwd = os.path.dirname(os.path.realpath(__file__))
path = Path("stats/")
path_to_stats = cwd / path
print("[*] Checking if 'stats' folder is present...")
if not os.path.exists(path_to_stats):
print("[*] Does not exist. Creating 'stats' folder...")
os.makedirs(path_to_stats)
else:
try:
unix_datetime_cases = datetime.utcfromtimestamp(
os.path.getmtime(os.path.join(
path_to_stats, "covid-cases.csv")))
unix_datetime_deaths = datetime.utcfromtimestamp(os.path.getmtime(
os.path.join(path_to_stats, "covid-deaths.csv")))
print("[@] 'covid-cases.csv' last modified: {}".format(
unix_datetime_cases))
print("[@] 'covid-deaths.csv' last modified: {}".format(
unix_datetime_deaths))
if unix_datetime_cases.date() and unix_datetime_deaths.date() == datetime.today().date():
print("[@] Local data is up-to-date. Skipping download.")
return None
else:
pass
except Exception:
print("[/] It appears that the data has not been downloaded.")
print("[*] Attempting to download COVID-19 cases...")
try:
response = requests.get(url_cases)
with open(os.path.join(path_to_stats, "covid-cases.csv"), 'wb') as f:
f.write(response.content)
except Exception:
traceback.print_exc()
print("[/] Failed to download COVID-19 cases.")
else:
print("[@] Successfully downloaded COVID-19 cases!")
print("[*] Attempting to download COVID-19 deaths...")
try:
response = requests.get(url_deaths)
with open(os.path.join(path_to_stats, "covid-deaths.csv"), 'wb') as f:
f.write(response.content)
except Exception:
print("[/] Failed to download COVID-19 deaths!")
else:
print("[@] Successfully downloaded COVID-19 deaths!")
def csv_parser():
# Get abs path
cwd = os.path.dirname(os.path.realpath(__file__))
path = Path("stats/")
path_to_stats = cwd / path
# Get path to files
covid_case_file_path = path_to_stats / "covid-cases.csv"
covid_death_file_path = path_to_stats / "covid-deaths.csv"
# Read CSV files
cases = pd.read_csv(covid_case_file_path).dropna()
deaths = pd.read_csv(covid_death_file_path).dropna()
# Get England data
cases_filtered_england = cases.loc[cases['Area name'] == "England"]
deaths_filtered_england = deaths.loc[deaths['Area name'] == "England"]
# Sort by date
cases_sorted = cases_filtered_england.sort_values(by='Specimen date')
deaths_sorted = deaths_filtered_england.sort_values(by='Reporting date')
# Convert date strings to datetime objects
cases_sorted['Specimen date'] = pd.to_datetime(cases_sorted['Specimen date'])
deaths_sorted['Reporting date'] = pd.to_datetime(deaths_sorted['Reporting date'])
# Return data from function
return cases_sorted, deaths_sorted
# First, get latest stats using getStats()
getStats()
# Then, parse the data: we only need the dates and their corresponding values.
imported = csv_parser()
cases = imported[0]
deaths = imported[1]
# Temporary variables
date_today = datetime.now().strftime('%d-%m-%Y')
cwd = os.path.dirname(os.path.realpath(__file__))
path_2 = Path("graphs/")
path_to_graphs = cwd / path_2
print("[*] Checking if 'graphs' folder is present...")
if not os.path.exists(path_to_graphs):
print("[*] Does not exist. Creating 'graphs' folder...")
os.makedirs(path_to_graphs)
with Spinner("Creating and saving the daily cases plot... "):
p1 = ggplot(cases, aes(x="Specimen date", y="Daily lab-confirmed cases", group = 1)) + geom_point() + geom_line() + labs(title = "Daily COVID-19 Cases") + scale_x_date(date_breaks = "3 days") + stat_smooth(method='mavg', method_args={'window': 3}, color='cyan', show_legend=True) + stat_smooth(method='mavg', method_args={'window': 7}, color='blue') + theme(axis_text_x=element_text(rotation=45, hjust=1))
p1.save(filename=('cases_daily_' + str(date_today)),path=path_to_graphs,height=5, width=20, units = 'in', dpi=1000, verbose = False)
with Spinner("Creating and saving the daily change in deaths plot... "):
p2 = ggplot(deaths, aes(x="Reporting date", y="Daily change in deaths", group = 1)) + geom_point() + geom_line() + labs(title = "Daily Change in Deaths") + scale_x_date(date_breaks = "3 days") + stat_smooth(method='mavg', method_args={'window': 3}, color='cyan') + stat_smooth(method='mavg', method_args={'window': 7}, color='blue') + theme(axis_text_x=element_text(rotation=45, hjust=1))
p2.save(filename = ('deaths_change_daily_' + str(date_today)),path=path_to_graphs,height=5, width=20, units = 'in', dpi=1000, verbose = False)
with Spinner("Creating and saving the cumulative cases plot... "):
p3 = ggplot(cases, aes(x="Specimen date", y="Cumulative lab-confirmed cases", group = 1)) + geom_point() + geom_line() + labs(title = "Cumulative Cases") + scale_x_date(date_breaks = "3 days") + theme(axis_text_x=element_text(rotation=45, hjust=1))
p3.save(filename = ('cumulative_cases_' + str(date_today)),path=path_to_graphs,height=5, width=20, units = 'in', dpi=1000, verbose = False)
with Spinner("Creating and saving the cumulative deaths plot... "):
p4 = ggplot(deaths, aes(x="Reporting date", y="Cumulative deaths", group = 1)) + geom_point() + geom_line() + labs(title = "Cumulative Deaths") + scale_x_date(date_breaks = "3 days") + theme(axis_text_x=element_text(rotation=45, hjust=1))
p4.save(filename = ('cumulative_deaths_' + str(date_today)),path=path_to_graphs,height=5, width=20, units = 'in', dpi=1000, verbose = False)
|
