108 lines
2.4 KiB
Python
108 lines
2.4 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
A hub for controlling IOT devices.
|
||
|
|
"""
|
||
|
|
import re
|
||
|
|
import json
|
||
|
|
|
||
|
|
import requests
|
||
|
|
from flask import Flask, render_template
|
||
|
|
|
||
|
|
|
||
|
|
class RelayDevice:
|
||
|
|
"""
|
||
|
|
Represents a relay receptacle device controlled by an ESP-01. The
|
||
|
|
ESP-01 acts a simple switch to turn each receptable on or off via
|
||
|
|
a relay. Each device only has two outputs, `OUT0` and `OUT2`.
|
||
|
|
"""
|
||
|
|
def __init__(self):
|
||
|
|
self.id = ""
|
||
|
|
self.description = ""
|
||
|
|
self.location = ""
|
||
|
|
self.ip_address = ""
|
||
|
|
self.sub_devices = {}
|
||
|
|
|
||
|
|
self.update()
|
||
|
|
|
||
|
|
def update(self):
|
||
|
|
"""
|
||
|
|
Queries the physical device and updates the internal model as
|
||
|
|
appropriate.
|
||
|
|
"""
|
||
|
|
for name, device in self.sub_devices.items():
|
||
|
|
res = requests.get(self.ip_address)
|
||
|
|
gpio0 = re.search(r"GPIO0: (\bLow|\bHigh)", res.text).groups()[0]
|
||
|
|
device.state = gpio0 == 'High'
|
||
|
|
|
||
|
|
def from_dict(self, data):
|
||
|
|
"""
|
||
|
|
Initializes self with data from JSON dict.
|
||
|
|
"""
|
||
|
|
self.id = data['id']
|
||
|
|
self.description = data['description']
|
||
|
|
self.location = data['location']
|
||
|
|
self.ip_address = data['ip_address']
|
||
|
|
self.sub_devices = {}
|
||
|
|
for sub_dev_type, sub_devs in data['sub_devices'].items():
|
||
|
|
if sub_dev_type == 'RelayOutlet':
|
||
|
|
self.sub_devices[sub_dev_type] = []
|
||
|
|
for sub_dev in sub_devs:
|
||
|
|
sub_dev = RelayOutlet().from_dict(sub_dev)
|
||
|
|
self.sub_devices[sub_dev_type].append(sub_dev)
|
||
|
|
else:
|
||
|
|
raise ValueError(f"Unknown sub_device type {dev_name}")
|
||
|
|
return self
|
||
|
|
|
||
|
|
|
||
|
|
class RelayOutlet:
|
||
|
|
"""
|
||
|
|
Represents a single outlet on a RelayDevice.
|
||
|
|
"""
|
||
|
|
def __init__(self):
|
||
|
|
self.id = ""
|
||
|
|
self.description = ""
|
||
|
|
self.state = False
|
||
|
|
|
||
|
|
def from_dict(self, data):
|
||
|
|
"""
|
||
|
|
Initializes self with data from JSON dict.
|
||
|
|
"""
|
||
|
|
self.id = data['id']
|
||
|
|
self.description = data['description']
|
||
|
|
self.state = data['state']
|
||
|
|
return self
|
||
|
|
|
||
|
|
|
||
|
|
def init_network():
|
||
|
|
"""
|
||
|
|
Initializes the network of IOT devices.
|
||
|
|
"""
|
||
|
|
with open("devices.json", 'r') as file:
|
||
|
|
data = file.read()
|
||
|
|
data = json.loads(data)
|
||
|
|
network = {}
|
||
|
|
for dev_type, devices in data.items():
|
||
|
|
if dev_type == 'RelayDevice':
|
||
|
|
network[dev_type] = []
|
||
|
|
for device in devices:
|
||
|
|
device = RelayDevice().from_dict(device)
|
||
|
|
network[dev_type].append(device)
|
||
|
|
else:
|
||
|
|
raise ValueError(f"Unknown device type {dev_name}")
|
||
|
|
return network
|
||
|
|
|
||
|
|
|
||
|
|
app = Flask(__name__)
|
||
|
|
network = init_network()
|
||
|
|
|
||
|
|
@app.route('/')
|
||
|
|
def index():
|
||
|
|
"""
|
||
|
|
The index page.
|
||
|
|
"""
|
||
|
|
return render_template('index.html', network=network)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
app.run(host='0.0.0.0', port=5300)
|