COVID-19 Data - RGB Matrix
# SPDX-FileCopyrightText: 2020 John Furcean | |
# SPDX-License-Identifier: MIT | |
import board | |
import busio | |
from digitalio import DigitalInOut | |
import adafruit_requests as requests | |
import adafruit_esp32spi.adafruit_esp32spi_socket as socket | |
from adafruit_bitmap_font import bitmap_font | |
from adafruit_esp32spi import adafruit_esp32spi | |
from adafruit_esp32spi import adafruit_esp32spi_wifimanager | |
import adafruit_display_text.label | |
import board | |
import displayio | |
import framebufferio | |
import rgbmatrix | |
import terminalio | |
import time | |
import neopixel | |
# Number of Seconds in an hour times the number of hours | |
# This dictates the data refresh time | |
TIME_THRESHOLD = 3600*4 | |
# URLS for the covid tracking API | |
US_DATA_URL = "https://api.covidtracking.com/v1/us/current.json" | |
STATE_DATA_URL = "https://api.covidtracking.com/v1/states/STATE/current.json" | |
# Dictionary of the states that you want to display | |
STATES = { | |
'mi' : 'MICHIGAN', | |
'fl' : 'FLORIDA' | |
} | |
COLORS = [0x0080ff,0xffffff,0x009900] | |
def fetch_data(URL,label): | |
''' | |
fetches data from a covidtracking.com | |
URL: a url of an api endpoint within covidtracking.com | |
label: a text label for the api endpoint (str) | |
Returns: output string of the cobined data (str) | |
''' | |
#Debugging print statements | |
print() | |
print("Fetching json from", URL) | |
r = requests.get(URL) | |
data = r.json() | |
r.close() | |
# The US API call returns the json dictionary within a list | |
if label == 'UNITED STATES': | |
data = data[0] | |
# check if the returned data has positive cases | |
if data['positive']: | |
output_string = f"CASES: {data['positive']:,} " | |
# check if the returned data has recovered cases | |
if data['recovered']: | |
output_string += f"RECOVERED: {data['recovered']:,} " | |
# check if the returned data has deaths | |
if data['death']: | |
output_string += f"DEATHS: {data['death']:,} " | |
# check if the returned data has both deaths and positive cases | |
if data['death'] and data['positive']: | |
# Estimating mortality from COVID-19 - WHO | |
# https://www.who.int/news-room/commentaries/detail/estimating-mortality-from-covid-19 | |
output_string += f"ESTIMATED CFR: {(data['death']/data['positive'])*100:.2f}% " | |
return output_string | |
def scroll(top_line,bottom_line): | |
''' | |
Scrolls two lines of text | |
top_line: top line of text (str) | |
bottom_line: bottom line of text (str) | |
''' | |
top_line.x = top_line.x - 3 | |
top_line_width = top_line.bounding_box[2] | |
if top_line.x < -top_line_width: | |
top_line.x = display.width | |
bottom_line.x = bottom_line.x - 3 | |
bottom_line_width = bottom_line.bounding_box[2] | |
# if the bottom line of text has finished scrolling move on to the next line of text | |
if bottom_line.x < -bottom_line_width: | |
bottom_line.x = display.width | |
top_line.x = display.width | |
top_line.index += 1 | |
bottom_line.index += 1 | |
# this is meant to match the label string to the data string | |
# so they can scroll the same length | |
def add_padding(label,length): | |
''' | |
duplicates a string until it matches a certain length | |
label: a text string (str) | |
usually a state or "UNITED STATES" | |
length: the lenght you want to duplicate the label up to (int) | |
Retruns: a duplicated label string that is the same size as the length | |
''' | |
padding_len = 6 | |
padded_label = label | |
while len(padded_label) < length: | |
if (len(padded_label) + len(label) + padding_len) < length: | |
padded_label += (" "*padding_len)+label | |
else: | |
padded_label += " "*(length - len(padded_label)) | |
return padded_label | |
#initialize the RGB Matrix | |
displayio.release_displays() | |
matrix = rgbmatrix.RGBMatrix( | |
width=64, bit_depth=4, | |
rgb_pins=[ | |
board.MTX_R1, | |
board.MTX_G1, | |
board.MTX_B1, | |
board.MTX_R2, | |
board.MTX_G2, | |
board.MTX_B2 | |
], | |
addr_pins=[ | |
board.MTX_ADDRA, | |
board.MTX_ADDRB, | |
board.MTX_ADDRC, | |
board.MTX_ADDRD | |
], | |
clock_pin=board.MTX_CLK, | |
latch_pin=board.MTX_LAT, | |
output_enable_pin=board.MTX_OE | |
) | |
display = framebufferio.FramebufferDisplay(matrix) | |
# Create two lines of text. | |
line1 = adafruit_display_text.label.Label( | |
terminalio.FONT, | |
color=0xff0000, | |
text=f"COVID-19", | |
max_glyphs = 100) | |
line1.x = 2 | |
line1.y = 6 | |
line1.index = 0 | |
line2 = adafruit_display_text.label.Label( | |
terminalio.FONT, | |
color=0x0080ff, | |
text=f"Tracking", | |
max_glyphs=100) | |
line2.x = 2 | |
line2.y = 20 | |
line2.index= 0 | |
# Put each line of text into a Group, then show that group. | |
g = displayio.Group() | |
g.append(line1) | |
g.append(line2) | |
display.show(g) | |
# Get wifi details and more from a secrets.py file | |
try: | |
from secrets import secrets | |
except ImportError: | |
print("WiFi secrets are kept in secrets.py, please add them there!") | |
raise | |
# If you are using a board with pre-defined ESP32 Pins: | |
esp32_cs = DigitalInOut(board.ESP_CS) | |
esp32_ready = DigitalInOut(board.ESP_BUSY) | |
esp32_reset = DigitalInOut(board.ESP_RESET) | |
spi = busio.SPI(board.SCK, board.MOSI, board.MISO) | |
esp = adafruit_esp32spi.ESP_SPIcontrol(spi, esp32_cs, esp32_ready, esp32_reset) | |
requests.set_socket(socket, esp) | |
status_light = neopixel.NeoPixel(board.NEOPIXEL, 1, brightness=0.2) | |
# Boolean to force the initial data fetch | |
has_data = False | |
while True: | |
#Get the current time | |
now = time.monotonic() | |
# check if the current time has passed some threshold | |
# or if no data has been fetched | |
if not has_data or now - data_time > TIME_THRESHOLD: | |
print("Checking AP Connection...") | |
while not esp.is_connected: | |
try: | |
esp.connect_AP(secrets["ssid"], secrets["password"]) | |
except RuntimeError as e: | |
print("could not connect to AP, retrying: ", e) | |
continue | |
line1.x = 2 | |
line2.x = 2 | |
line1.text = "Fetching" | |
line2.text = "Data..." | |
data_strings = [] | |
label_strings = [] | |
try: | |
#fetch the data of the United States | |
us_string = fetch_data(US_DATA_URL,'UNITED STATES') | |
data_strings.append(us_string) | |
# match the size of the label string to | |
# the same size as the data string | |
us_label = add_padding('UNITED STATES',len(us_string)) | |
label_strings.append(us_label) | |
#loop over the states in the STATES dictionary defined at the top | |
for state_abv,state in STATES.items(): | |
#generate the url for the current state | |
state_url = STATE_DATA_URL.replace('STATE',state_abv) | |
#fetch data for the current state | |
data_string = fetch_data(state_url,state) | |
data_strings.append(data_string) | |
# match the size of the label string to | |
# the same size as the data string | |
label_string = add_padding(state,len(data_string)) | |
label_strings.append(label_string) | |
#get time of data fetch | |
data_time = time.monotonic() | |
has_data = True | |
except RuntimeError as e: | |
print("Couldn't fetch data!") | |
continue | |
line1.text = "COVID-19" | |
line2.text = data_strings[0] | |
# if we reached the last string in the data set, reset it. | |
if line2.index >= len(data_strings): | |
line2.index = 0 | |
# get the current label and matching data | |
label = label_strings[line2.index%len(data_strings)] | |
data_string = data_strings[line2.index%len(data_strings)] | |
# set the scrolling lines to the current state/united states and its data | |
line1.text = label | |
line2.text = data_string | |
# change the color depening on the current index | |
line2.color = COLORS[line2.index%len(COLORS)] | |
#scroll the two lines | |
scroll(line1,line2) | |
display.refresh(minimum_frames_per_second=0) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment