73 lines
2.8 KiB
Python
Executable File
73 lines
2.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
"""
|
|
This script does the following:
|
|
* gets temperature and humidity from two connected DHT22 sensors
|
|
* gets the energy gain of a PV system via an intermediate Tasmota smart plug
|
|
* sends an APRS message containing the gathered information to APRS-IS
|
|
"""
|
|
|
|
import requests
|
|
import aprslib
|
|
import Adafruit_DHT
|
|
|
|
TASMOTA_IP = "192.168.1.100" # change to IP address of Tasmota device
|
|
DHT_SENSOR = Adafruit_DHT.DHT22
|
|
SENSOR_RACK = 27 # change to corresponding GIO Pin
|
|
SENSOR_OUTSIDE = 17 # change to corresponding GIO Pin
|
|
APRS_SYMBOL = "-" # pick your APRS symbol here
|
|
CALL = "XXXXX-15" # enter your call sign
|
|
PASSWORD = "12345" # generate a passcode for your call and enter it here
|
|
PORT = 14580 # should be okay
|
|
GPS_COORDINATES = "585.36N/04041.46W" # enter your coordinates here
|
|
|
|
|
|
def get_energy_data():
|
|
"""grabs data from the Tasmota smart plug via HTTP request"""
|
|
resp = requests.get(url="http://{}/cm?cmnd=STATUS+8".format(TASMOTA_IP))
|
|
data = resp.json()
|
|
energy = data["StatusSNS"]["ENERGY"]
|
|
# Balkonkraftwerke (certain small PV systems) only generate 600W...
|
|
if energy["Power"] > 600:
|
|
energy["Power"] = 600
|
|
energy["Total"] = str(format(energy["Total"], '.1f'))
|
|
energy["Yesterday"] = str(format(energy["Yesterday"], '.1f'))
|
|
energy["Today"] = str(format(energy["Today"], '.1f'))
|
|
return energy
|
|
|
|
|
|
def send_aprs(weather_data, energy_data):
|
|
"""sends the APRS message to APRS-IS"""
|
|
# a valid passcode for the callsign is required in order to send
|
|
ais = aprslib.IS(CALL, passwd=PASSWORD, port=PORT)
|
|
ais.connect()
|
|
# send a single status message
|
|
msg = "PV Power:{} W, Today:{} kWh, Yesterday:{} kWh, Total:{} kWh - {}".format(\
|
|
energy_data["Power"], energy_data["Today"], energy_data["Yesterday"], \
|
|
energy_data["Total"], weather_data)
|
|
print(msg)
|
|
ais.sendall("{}>APRS,TCPIP*:={}{} {}".format(CALL, GPS_COORDINATES, APRS_SYMBOL, msg))
|
|
ais.sendall("{}>APRS,TCPIP*:>".format(CALL))
|
|
|
|
|
|
def get_weather_data():
|
|
"""grabs temperature and humidity from the attached DHT22 sensors"""
|
|
humidity_rack, temperature_rack = Adafruit_DHT.read_retry(DHT_SENSOR, SENSOR_RACK)
|
|
humidity_outside, temperature_outside = Adafruit_DHT.read_retry(DHT_SENSOR, SENSOR_OUTSIDE)
|
|
ret_val = "WX: "
|
|
if humidity_rack is not None and temperature_rack is not None:
|
|
ret_val = ret_val + "Rack:{0:0.1f}C/{1:0.1f}%".format(\
|
|
temperature_rack, humidity_rack)
|
|
else:
|
|
print("Failed to retrieve data from rack sensor")
|
|
if humidity_outside is not None and temperature_outside is not None:
|
|
ret_val = ret_val + " Outside:{0:0.1f}C/{1:0.1f}%".format(\
|
|
temperature_outside, humidity_outside)
|
|
else:
|
|
print("Failed to retrieve data from outside sensor")
|
|
return ret_val
|
|
|
|
|
|
if __name__ == "__main__":
|
|
send_aprs(get_weather_data(), get_energy_data())
|