style(examples):Flip display vertically (back to default)

[EVA-2020-02-2.git] / examples / all-in-one-enviro-mini-bk.py

#!/usr/bin/env python3

import time
import colorsys
import os
import sys
import ST7735
import schedule # https://pypi.org/project/schedule/ via https://stackoverflow.com/a/16786600
import datetime # http://stackoverflow.com/questions/2150739/ddg#28147286
try:
    # Transitional fix for breaking change in LTR559
    from ltr559 import LTR559
    ltr559 = LTR559()
except ImportError:
    import ltr559

from bme280 import BME280
from enviroplus import gas
from subprocess import PIPE, Popen
from PIL import Image
from PIL import ImageDraw
from PIL import ImageFont
from fonts.ttf import RobotoMedium as UserFont
import logging
import numpy as np

logging.basicConfig(
    format='%(asctime)s.%(msecs)03d %(levelname)-8s %(message)s',
    level=logging.INFO,
    datefmt='%Y-%m-%d %H:%M:%S')

logging.info("""all-in-one.py - Displays readings from all of Enviro plus' sensors
Press Ctrl+C to exit!
""")

# Script Constants
varLenBufferTTL = int((2*24*60*60)*10**9) # time-to-live in nanoseconds

# BME280 temperature/pressure/humidity sensor
bme280 = BME280()

# Create ST7735 LCD display class
st7735 = ST7735.ST7735(
    port=0,
    cs=1,
    dc=9,
    backlight=12,
    rotation=270,
    #rotation=90, # flip upside down wrt enviro+ default orientation
    spi_speed_hz=10000000
)

# Initialize display
st7735.begin()

WIDTH = st7735.width
HEIGHT = st7735.height

# Set up canvas and font
img = Image.new('RGB', (WIDTH, HEIGHT), color=(0, 0, 0))
draw = ImageDraw.Draw(img)
path = os.path.dirname(os.path.realpath(__file__))
font_size = 20
font = ImageFont.truetype(UserFont, font_size)
font2_size = 15
font2 = ImageFont.truetype(UserFont, font2_size)

message = ""

# The position of the top bar
top_pos = 25


# Displays data and text on the 0.96" LCD
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
    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)
    draw.rectangle((0, 0, WIDTH, HEIGHT), (255, 255, 255))
    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)]
        # Draw a 1-pixel wide rectangle of colour
        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))
    # Write the text at the top in black
    draw.text((0, 0), message, font=font, fill=(0, 0, 0))
    st7735.display(img)

# Displays data and text on the 0.96" LCD
def display_text2(variable, data, unit, values):
    # Scale the values for the variable between 0 and 1
    print('DEBUG:len(values[' + str(variable) + ']):' + str(len(values[variable])))
    #print('DEBUG:values[' + str(variable) + ']:' + str(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)
    #message = "{}: {:.1f} {}".format(variable[:4], values[variable][-1], unit)
    logging.info(message)
    draw.rectangle((0, 0, WIDTH, HEIGHT), (255, 255, 255))
    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)]
        # Draw a 1-pixel wide rectangle of colour
        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))
    # Write the text at the top in black
    draw.text((0, 0), message, font=font, fill=(0, 0, 0))
    # Write text (test)
    maxMsg = "MAX:{:.1f}".format(vmax)
    durMsg = "HR:{:.1f}".format(span_time_h)
    minMsg = "MIN:{:.1f}".format(vmin)
    maxMsg_y = int( ((HEIGHT - top_pos)/(HEIGHT))*(HEIGHT*(1/4)) + top_pos - (font2_size/2) )
    durMsg_y = int( ((HEIGHT - top_pos)/(HEIGHT))*(HEIGHT*(2/4)) + top_pos - (font2_size/2) )
    minMsg_y = int( ((HEIGHT - top_pos)/(HEIGHT))*(HEIGHT*(3/4)) + top_pos - (font2_size/2) )
    maxMsg_x = int( WIDTH*(3/100) )
    durMsg_x = int( WIDTH*(3/100) )
    minMsg_x = int( WIDTH*(3/100) )
    draw.text((maxMsg_x, maxMsg_y), maxMsg, font=font2, fill=(0, 0, 0))
    draw.text((durMsg_x, durMsg_y), durMsg, font=font2, fill=(0, 0, 0))
    draw.text((minMsg_x, minMsg_y), minMsg, font=font2, fill=(0, 0, 0))
    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)
    output, _error = process.communicate()
    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 = 0  # The starting mode
last_page = 0
light = 1

