X-Git-Url: https://zdv2.bktei.com/gitweb/EVA-2020-02-2.git/blobdiff_plain/20442c9a53edb05b925faa07eb8360bd46442978..6c6e55301f85a23a4d393cf5244f0f5ba441e309:/examples/combined.py?ds=sidebyside diff --git a/examples/combined.py b/examples/combined.py index 43668a5..4b8fbdd 100755 --- a/examples/combined.py +++ b/examples/combined.py @@ -2,7 +2,6 @@ import time import colorsys -import os import sys import ST7735 try: @@ -38,6 +37,7 @@ bme280 = BME280() # PMS5003 particulate sensor pms5003 = PMS5003() +time.sleep(1.0) # Create ST7735 LCD display class st7735 = ST7735.ST7735( @@ -106,23 +106,23 @@ units = ["C", # with NO WARRANTY. The authors of this example code claim # NO RESPONSIBILITY if reliance on the following values or this # code in general leads to ANY DAMAGES or DEATH. -limits = [[4,18,28,35], - [250,650,1013.25,1015], - [20,30,60,70], - [-1,-1,30000,100000], - [-1,-1,40,50], - [-1,-1,450,550], - [-1,-1,200,300], - [-1,-1,50,100], - [-1,-1,50,100], - [-1,-1,50,100]] +limits = [[4, 18, 28, 35], + [250, 650, 1013.25, 1015], + [20, 30, 60, 70], + [-1, -1, 30000, 100000], + [-1, -1, 40, 50], + [-1, -1, 450, 550], + [-1, -1, 200, 300], + [-1, -1, 50, 100], + [-1, -1, 50, 100], + [-1, -1, 50, 100]] # RGB palette for values on the combined screen -palette = [(0,0,255), # Dangerously Low - (0,255,255), # Low - (0,255,0), # Normal - (255,255,0), # High - (255,0,0)] # Dangerously High +palette = [(0, 0, 255), # Dangerously Low + (0, 255, 255), # Low + (0, 255, 0), # Normal + (255, 255, 0), # High + (255, 0, 0)] # Dangerously High values = {} @@ -132,8 +132,9 @@ def display_text(variable, data, unit): # Maintain length of list values[variable] = values[variable][1:] + [data] # Scale the values for the variable between 0 and 1 - colours = [(v - min(values[variable]) + 1) / (max(values[variable]) - - min(values[variable]) + 1) for v in values[variable]] + vmin = min(values[variable]) + vmax = max(values[variable]) + colours = [(v - vmin + 1) / (vmax - vmin + 1) for v in values[variable]] # Format the variable name and value message = "{}: {:.1f} {}".format(variable[:4], data, unit) logging.info(message) @@ -141,18 +142,17 @@ def display_text(variable, data, unit): for i in range(len(colours)): # Convert the values to colours from red to blue colour = (1.0 - colours[i]) * 0.6 - r, g, b = [int(x * 255.0) for x in colorsys.hsv_to_rgb(colour, - 1.0, 1.0)] + r, g, b = [int(x * 255.0) for x in colorsys.hsv_to_rgb(colour, 1.0, 1.0)] # Draw a 1-pixel wide rectangle of colour - draw.rectangle((i, top_pos, i+1, HEIGHT), (r, g, b)) + draw.rectangle((i, top_pos, i + 1, HEIGHT), (r, g, b)) # Draw a line graph in black - line_y = HEIGHT - (top_pos + (colours[i] * (HEIGHT - top_pos)))\ - + top_pos - draw.rectangle((i, line_y, i+1, line_y+1), (0, 0, 0)) + line_y = HEIGHT - (top_pos + (colours[i] * (HEIGHT - top_pos))) + top_pos + draw.rectangle((i, line_y, i + 1, line_y + 1), (0, 0, 0)) # Write the text at the top in black draw.text((0, 0), message, font=font, fill=(0, 0, 0)) st7735.display(img) + # Saves the data to be used in the graphs later and prints to the log def save_data(idx, data): variable = variables[idx] @@ -167,24 +167,23 @@ def save_data(idx, data): def display_everything(): draw.rectangle((0, 0, WIDTH, HEIGHT), (0, 0, 0)) column_count = 2 - row_count = (len(variables)/column_count) + row_count = (len(variables) / column_count) for i in range(len(variables)): variable = variables[i] data_value = values[variable][-1] unit = units[i] - x = x_offset + ((WIDTH/column_count) * (i / row_count)) - y = y_offset + ((HEIGHT/row_count) * (i % row_count)) + x = x_offset + ((WIDTH / column_count) * (i / row_count)) + y = y_offset + ((HEIGHT / row_count) * (i % row_count)) message = "{}: {:.1f} {}".format(variable[:4], data_value, unit) lim = limits[i] rgb = palette[0] for j in range(len(lim)): if data_value > lim[j]: - rgb = palette[j+1] + rgb = palette[j + 1] draw.text((x, y), message, font=smallfont, fill=rgb) st7735.display(img) - # Get the temperature of the CPU for compensation def get_cpu_temperature(): process = Popen(['vcgencmd', 'measure_temp'], stdout=PIPE, universal_newlines=True) @@ -192,156 +191,158 @@ def get_cpu_temperature(): return float(output[output.index('=') + 1:output.rindex("'")]) -# Tuning factor for compensation. Decrease this number to adjust the -# temperature down, and increase to adjust up -factor = 2.25 - -cpu_temps = [get_cpu_temperature()] * 5 - -delay = 0.5 # Debounce the proximity tap -mode = 10 # The starting mode -last_page = 0 -light = 1 +def main(): + # Tuning factor for compensation. Decrease this number to adjust the + # temperature down, and increase to adjust up + factor = 2.25 + + cpu_temps = [get_cpu_temperature()] * 5 + + delay = 0.5 # Debounce the proximity tap + mode = 10 # The starting mode + last_page = 0 + + for v in variables: + values[v] = [1] * WIDTH + + # The main loop + try: + while True: + proximity = ltr559.get_proximity() + + # If the proximity crosses the threshold, toggle the mode + if proximity > 1500 and time.time() - last_page > delay: + mode += 1 + mode %= (len(variables) + 1) + last_page = time.time() + + # One mode for each variable + if mode == 0: + # variable = "temperature" + unit = "C" + cpu_temp = get_cpu_temperature() + # Smooth out with some averaging to decrease jitter + cpu_temps = cpu_temps[1:] + [cpu_temp] + avg_cpu_temp = sum(cpu_temps) / float(len(cpu_temps)) + raw_temp = bme280.get_temperature() + data = raw_temp - ((avg_cpu_temp - raw_temp) / factor) + display_text(variables[mode], data, unit) -for v in variables: - values[v] = [1] * WIDTH + if mode == 1: + # variable = "pressure" + unit = "hPa" + data = bme280.get_pressure() + display_text(variables[mode], data, unit) -# The main loop -try: - while True: - proximity = ltr559.get_proximity() - - # If the proximity crosses the threshold, toggle the mode - if proximity > 1500 and time.time() - last_page > delay: - mode += 1 - mode %= (len(variables)+1) - last_page = time.time() - - # One mode for each variable - if mode == 0: - # variable = "temperature" - unit = "C" - cpu_temp = get_cpu_temperature() - # Smooth out with some averaging to decrease jitter - cpu_temps = cpu_temps[1:] + [cpu_temp] - avg_cpu_temp = sum(cpu_temps) / float(len(cpu_temps)) - raw_temp = bme280.get_temperature() - data = raw_temp - ((avg_cpu_temp - raw_temp) / factor) - display_text(variables[mode], data, unit) - - if mode == 1: - # variable = "pressure" - unit = "hPa" - data = bme280.get_pressure() - display_text(variables[mode], data, unit) - - if mode == 2: - # variable = "humidity" - unit = "%" - data = bme280.get_humidity() - display_text(variables[mode], data, unit) - - if mode == 3: - # variable = "light" - unit = "Lux" - if proximity < 10: - data = ltr559.get_lux() - else: - data = 1 - display_text(variables[mode], data, unit) - - if mode == 4: - # variable = "oxidised" - unit = "kO" - data = gas.read_all() - data = data.oxidising / 1000 - display_text(variables[mode], data, unit) - - if mode == 5: - # variable = "reduced" - unit = "kO" - data = gas.read_all() - data = data.reducing / 1000 - display_text(variables[mode], data, unit) - - if mode == 6: - # variable = "nh3" - unit = "kO" - data = gas.read_all() - data = data.nh3 / 1000 - display_text(variables[mode], data, unit) - - if mode == 7: - # variable = "pm1" - unit = "ug/m3" - try: - data = pms5003.read() - except pmsReadTimeoutError: - logging.warn("Failed to read PMS5003") - else: - data = float(data.pm_ug_per_m3(1.0)) + if mode == 2: + # variable = "humidity" + unit = "%" + data = bme280.get_humidity() display_text(variables[mode], data, unit) - if mode == 8: - # variable = "pm25" - unit = "ug/m3" - try: - data = pms5003.read() - except pmsReadTimeoutError: - logging.warn("Failed to read PMS5003") - else: - data = float(data.pm_ug_per_m3(2.5)) + if mode == 3: + # variable = "light" + unit = "Lux" + if proximity < 10: + data = ltr559.get_lux() + else: + data = 1 display_text(variables[mode], data, unit) - if mode == 9: - # variable = "pm10" - unit = "ug/m3" - try: - data = pms5003.read() - except pmsReadTimeoutError: - logging.warn("Failed to read PMS5003") - else: - data = float(data.pm_ug_per_m3(10)) + if mode == 4: + # variable = "oxidised" + unit = "kO" + data = gas.read_all() + data = data.oxidising / 1000 display_text(variables[mode], data, unit) - if mode == 10: - # Everything on one screen - cpu_temp = get_cpu_temperature() - # Smooth out with some averaging to decrease jitter - cpu_temps = cpu_temps[1:] + [cpu_temp] - avg_cpu_temp = sum(cpu_temps) / float(len(cpu_temps)) - raw_temp = bme280.get_temperature() - raw_data = raw_temp - ((avg_cpu_temp - raw_temp) / factor) - save_data(0, raw_data) - display_everything() - raw_data = bme280.get_pressure() - save_data(1, raw_data) - display_everything() - raw_data = bme280.get_humidity() - save_data(2, raw_data) - if proximity < 10: - raw_data = ltr559.get_lux() - else: - raw_data = 1 - save_data(3, raw_data) - display_everything() - gas_data = gas.read_all() - save_data(4, gas_data.oxidising / 1000) - save_data(5, gas_data.reducing / 1000) - save_data(6, gas_data.nh3 / 1000) - display_everything() - pms_data = None - try: - pms_data = pms5003.read() - except pmsReadTimeoutError: - logging.warn("Failed to read PMS5003") - else: - save_data(7, float(pms_data.pm_ug_per_m3(1.0))) - save_data(8, float(pms_data.pm_ug_per_m3(2.5))) - save_data(9, float(pms_data.pm_ug_per_m3(10))) - display_everything() + if mode == 5: + # variable = "reduced" + unit = "kO" + data = gas.read_all() + data = data.reducing / 1000 + display_text(variables[mode], data, unit) + if mode == 6: + # variable = "nh3" + unit = "kO" + data = gas.read_all() + data = data.nh3 / 1000 + display_text(variables[mode], data, unit) -# Exit cleanly -except KeyboardInterrupt: - sys.exit(0) + if mode == 7: + # variable = "pm1" + unit = "ug/m3" + try: + data = pms5003.read() + except pmsReadTimeoutError: + logging.warn("Failed to read PMS5003") + else: + data = float(data.pm_ug_per_m3(1.0)) + display_text(variables[mode], data, unit) + + if mode == 8: + # variable = "pm25" + unit = "ug/m3" + try: + data = pms5003.read() + except pmsReadTimeoutError: + logging.warn("Failed to read PMS5003") + else: + data = float(data.pm_ug_per_m3(2.5)) + display_text(variables[mode], data, unit) + + if mode == 9: + # variable = "pm10" + unit = "ug/m3" + try: + data = pms5003.read() + except pmsReadTimeoutError: + logging.warn("Failed to read PMS5003") + else: + data = float(data.pm_ug_per_m3(10)) + display_text(variables[mode], data, unit) + if mode == 10: + # Everything on one screen + cpu_temp = get_cpu_temperature() + # Smooth out with some averaging to decrease jitter + cpu_temps = cpu_temps[1:] + [cpu_temp] + avg_cpu_temp = sum(cpu_temps) / float(len(cpu_temps)) + raw_temp = bme280.get_temperature() + raw_data = raw_temp - ((avg_cpu_temp - raw_temp) / factor) + save_data(0, raw_data) + display_everything() + raw_data = bme280.get_pressure() + save_data(1, raw_data) + display_everything() + raw_data = bme280.get_humidity() + save_data(2, raw_data) + if proximity < 10: + raw_data = ltr559.get_lux() + else: + raw_data = 1 + save_data(3, raw_data) + display_everything() + gas_data = gas.read_all() + save_data(4, gas_data.oxidising / 1000) + save_data(5, gas_data.reducing / 1000) + save_data(6, gas_data.nh3 / 1000) + display_everything() + pms_data = None + try: + pms_data = pms5003.read() + except pmsReadTimeoutError: + logging.warn("Failed to read PMS5003") + else: + save_data(7, float(pms_data.pm_ug_per_m3(1.0))) + save_data(8, float(pms_data.pm_ug_per_m3(2.5))) + save_data(9, float(pms_data.pm_ug_per_m3(10))) + display_everything() + + # Exit cleanly + except KeyboardInterrupt: + sys.exit(0) + + +if __name__ == "__main__": + main()