Für diese Projekt setzen wir voraus, dass eine Python Entwicklungsumgebung eingerichtet ist und ein grundsätzliches Verständnis der Python Programmiersprache vorhanden ist.
Falls dies nicht der Fall ist sollte hier begonnen werden. Informationen über die Tinkerforge API sind dann hier zu finden.
Wir setzen uns folgende Ziele für dieses Projekt:
Da dieses Projekt wahrscheinlich 24/7 laufen wird, wollen wir sicherstellen, dass das Programm möglichst robust gegen externe Einflüsse ist. Das Programm sollte weiterhin funktionieren falls
Im Folgenden werden wir Schritt für Schritt zeigen wie diese Ziele erreicht werden können.
Als Erstes legen wir fest wohin unser Programm sich verbinden soll:
HOST = "localhost"
PORT = 4223
Falls eine WIFI Extension verwendet wird, oder der Brick Daemon auf einem anderen PC läuft, dann muss "localhost" durch die IP Adresse oder den Hostnamen der WIFI Extension oder des anderen PCs ersetzt werden.
Nach dem Start des Programms müssen der CALLBACK_ENUMERATE
Callback
und der CALLBACK_CONNECTED
Callback registriert und ein erstes
Enumerate ausgelöst werden:
def __init__(self):
self.ipcon = IPConnection()
self.ipcon.connect(WeatherStation.HOST, WeatherStation.PORT)
self.ipcon.register_callback(IPConnection.CALLBACK_ENUMERATE,
self.cb_enumerate)
self.ipcon.register_callback(IPConnection.CALLBACK_CONNECTED,
self.cb_connected)
self.ipcon.enumerate()
Der Enumerate Callback wird ausgelöst wenn ein Brick per USB angeschlossen wird
oder wenn die enumerate()
Funktion aufgerufen wird. Dies ermöglicht es die
Bricks und Bricklets im Stapel zu erkennen ohne im Voraus ihre UIDs kennen zu
müssen.
Der Connected Callback wird ausgelöst wenn die Verbindung zur WIFI Extension oder zum Brick Daemon hergestellt wurde. In diesem Callback muss wiederum ein Enumerate angestoßen werden, wenn es sich um ein Auto-Reconnect handelt:
def cb_connected(self, connected_reason):
if connected_reason == IPConnection.CONNECT_REASON_AUTO_RECONNECT:
self.ipcon.enumerate()
Ein Auto-Reconnect bedeutet, dass die Verbindung zur WIFI Extension oder zum Brick Daemon verloren gegangen ist und automatisch wiederhergestellt werden konnte. In diesem Fall kann es sein, dass die Bricklets ihre Konfiguration verloren haben und wir sie neu konfigurieren müssen. Da die Konfiguration beim Enumerate (siehe unten) durchgeführt wird, lösen wir einfach noch ein Enumerate aus.
Schritt 1 zusammengefügt:
class WeatherStation:
HOST = "localhost"
PORT = 4223
def __init__(self):
self.ipcon = IPConnection()
self.ipcon.connect(WeatherStation.HOST, WeatherStation.PORT)
self.ipcon.register_callback(IPConnection.CALLBACK_ENUMERATE,
self.cb_enumerate)
self.ipcon.register_callback(IPConnection.CALLBACK_CONNECTED,
self.cb_connected)
self.ipcon.enumerate()
def cb_connected(self, connected_reason):
if connected_reason == IPConnection.CONNECT_REASON_AUTO_RECONNECT:
self.ipcon.enumerate()
Während des Enumerierungsprozesse sollen alle messenden Bricklets konfiguriert werden. Dadurch ist sichergestellt, dass sie neu konfiguriert werden nach einem Verbindungsabbruch oder einer Unterbrechung der Stromversorgung.
Die Konfiguration soll beim ersten Start (ENUMERATION_TYPE_CONNECTED
)
durchgeführt werden und auch bei jedem extern ausgelösten Enumerate
(ENUMERATION_TYPE_AVAILABLE
):
def cb_enumerate(self, uid, connected_uid, position, hardware_version,
firmware_version, device_identifier, enumeration_type):
if enumeration_type == IPConnection.ENUMERATION_TYPE_CONNECTED or \
enumeration_type == IPConnection.ENUMERATION_TYPE_AVAILABLE:
Die Konfiguration des LCD 20x4 ist einfach, wir löschen den aktuellen Inhalt des Displays und schalten das Backlight ein:
if device_identifier == LCD20x4.DEVICE_IDENTIFIER:
self.lcd = LCD20x4(uid, self.ipcon)
self.lcd.clear_display()
self.lcd.backlight_on()
Das Ambient Light, Humidity und Barometer Bricklet werden so eingestellt, dass sie uns ihre jeweiligen Messwerte höchsten mit einer Periode von 1000ms (1s) mitteilen:
elif device_identifier == AmbientLight.DEVICE_IDENTIFIER:
self.al = AmbientLight(uid, self.ipcon)
self.al.set_illuminance_callback_period(1000)
self.al.register_callback(self.al.CALLBACK_ILLUMINANCE,
self.cb_illuminance)
elif device_identifier == Humidity.DEVICE_IDENTIFIER:
self.hum = Humidity(uid, self.ipcon)
self.hum.set_humidity_callback_period(1000)
self.hum.register_callback(self.hum.CALLBACK_HUMIDITY,
self.cb_humidity)
elif device_identifier == Barometer.DEVICE_IDENTIFIER:
self.baro = Barometer(uid, self.ipcon)
self.baro.set_air_pressure_callback_period(1000)
self.baro.register_callback(self.baro.CALLBACK_AIR_PRESSURE,
self.cb_air_pressure)
Dies bedeutet, dass die Bricklets die cb_illuminance
, cb_humidity
und cb_air_pressure
Callback-Funktionen immer dann aufrufen wenn sich der
Messwert verändert hat, aber höchsten alle 1000ms.
Schritt 2 zusammengefügt:
def cb_enumerate(self, uid, connected_uid, position, hardware_version,
firmware_version, device_identifier, enumeration_type):
if enumeration_type == IPConnection.ENUMERATION_TYPE_CONNECTED or \
enumeration_type == IPConnection.ENUMERATION_TYPE_AVAILABLE:
if device_identifier == LCD20x4.DEVICE_IDENTIFIER:
self.lcd = LCD20x4(uid, self.ipcon)
self.lcd.clear_display()
self.lcd.backlight_on()
elif device_identifier == AmbientLight.DEVICE_IDENTIFIER:
self.al = AmbientLight(uid, self.ipcon)
self.al.set_illuminance_callback_period(1000)
self.al.register_callback(self.al.CALLBACK_ILLUMINANCE,
self.cb_illuminance)
elif device_identifier == Humidity.DEVICE_IDENTIFIER:
self.hum = Humidity(uid, self.ipcon)
self.hum.set_humidity_callback_period(1000)
self.hum.register_callback(self.hum.CALLBACK_HUMIDITY,
self.cb_humidity)
elif device_identifier == Barometer.DEVICE_IDENTIFIER:
self.baro = Barometer(uid, self.ipcon)
self.baro.set_air_pressure_callback_period(1000)
self.baro.register_callback(self.baro.CALLBACK_AIR_PRESSURE,
self.cb_air_pressure)
Wir wollen eine hübsche Darstellung der Messwerte auf dem Display. Zum Beispiel:
Illuminanc 137.39 lx
Humidity 34.10 %
Air Press 987.70 mb
Temperature 22.64 °C
Die Dezimaltrennzeichen und die Einheiten sollen in jeweils einer Spalte übereinander stehen. Daher verwenden wird zwei Zeichen für jede Einheit, zwei Nachkommastellen und kürzen die Namen so, dass sie in den restlichen Platz der jeweiligen Zeile passen. Das ist auch der Grund, warum dem "Illuminanc" das letzte "e" fehlt.
text = '%6.2f' % value
Der obige Ausdruck wandelt eine Fließkommazahl in eine Zeichenkette um, gemäß der gegebenen Formatspezifikation. Das Ergebnis ist dann mindestens 6 Zeichen lang mit 2 Nachkommastellen. Fall es weniger als 6 Zeichen sind wird von Links mit Leerzeichen aufgefüllt.
def cb_illuminance(self, illuminance):
text = 'Illuminanc %6.2f lx' % (illuminance/10.0)
self.lcd.write_line(0, 0, text)
def cb_humidity(self, humidity):
text = 'Humidity %6.2f %%' % (humidity/10.0)
self.lcd.write_line(1, 0, text)
def cb_air_pressure(self, air_pressure):
text = 'Air Press %7.2f mb' % (air_pressure/1000.0)
self.lcd.write_line(2, 0, text)
Es fehlt noch die Temperatur. Das Barometer Bricklet kann auch die Temperatur
messen, aber es hat dafür keinen Callback. Als einfacher Workaround können wir
die Temperatur in der cb_air_pressure
Callback-Funktion abfragen:
def cb_air_pressure(self, air_pressure):
text = 'Air Press %7.2f mb' % (air_pressure/1000.0)
self.lcd.write_line(2, 0, text)
# \xDF == ° on LCD 20x4 charset
text = 'Temperature %5.2f \xDFC' % (self.baro.get_chip_temperature()/100.0)
self.lcd.write_line(3, 0, text)
Schritt 3 zusammengefügt:
def cb_illuminance(self, illuminance):
text = 'Illuminanc %6.2f lx' % (illuminance/10.0)
self.lcd.write_line(0, 0, text)
def cb_humidity(self, humidity):
text = 'Humidity %6.2f %%' % (humidity/10.0)
self.lcd.write_line(1, 0, text)
def cb_air_pressure(self, air_pressure):
text = 'Air Press %7.2f mb' % (air_pressure/1000.0)
self.lcd.write_line(2, 0, text)
# \xDF == ° on LCD 20x4 charset
text = 'Temperature %5.2f \xDFC' % (self.baro.get_chip_temperature()/100.0)
self.lcd.write_line(3, 0, text)
Das ist es. Wenn wir diese drei Schritte zusammen in eine Datei kopieren und ausführen, dann hätten wir jetzt eine funktionierenden Wetterstation.
Es gibt einige offensichtliche Möglichkeiten die Ausgabe der Messdaten noch zu
verbessern. Die Namen könnten dynamisch exakt gekürzt werden, abhängig vom
aktuell freien Raum der jeweiligen Zeile. Auch könnten die Namen können noch
ins Deutsche übersetzt werden. Ein anderes Problem ist die Abfrage der
Temperatur in der cb_air_pressure
Callback-Funktion. Wenn sich der Luftdruck
nicht ändert dann wird auch die Anzeige der Temperatur nicht aktualisiert, auch
wenn sich diese eigentlich geändert hat. Es wäre besser die Temperatur jede Sekunde in einem eigenen Thread anzufragen.
Aber wir wollen das Programm für den Anfang einfach halten.
Wie dem auch sei, wir haben noch nicht alle Ziele erreicht. Das Programm ist noch nicht robust genug. Was passiert wenn die Verbindung beim Start des Programms nicht hergestellt werden kann, oder wenn das Enumerate nach einem Auto-Reconnect nicht funktioniert?
Wir brauchen noch Fehlerbehandlung!
Beim Start des Programms versuchen wir solange die Verbindung herzustellen, bis es klappt:
while True:
try:
self.ipcon.connect(WeatherStation.HOST, WeatherStation.PORT)
break
except Error as e:
log.error('Connection Error: ' + str(e.description))
time.sleep(1)
except socket.error as e:
log.error('Socket error: ' + str(e))
time.sleep(1)
und es wird solange versucht ein Enumerate zu starten bis auch dis geklappt hat:
while True:
try:
self.ipcon.enumerate()
break
except Error as e:
log.error('Enumerate Error: ' + str(e.description))
time.sleep(1)
Mit diesen Änderungen kann das Programm schon gestartet werden bevor die Wetterstation angeschlossen ist.
Es muss auch sichergestellt werden, dass wir nur auf das LCD schreiben nachdem es initialisiert wurde:
def cb_illuminance(self, illuminance):
if self.lcd is not None:
text = 'Illuminanc %6.2f lx' % (illuminance/10.0)
self.lcd.write_line(0, 0, text)
log.info('Write to line 0: ' + text)
und es müssen mögliche Fehler während des Enumerierungsprozesses behandelt werden:
if device_identifier == AmbientLight.DEVICE_IDENTIFIER:
try:
self.al = AmbientLight(uid, self.ipcon)
self.al.set_illuminance_callback_period(1000)
self.al.register_callback(self.al.CALLBACK_ILLUMINANCE,
self.cb_illuminance)
log.info('Ambient Light initialized')
except Error as e:
log.error('Ambient Light init failed: ' + str(e.description))
self.al = None
Zusätzlich wollen wir noch ein paar Logausgaben einfügen. Diese ermöglichen es später herauszufinden was ein Problem ausgelöst hat, wenn die Wetterstation nach einer Weile möglicherweise nicht mehr funktioniert wie erwartet.
Zum Beispiel, wenn die Wetterstation über WLAN angebunden ist und häufig Auto-Reconnects auftreten, dann ist wahrscheinlich die WLAN Verbindung nicht sehr stabil.
Jetzt sind alle für diese Projekt gesteckten Ziele erreicht.
Das gesamte Programm für die Wetterstation (download):
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import socket
import sys
import time
import math
import logging as log
log.basicConfig(level=log.INFO)
from tinkerforge.ip_connection import IPConnection
from tinkerforge.ip_connection import Error
from tinkerforge.bricklet_lcd_20x4 import BrickletLCD20x4
from tinkerforge.bricklet_ambient_light import BrickletAmbientLight
from tinkerforge.bricklet_ambient_light_v2 import BrickletAmbientLightV2
from tinkerforge.bricklet_ambient_light_v3 import BrickletAmbientLightV3
from tinkerforge.bricklet_humidity import BrickletHumidity
from tinkerforge.bricklet_humidity_v2 import BrickletHumidityV2
from tinkerforge.bricklet_barometer import BrickletBarometer
from tinkerforge.bricklet_barometer_v2 import BrickletBarometerV2
class WeatherStation:
HOST = "localhost"
PORT = 4223
ipcon = None
lcd = None
al = None
al_v2 = None
al_v3 = None
hum = None
hum_v2 = None
baro = None
baro_v2 = None
def __init__(self):
self.ipcon = IPConnection()
while True:
try:
self.ipcon.connect(WeatherStation.HOST, WeatherStation.PORT)
break
except Error as e:
log.error('Connection Error: ' + str(e.description))
time.sleep(1)
except socket.error as e:
log.error('Socket error: ' + str(e))
time.sleep(1)
self.ipcon.register_callback(IPConnection.CALLBACK_ENUMERATE,
self.cb_enumerate)
self.ipcon.register_callback(IPConnection.CALLBACK_CONNECTED,
self.cb_connected)
while True:
try:
self.ipcon.enumerate()
break
except Error as e:
log.error('Enumerate Error: ' + str(e.description))
time.sleep(1)
def cb_illuminance(self, illuminance):
if self.lcd is not None:
text = 'Illuminanc %6.2f lx' % (illuminance/10.0)
self.lcd.write_line(0, 0, text)
log.info('Write to line 0: ' + text)
def cb_illuminance_v2(self, illuminance):
if self.lcd is not None:
text = 'Illumina %8.2f lx' % (illuminance/100.0)
self.lcd.write_line(0, 0, text)
log.info('Write to line 0: ' + text)
def cb_illuminance_v3(self, illuminance):
if self.lcd is not None:
text = 'Illumina %8.2f lx' % (illuminance/100.0)
self.lcd.write_line(0, 0, text)
log.info('Write to line 0: ' + text)
def cb_humidity(self, humidity):
if self.lcd is not None:
text = 'Humidity %6.2f %%' % (humidity/10.0)
self.lcd.write_line(1, 0, text)
log.info('Write to line 1: ' + text)
def cb_humidity_v2(self, humidity):
if self.lcd is not None:
text = 'Humidity %6.2f %%' % (humidity/100.0)
self.lcd.write_line(1, 0, text)
log.info('Write to line 1: ' + text)
def cb_air_pressure(self, air_pressure):
if self.lcd is not None:
text = 'Air Press %7.2f mb' % (air_pressure/1000.0)
self.lcd.write_line(2, 0, text)
log.info('Write to line 2: ' + text)
try:
temperature = self.baro.get_chip_temperature()
except Error as e:
log.error('Could not get temperature: ' + str(e.description))
return
# \xDF == ° on LCD 20x4 charset
text = 'Temperature %5.2f \xDFC' % (temperature/100.0)
self.lcd.write_line(3, 0, text)
log.info('Write to line 3: ' + text.replace('\xDF', '°'))
def cb_air_pressure_v2(self, air_pressure):
if self.lcd is not None:
text = 'Air Press %7.2f mb' % (air_pressure/1000.0)
self.lcd.write_line(2, 0, text)
log.info('Write to line 2: ' + text)
try:
temperature = self.baro_v2.get_temperature()
except Error as e:
log.error('Could not get temperature: ' + str(e.description))
return
# \xDF == ° on LCD 20x4 charset
text = 'Temperature %5.2f \xDFC' % (temperature/100.0)
self.lcd.write_line(3, 0, text)
log.info('Write to line 3: ' + text.replace('\xDF', '°'))
def cb_enumerate(self, uid, connected_uid, position, hardware_version,
firmware_version, device_identifier, enumeration_type):
if enumeration_type == IPConnection.ENUMERATION_TYPE_CONNECTED or \
enumeration_type == IPConnection.ENUMERATION_TYPE_AVAILABLE:
if device_identifier == BrickletLCD20x4.DEVICE_IDENTIFIER:
try:
self.lcd = BrickletLCD20x4(uid, self.ipcon)
self.lcd.clear_display()
self.lcd.backlight_on()
log.info('LCD 20x4 initialized')
except Error as e:
log.error('LCD 20x4 init failed: ' + str(e.description))
self.lcd = None
elif device_identifier == BrickletAmbientLight.DEVICE_IDENTIFIER:
try:
self.al = BrickletAmbientLight(uid, self.ipcon)
self.al.set_illuminance_callback_period(1000)
self.al.register_callback(self.al.CALLBACK_ILLUMINANCE,
self.cb_illuminance)
log.info('Ambient Light initialized')
except Error as e:
log.error('Ambient Light init failed: ' + str(e.description))
self.al = None
elif device_identifier == BrickletAmbientLightV2.DEVICE_IDENTIFIER:
try:
self.al_v2 = BrickletAmbientLightV2(uid, self.ipcon)
self.al_v2.set_configuration(self.al_v2.ILLUMINANCE_RANGE_64000LUX,
self.al_v2.INTEGRATION_TIME_200MS)
self.al_v2.set_illuminance_callback_period(1000)
self.al_v2.register_callback(self.al_v2.CALLBACK_ILLUMINANCE,
self.cb_illuminance_v2)
log.info('Ambient Light 2.0 initialized')
except Error as e:
log.error('Ambient Light 2.0 init failed: ' + str(e.description))
self.al_v2 = None
elif device_identifier == BrickletAmbientLightV3.DEVICE_IDENTIFIER:
try:
self.al_v3 = BrickletAmbientLightV3(uid, self.ipcon)
self.al_v3.set_configuration(self.al_v3.ILLUMINANCE_RANGE_64000LUX,
self.al_v3.INTEGRATION_TIME_200MS)
self.al_v3.set_illuminance_callback_configuration(1000, False, 'x', 0, 0)
self.al_v3.register_callback(self.al_v3.CALLBACK_ILLUMINANCE,
self.cb_illuminance_v3)
log.info('Ambient Light 3.0 initialized')
except Error as e:
log.error('Ambient Light 3.0 init failed: ' + str(e.description))
self.al_v3 = None
elif device_identifier == BrickletHumidity.DEVICE_IDENTIFIER:
try:
self.hum = BrickletHumidity(uid, self.ipcon)
self.hum.set_humidity_callback_period(1000)
self.hum.register_callback(self.hum.CALLBACK_HUMIDITY,
self.cb_humidity)
log.info('Humidity initialized')
except Error as e:
log.error('Humidity init failed: ' + str(e.description))
self.hum = None
elif device_identifier == BrickletHumidityV2.DEVICE_IDENTIFIER:
try:
self.hum_v2 = BrickletHumidityV2(uid, self.ipcon)
self.hum_v2.set_humidity_callback_configuration(1000, True, 'x', 0, 0)
self.hum_v2.register_callback(self.hum_v2.CALLBACK_HUMIDITY,
self.cb_humidity_v2)
log.info('Humidity 2.0 initialized')
except Error as e:
log.error('Humidity 2.0 init failed: ' + str(e.description))
self.hum_v2 = None
elif device_identifier == BrickletBarometer.DEVICE_IDENTIFIER:
try:
self.baro = BrickletBarometer(uid, self.ipcon)
self.baro.set_air_pressure_callback_period(1000)
self.baro.register_callback(self.baro.CALLBACK_AIR_PRESSURE,
self.cb_air_pressure)
log.info('Barometer initialized')
except Error as e:
log.error('Barometer init failed: ' + str(e.description))
self.baro = None
elif device_identifier == BrickletBarometerV2.DEVICE_IDENTIFIER:
try:
self.baro_v2 = BrickletBarometerV2(uid, self.ipcon)
self.baro_v2.set_air_pressure_callback_configuration(1000, False, 'x', 0, 0)
self.baro_v2.register_callback(self.baro_v2.CALLBACK_AIR_PRESSURE,
self.cb_air_pressure_v2)
log.info('Barometer initialized')
except Error as e:
log.error('Barometer init failed: ' + str(e.description))
self.baro_v2 = None
def cb_connected(self, connected_reason):
if connected_reason == IPConnection.CONNECT_REASON_AUTO_RECONNECT:
log.info('Auto Reconnect')
while True:
try:
self.ipcon.enumerate()
break
except Error as e:
log.error('Enumerate Error: ' + str(e.description))
time.sleep(1)
if __name__ == "__main__":
log.info('Weather Station: Start')
weather_station = WeatherStation()
if sys.version_info < (3, 0):
input = raw_input # Compatibility for Python 2.x
input('Press key to exit\n')
if weather_station.ipcon != None:
weather_station.ipcon.disconnect()
log.info('Weather Station: End')