homecontrol-dash/homecontrol-dash.py

#!/usr/bin/env python3
import dash
import dash_bootstrap_components as dbc
import dash_core_components as dcc
import dash_html_components as html
from dash.dependencies import Input, Output
import time
import requests
from math import log, log10

URL_BASE="http://innocence:5000"

def get_sensors():
    ret = {}
    try:
        url = f"{URL_BASE}/sensor/get"
        res = requests.get(url)
        ret = res.json()
    except Exception as ex:
        print(f"Exception Type: {type(ex).__name__}, args:\n{ex.args}")

    return ret


def get_values(sensorId, min_ts, max_ts, limit):
    ret = {}
    if sensorId:
        try:
            url = f"{URL_BASE}/sensor/get_values/{sensorId}?min_ts={min_ts}&max_ts={max_ts}&limit={limit}"
            # print(url)
            res = requests.get(url)
            ret = res.json()[sensorId]
        except Exception as ex:
            print(f"Exception Type: {type(ex).__name__}, args:\n{ex.args}")
    return ret

def pTime(value, fmt="%Y/%m/%d %H:%M"):
    if value > 0:
        return time.strftime(fmt, time.localtime(float(value)))
    else:
        return 0


sensors = get_sensors()
sensor = next(iter(sensors), None)
tabs = []
for s in sensors:
    sensorType = sensors[s]["sensorType"]
    tabs.append(dcc.Tab(label=sensorType, value=s))

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.layout = html.Div(children=[
    html.H1(children='dashboard.ykonni.de'),
    dbc.Row([
        dbc.Col(dcc.Tabs(id="tabs-select-sensor", value=sensor, children=tabs))
    ]),
    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'}
            ),
        ]),
    ]),
])

@app.callback(
        [Output('slider-min', 'min'), Output('slider-min', 'value'),
            Output('slider-min', 'max')],
        [Input('tabs-select-sensor', 'value')])
def update_slider_min(sensorId):
    res = get_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())
    sensorType = None
    if "values" in res:
        min_ts = int(res["values"][0]["ts"])
        sensorType = res["sensorType"]

    # print(f"min: [{min_ts} [{cur_ts}] {max_ts}]"
    return min_ts, cur_ts, max_ts


@app.callback(
        Output('slider-min-txt', 'children'),
        [Input('slider-min', 'value')])
def update_slider_min_txt(value):
    return f"From: {pTime(value)}"

@app.callback(
        [Output('slider-max', 'min'), Output('slider-max', 'value'),
            Output('slider-max', 'max')],
        [Input('tabs-select-sensor', 'value')])
def update_slider_max(sensorId):
    res = get_values(sensorId, 0, int(time.time()), 1)
    min_ts = 0
    max_ts = int(time.time())
    sensorType = None
    if "values" in res:
        min_ts = int(res["values"][0]["ts"])
        sensorType = res["sensorType"]

    # print(f"max: [{min_ts} [{max_ts}] {max_ts}]"
    return min_ts, max_ts, max_ts


@app.callback(
        Output('slider-max-txt', 'children'),
        [Input('slider-max', 'value')])
def update_slider_max_txt(value):
    return f"To: {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 = get_values(sensorId, min_ts, max_ts, limit)
    if "values" in res:
        v = res["values"]
        s = res["sensorType"]
        x = [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)' }
    }

if __name__ == '__main__':

    app.run_server(port=8081,debug=True)