Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
COVID-19 tracker on a PiminiTFT
[Unit]
Description=COVID-19 Tracker Service
After=multi-user.target
[Service]
User=pi
Type=simple
WorkingDirectory=/home/pi/CovidTracker
ExecStart=/usr/bin/python3 /home/pi/CovidTracker/main.py --update 3600
Restart=on-failure
[Install]
WantedBy=multi-user.target
import argparse
import asyncio
import board
import datetime as dt
import digitalio
import json
import logging
import requests
from systemd import journal
from adafruit_rgb_display.rgb import color565
import adafruit_rgb_display.st7789 as st7789
from PIL import Image, ImageDraw, ImageFont
from stats import CovidStats
#COVID_19_API = "https://covid19api.herokuapp.com/"
# https://github.com/ExpDev07/coronavirus-tracker-api
COVID_19_API = "https://coronavirus-tracker-api.herokuapp.com/v2/"
HEADER = "COVID-19"
class CovidTracker:
def __init__(self, refresh_rate=600, width=240, height=240):
self.get_stats_interval_sec = refresh_rate
self.width = width
self.height = height
self.logger = logging.getLogger(__name__)
self.logger.propagate = False
self.logger.addHandler(journal.JournaldLogHandler())
self.logger.setLevel(logging.INFO)
cs_pin = digitalio.DigitalInOut(board.CE0)
dc_pin = digitalio.DigitalInOut(board.D25)
reset_pin = None
BAUDRATE = 64000000
self.display = st7789.ST7789(
board.SPI(),
cs=cs_pin,
dc=dc_pin,
rst=reset_pin,
baudrate=BAUDRATE,
width=width,
height=height,
y_offset=90,
rotation=180
)
backlight = digitalio.DigitalInOut(board.D22)
backlight.switch_to_output()
backlight.value = True
self.buttonA = digitalio.DigitalInOut(board.D23)
self.buttonB = digitalio.DigitalInOut(board.D24)
self.buttonA.switch_to_input()
self.buttonB.switch_to_input()
self._button_hold_time = 2.0
self.image = Image.new("RGB", (width, height))
self.draw = ImageDraw.Draw(self.image)
self.draw.rectangle((0, 0, width, height), outline=0, fill=(0, 0, 0))
self.display.image(self.image)#, rotation)
self.font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 24)
self.stats = CovidStats()
loop = asyncio.get_event_loop()
loop.create_task(self.update_loop())
# loop.create_task(self.button_loop())
def update_display(self):
if self.stats.new_data:
self.logger.info('New data at ' + self.stats.last_updated.strftime('%d-%b-%Y (%H:%M)'))
self.draw.rectangle((0, 0, self.width, self.height), outline=0, fill=0)
hoffest = (240 - self.font.getsize(HEADER)[0]) / 2
cases = self.stats.confirmed
deaths = self.stats.deaths
padding = 10
x = 0
y = padding
self.draw.text((hoffest, y), HEADER, font=self.font, fill="#FFFFFF")
y += self.font.getsize(HEADER)[1]+padding
self.draw.text((x, y), cases, font=self.font, fill="#FF0000")
y += self.font.getsize(cases)[1]*3 + padding
self.draw.text((x, y), deaths, font=self.font, fill="#FF0000")
self.display.image(self.image)#, rotation)
async def get_stats_from_server(self) -> None:
self.logger.debug("Fetching update")
r = requests.get(COVID_19_API + 'latest')
if r.status_code == 200:
data = json.loads(r.content)
confirmed_world = int(data["latest"]["confirmed"])
deaths_world = int(data["latest"]["deaths"])
recovered_world = int(data["latest"]["recovered"])
r = requests.get(COVID_19_API + 'locations/223')
if r.status_code == 200:
data = json.loads(r.content)
last_update = dt.datetime.strptime(data["location"]["last_updated"], '%Y-%m-%dT%H:%M:%S.%fZ')
confirmed_uk = int(data["location"]["latest"]["confirmed"])
confirmed_uk_timeline = data["location"]["timelines"]["confirmed"]
deaths_uk = int(data["location"]["latest"]["deaths"])
deaths_uk_timeline = data["location"]["timelines"]["deaths"]
recovered_uk = int(data["location"]["latest"]["recovered"])
r = requests.get(COVID_19_API + 'locations/200') #?country_code=ZA;timelines=1')
if r.status_code == 200:
data = json.loads(r.content)
confirmed_sa = int(data["location"]["latest"]["confirmed"])
confirmed_sa_timeline = data["location"]["timelines"]["confirmed"]
deaths_sa = int(data["location"]["latest"]["deaths"])
deaths_sa_timeline = data["location"]["timelines"]["deaths"]
recovered_sa = int(data["location"]["latest"]["recovered"])
self.stats.update(confirmed_world, confirmed_uk, confirmed_sa, deaths_world, deaths_uk, deaths_sa, last_update)
self.update_display()
self.logger.debug("Confirmed: {0} UK: {1} SA {2}".format(confirmed_world, confirmed_uk, confirmed_sa))
self.logger.debug("Deaths: {0} UK: {1} SA {2}".format(deaths_world, deaths_uk, deaths_sa))
else:
self.logger.error("Error refreshing data")
async def update_loop(self) -> None:
while True:
await self.get_stats_from_server()
await asyncio.sleep(self.get_stats_interval_sec)
async def button_loop(self) -> None:
last_a = True
last_b = True
while True:
if last_a and not self.buttonA.value:
self._t_a_pressed = time.time()
self._hold_a_fired = False
self._t_a_repeat = time.time()
Thread(target=self._button_press_handler).start()
if not last_a and self.buttonA.value:
Thread(target=self._button_release_handler, args=(self._hold_a_fired,)).start()
if not self.buttonA.value:
if not self._hold_a_fired and (time.time() - self._t_a_pressed) > self._button_hold_time:
Thread(target=self._button_hold_handler).start()
self._hold_a_fired = True
last_a = self.buttonA.value
await asyncio.sleep(0.05)
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument('--update', type=float, default=600, help='Rate at which to update the data')
args = parser.parse_args()
try:
ct = CovidTracker(refresh_rate=args.update)
loop = asyncio.get_event_loop()
loop.run_forever()
except KeyboardInterrupt:
pass
import datetime as dt
class CovidStats:
def __init__(self):
self.last_updated = dt.datetime.now()
self.last_cases = 0
self.last_uk_cases = 0
self.last_sa_cases = 0
self.last_deaths = 0
self.last_uk_deaths = 0
self.last_sa_deaths = 0
self.new_cases = None
self.new_uk_cases = None
self.new_sa_cases = None
self.new_deaths = None
self.new_uk_deaths = None
self.new_sa_deaths = None
self.updated = True
def update(self, cw, cuk, csa, dw, duk, dsa, update):
if self.last_updated != update:
self.last_updated = update
self.updated = True
if cw != self.last_cases:
if self.new_cases is not None:
self.new_cases = cw - self.last_cases
else:
self.new_cases = 0
self.last_cases = cw
if cuk != self.last_uk_cases:
if self.new_uk_cases is not None:
self.new_uk_cases = cuk - self.last_uk_cases
else:
self.new_uk_cases = 0
self.last_uk_cases = cuk
if csa != self.last_sa_cases:
if self.new_sa_cases is not None:
self.new_sa_cases = csa - self.last_sa_cases
else:
self.new_sa_cases = 0
self.last_sa_cases = csa
if dw != self.last_deaths:
if self.new_deaths is not None:
self.new_deaths = dw - self.last_deaths
else:
self.new_deaths = 0
self.last_deaths = dw
if duk != self.last_uk_deaths:
if self.new_uk_deaths is not None:
self.new_uk_deaths = duk - self.last_uk_deaths
else:
self.new_uk_deaths = 0
self.last_uk_deaths = duk
if dsa != self.last_sa_deaths:
if self.new_sa_deaths is not None:
self.new_sa_deaths = dsa - self.last_sa_deaths
else:
self.new_sa_deaths = 0
self.last_sa_deaths = dsa
def new_data(self):
if self.updated:
self.updated = False
return True
else:
return False
@property
def confirmed(self):
return "Confirmed: {0}\n UK: {1} ({2})\n SA: {3} ({4})".format(
self.last_cases, self.last_uk_cases, self.new_uk_cases,
self.last_sa_cases, self.new_sa_cases
)
@property
def deaths(self):
return "Deaths: {0}\n UK: {1} ({2})\n SA: {3} ({4})".format(
self.last_deaths, self.last_uk_deaths, self.new_uk_deaths,
self.last_sa_deaths, self.new_sa_deaths
)
def __repr__(self):
fmt = """
Confirmed Cases:
Total: {cw:d}
UK: {cuk:d}
SA: {csa:d}
Deaths:
Total: {dw:d}
UK: {duk:d}
SA: {dw:d}
"""
return fmt.format(
cw=self.last_cases,
cuk=self.last_uk_cases,
csa=self.last_sa_cases,
dw=self.last_deaths,
duk=self.last_uk_deaths,
dsa=self.last_sa_deaths
)
__str__ = __repr__
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment