Juice/models.py

203 lines
4.5 KiB
Python

#!/usr/bin/env python3
"""
Data models for different device types managed by the Juice IOT hub.
"""
import re
import json
import requests
class Network(list):
"""
Represents a network of IOT devices.
"""
def find(self, device_id):
"""
Searches through the network and returns the matching device,
or None.
"""
for device in self:
if device.id == device_id:
break
else:
return None
return device
class Device:
"""
Represents a generic device on the Network.
"""
def find(self, sub_device_id):
"""
Searches through the devices sub_devices and returns the matching
sub_device, or None.
"""
for sub_device in self.sub_devices:
if sub_device.id == sub_device_id:
break
else:
return None
return sub_device
class RelayDevice(Device):
"""
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_device_id):
"""
Toggles the state of a sub device.
"""
sub_device = self.find(sub_device_id)
if not sub_device:
return None
params = {sub_device.gpio: 'toggle'}
res = requests.get('http://' + self.ip_address + '/gpio',params=params)
res.raise_for_status()
state = re.search(
rf"GPIO{sub_device.gpio}: (\bLow|\bHigh)",
res.text
).groups()[0] == 'High'
sub_device.state = state
return 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 LightStrip(Device):
"""
Represents an RGB LED light strip controller.
"""
def __init__(self):
self.id = ""
self.type = "LightStrip"
self.description = ""
self.location = ""
self.ip_address = ""
self.locked = False
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 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 = Network()
for device_dict in data:
if device_dict.get('type') == 'RelayDevice':
device = RelayDevice().from_dict(device_dict)
network.append(device)
elif device_dict.get('type') == 'LightStrip':
device = LightStrip().from_dict(device_dict)
network.append(device)
else:
raise ValueError(f"Unknown device type {device_dict.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'))
network = init_network()