135 lines
3.9 KiB
Text
135 lines
3.9 KiB
Text
|
|
from machine import Pin, ADC, Timer
|
|
import network
|
|
import time
|
|
import ujson
|
|
from uhashlib import sha256
|
|
import binascii
|
|
import math
|
|
import urequests
|
|
import ntptime
|
|
|
|
|
|
######################################################## CONFIG ########################################################
|
|
# Wifi settings:
|
|
# WIFI_ESSID = 'Your wifi ESSID (name)'
|
|
# WIFI_PSK = 'Your wifi key'
|
|
#
|
|
# Notification settings:
|
|
# NOTIFICATION_URL = 'https://automation.jaseg.de/notify/klingel'
|
|
# NOTIFICATION_SECRET = b'Your notification proxy secret for this endpoint'
|
|
#
|
|
# NOTIFICATION_COOLDOWN = 60 # how long to wait after sending a notification before sending the next, in seconds
|
|
# HEARTBEAT_INTERVAL = 60 # seconds
|
|
#
|
|
# Detection settings
|
|
# MEAN_LEN = 8 # Window length for DC offset determination in seconds (1024ms to be exact)
|
|
# RMS_THRESHOLD = 1000 # Threshold for rms detection threshold over 1s window in ADC counts
|
|
######################################################### END ##########################################################
|
|
|
|
|
|
def wifi_connect():
|
|
iface = network.WLAN(network.STA_IF)
|
|
if not iface.isconnected():
|
|
print('Connecting to wifi... ')
|
|
iface.active(True)
|
|
iface.connect(WIFI_ESSID, WIFI_PSK)
|
|
for i in range(20):
|
|
if iface.isconnected():
|
|
print('Wifi connected. IP config: ', iface.ifconfig())
|
|
break
|
|
time.sleep(0.5)
|
|
else:
|
|
print("Couldn't connect to wifi.")
|
|
|
|
|
|
buf = [0] * 1024
|
|
capture = None
|
|
mean = 0
|
|
rms = 0
|
|
sample_tim = Timer(-1)
|
|
|
|
def start_sampling():
|
|
global sample_tim
|
|
buf_pos = 0
|
|
buf_sum = 0
|
|
mean_acc = []
|
|
|
|
adc = ADC(Pin(34))
|
|
adc.atten(ADC.ATTN_11DB)
|
|
|
|
def sample_cb(tim):
|
|
global buf, mean, rms, capture
|
|
nonlocal adc, buf_pos, buf_sum, mean_acc
|
|
|
|
val = adc.read()
|
|
buf[buf_pos] = val
|
|
buf_sum += val
|
|
|
|
buf_pos += 1
|
|
if buf_pos == len(buf):
|
|
buf_pos = 0
|
|
mean_acc = [buf_sum/len(buf)] + mean_acc[:MEAN_LEN-1]
|
|
mean = sum(mean_acc)/len(mean_acc)
|
|
buf_sum = 0
|
|
|
|
rms = math.sqrt( sum( (x-mean)**2 for x in buf )/len(buf) )
|
|
capture = list(buf) # Make a copy
|
|
|
|
sample_tim.init(period=1, mode=Timer.PERIODIC, callback=sample_cb) # period in ms
|
|
|
|
def uhmac(key, data):
|
|
blocksize = 64
|
|
key += bytes(64 - len(key))
|
|
tx = lambda s, x: bytes( b ^ x for b in s )
|
|
outer = sha256(tx(key, 0x5C))
|
|
inner = sha256(tx(key, 0x36))
|
|
inner.update(data)
|
|
outer.update(inner.digest())
|
|
return outer.digest()
|
|
|
|
def unix_time():
|
|
return int(time.time()) + 946684800 # ESP32 counts from 2000-01-01, unix from 1970-01-01
|
|
|
|
def usign(secret, scope, payload=None, seq=None):
|
|
payload = {'time': unix_time(), 'scope': scope, 'd': payload}
|
|
if seq is not None:
|
|
payload['seq'] = seq
|
|
|
|
payload = ujson.dumps(payload).encode()
|
|
auth = binascii.hexlify(uhmac(secret, payload))
|
|
|
|
return ujson.dumps({'payload': payload, 'auth': auth})
|
|
|
|
def notify(scope, **kwargs):
|
|
data = usign(NOTIFICATION_SECRET, scope, kwargs)
|
|
print(unix_time(), 'Notifying', NOTIFICATION_URL)
|
|
urequests.post(NOTIFICATION_URL, data=data, headers={'Content-Type': 'application/json'})
|
|
|
|
def loop():
|
|
global rms, capture
|
|
last_notification, last_heartbeat = 0, 0
|
|
while True:
|
|
now = unix_time()
|
|
if (now - last_notification) > NOTIFICATION_COOLDOWN and rms > RMS_THRESHOLD:
|
|
wifi_connect()
|
|
old_capture = capture
|
|
rms = 0
|
|
while rms == 0:
|
|
time.sleep(0.1)
|
|
rms = 0
|
|
notify('default', rms=rms, capture=[old_capture, capture])
|
|
last_notification = now
|
|
|
|
if (now - last_heartbeat) > HEARTBEAT_INTERVAL:
|
|
notify('heartbeat')
|
|
last_heartbeat = now
|
|
|
|
time.sleep(0.1)
|
|
|
|
wifi_connect()
|
|
ntptime.settime()
|
|
start_sampling()
|
|
notify('boot')
|
|
loop()
|
|
|