# Create a values dict to store the data
variables = ["temperature",
             "pressure",
             "humidity",
             "light"]
values = {} # Initialize values dictionary
for v in variables:
    values[v] = [1] * WIDTH # Init a WIDTH-length list as value for each string in variables

# Create a varLenBuffer dict to store recent data
varLenBuffer = {}
for v in variables:
    varLenBuffer[v] = [] # Init an empty list for each string in variables

# Create a varLenBufferFlt dict to store recent data as floats only
varLenBufferFlt = {}
for v in variables:
    varLenBufferFlt[v] = [] # Init an empty list for each string in variables
    
# Create a fixLenBuffer dict to store data for displaying
fixLenBuffer = {}
for v in variables:
    fixLenBuffer[v] = [] # Init an empty list for each string in variables
    
pollDelay = 5.0

def pollSensors():
    # Desc: Update variables containing latest sensor tuples
    # Output: (time [ns], unit, float)
    #         now_temp_tuple (°C)
    #         now_pressure_tuple (hPa)
    #         now_humidity_tuple (%)
    #         now_illuminance_tuple (lux)
    # Depends: time, bme280, ltr559, get_cpu_temperature()

    # Tell function to modify these global variables
    global now_temp_tuple
    global now_pressure_tuple
    global now_humidity_tuple
    global now_illuminance_tuple
    # Initialize
    cpu_temps = []
    poll_time_ns_start = time.time_ns() # Get time reading (unix spoech, nanoseconds)
    # Get temperature reading
    cpu_temp = get_cpu_temperature() # get °C from CPU
    # 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() # get °C from BME280 sensor
    now_temp = raw_temp - ((avg_cpu_temp - raw_temp) / factor)
    now_temp_tuple = (time.time_ns(), '°C', now_temp)
    # Get pressure reading
    now_time_ns = time.time_ns() # Get time reading (unix epoch, nanoseconds)
    now_pressure = bme280.get_pressure() # get hPa from BME280 sensor
    now_pressure_tuple = (time.time_ns(), 'hPa', now_pressure)
    # Get humidity reading
    now_humidity = bme280.get_humidity() # get % humidity from BME280 sensor
    now_humidity_tuple = (time.time_ns(), '%', now_humidity)
    # Get light reading
    proximity = ltr559.get_proximity() # get proximity reading
    if proximity < 10:
        now_illuminance = ltr559.get_lux() # get lux reading from LTR-559 sensor if nothing is nearby
        now_illuminance_tuple = (time.time_ns(), 'lux', now_illuminance)
    poll_time_ns_end = time.time_ns()
    #print('DEBUG:poll time (s):' + str((poll_time_ns_end - poll_time_ns_start)/1000000000))


def dateIso8601Str():
    nowUTC = datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc).isoformat()
    return str(nowUTC)

