#!/usr/bin/env python3 """ A hub for controlling IOT devices. """ import re import copy import json import requests from flask import Flask, render_template, request, abort 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 toggle(self, sub_dev_id): """ Toggles the state of a sub device. """ for sub_dev in sum(self.sub_devices.values(), []): if sub_dev.id == sub_dev_id: break else: return None params = {sub_dev.gpio: 'toggle'} res = requests.get('http://' + self.ip_address + '/gpio',params=params) res.raise_for_status() state = re.search( rf"GPIO{sub_dev.gpio}: (\bLow|\bHigh)", res.text ).groups()[0] == 'High' sub_dev.state = state return json.dumps({sub_dev_id: state}) def from_dict(self, data): """ Initializes self with data from JSON dict. """ for key, value in data.items(): setattr(self, key, value) 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.gpio = "" self.state = False def from_dict(self, data): """ Initializes self with data from JSON dict. """ for key, value in data.items(): setattr(self, key, value) return self class DeviceEncoder(json.JSONEncoder): """ A custom json encoder for dumping devices to file. """ def default(self, o): return vars(o) def init_network(filepath="devices.json"): """ Initializes the network of IOT devices. """ with open(filepath, '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 def save_network(filepath="devices.json"): """ Dumps the network to file. """ with open(filepath, 'w') as file: file.write(json.dumps(network, cls=DeviceEncoder, indent='\t')) app = Flask(__name__) network = init_network() @app.route('/') def index(): """ The index page. """ return render_template('index.html', network=network) @app.route('/toggle') def toggle(): """ Toggles the state of a RelayDevice. """ device_id = request.args.get('device_id') sub_dev_id = request.args.get('sub_dev_id') for device in sum(network.values(), []): if device.id == device_id: break else: abort(404) res = device.toggle(sub_dev_id) if not res: abort(404) save_network() return res if __name__ == '__main__': app.run(host='0.0.0.0', port=5300)