iou1name/Juice
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

split data models into separate module

Browse Source
This commit is contained in:
iou1name 2019-07-02 08:56:17 -04:00
parent c6fe1216c2
commit bd61bb6819
2 changed files with 148 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

147
juice.py
View File

@ -7,142 +7,15 @@ import re
import copy
import json
import requests
from flask import Flask, render_template, request, abort, jsonify
from flask import Blueprint
import auth
import config
import models
from auth import auth_required
from tools import make_error
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.type = "RelayDevice"
self.description = ""
self.location = ""
self.ip_address = ""
self.locked = False
self.sub_devices = []
self.sub_devices.append(RelayOutlet())
self.sub_devices.append(RelayOutlet())
self.sub_devices[0].id = 'OUT0'
self.sub_devices[0].gpio = '0'
self.sub_devices[1].id = 'OUT2'
self.sub_devices[1].gpio = '2'
self.update()
def update(self):
"""
Queries the physical device and updates the internal model as
appropriate.
"""
if not self.ip_address:
return
for sub_dev in self.sub_devices:
res = requests.get(self.ip_address)
gpio0 = re.search(r"GPIO0: (\bLow|\bHigh)", res.text).groups()[0]
sub_dev.state = gpio0 == 'High'
def toggle(self, sub_dev_id):
"""
Toggles the state of a sub device.
"""
for sub_dev in self.sub_devices:
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)
self.sub_devices = []
for sub_dev_dict in data.get('sub_devices'):
if sub_dev_dict.get('type') == 'RelayOutlet':
sub_dev = RelayOutlet().from_dict(sub_dev_dict)
self.sub_devices.append(sub_dev)
else:
typ = sub_dev.get('type')
raise ValueError(f"Unknown sub_device type {typ}")
return self
class RelayOutlet:
"""
Represents a single outlet on a RelayDevice.
"""
def __init__(self):
self.id = ""
self.type = "RelayOutlet"
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 device_dict in data:
if device_dict.get('type') == 'RelayDevice':
device = RelayDevice().from_dict(device_dict)
network.append(device)
else:
raise ValueError(f"Unknown device type {device.get('type')}")
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('/')
@ -151,6 +24,7 @@ def index():
"""
The index page.
"""
global network
init_state = {}
for device in network:
device_state = {}
@ -166,6 +40,7 @@ def toggle():
"""
Toggles the state of a RelayDevice.
"""
global network
device_id = request.args.get('device_id')
sub_dev_id = request.args.get('sub_dev_id')
@ -177,7 +52,7 @@ def toggle():
res = device.toggle(sub_dev_id)
if not res:
return make_error(404, "sub_dev_id not found")
save_network()
models.save_network(network)
return res
@app_views.route('/edit')
@ -186,6 +61,7 @@ def edit():
"""
Edits the text of a particular field.
"""
global network
device_id = request.args.get('device_id')
sub_dev_id = request.args.get('sub_dev_id')
field = request.args.get('field')
@ -221,7 +97,7 @@ def edit():
'field': field,
'value': value
}
save_network()
models.save_network(network)
return json.dumps(data)
@ -232,6 +108,7 @@ def new_device():
Allows adding a new device. Accepts device_type parameter, returns
the device_id.
"""
global network
device_type = request.args.get('device_type')
if device_type == 'RelayDevice':
@ -251,7 +128,7 @@ def new_device():
num = str(int(num[0]) + 1).zfill(2)
device.id = device_type + num
network.append(device)
save_network()
models.save_network(network)
data = {'device_id': device.id}
return json.dumps(data)
@ -262,6 +139,7 @@ def lock_device():
"""
Locks or unlocks a device to prevent or allow editing it's fields.
"""
global network
device_id = request.args.get('device_id')
locked = request.args.get('locked') == 'true'
@ -275,7 +153,7 @@ def lock_device():
device.locked = True
else:
device.locked = False
save_network()
models.save_network(network)
return jsonify(device_id=device.id, locked=device.locked)
@ -286,6 +164,7 @@ def delete():
"""
Deletes a device.
"""
global network
device_id = request.args.get('device_id')
for device in network:
@ -295,7 +174,7 @@ def delete():
return make_error(404, "device_id not found")
network.remove(device)
save_network()
models.save_network(network)
return jsonify(True)
@ -312,7 +191,7 @@ else:
app.secret_key = secret_key
with open('secret_key', 'wb') as file:
file.write(secret_key)
network = init_network()
network = models.init_network()
if __name__ == '__main__':

135
models.py Normal file
View File

@ -0,0 +1,135 @@
#!/usr/bin/env python3
"""
Data models for different device types managed by the Juice IOT hub.
"""
import re
import json
import requests
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.type = "RelayDevice"
self.description = ""
self.location = ""
self.ip_address = ""
self.locked = False
self.sub_devices = []
self.sub_devices.append(RelayOutlet())
self.sub_devices.append(RelayOutlet())
self.sub_devices[0].id = 'OUT0'
self.sub_devices[0].gpio = '0'
self.sub_devices[1].id = 'OUT2'
self.sub_devices[1].gpio = '2'
self.update()
def update(self):
"""
Queries the physical device and updates the internal model as
appropriate.
"""
if not self.ip_address:
return
for sub_dev in self.sub_devices:
res = requests.get(self.ip_address)
gpio0 = re.search(r"GPIO0: (\bLow|\bHigh)", res.text).groups()[0]
sub_dev.state = gpio0 == 'High'
def toggle(self, sub_dev_id):
"""
Toggles the state of a sub device.
"""
for sub_dev in self.sub_devices:
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)
self.sub_devices = []
for sub_dev_dict in data.get('sub_devices'):
if sub_dev_dict.get('type') == 'RelayOutlet':
sub_dev = RelayOutlet().from_dict(sub_dev_dict)
self.sub_devices.append(sub_dev)
else:
typ = sub_dev.get('type')
raise ValueError(f"Unknown sub_device type {typ}")
return self
class RelayOutlet:
"""
Represents a single outlet on a RelayDevice.
"""
def __init__(self):
self.id = ""
self.type = "RelayOutlet"
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 device_dict in data:
if device_dict.get('type') == 'RelayDevice':
device = RelayDevice().from_dict(device_dict)
network.append(device)
else:
raise ValueError(f"Unknown device type {device.get('type')}")
return network
def save_network(network, filepath="devices.json"):
"""
Dumps the network to file.
"""
with open(filepath, 'w') as file:
file.write(json.dumps(network, cls=DeviceEncoder, indent='\t'))