BlockyGrapher/dnsmonitor.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2014-18 Richard Hull and contributors
# See LICENSE.rst for details.
# PYTHON_ARGCOMPLETE_OK

import math
import time
import datetime
import random
import urllib.request
import re
import sys
import random
import signal
#from demo_opts import get_device
from luma.core import cmdline, error
from luma.core.render import canvas
from pathlib import Path
from PIL import ImageFont

def make_font(name, size):
    return ImageFont.truetype(name, size)

def display_settings(device, args):
    """
    Display a short summary of the settings.

    :rtype: str
    """
    iface = ''
    display_types = cmdline.get_display_types()
    if args.display not in display_types['emulator']:
        iface = f'Interface: {args.interface}\n'

    lib_name = cmdline.get_library_for_display_type(args.display)
    if lib_name is not None:
        lib_version = cmdline.get_library_version(lib_name)
    else:
        lib_name = lib_version = 'unknown'

    import luma.core
    version = f'luma.{lib_name} {lib_version} (luma.core {luma.core.__version__})'

    return f'Version: {version}\nDisplay: {args.display}\n{iface}Dimensions: {device.width} x {device.height}\n{"-" * 60}'

def get_device(actual_args=None):
    """
    Create device from command-line arguments and return it.
    """
    if actual_args is None:
        actual_args = sys.argv[1:]
    parser = cmdline.create_parser(description='luma.examples arguments')
    args = parser.parse_args(actual_args)

    if args.config:
        # load config from file
        config = cmdline.load_config(args.config)
        args = parser.parse_args(config + actual_args)

    # create device
    try:
        device = cmdline.create_device(args)
        print(display_settings(device, args))
        return device

    except error.Error as e:
        parser.error(e)
        return None

def sigterm_handler(signum, frame):
    sys.exit(0)

def scalepoints(points):
    if device.height == 32:
        h = 31
    else:
        h = 46
    sclpoint = [0] * 15
    scl = 1
    maxpnt = 0
    i = 0
    while i <= 14:
        if points[i] > maxpnt:
            maxpnt = points[i]
        i += 1
    while h * scl < maxpnt:
        scl += 1
    i = 0
    while i <= 14:
        sclpoint[i] = int(points[i] / scl)
        i += 1
    maxcanv = h * scl
    return sclpoint, maxcanv

def movepoints(points, insertq, restarts, insertr):
    i = 1
    tmp = points[0]
    tmp2 = 0
    points[0] = insertq
    while i < len(points):
        if i == len(points) - 1:
            points[i] = tmp
        else:
            tmp2 = points[i]
            points[i] = tmp
            tmp = tmp2
        i += 1
    i = 1
    tmp3 = restarts[0]
    tmp4 = False
    restarts[0] = insertr
    while i < len(restarts):
        if i == len(restarts) - 1:
            restarts[i] = tmp3
        else:
            tmp4 = restarts[i]
            restarts[i] = tmp3
            tmp3 = tmp4
        i += 1
    return points, restarts

def parsedata(prev, flaunch):
    try:
        data = str(urllib.request.urlopen("http://10.0.0.28:4500/metrics").read()).split("\\n")
        count = 0
        i = 0
        while data[i] != "# TYPE blocky_query_total counter":
            i += 1
        i += 1
        while True:
            if data[i] == "# HELP blocky_request_duration_ms Request duration distribution":
                break
            else:
                count += int(re.sub('blocky_query_total{.*?} ', '', data[i]))
            i += 1
    except:
        if flaunch == True:
            return -1
        else:
            return 0, -1
    if flaunch == True:
        return count
    else:
        return count, count - prev

def drawout(sclpo, maxcanv, failture, restart):
    match device.height:
        case 32:
            with canvas(device) as draw:
                d1 = 1
                d2 = device.width - 66
                now = datetime.datetime.now()
                while d1 < len(sclpo):
                    draw.line((d2 + 10, 31 - sclpo[d1 - 1], d2, 31 - sclpo[d1]), fill="white")
                    d1 += 1
                    d2 -= 10
                draw.text((device.width - 53, device.height - 19), now.strftime("%H:%M"), font=regfont, fill="white")
                draw.text((0, 0), str(maxcanv), font=smafont, fill="white")
                draw.text((0, device.height / 2 - 4), str(int(maxcanv / 2)), font=smafont, fill="white")
                draw.text((0, device.height - 9), "0", font=smafont, fill="white")
                if failture:
                    draw.text((device.width - 50, 0), "\uf071", font=icons, fill="white")
                if restart:
                    draw.text((device.width - 30, 0), "\uf021", font=icons, fill="white")
        case 64:
            with canvas(device) as draw:
                d1 = 1
                d2 = device.width - 10
                now = datetime.datetime.now()
                while d1 < len(sclpo):
                    draw.line((d2 + 10, 47 - sclpo[d1 - 1], d2, 47 - sclpo[d1]), fill="white")
                    d1 += 1
                    d2 -= 10
                draw.text((device.width - 53, device.height - 19), now.strftime("%H:%M"), font=regfont, fill="white")
                draw.text((0, 0), str(maxcanv), font=smafont, fill="white")
                draw.text((0, (device.height - 19) / 2 - 2), str(int(maxcanv / 2)), font=smafont, fill="white")
                draw.text((0, device.height - 26), "0", font=smafont, fill="white")
                if failture:
                    draw.text((2, device.height - 14), "\uf071", font=icons, fill="white")
                if restart:
                    draw.text((22, device.height - 14), "\uf021", font=icons, fill="white")
        case _:
            with canvas(device) as draw:
                draw.text((0,0), "Display not supported", font=smafont, fill="white")

def getdata(oldtotal):
    total, query = parsedata(oldtotal, False)
    if query == -1:
        return oldtotal, 0, True, False
    if total < oldtotal:
        return total, 0, False, True
    return total, query, False, False

def main():
    print("Blocky Graph Monitor for OLED v0.5 B")
    print("(C) Nikopol 2024")
    today_last_time = "NaN"
    last_hrs = "NaN"
    i = 0
    failture = False
    restart = False
    rstate = False
    device.contrast(0)
    pointsmap = [0] * 15
    restartmap = [False] * len(pointsmap)
    total = parsedata(0, True)
    if total == -1:
        failture = True
    while True:
        now = datetime.datetime.now()
        today_time = now.strftime("%H:%M")
        hrs = now.strftime("%H")
        #if today_time != today_last_time:
        if 1:
            today_last_time = today_time
            #if hrs != last_hrs:
            if 1:
                last_hrs = hrs
                total, query, failture, rstate = getdata(total)
                #rstate = True
                print(total)
                print(query)  
                pointsmap, restartmap = movepoints(pointsmap, query, restartmap, rstate)
                sclpo, maxcanv = scalepoints(pointsmap)
                restart = False
                while i < len(restartmap):
                    print(restartmap[i])
                    if restartmap[i] == True:
                        restart = True         
                    i += 1
                i = 0
            drawout(sclpo, maxcanv, failture, restart)
        time.sleep(1)


if __name__ == "__main__":
    try:
        print("\n".join(sys.argv))
        device = get_device()
        regfont = make_font("TerminusTTF-4.49.3.ttf", 22)
        smafont = make_font("TerminusTTF-4.49.3.ttf", 12)
        icons = make_font("fontawesome-webfont.ttf", 15)
        signal.signal(signal.SIGTERM, sigterm_handler)
        main()
    except KeyboardInterrupt:
        pass