[EVA-2020-02-2.git] / examples / combined.py

#!/usr/bin/env python

import time
import colorsys
import os
import sys
import ST7735
try:
    # Transitional fix for breaking change in LTR559
    from ltr559 import LTR559
    ltr559 = LTR559()
except ImportError:
    import ltr559

from bme280 import BME280
from pms5003 import PMS5003, ReadTimeoutError as pmsReadTimeoutError
from enviroplus import gas
from subprocess import PIPE, Popen
from PIL import Image
from PIL import ImageDraw
from PIL import ImageFont
import logging

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!

""")

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

# PMS5003 particulate sensor
pms5003 = PMS5003()

# Create ST7735 LCD display class
st7735 = ST7735.ST7735(
    port=0,
    cs=1,
    dc=9,
    backlight=12,
    rotation=270,
    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 = ImageFont.truetype(path + "/fonts/Asap/Asap-Bold.ttf", 20)
smallfont = ImageFont.truetype(path + "/fonts/Asap/Asap-Bold.ttf", 10)
x_offset = 2
y_offset = 2

message = ""

# The position of the top bar
top_pos = 25

# Create a values dict to store the data
variables = ["temperature",
             "pressure",
             "humidity",
             "light",
             "oxidised",
             "reduced",
             "nh3",
             "pm1",
             "pm25",
             "pm10"]

units = ["C",
         "hPa",
         "%",
         "Lux",
         "kO",
         "kO",
         "kO",
         "ug/m3",
         "ug/m3",
         "ug/m3"]

# Define your own warning limits
# The limits definition follows the order of the variables array
# Example limits explanation for temperature:
# [4,18,28,35] means
# [-273.15 .. 4] -> Dangerously Low
# (4 .. 18]      -> Low
# (18 .. 28]     -> Normal
# (28 .. 35]     -> High
# (35 .. MAX]    -> Dangerously High
# DISCLAIMER: The limits provided here are just examples and come
# 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]]

# 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

values = {}


# 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
    colours = [(v - min(values[variable]) + 1) / (max(values[variable])
               - min(values[variable]) + 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)

# Saves the data to be used in the graphs later and prints to the log
def save_data(idx, data):
    variable = variables[idx]
    # Maintain length of list
    values[variable] = values[variable][1:] + [data]
    unit = units[idx]
    message = "{}: {:.1f} {}".format(variable[:4], data, unit)
    logging.info(message)


# Displays all the text on the 0.96" LCD
def display_everything():
    draw.rectangle((0, 0, WIDTH, HEIGHT), (0, 0, 0))
    column_count = 2
    row_count = (len(variables)/column_count)
    for i in xrange(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))
        message = "{}: {:.1f} {}".format(variable[:4], data_value, unit)
        lim = limits[i]
        rgb = palette[0]
        for j in xrange(len(lim)):
            if data_value > lim[j]:
                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)
    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 = 1.95

cpu_temps = [get_cpu_temperature()] * 5

delay = 0.5  # Debounce the proximity tap
mode = 10    # The starting mode
last_page = 0
light = 1

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)

        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))
                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)
Commit	Line	Data
71dc2962 K	1	#!/usr/bin/env python
	2
	3	import time
	4	import colorsys
	5	import os
	6	import sys
	7	import ST7735
	8	try:
	9	# Transitional fix for breaking change in LTR559
	10	from ltr559 import LTR559
	11	ltr559 = LTR559()
	12	except ImportError:
	13	import ltr559
	14
	15	from bme280 import BME280
	16	from pms5003 import PMS5003, ReadTimeoutError as pmsReadTimeoutError
	17	from enviroplus import gas
	18	from subprocess import PIPE, Popen
	19	from PIL import Image
	20	from PIL import ImageDraw
	21	from PIL import ImageFont
	22	import logging
	23
	24	logging.basicConfig(
	25	format='%(asctime)s.%(msecs)03d %(levelname)-8s %(message)s',
	26	level=logging.INFO,
	27	datefmt='%Y-%m-%d %H:%M:%S')
	28
	29	logging.info("""all-in-one.py - Displays readings from all of Enviro plus' sensors
	30
	31	Press Ctrl+C to exit!
	32
	33	""")
	34
	35	# BME280 temperature/pressure/humidity sensor
	36	bme280 = BME280()
	37
	38	# PMS5003 particulate sensor
	39	pms5003 = PMS5003()
	40
	41	# Create ST7735 LCD display class
	42	st7735 = ST7735.ST7735(
	43	port=0,
	44	cs=1,
	45	dc=9,
	46	backlight=12,
	47	rotation=270,
	48	spi_speed_hz=10000000
	49	)
	50
	51	# Initialize display
	52	st7735.begin()
	53
	54	WIDTH = st7735.width
	55	HEIGHT = st7735.height
	56
	57	# Set up canvas and font
	58	img = Image.new('RGB', (WIDTH, HEIGHT), color=(0, 0, 0))
	59	draw = ImageDraw.Draw(img)
	60	path = os.path.dirname(os.path.realpath(__file__))
	61	font = ImageFont.truetype(path + "/fonts/Asap/Asap-Bold.ttf", 20)
	62	smallfont = ImageFont.truetype(path + "/fonts/Asap/Asap-Bold.ttf", 10)
	63	x_offset = 2
	64	y_offset = 2
65
66	message = ""
67
68	# The position of the top bar
69	top_pos = 25
70
71	# Create a values dict to store the data
72	variables = ["temperature",
73	"pressure",
74	"humidity",
75	"light",
76	"oxidised",
77	"reduced",
78	"nh3",
79	"pm1",
80	"pm25",
81	"pm10"]
82
83	units = ["C",
84	"hPa",
85	"%",
86	"Lux",
87	"kO",
88	"kO",
89	"kO",
90	"ug/m3",
91	"ug/m3",
92	"ug/m3"]
93
94	# Define your own warning limits
95	# The limits definition follows the order of the variables array
96	# Example limits explanation for temperature:
97	# [4,18,28,35] means
98	# [-273.15 .. 4] -> Dangerously Low
99	# (4 .. 18] -> Low
100	# (18 .. 28] -> Normal
101	# (28 .. 35] -> High
102	# (35 .. MAX] -> Dangerously High
103	# DISCLAIMER: The limits provided here are just examples and come
104	# with NO WARRANTY. The authors of this example code claim
105	# NO RESPONSIBILITY if reliance on the following values or this
106	# code in general leads to ANY DAMAGES or DEATH.
107	limits = [[4,18,28,35],
108	[250,650,1013.25,1015],
109	[20,30,60,70],
110	[-1,-1,30000,100000],
111	[-1,-1,40,50],
112	[-1,-1,450,550],
113	[-1,-1,200,300],
114	[-1,-1,50,100],
115	[-1,-1,50,100],
116	[-1,-1,50,100]]
117
118	# RGB palette for values on the combined screen
119	palette = [(0,0,255), # Dangerously Low
120	(0,255,255), # Low
121	(0,255,0), # Normal
122	(255,255,0), # High
123	(255,0,0)] # Dangerously High
124
125	values = {}
126
127
128	# Displays data and text on the 0.96" LCD
129	def display_text(variable, data, unit):
130	# Maintain length of list
131	values[variable] = values[variable][1:] + [data]
132	# Scale the values for the variable between 0 and 1
133	colours = [(v - min(values[variable]) + 1) / (max(values[variable])
134	- min(values[variable]) + 1) for v in values[variable]]
135	# Format the variable name and value
136	message = "{}: {:.1f} {}".format(variable[:4], data, unit)
137	logging.info(message)
138	draw.rectangle((0, 0, WIDTH, HEIGHT), (255, 255, 255))
139	for i in range(len(colours)):
140	# Convert the values to colours from red to blue
141	colour = (1.0 - colours[i]) * 0.6
142	r, g, b = [int(x * 255.0) for x in colorsys.hsv_to_rgb(colour,
143	1.0, 1.0)]
144	# Draw a 1-pixel wide rectangle of colour
145	draw.rectangle((i, top_pos, i+1, HEIGHT), (r, g, b))
146	# Draw a line graph in black
147	line_y = HEIGHT - (top_pos + (colours[i] * (HEIGHT - top_pos)))\
148	+ top_pos
149	draw.rectangle((i, line_y, i+1, line_y+1), (0, 0, 0))
150	# Write the text at the top in black
151	draw.text((0, 0), message, font=font, fill=(0, 0, 0))
152	st7735.display(img)
153
154	# Saves the data to be used in the graphs later and prints to the log
155	def save_data(idx, data):
156	variable = variables[idx]
157	# Maintain length of list
158	values[variable] = values[variable][1:] + [data]
159	unit = units[idx]
160	message = "{}: {:.1f} {}".format(variable[:4], data, unit)
161	logging.info(message)
162
163
164	# Displays all the text on the 0.96" LCD
165	def display_everything():
166	draw.rectangle((0, 0, WIDTH, HEIGHT), (0, 0, 0))
167	column_count = 2
168	row_count = (len(variables)/column_count)
169	for i in xrange(len(variables)):
170	variable = variables[i]
171	data_value = values[variable][-1]
172	unit = units[i]
173	x = x_offset + ((WIDTH/column_count) * (i / row_count))
174	y = y_offset + ((HEIGHT/row_count) * (i % row_count))
175	message = "{}: {:.1f} {}".format(variable[:4], data_value, unit)
176	lim = limits[i]
177	rgb = palette[0]
178	for j in xrange(len(lim)):
179	if data_value > lim[j]:
180	rgb = palette[j+1]
181	draw.text((x, y), message, font=smallfont, fill=rgb)
182	st7735.display(img)
183
184
185
186	# Get the temperature of the CPU for compensation
187	def get_cpu_temperature():
188	process = Popen(['vcgencmd', 'measure_temp'], stdout=PIPE, universal_newlines=True)
189	output, _error = process.communicate()
190	return float(output[output.index('=') + 1:output.rindex("'")])
191
192
193	# Tuning factor for compensation. Decrease this number to adjust the
194	# temperature down, and increase to adjust up
195	factor = 1.95
196
197	cpu_temps = [get_cpu_temperature()] * 5
198
199	delay = 0.5 # Debounce the proximity tap
200	mode = 10 # The starting mode
201	last_page = 0
202	light = 1
203
204	for v in variables:
205	values[v] = [1] * WIDTH
206
207	# The main loop
208	try:
209	while True:
210	proximity = ltr559.get_proximity()
211
212	# If the proximity crosses the threshold, toggle the mode
213	if proximity > 1500 and time.time() - last_page > delay:
214	mode += 1
215	mode %= (len(variables)+1)
216	last_page = time.time()
217
218	# One mode for each variable
219	if mode == 0:
220	# variable = "temperature"
221	unit = "C"
222	cpu_temp = get_cpu_temperature()
223	# Smooth out with some averaging to decrease jitter
224	cpu_temps = cpu_temps[1:] + [cpu_temp]
225	avg_cpu_temp = sum(cpu_temps) / float(len(cpu_temps))
226	raw_temp = bme280.get_temperature()
227	data = raw_temp - ((avg_cpu_temp - raw_temp) / factor)
228	display_text(variables[mode], data, unit)
229
230	if mode == 1:
231	# variable = "pressure"
232	unit = "hPa"
233	data = bme280.get_pressure()
234	display_text(variables[mode], data, unit)
235
236	if mode == 2:
237	# variable = "humidity"
238	unit = "%"
239	data = bme280.get_humidity()
240	display_text(variables[mode], data, unit)
241
242	if mode == 3:
243	# variable = "light"
244	unit = "Lux"
245	if proximity < 10:
246	data = ltr559.get_lux()
247	else:
248	data = 1
249	display_text(variables[mode], data, unit)
250
251	if mode == 4:
252	# variable = "oxidised"
253	unit = "kO"
254	data = gas.read_all()
255	data = data.oxidising / 1000
256	display_text(variables[mode], data, unit)
257
258	if mode == 5:
259	# variable = "reduced"
260	unit = "kO"
261	data = gas.read_all()
262	data = data.reducing / 1000
263	display_text(variables[mode], data, unit)
264
265	if mode == 6:
266	# variable = "nh3"
267	unit = "kO"
268	data = gas.read_all()
269	data = data.nh3 / 1000
270	display_text(variables[mode], data, unit)
271
272	if mode == 7:
273	# variable = "pm1"
274	unit = "ug/m3"
275	try:
276	data = pms5003.read()
277	except pmsReadTimeoutError:
278	logging.warn("Failed to read PMS5003")
279	else:
280	data = float(data.pm_ug_per_m3(1.0))
281	display_text(variables[mode], data, unit)
282
283	if mode == 8:
284	# variable = "pm25"
285	unit = "ug/m3"
286	try:
287	data = pms5003.read()
288	except pmsReadTimeoutError:
289	logging.warn("Failed to read PMS5003")
290	else:
291	data = float(data.pm_ug_per_m3(2.5))
292	display_text(variables[mode], data, unit)
293
294	if mode == 9:
295	# variable = "pm10"
296	unit = "ug/m3"
297	try:
298	data = pms5003.read()
299	except pmsReadTimeoutError:
300	logging.warn("Failed to read PMS5003")
301	else:
302	data = float(data.pm_ug_per_m3(10))
303	display_text(variables[mode], data, unit)
304	if mode == 10:
305	# Everything on one screen
306	cpu_temp = get_cpu_temperature()
307	# Smooth out with some averaging to decrease jitter
308	cpu_temps = cpu_temps[1:] + [cpu_temp]
309	avg_cpu_temp = sum(cpu_temps) / float(len(cpu_temps))
310	raw_temp = bme280.get_temperature()
311	raw_data = raw_temp - ((avg_cpu_temp - raw_temp) / factor)
312	save_data(0, raw_data)
313	display_everything()
314	raw_data = bme280.get_pressure()
315	save_data(1, raw_data)
316	display_everything()
317	raw_data = bme280.get_humidity()
318	save_data(2, raw_data)
319	if proximity < 10:
320	raw_data = ltr559.get_lux()
321	else:
322	raw_data = 1
323	save_data(3, raw_data)
324	display_everything()
325	gas_data = gas.read_all()
326	save_data(4, gas_data.oxidising / 1000)
327	save_data(5, gas_data.reducing / 1000)
328	save_data(6, gas_data.nh3 / 1000)
329	display_everything()
330	pms_data = None
331	try:
332	pms_data = pms5003.read()
333	except pmsReadTimeoutError:
334	logging.warn("Failed to read PMS5003")
335	else:
336	save_data(7, float(pms_data.pm_ug_per_m3(1.0)))
337	save_data(8, float(pms_data.pm_ug_per_m3(2.5)))
338	save_data(9, float(pms_data.pm_ug_per_m3(10)))
339	display_everything()
340
341
342
343	# Exit cleanly
344	except KeyboardInterrupt:
345	sys.exit(0)