Source code for

import sys
import traceback

import json
import requests
import humanize
import schedule
import dateutil.parser

from datetime import datetime, timezone

from import BaseCommand
from django.conf import settings

from influxdb import InfluxDBClient

from ...models import Node, Timer, Device
from ...api.serializers import NodeSerializer
from ...device_templates.mqtt_device import init_mqtt


[docs]def broadcast_gotify_message(title, message, node, payload): return if "title" not in payload: payload['title'] = title if "message" not in payload: payload['message'] = message if "node" not in payload: payload['node'] = NodeSerializer(node).to_representation(node) data = { "priority": 0, "title": title, "message": message, "extras": { "client::display": { "contentType": "text/markdown" }, "android::action": { "onReceive": { "intentUrl": "festivalgrid://" + json.dumps(payload) } } }, } return"%s:%d/message?token=%s" % (settings.GOTIFY_HOST, int(settings.GOTIFY_PORT), settings.GOTIFY_TOKEN), json=data)
[docs]class Command(BaseCommand): help = 'FestivalGrid worker' node_last_message = {} influx_client = None nodes_template = {}
[docs] def broadcast_message(self, id, title, message, node=None, payload={}): if node: if node in self.node_last_message and self.node_last_message[node] == id: return self.node_last_message[node] = id broadcast_gotify_message(title, message, node, payload)
[docs] def monitor_nodes(self): nodes = Node.objects.filter(device__isnull=False).order_by('name') now = self.stdout.write("\n---------------------- [%s] ----------------------\n\n" % now.strftime("%H:%M:%S")) self.stdout.write(" name power timestamp stale status\n")) for node in nodes: sql = 'SELECT last("Power") FROM "%d"' % points = self.influx_client.query(sql).get_points() for point in points: power = int(point["last"]) timestamp = dateutil.parser.parse(point["time"]) timedelta = now - timestamp log = ('OK', if node.cut_power: log = ("POWER OFF", else: if power > node.shutdown_level: log = ("EMERGENCY SHUTDOWN", node.cut_power = True self.broadcast_message( log[0], "Emergency Shutdown: %s" % node, "%s has been shutdown because power consumption exceeded %d W (was %d W)." % (node, node.shutdown_level, power), node, {"color": "#ffaaaa"} ) elif power > node.warning_level: log = ("POWER LEVEL WARNING", self.broadcast_message( log[0], "Warning: %s @ %d W" % (node, power), "Warning level: %d W | Shutdown level: %d W" % (node.warning_level, node.shutdown_level), node, {"color": "#ffcc6c"} ) if timedelta.total_seconds() > 60: log = ("STALE", self.broadcast_message( log[0], "Stale: %s" % node, "%s has not sent and updates for %s." % (, humanize.naturaldelta(timedelta)), node, {"color": "#ffcc6c"} ) self.stdout.write(log[1](" %s %5s W %s %5s s %s" % (, power, timestamp.strftime("%H:%M:%S"), int(timedelta.total_seconds()), log[0] )))
[docs] def poll_devices(self): for node in Node.objects.filter(device__template__contains="demo"): # ".polling" try: data = node.device.get_template().handle_polling(node) if data: self.stdout.write(" data: %s" % data) self.influx_client.write_points([data]) except Device.DoesNotExist: pass
[docs] def timer_action(self, timer): self.stdout.write('running timer action!') if timer.action == "TOGGLE": timer.node.cut_power = not timer.node.cut_power else: timer.node.cut_power = timer.action == 'OFF'
[docs] def schedule_timers(self): schedule.clear('timers') self.stdout.write("(re)scheduling...") for timer in Timer.objects.filter(enabled=True): self.stdout.write(" %s %s" % (, timer.time)) schedule.every(), timer).tag('timers')
[docs] def cache_node_templates(self): self.nodes_template = {} for node in Node.objects.filter(device__mqtt_topic__isnull=False): try: self.nodes_template[node.device.mqtt_topic] = node.device.get_template() except Device.DoesNotExist: pass
[docs] def on_message(self, mosq, userdata, msg): self.stdout.write('\ntopic: %s' % msg.topic)) self.stdout.write(' payload: %s' % msg.payload) if "setup_changed" in msg.topic: self.stdout.write("\n Setup changed. Re-caching node templates...") self.cache_node_templates() self.schedule_timers() return influx_name = msg.topic.split('/')[1] if influx_name in self.nodes_template: try: data = self.nodes_template[influx_name].digest_message(msg.topic, msg.payload) if data: self.stdout.write(" data: %s" % data) self.influx_client.write_points([data]) except: traceback.print_exc()
[docs] def on_connect(self, client, userdata, flags, result_code): self.stdout.write("connected to %s:%d" % (settings.MQTT_HOST, settings.MQTT_PORT)) client.subscribe('festivalgrid/+/setup_changed', 2) for mqtt_topic in self.nodes_template.keys(): topic = 'festivalgrid/%s/#' % mqtt_topic self.stdout.write(' subscribing to: %s' % topic) client.subscribe(topic, 2)
[docs] def on_disconnect(self, mosq, userdata, rc): sys.exit()
[docs] def handle(self, *args, **options): self.cache_node_templates() self.influx_client = InfluxDBClient(settings.INFLUXDB_HOST, settings.INFLUXDB_PORT, settings.INFLUXDB_USERNAME, settings.INFLUXDB_PASSWORD, timeout=1, retries=0) self.stdout.write("connected to influxdb.") self.influx_client.create_database(settings.INFLUXDB_DATABASE_NAME) self.influx_client.switch_database(settings.INFLUXDB_DATABASE_NAME) schedule.every(10) schedule.every(10) schedule.run_all() self.schedule_timers() mqtt_client = init_mqtt('festivalgrid-worker', self.on_message, self.on_connect, self.on_disconnect) while True: mqtt_client.loop(timeout=1) schedule.run_pending()