#!/usr/bin/env python3 """ A hub for controlling IOT devices. """ import os import re import copy import json import requests from flask import Flask, render_template, request, abort, jsonify from flask import Blueprint import config import auth from auth import auth_required 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({'ok': True, 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_views = Blueprint('app_views', __name__) @app_views.route('/') @auth_required def index(): """ The index page. """ return render_template('index.html', network=network) @app_views.route('/toggle') @auth_required 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: return make_error(404, "device_id not found") res = device.toggle(sub_dev_id) if not res: return make_error(404, "sub_dev_id not found") save_network() return res @app_views.route('/edit') @auth_required def edit(): """ Edits the text of a particular field. """ device_id = request.args.get('device_id') sub_dev_id = request.args.get('sub_dev_id') field = request.args.get('field') value = request.args.get('value') for device in sum(network.values(), []): if device.id == device_id: break else: return make_error(404, "device_id not found") if sub_dev_id: for sub_dev in sum(device.sub_devices.values(), []): if sub_dev.id == sub_dev_id: break else: return make_error(404, "sub_dev_id not found") if hasattr(sub_dev, field): setattr(sub_dev, field, value) else: return make_error(404, "sub_device field not found") else: if hasattr(device, field): setattr(device, field, value) else: return make_error(404, "device field not found") data = { 'ok': True, 'field': field, 'value': value } save_network() return json.dumps(data) def make_error(code, message): """ Returns a JSON error. """ res = jsonify(ok=False, status=code, message=message) res.status_code = code return res app = Flask(__name__) app.register_blueprint(app_views, url_prefix=config.url_prefix) app.register_blueprint(auth.auth_views, url_prefix=config.url_prefix) if os.path.isfile('secret_key'): with open('secret_key', 'rb') as file: app.secret_key = file.read() else: secret_key = os.urandom(32) app.secret_key = secret_key with open('secret_key', 'wb') as file: file.write(secret_key) network = init_network() if __name__ == '__main__': app.run(host='0.0.0.0', port=5300)