def medianSubset(listIn: list = [], listOutLen: int = 0) -> list:
    # Input: int:  listOutLen: quantity of elements in output list
    #        list: listIn: input list consisting of integers or floats
    # Output: list: ints/floats of specified size
    # Ref/Attrib: PEP 3107 typing https://stackoverflow.com/a/21384492
    # Version: 0.1.0
    #print('DEBUG:listOutLen:' + str(listOutLen))
    #print('DEBUG:listIn:' + str(listIn))

    # Exit for invalid input
    if not isinstance(listOutLen, int):
        raise ValueError('ERROR:Not a valid int:' + str(listOutLen))
    else:
        if not listOutLen > 0:
            raise ValueError('ERROR:Invalid value:' + str(listOutLen))
    if not isinstance(listIn, list):
        raise ValueError('ERROR:Not a valid list:' + str(listOutLen))
    if not all([( (isinstance(x,int)) or (isinstance(x,float)) ) for x in listIn]):
        raise ValueError('ERROR:Input list contains something besides integers or floating point numbers.')
    
    # Initialize listOut
    listOut = [None] * listOutLen
    #print('DEBUG:listOut:' + str(listOut))
    
    # Calc listIn length
    listInLen = len(listIn)
    #print('DEBUG:listInLen:' + str(listInLen))

    # Calc subset length float
    subsetLenFloat = ( (max([listInLen,listOutLen]) - 1) /min([listInLen,listOutLen]))
    subsetIndRatio = ( (listInLen)/(listOutLen) )
    #print('DEBUG:subsetLenFloat: %.5f' % subsetLenFloat)
    #print('DEBUG:subsetLenFloat2: %.5f' % subsetIndRatio)
    
    # Iterate for each element in listOut
    for i_out in range(listOutLen):
        #print('DEBUG:i_out:' + str(i_out))
        ## Decide to expand or reduce listIn to produce listOut
        if listInLen > listOutLen:
            ### reduce listIn to listOut
            #print('DEBUG:listOutLen:' + str(listOutLen))
            #print('DEBUG:listInLen:' + str(listInLen))            
            if i_out == 0:
                #### Initialize subsetIndLo in first loop
                subsetIndLo = int(0)
                #print('DEBUG:subsetIndLo:' + str(subsetIndLo))
            #print('DEBUG:i_out:' + str(i_out))
            #### Calc indices of i_out'th subset of listIn
            subsetIndHi = (listInLen - 1) * (i_out + 1) // listOutLen
            subsetLen = subsetIndHi - subsetIndLo + 1
            #print('DEBUG:subsetIndLo:' + str(subsetIndLo))
            #print('DEBUG:subsetIndHi:' + str(subsetIndHi))
            #print('DEBUG:subsetLen:' + str(subsetLen))
            #### Extract subset from listIn using indices inclusively
            subset = listIn[ int(subsetIndLo) : int(subsetIndHi)+1 ]
            #print('DEBUG:subset:' + str(subset))
            #### Calculate median for subset
            subsetMedian = np.median(subset)
            #print('DEBUG:subset median:' + str(subsetMedian))
            #### Set listOut element
            listOut[i_out] = subsetMedian
            #### Housekeeping
            ##### Update subsetIndLo for next loop
            subsetIndLo = subsetIndHi + 1
            #print('DEBUG:Updated subsetIndLo:' + str(subsetIndLo))
        elif listOutLen > listInLen:
            ### Expand listIn to listOut
            #print('DEBUG:listOutLen:' + str(listOutLen))
            #print('DEBUG:listInLen:' + str(listInLen))
            #### Identify index list of lists mapping listIn to ListOut
            expandIndex = int(i_out / subsetLenFloat)
            expandIndex = min([expandIndex,(listInLen - 1)])
            #print('DEBUG:expandIndex:' + str(expandIndex))
            listOut[i_out] = listIn[expandIndex]
            #print('DEBUG:listOut[i_out]:' + str(listOut[i_out]))
        elif listOutLen == listInLen:
            listOut = listIn
            #print('DEBUG:end for loop===========')
    return listOut

def updateBuffer():
    global now_temp_tuple
    global now_pressure_tuple
    global now_humidity_tuple
    global now_illuminance_tuple
    global varLenBuffer
    global fixLenBuffer
    global fixLenBufferFlt
    global span_time_h
    #print('DEBUG:This is the updateBuffer() function.')
    #print('DEBUG:===========================================================')
    #print('DEBUG:===========================================================')
    # Capture new sensor tuples
    pollSensors()
    #print('DEBUG:now_temp_tuple:' + str(now_temp_tuple))
    #print('DEBUG:now_pressure_tuple:' + str(now_pressure_tuple))
    #print('DEBUG:now_humidity_tuple:' + str(now_humidity_tuple))
    #print('DEBUG:now_illuminance_tuple:' + str(now_illuminance_tuple))    

    # Append new sensor tuples to varying-length buffer
    ## Temperature
    varLenBuffer[variables[0]].append(now_temp_tuple)
    ## Pressure
    varLenBuffer[variables[1]].append(now_pressure_tuple)
    ## Humidity
    varLenBuffer[variables[2]].append(now_humidity_tuple)
    ## Illuminance
    varLenBuffer[variables[3]].append(now_illuminance_tuple)
    #print('DEBUG:varLenBuffer:' + str(varLenBuffer))

    # Trim outdated sensor tuples from varying-length buffer
    ## iterate through each tuple list and remove old tuples
    varLenBufferTemp = []
    for v in variables:
        #varLenBufferTemp = varLenBuffer[v].copy()
        now_time_ns = time.time_ns() # get ns timestamp of now
        thn_time_ns = varLenBuffer[v][0][0] # get ns timestamp of earliest tuple
        dif_time_ns = now_time_ns - thn_time_ns # calc nanosecond difference
        #print('DEBUG:varLenBufferTTL:' + str(varLenBufferTTL))
        #print('DEBUG:now:' + str(now_time_ns))
        #print('DEBUG:thn:' + str(thn_time_ns))
        #print('DEBUG:dif:' + str(dif_time_ns))
        #print('DEBUG:dif(s):' + str(dif_time_ns / 1000000000))
        if dif_time_ns > varLenBufferTTL:
            varLenBuffer[v].pop(0) # discard earliest tuple if age > varLenBufferTTL
        print('DEBUG:Len of varLenBuffer[' + str(v) + ']:' + str(len(varLenBuffer[v])))
        #print('DEBUG:*******************************************')
        #print('DEBUG:varLenBuffer[variables[' + str(v) + ']]:' + str(varLenBuffer[v]))
        #print('DEBUG:*******************************************')

    # Calculate buffer time span in hours
    ## Get earliest timestamp (use temperature tuples)
    first_time_ns = varLenBuffer[variables[0]][0][0]
    last_time_ns = varLenBuffer[variables[0]][-1][0]
    span_time_ns = int(last_time_ns - first_time_ns)
    span_time_h = float(span_time_ns / (10**9*60*60)) # nanoseconds to hours
        
    # Convert tuple buffer into float buffer
    for v in variables:
        varLenBufferFlt[v].clear() # clear old float list
        #print('DEBUG:v:' + str(v))
        for t in varLenBuffer[v]:
            #print('DEBUG:t:' + str(t))
            #print('DEBUG:t[2]:' + str(t[2]))
            #print('DEBUG:------------------------------------------')
            #print('DEBUG:varLenBufferFlt[' + str(v) + ']:' + str(varLenBufferFlt[v]))
            #print('DEBUG:------------------------------------------')
            if isinstance(t[2], float):
                varLenBufferFlt[v].append(float(t[2])) # build new float list
            else:
                varLenBufferFlt[v].append(float(-273)) # add obvious zero otherwise
                #print('DEBUG:varLenBufferFlt[' + str(v) + ']:' + str(varLenBufferFlt[v]))

    # Compress/expand buffer to fixed-length buffer
    for v in variables:
        #print('DEBUG:varLenBufferFlt[0]:' + str(varLenBufferFlt[variables[0]]))
        fixLenBuffer[v] = medianSubset(varLenBufferFlt[v], WIDTH)
        print('DEBUG:Len of fixLenBuffer[' + str(v) + ']:' + str(len(fixLenBuffer[v])))
        #print('DEBUG:fixLenBuffer[' + str(v) + ']:' + str(fixLenBuffer[v]))

        
