163 lines
4.6 KiB
Python
163 lines
4.6 KiB
Python
import utime
|
|
import ujson
|
|
import network
|
|
import urequests
|
|
import machine
|
|
from machine import Pin
|
|
|
|
threshold = 15 # cm
|
|
sound_speed = 340 * 100 # m/s * centi = cm/s
|
|
|
|
|
|
# TODO: add if open for more than 1 minute, send alert
|
|
class App:
|
|
ssid: str
|
|
password: str
|
|
server: str
|
|
token: str
|
|
|
|
opened: bool = False
|
|
previously_opened: bool = False
|
|
ultra_opened_counter: int = 0
|
|
ultra_opened_threshold: int = 3
|
|
|
|
led = Pin(15, Pin.OUT)
|
|
trigger = Pin(2, Pin.OUT)
|
|
echo = Pin(3, Pin.IN)
|
|
|
|
def __init__(self):
|
|
print("Loading .env.json...")
|
|
with open(".env.json") as stream:
|
|
data = ujson.load(stream)
|
|
|
|
network = data.get("network")
|
|
if network is None:
|
|
print("'network' missing in env")
|
|
exit(1)
|
|
|
|
server = data.get("server")
|
|
if server is None:
|
|
print("'server' missing in env")
|
|
exit(1)
|
|
|
|
self.ssid = network.get("ssid")
|
|
self.password = network.get("password")
|
|
self.server = server.get("address")
|
|
self.token = server.get("token")
|
|
|
|
if self.ssid is None or \
|
|
self.password is None or \
|
|
self.server is None or \
|
|
self.token is None:
|
|
print("missing properties in env")
|
|
exit(1)
|
|
|
|
def connect(self):
|
|
print("Connecting to wifi...")
|
|
wlan = network.WLAN(network.STA_IF)
|
|
wlan.active(False)
|
|
utime.sleep_ms(250)
|
|
wlan.active(True)
|
|
utime.sleep_ms(250)
|
|
|
|
wlan.connect(self.ssid, self.password)
|
|
utime.sleep_ms(100)
|
|
|
|
i = 0
|
|
while not wlan.isconnected():
|
|
print(f"Waiting for connection{(i % 3 + 1) * "."}{3 * " "}", end="\r")
|
|
i += 1
|
|
for _ in range(4):
|
|
self.led.toggle()
|
|
utime.sleep_ms(250)
|
|
|
|
if i % 10 == 0:
|
|
print("Attempting to restart connection...")
|
|
wlan.connect(self.ssid, self.password)
|
|
for i in range(10):
|
|
self.led.toggle()
|
|
utime.sleep_ms(50)
|
|
|
|
print(f"Connected with IP {wlan.ifconfig()[0]}")
|
|
self.update_server()
|
|
|
|
def update_server(self):
|
|
print("Updating state...", end="\r")
|
|
|
|
data = {"opened": self.opened}
|
|
raw = ujson.dumps(data)
|
|
try:
|
|
r = urequests.post(f"{self.server}/open",
|
|
headers={"Authorization": self.token, "Content-Type": "application/json"},
|
|
data=raw)
|
|
print(f"State updated [{r.status_code}]")
|
|
except Exception as e:
|
|
print(f"Error occurred: {e}")
|
|
|
|
def ultra(self):
|
|
self.trigger.low()
|
|
utime.sleep_us(2)
|
|
self.trigger.high()
|
|
utime.sleep_us(10)
|
|
self.trigger.low()
|
|
|
|
# echo goes high as soon as sound wave is sent
|
|
# and returns to low once sound gets back
|
|
sent_time = utime.ticks_us()
|
|
while self.echo.value() == 0:
|
|
sent_time = utime.ticks_us() # us
|
|
|
|
received_time = utime.ticks_us()
|
|
while self.echo.value() == 1:
|
|
received_time = utime.ticks_us() # us
|
|
|
|
delta_time = received_time - sent_time # us
|
|
distance = delta_time / 1_000_000 * sound_speed / 2
|
|
print(f"distance: {distance} cm")
|
|
|
|
if distance < threshold:
|
|
# door closed
|
|
self.led.low()
|
|
self.ultra_opened_counter = 0
|
|
|
|
# was it closed just now?
|
|
if self.previously_opened:
|
|
self.update_server()
|
|
self.previously_opened = False
|
|
else:
|
|
# door opened
|
|
self.led.high()
|
|
self.ultra_opened_counter += 1
|
|
|
|
# was it opened just now?
|
|
if not self.previously_opened:
|
|
self.update_server()
|
|
self.previously_opened = True
|
|
|
|
def run(self):
|
|
print("Starting door alarm...")
|
|
self.connect()
|
|
|
|
i = 0
|
|
# T = 500ms f = 2 Hz
|
|
while True:
|
|
try:
|
|
self.ultra()
|
|
|
|
if i >= 20: # every 10 seconds, blink for 100ms
|
|
self.led.toggle()
|
|
utime.sleep_ms(100)
|
|
self.led.toggle()
|
|
utime.sleep_ms(400)
|
|
i = 0
|
|
else:
|
|
utime.sleep_ms(500)
|
|
except Exception as e:
|
|
print(f"Fatal exception occurred: {e}")
|
|
|
|
i += 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
App().run()
|