#!/usr/bin/env python3 import argparse import dash import dash_bootstrap_components as dbc import dash_core_components as dcc import dash_daq as daq import dash_html_components as html from dash.dependencies import Input, Output import json import logging import os import time import requests from math import log, log10 from xdg import XDG_CONFIG_HOME CONFIG_FILE = f"{XDG_CONFIG_HOME}/hcdash.json" ADDRESS="http://localhost:8200" PORT=8201 def setup(): # arguments parser = argparse.ArgumentParser(description='homecontrol') parser.add_argument("-a", "--address", dest="address", type=str, help="homecontrol address") parser.add_argument("-p", "--port", dest="port", type=str, help="dashboard port") parser.add_argument("-c", "--config", dest="config", type=str, help="config file", default=CONFIG_FILE) parser.add_argument("-d", "--debug", dest="debug", action="store_true", help="debug mode") # parse arguments args = parser.parse_args() # initialize config if not os.path.exists(XDG_CONFIG_HOME): os.makedirs(XDG_CONFIG_HOME) config = {} try: config_file = open(args.config, "r") config = json.load(config_file) except: pass # fill new keys with defaults if not config.get("address"): config["address"] = ADDRESS if not config.get("port"): config["port"] = PORT # overwrite with arguments if args.address: config["address"] = args.address if args.port: config["port"] = args.port # save to file with open(args.config, 'w') as config_file: json.dump(config, config_file) # temporary option if args.debug: config["debug"] = args.debug else: config["debug"] = False return config class HCDash: def __init__(self, config): self.config = config # logging FORMAT="[%(asctime)13s :: %(name)18s :: %(levelname)7s] %(message)s" DATEFMT="%Y%m%d %H:%M:%S" logging.basicConfig(level=logging.INFO, format=FORMAT, datefmt=DATEFMT) self.logger = logging.getLogger("HCDash") if self.config["debug"]: self.logger.setLevel(logging.DEBUG) def get_actors(self): ret = {} try: url = f"{self.config.get('address')}/actor/get" res = requests.get(url) tmp = res.json() # 10 minutes for a in tmp: if tmp[a].get("ts") > (time.time() - 60*10): ret[a] = tmp[a] self.logger.debug(f"get_actors: {ret}") except Exception as ex: self.logger.error(f"Exception get_actors, type: {type(ex).__name__}, args:\n{ex.args}") finally: return ret def get_sensors(self): ret = {} try: url = f"{self.config.get('address')}/sensor/get" res = requests.get(url) ret = res.json() self.logger.debug(f"get_sensors: {ret}") except Exception as ex: self.logger.error(f"Exception Type: {type(ex).__name__}, args:\n{ex.args}") finally: return ret def set_level(self, actorId, level): req = {"id": actorId, "command": "set_level", "data": {"level": level}} self.logger.debug(f"set_level: {req}") requests.post(f"{self.config.get('address')}/actor/command", json=req) def get_sensor_values(self, sensorId, min_ts, max_ts, limit): ret = {} if sensorId: try: url = f"{self.config.get('address')}/sensor/get_values/{sensorId}?min_ts={min_ts}&max_ts={max_ts}&limit={limit}" res = requests.get(url) ret = res.json()[sensorId] except Exception as ex: self.logger.error(f"Exception Type: {type(ex).__name__}, args:\n{ex.args}") return ret def get_actor_levels(self, actorId, min_ts, max_ts, limit): self.logger.debug(f"get_actor_levels: {actorId}") ret = {} if actorId: try: url = f"{self.config.get('address')}/actor/get_levels/{actorId}?min_ts={min_ts}&max_ts={max_ts}&limit={limit}" res = requests.get(url) ret = res.json()[actorId] except Exception as ex: self.logger.error(f"Exception Type: {type(ex).__name__}, args:\n{ex.args}") return ret def pTime(self, value, fmt="%Y/%m/%d %H:%M"): if value > 0: return time.strftime(fmt, time.localtime(float(value))) else: return 0 def main(self): tab = "sensors" tabs = [ dcc.Tab(label="sensors", value="sensors"), dcc.Tab(label="actors", value="actors") ] external_stylesheets = [dbc.themes.BOOTSTRAP] meta_tags=[{"name": "viewport", "content": "width=device-width, initial-scale=1"}] app = dash.Dash(__name__, external_stylesheets=external_stylesheets, meta_tags=meta_tags) app.title = "dashboard.ykonni.de" app.config.suppress_callback_exceptions = True app.layout = html.Div(children=[ html.H1(children='dashboard.ykonni.de'), dbc.Row([ dbc.Col(dcc.Tabs(id="tabs-select-class", value=tab, children=tabs)) ]), html.Div(id="tabs-content") ]) # tab callbacks @app.callback( Output("tabs-content", "children"), [Input("tabs-select-class", "value")]) def update_tabs(value): self.logger.debug(f"update_tabs: {value}") if value == "sensors": sensors = self.get_sensors() sensor = next(iter(sensors), None) sensor_tabs = [] for s in sensors: sensorType = sensors[s]["sensorType"] sensor_tabs.append(dcc.Tab(label=sensorType, value=s)) return html.Div([ dbc.Row([ dbc.Col(dcc.Tabs(id="tabs-select-sensor", value=sensor, children=sensor_tabs)) ]), html.Div(id="sensor-data"), ]) elif value == "actors": actors = self.get_actors() actor = next(iter(actors), None) actor_tabs = [] for a in actors: actorType = actors[a]["actorType"] actor_tabs.append(dcc.Tab(label=actorType, value=a)) return html.Div([ dbc.Row([ dbc.Col(dcc.Tabs(id="tabs-select-actor", value=actor, children=actor_tabs)) ]), html.Div(id="actor-data"), ]) else: return html.Div([ html.H3("undefined") ]), None, None @app.callback( [Output("sensor-data", "children"), Output('slider-min', 'min'), Output('slider-min', 'value'), Output('slider-min', 'max'), Output('slider-max', 'min'), Output('slider-max', 'value'), Output('slider-max', 'max')], [Input("tabs-select-sensor", "value")]) def update_sensor(sensorId): self.logger.debug(f"update_sensor: {sensorId}") if sensorId: sensor_data = [ dbc.Row([ dbc.Col( dcc.Graph( id='graph-sensor-values', figure={ 'data': [ {'x': [0], 'y': [0], 'mode': 'line', 'name': 'None'} ], 'layout': { 'title': 'initial values' } } ) ) ]), dbc.Row([ dbc.Col([ html.Div(id='slider-min-txt', style={'marginLeft': '5em', 'marginRight': '3em'}), html.Div([ dcc.Slider(id='slider-min', min=0, max=round(time.time()), step=600, value=0 ), ], style={'marginLeft': '5em', 'marginRight': '3em'} ), ]), dbc.Col([ html.Div(id='slider-max-txt', style={'marginLeft': '3em', 'marginRight': '3em'}), html.Div([ dcc.Slider(id='slider-max', min=0, max=round(time.time()), step=600, value=round(time.time())), ], style={'marginLeft': '3em', 'marginRight': '3em'} ), ]), dbc.Col([ html.Div(id='slider-limit-txt', style={'marginLeft': '3em', 'marginRight': '5em'}), html.Div([ dcc.Slider(id='slider-limit', min=0, max=1000, step=10, value=0 ), ], style={'marginLeft': '3em', 'marginRight': '5em'} ), ]) ]), ] res = self.get_sensor_values(sensorId, 0, int(time.time()), 1) min_ts = 0 # default last 7 days cur_ts = round(time.time() - (60*60*24*7)) max_ts = int(time.time()) if "values" in res: min_ts = int(res["values"][0]["ts"]) # sensor_data, min: [min, cur, max], max: [min, max, max] return sensor_data, min_ts, cur_ts, max_ts, min_ts, max_ts, max_ts @app.callback( Output('slider-min-txt', 'children'), [Input('slider-min', 'value')]) def update_slider_min_txt(value): return f"From: {self.pTime(value)}" @app.callback( Output('slider-max-txt', 'children'), [Input('slider-max', 'value')]) def update_slider_max_txt(value): return f"To: {self.pTime(value)}" @app.callback( Output('slider-limit-txt', 'children'), [Input('slider-limit', 'value')]) def update_slider_limit_txt(value): return f"Limit: {value if value > 0 else None}" @app.callback( Output('graph-sensor-values', 'figure'), [Input('tabs-select-sensor', 'value'), Input('slider-min', 'value'), Input('slider-max', 'value'), Input('slider-limit', 'value')]) def update_graph_sensor_values(sensorId, min_ts, max_ts, limit): res = {} if min_ts > 0: res = self.get_sensor_values(sensorId, min_ts, max_ts, limit) if "values" in res: v = res["values"] s = res["sensorType"] x = [self.pTime(v[i]["ts"], fmt="%m%d-%H%M") for i in range(len(v))] if s == "luminance": y = [log10(v[i]["value"]/5+1) for i in range(len(v))] else: y = [v[i]["value"] for i in range(len(v))] else: s = sensorId x = [0] y = [0] return { 'data': [ {'x': x, 'y': y, 'mode': 'line', 'name': f'{s}'} ], 'layout': { 'title': f'Data for sensor: {s} ({len(x)} elements)' } } ## actor callbacks @app.callback( [Output("actor-data", "children"), Output("slider-actor-min", "min"), Output("slider-actor-min", "value"), Output("slider-actor-min", "max"), Output("slider-actor-max", "min"), Output("slider-actor-max", "value"), Output("slider-actor-max", "max")], [Input("tabs-select-actor", "value")]) def update_actor(actorId): self.logger.debug(f"update_actor: {actorId}") if actorId: res = self.get_actors() l = res[actorId].get("level") or 0 level = f"#{format(l, '06x')}" actor_data = [ dbc.Row([ dbc.Col( dcc.Graph( id='graph-actor-values', figure={ 'data': [ {'x': [0], 'y': [0], 'mode': 'line', 'name': 'None'} ], 'layout': { 'title': 'initial values' } } ) ) ]), dbc.Row([ dbc.Col([ html.Div(id='slider-actor-min-txt', style={'marginLeft': '5em', 'marginRight': '3em'}), html.Div([ dcc.Slider(id='slider-actor-min', min=0, max=round(time.time()), step=600, value=0 ), ], style={'marginLeft': '5em', 'marginRight': '3em'} ), ]), dbc.Col([ html.Div(id='slider-actor-max-txt', style={'marginLeft': '3em', 'marginRight': '3em'}), html.Div([ dcc.Slider(id='slider-actor-max', min=0, max=round(time.time()), step=600, value=round(time.time())), ], style={'marginLeft': '3em', 'marginRight': '3em'} ), ]), dbc.Col([ html.Div(id='slider-actor-limit-txt', style={'marginLeft': '3em', 'marginRight': '5em'}), html.Div([ dcc.Slider(id='slider-actor-limit', min=0, max=1000, step=10, value=0 ), ], style={'marginLeft': '3em', 'marginRight': '5em'} ), ]) ]), dbc.Row([ daq.ColorPicker( id="color-picker", label="Color Picker", size=360, value=dict(hex=level) ), html.P(id="empty") ]), ] res = self.get_actor_levels(actorId, 0, int(time.time()), 1) min_ts = 0 # default last 7 days cur_ts = round(time.time() - (60*60*24*7)) max_ts = int(time.time()) if "levels" in res: min_ts = int(res["levels"][0]["ts"]) # actor_data, min: [min, cur, max], max: [min, max, max] return actor_data, min_ts, cur_ts, max_ts, min_ts, max_ts, max_ts @app.callback( Output('slider-actor-min-txt', 'children'), [Input('slider-actor-min', 'value')]) def update_slider_actor_min_txt(value): # self.logger.debug(f"update_slider_actor_min_txt: {value}") return f"From: {self.pTime(value)}" @app.callback( Output('slider-actor-max-txt', 'children'), [Input('slider-actor-max', 'value')]) def update_slider_actor_max_txt(value): # self.logger.debug(f"update_slider_actor_max_txt: {value}") return f"To: {self.pTime(value)}" @app.callback( Output('slider-actor-limit-txt', 'children'), [Input('slider-actor-limit', 'value')]) def update_slider_actor_limit_txt(value): return f"Limit: {value if value > 0 else None}" @app.callback( Output("empty", "value"), [Input("tabs-select-actor", "value"), Input("color-picker", "value")]) def set_level(actorId, level): l = f"0x{level.get('hex').replace('#', '')}" if l: self.set_level(actorId, l) return "" @app.callback( Output('graph-actor-values', 'figure'), [Input('tabs-select-actor', 'value'), Input('slider-actor-min', 'value'), Input('slider-actor-max', 'value'), Input('slider-actor-limit', 'value')]) def update_graph_actor(actorId, min_ts, max_ts, limit): self.logger.debug(f"update_graph_actor: {actorId}, {min_ts}, {max_ts}, {limit}") res = {} if min_ts > 0: res = self.get_actor_levels(actorId, min_ts, max_ts, limit) if "levels" in res: v = res["levels"] s = res["actorType"] x = [self.pTime(v[i]["ts"], fmt="%m%d-%H%M") for i in range(len(v))] y = [v[i]["value"] for i in range(len(v))] # y = [[(int(v[i]["value"]) >> 16) & 0xff, (int(v[i]["value"]) >> 8) & 0xff, # int(v[i]["value"]) & 0xff] for i in range(len(v))] else: s = actorId x = [0] y = [0] return { 'data': [ {'x': x, 'y': y, 'mode': 'markers', 'name': f'{s}'} ], 'layout': { 'title': f'Data for sensor: {s} ({len(x)} elements)' } } app.run_server(host="0.0.0.0", port=self.config.get("port"), debug=self.config.get("debug")) def main(): config = setup() hcd = HCDash(config) hcd.main() if __name__ == '__main__': main()