# The main loop
try:
    # Schedule tasks
    # schedule.every(1).second.do(updateBuffer)
    schedule.every().minute.at(":00").do(updateBuffer)
    # schedule.every().minute.at(":05").do(updateBuffer)
    # schedule.every().minute.at(":10").do(updateBuffer)
    # schedule.every().minute.at(":15").do(updateBuffer)
    # schedule.every().minute.at(":20").do(updateBuffer)
    # schedule.every().minute.at(":25").do(updateBuffer)
    # schedule.every().minute.at(":30").do(updateBuffer)
    # schedule.every().minute.at(":35").do(updateBuffer)
    # schedule.every().minute.at(":40").do(updateBuffer)
    # schedule.every().minute.at(":45").do(updateBuffer)
    # schedule.every().minute.at(":50").do(updateBuffer)
    # schedule.every().minute.at(":55").do(updateBuffer)
    pollSensors() # initial run to start up sensors
    time.sleep(1) # pause to give sensors time to initialize
    updateBuffer() # initial run

    while True:
        proximity = ltr559.get_proximity()

        # If the proximity crosses the threshold, toggle the display mode
        if proximity > 1500 and time.time() - last_page > delay:
            mode += 1
            mode %= len(variables)
            last_page = time.time()

        # Run scheduled tasks
        schedule.run_pending()
        
        # One display mode for each variable
        if mode == 0:
            # variable = "temperature"
            ## run function to display latest temp values in variables[mode] list
            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)
            display_text2(variables[mode],data,unit,fixLenBuffer)

        if mode == 1:
            # variable = "pressure"
            unit = "hPa"
            data = bme280.get_pressure()
            #display_text(variables[mode], data, unit)
            display_text2(variables[mode],data,unit,fixLenBuffer)
            
        if mode == 2:
            # variable = "humidity"
            unit = "%"
            data = bme280.get_humidity()
            #display_text(variables[mode], data, unit)
            display_text2(variables[mode],data,unit,fixLenBuffer)
            
        if mode == 3:
            # variable = "light"
            unit = "Lux"
            if proximity < 10:
                data = ltr559.get_lux()
            else:
                data = 1
            #display_text(variables[mode], data, unit)
            display_text2(variables[mode],data,unit,fixLenBuffer)
      
        

# Exit cleanly
except KeyboardInterrupt:
    sys.exit(0)