rest_system_monitor.py

ha_sensors.py

import json
from threading import Lock

import requests
import time
from zeroconf import ServiceBrowser, ServiceStateChange
from zeroconf import Zeroconf

IDENTIFIER = "restsysadmin"
HA_URL = "http://192.168.1.8:8123"
HA_TOKEN = "...insert here....."


class HASensorPusher:
    def __init__(self, identifier=IDENTIFIER):
        self.zero = None
        self.browser = None
        self.nodes = {}
        self.running = False
        self.identifier = identifier.encode("utf-8")
        self.ttl = 10  # seconds between rounds of sensor readings
        self.sleep_time = 0.3  # seconds between individual sensor readings
        self.lock = Lock()

    @staticmethod
    def ha_binary_sensor_update(device_id, state="on", attrs=None):
        attrs = attrs or {"friendly_name": device_id,
                          "state_color": True,
                          "device_class": "presence"}
        try:
            response = requests.post(
                f"{HA_URL}/api/states/binary_sensor.{IDENTIFIER}_{device_id}",
                headers={
                    "Authorization": f"Bearer {HA_TOKEN}",
                    "content-type": "application/json",
                },
                data=json.dumps({"state": state, "attributes": attrs}),
            )
            print(response.text)
        except:
            print("error updating sensor")

    @staticmethod
    def ha_sensor_update(device_id, state="on", attrs=None):
        attrs = attrs or {"friendly_name": device_id,
                          "state_color": True,
                          "device_class": "presence"}
        try:
            response = requests.post(
                f"{HA_URL}/api/states/sensor.{IDENTIFIER}_{device_id}",
                headers={
                    "Authorization": f"Bearer {HA_TOKEN}",
                    "content-type": "application/json",
                },
                data=json.dumps({"state": state, "attributes": attrs}),
            )
            print(response.text)
        except:
            print("error updating sensor")

    def on_new_node(self, node):
        device_id = node["name"]
        print("found", node)
        self.on_node_update(node)
        self.get_readings()
        self.get_info(device_id)  # one time sensor readings

    def on_lost_node(self, node):
        print("lost", node)
        self.on_node_update(node)

    def on_node_update(self, node):
        device_id = node["name"]
        self.nodes[device_id] = node
        self.ha_binary_sensor_update(device_id, state=node["state"],
                                     attrs={"friendly_name": node["name"],
                                            "state_color": True,
                                            "device_class": "presence"})

    def on_service_state_change(self, zeroconf, service_type, name,
                                state_change):
        info = zeroconf.get_service_info(service_type, name)
        if info and info.properties:
            for key, value in info.properties.items():
                if key == b"type" and value == self.identifier:
                    host = info._properties[b"host"].decode("utf-8")
                    port = info._properties[b"port"].decode("utf-8")
                    name = info._properties[b"name"].decode("utf-8")
                    node = {"host": host, "port": port, "name": name}
                    node["last_seen"] = time.time()
                    if state_change is ServiceStateChange.Added:
                        node["state"] = "on"
                        self.on_new_node(node)
                    elif ServiceStateChange.Removed:
                        node["state"] = "off"
                        self.on_lost_node(node)
                    else:
                        node["state"] = "on"
                        self.on_node_update(node)

    def start(self):
        self.zero = Zeroconf()
        self.browser = ServiceBrowser(self.zero, "_http._tcp.local.",
                                      handlers=[self.on_service_state_change])
        self.running = True
        while self.running:
            try:
                self.get_readings()
                time.sleep(self.ttl)
            except KeyboardInterrupt:
                break
        self.stop()

    def stop(self):
        if self.zero:
            self.zero.close()
        self.zero = None
        self.browser = None
        self.running = False

    def get_info(self, device_id):
        data = self.nodes[device_id]
        url = f"{data['host']}:{data['port']}"
        readings = {
            "os": f"{url}/os_name",
            "boot_time": f"{url}/boot_time",
            "system": f"{url}/system",
            "release": f"{url}/release",
            "machine": f"{url}/machine",
            "architecture": f"{url}/architecture",
            "cpu_count": f"{url}/cpu_count",
            # "fans": f"{url}/fans",
            # "network_interfaces": f"{url}/network_interfaces",
            "external_ip": f"{url}/external_ip",
            # "procs": f"{url}/procs",
            "is_systemd": f"{url}/is_systemd",
            "is_dbus": f"{url}/is_dbus"
        }
        for sensor_name, url in readings.items():
            self.read_sensor(device_id, sensor_name, url)
            time.sleep(self.sleep_time)

    def read_sensor(self, device_id, sensor_name, url):
        sensor_id = f"{device_id}_{sensor_name}"
        try:
            res = requests.get("http://" + url).text
            try:
                if "/is_" in url:
                    res = str(res) == "1"
                    try:
                        self.ha_binary_sensor_update(sensor_id, state="on" if res else "off",
                                                     attrs={"friendly_name": sensor_name,
                                                            "state_color": True,
                                                            "device_class": "presence"})
                    except Exception as e:
                        print(e)
                elif url.endswith("_count"):
                    res = int(res)
                    a = {"friendly_name": sensor_name,
                         "state_color": True}
                    self.ha_sensor_update(sensor_id, state=str(res), attrs=a)
                else:
                    res = float(res)  # numeric reading
                    a = {"friendly_name": sensor_name,
                         "state_color": True}
                    if "temperature" in url:
                        a["unit_of_measurement"] = "°C"
                    self.ha_sensor_update(sensor_id, state=str(res), attrs=a)

            except Exception as e:
                # text
                a = {"friendly_name": sensor_name,
                     "state_color": True}
                self.ha_sensor_update(sensor_id, state=str(res), attrs=a)
        except:
            pass

    def get_readings(self):
        with self.lock:
            for device_id, data in dict(self.nodes).items():
                if data["state"] == "on":
                    url = f"{data['host']}:{data['port']}"
                    readings = {
                        "cpu_usage": f"{url}/cpu_usage",
                        "cpu_freq": f"{url}/cpu_freq",
                        "cpu_freq_max": f"{url}/cpu_freq_max",
                        "cpu_freq_min": f"{url}/cpu_freq_min",
                        "cpu_temperature": f"{url}/cpu_temperature",
                        "memory_usage": f"{url}/memory_usage",
                        "memory_total": f"{url}/memory_total",
                        "swap_usage": f"{url}/swap_usage",
                        "swap_total": f"{url}/swap_total",
                        "disk_usage": f"{url}/disk_usage",
                        "disk_percent": f"{url}/disk_percent",
                        "disk_total": f"{url}/disk_total",
                        "battery_percent": f"{url}/battery_percent",
                        "external_ip": f"{url}/external_ip",
                        "is_kdeconnect": f"{url}/is_kdeconnect",
                        "is_pulseaudio": f"{url}/is_pulseaudio",
                        "is_pipewire": f"{url}/is_pipewire",
                        "is_plasma": f"{url}/is_plasma",
                        "is_hivemind": f"{url}/is_hivemind",
                        "is_ovos": f"{url}/is_ovos",
                        "is_spotify": f"{url}/is_spotify",
                        "is_firefox": f"{url}/is_firefox",
                        # "pulseaudio_info": f"{url}/pulseaudio/info",
                        "pulseaudio_version": f"{url}/pulseaudio/version",
                        "pulseaudio_channel_count": f"{url}/pulseaudio/channel_count",
                        "pulseaudio_default_sink": f"{url}/pulseaudio/default_sink",
                        "pulseaudio_default_source": f"{url}/pulseaudio/default_source",
                        "pulseaudio_hostname": f"{url}/pulseaudio/hostname",
                        # "pulseaudio_list_cards": f"{url}/pulseaudio/list_cards",
                        # "pulseaudio_list_sources": f"{url}/pulseaudio/list_sources",
                        # "pulseaudio_list_sinks": f"{url}/pulseaudio/list_sinks",
                        # "pulseaudio_list_input_sinks": f"{url}/pulseaudio/list_input_sinks",
                        "pulseaudio_is_playing": f"{url}/pulseaudio/is_playing",
                        "brightness": f"{url}/brightness"
                    }
                    for sensor_name, url in readings.items():
                        self.read_sensor(device_id, sensor_name, url)
                        time.sleep(self.sleep_time)


if __name__ == "__main__":
    ha = HASensorPusher()
    ha.start()
    while True:
        try:
            time.sleep(1)
        except KeyboardInterrupt:
            break
    ha.stop()

requirements.txt

psutil
pulsectl
flaskpulsectl
zeroconf

rest_system_monitor.py

import getpass
import os
import platform
import shutil
import signal
import subprocess
import urllib.request

import psutil
from flask import Flask

DEVELOPMENT_ENV = True
PORT = 5000
NAME = "AsusTUF"
IDENTIFIER = "restsysadmin"
ICON = "mdi:"

try:
    import pulsectl

    pulse = pulsectl.Pulse('restsysmon')
except:
    pulse = None

try:
    import screen_brightness_control as sbc
except:
    sbc = None

try:
    # if zero conf is installed, enable auto discovery

    from zeroconf import NonUniqueNameException
    from zeroconf import Zeroconf, ServiceInfo
    import ipaddress
    import socket


    def _get_ip():
        # taken from https://stackoverflow.com/a/28950776/13703283
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            # doesn't even have to be reachable
            s.connect(('10.255.255.255', 1))
            IP = s.getsockname()[0]
        except Exception:
            IP = '127.0.0.1'
        finally:
            s.close()
        return IP


    class ZeroConfAnnounce:
        IP = _get_ip()
        zero = Zeroconf()
        info = ServiceInfo(
            "_http._tcp.local.",
            f" - {NAME}._http._tcp.local.",
            addresses=[ipaddress.ip_address(IP).packed],
            port=PORT,
            properties={"type": "restsysadmin",
                        "name": NAME,
                        "host": IP,
                        "port": PORT},
        )
        try:
            zero.register_service(info)
            print(f"Announcing node via Zeroconf")
        except NonUniqueNameException:  # TODO - why is this being called twice ???
            pass

        @classmethod
        def shutdown(cls):
            if cls.zero:
                cls.zero.unregister_service(cls.info)
                cls.zero.close()
            cls.zero = None
except:
    # no auto discovery
    ZeroConfAnnounce = None

app = Flask(__name__)


@app.route("/")
def index():
    return NAME


# notify platform
@app.route("/notify/<message>")
def notify(message):
    subprocess.call(['notify-send', message])
    return message


# system info
@app.route("/os_name")
def os_name():
    return os.name


@app.route("/boot_time")
def boot_time():
    return str(psutil.boot_time())


@app.route("/system")
def system():
    return platform.system()


@app.route("/release")
def release():
    return platform.release()


@app.route("/machine")
def machine():
    return platform.machine()


@app.route("/architecture")
def architecture():
    return str(platform.architecture()[0])


# sensors
@app.route("/cpu_count")
def cpu_count():
    return str(os.cpu_count())


@app.route("/cpu_usage")
def cpu_usage():
    return str(psutil.cpu_percent(1))


@app.route("/cpu_freq")
def cpu_freq():
    return str(psutil.cpu_freq().current)


@app.route("/cpu_freq_max")
def cpu_freq_max():
    return str(psutil.cpu_freq().max)


@app.route("/cpu_freq_min")
def cpu_freq_min():
    return str(psutil.cpu_freq().min)


@app.route("/cpu_temperature")
def cpu_temperature():
    return str(psutil.sensors_temperatures()['coretemp'][0].current)


@app.route("/memory_usage")
def memory_usage():
    return str(psutil.virtual_memory()[2])


@app.route("/memory_total")
def memory_total():
    return str(psutil.virtual_memory()[0])


@app.route("/swap_usage")
def swap_usage():
    return str(psutil.swap_memory()[3])


@app.route("/swap_total")
def swap_total():
    return str(psutil.swap_memory()[0])


@app.route("/disk_usage")
def disk_usage():
    total, used, free = shutil.disk_usage("/")
    return str(used)


@app.route("/disk_percent")
def disk_percent():
    total, used, free = shutil.disk_usage("/")
    return str(used * 100 / total)


@app.route("/disk_total")
def disk_total():
    total, used, free = shutil.disk_usage("/")
    return str(total)


@app.route("/battery_percent")
def battery_percent():
    battery = psutil.sensors_battery()
    return str(battery.percent)


@app.route("/fans")
def fans():
    return {label: [f.label for f in fans]
            for label, fans in psutil.sensors_fans().items()}


@app.route("/fan/<label>/<fan_name>")
def get_fan(label, fan_name):
    fans = psutil.sensors_fans().items()
    if label in fans:
        for f in fans[label]:
            if f.label == fan_name:
                return str(f.current)
    return "0"


# network
@app.route("/network_interfaces")
def network():
    return list(psutil.net_if_addrs().keys())


@app.route("/network/<interface>")
def network_ip(interface):
    ifs = psutil.net_if_addrs()
    if interface in ifs:
        return ifs[interface][0].address
    return ""


@app.route("/external_ip")
def external_ip():
    return urllib.request.urlopen('https://api.ipify.org').read().decode('utf8')


# running processes
@app.route("/procs")
def process():
    return {p.pid: p.info['name']
            for p in psutil.process_iter(['name', 'username'])
            if p.info['username'] == getpass.getuser()}


def _find_proc(name):
    for p in psutil.process_iter(["name", "exe", "cmdline"]):
        if name == p.info['name'] or \
                p.info['exe'] and os.path.basename(p.info['exe']) == name or \
                p.info['cmdline'] and p.info['cmdline'][0] == name:
            return p


@app.route("/get_process/<name>")
def get_process(name):
    p = _find_proc(name)
    if p:
        return {"pid": str(p.pid),
                "status": p.status().upper(),
                "name": p.name(),
                "exe": p.info["exe"],
                "cmdline": " ".join(p.info["cmdline"])}
    return {}


@app.route(f"/process_status/<name>")
def process_status(name):
    p = _find_proc(name)
    if p:
        return p.status().upper()
    return "not running".upper()


@app.route("/process_kill/<pid>")
def process_kill(pid):
    include_parent = True
    sig = signal.SIGTERM
    timeout = None

    try:
        assert pid != os.getpid(), "won't kill myself"
        parent = psutil.Process(pid)
        children = parent.children(recursive=True)
        if include_parent:
            children.append(parent)
        for p in children:
            try:
                p.send_signal(sig)
            except psutil.NoSuchProcess:
                pass
        gone, alive = psutil.wait_procs(children, timeout=timeout,
                                        callback=None)
        return str(gone)
    except Exception as e:
        return "0"


# processes of interest indicating system capabilities
@app.route("/is_systemd")
def is_systemd():
    p = _find_proc("systemd")
    if p:
        return "1"
    return "0"


@app.route("/is_dbus")
def is_dbus():
    p = _find_proc("dbus-daemon")
    if p:
        return "1"
    return "0"


@app.route("/is_kdeconnect")
def is_kdeconnect():
    p = _find_proc("kdeconnectd")
    if p:
        return "1"
    return "0"


@app.route("/is_pulseaudio")
def is_pulseaudio():
    p = _find_proc("pulseaudio")
    if p:
        return "1"
    return "0"


@app.route("/is_pipewire")
def is_pipewire():
    p = _find_proc("pipewire")
    if p:
        return "1"
    return "0"


@app.route("/is_plasma")
def is_plasma():
    p = _find_proc("plasmashell")
    if p:
        return "1"
    return "0"


@app.route("/is_hivemind")
def is_hivemind():
    p = _find_proc("hivemind-core")
    if p:
        return "1"
    return "0"


@app.route("/is_ovos")
def is_ovos():
    p = _find_proc("ovos-core")
    if p:
        return "1"
    return "0"


@app.route("/is_spotify")
def is_spotify():
    # TODO - raspotify / spotifyd etc
    p = _find_proc("spotify")
    if p:
        return "1"
    return "0"


@app.route("/is_firefox")
def is_firefox():
    # TODO - other browsers
    p = _find_proc("firefox")
    if p:
        return "1"
    return "0"


# Pulseaudio
@app.route("/pulseaudio/info")
def pa_info():
    info = pulse.server_info().__dict__
    info.pop("c_struct_fields")
    return info


@app.route("/pulseaudio/version")
def pa_version():
    return pulse.server_info().server_version


@app.route("/pulseaudio/channel_count")
def pa_channel_count():
    return str(pulse.server_info().channel_count)


@app.route("/pulseaudio/default_sink")
def pa_sink():
    return pulse.server_info().default_sink_name


@app.route("/pulseaudio/default_source")
def pa_source():
    return pulse.server_info().default_source_name


@app.route("/pulseaudio/hostname")
def pa_hostname():
    return pulse.server_info().host_name


@app.route("/pulseaudio/list_cards")
def pa_list_cards():
    sinks = []

    for s in pulse.card_list():
        sinks.append({
            "index": s.index,
            "profile": s.profile_active.name,
            "name": s.name,
            "card_name": s.proplist.get("alsa.card_name") or s.name,
            "description": s.proplist.get("device.description", ""),
            "form_factor": s.proplist.get("device.form_factor", "unknown")
        })
    return sinks


@app.route("/pulseaudio/list_sources")
def pa_list_sources():
    sinks = []

    for s in pulse.source_list():
        volumes = [v * 100 for v in s.volume.__dict__["values"]]

        sinks.append({
            "index": s.index,
            "mute": s.mute,
            "name": s.name,
            "card": s.card,
            "channel_count": s.channel_count,
            "channel_list": s.channel_list,
            "driver": s.driver,
            "description": s.description,
            "state": str(s.state).split("state=")[-1][:-1],
            "volumes": volumes
        })
    return sinks


@app.route("/pulseaudio/list_sinks")
def pa_list_sinks():
    sinks = []

    for s in pulse.sink_list():
        volumes = [v * 100 for v in s.volume.__dict__["values"]]

        sinks.append({
            "index": s.index,
            "mute": s.mute,
            "name": s.name,
            "card": s.card,
            "channel_count": s.channel_count,
            "channel_list": s.channel_list,
            "driver": s.driver,
            "description": s.description,
            "state": str(s.state).split("state=")[-1][:-1],
            "volumes": volumes
        })
    return sinks


@app.route("/pulseaudio/list_input_sinks")
def pa_list_input_sinks():
    sinks = []
    from pprint import pprint
    for s in pulse.sink_input_list():
        pprint(s.__dict__)
        pprint(s.volume.__dict__)
        volumes = [v * 100 for v in s.volume.__dict__["values"]]

        sinks.append({
            "index": s.index,
            "mute": s.mute,
            "name": s.name,
            "playing": 1 if not s.corked else 0,
            "media_name": s.proplist.get("media.name") or "",
            "application": s.proplist.get('application.name') or
                           s.proplist.get('application.binary') or
                           s.proplist.get('application.icon_name') or "",
            "sink": s.sink,
            "channel_count": s.channel_count,
            "channel_list": s.channel_list,
            "driver": s.driver,
            "volumes": volumes
        })
    return sinks


@app.route("/pulseaudio/is_playing")
def pa_is_playing():
    if any(not s.corked for s in pulse.sink_input_list()):
        return "1"
    return "0"


# Screen
@app.route("/brightness")
def get_brightness():
    # get the brightness
    brightness = sbc.get_brightness()
    return str(brightness[0])


@app.route("/monitor/<monitor>/brightness")
def monitor_brightness(monitor):
    # get the brightness for the primary monitor
    primary = sbc.get_brightness(display=monitor)
    return str(primary)


@app.route("/monitor/<monitor>/set_brightness/<value>")
def set_monitor_brightness(monitor, value):
    # set the brightness to 100% for the primary monitor
    sbc.set_brightness(value, display=monitor)
    return f"{monitor} - {value} %"


@app.route("/set_brightness/<value>")
def set_brightness(value):
    # set the brightness to 100%
    sbc.set_brightness(value, display=0)
    return str(value)


@app.route("/monitors")
def list_monitors():
    # show the current brightness for each detected monitor
    return sbc.list_monitors()


if __name__ == "__main__":

    app.run(debug=DEVELOPMENT_ENV, host="0.0.0.0", port=PORT)

    if ZeroConfAnnounce is not None:
        ZeroConfAnnounce.shutdown()

zeroconf_scanner.py

import time

from zeroconf import ServiceBrowser, ServiceStateChange
from zeroconf import Zeroconf


IDENTIFIER = "restsysadmin"


class ZeroScanner:
    def __init__(self, identifier=IDENTIFIER):
        self.zero = None
        self.browser = None
        self.nodes = {}
        self.running = False
        self.identifier = identifier.encode("utf-8")

    def get_nodes(self):
        return self.nodes

    def on_new_node(self, node):
        print(node)
        node["last_seen"] = time.time()
        self.nodes[f'{node["name"]}:{node["host"]}'] = node

    def on_node_update(self, node):
        node["last_seen"] = time.time()
        self.nodes[f'{node["name"]}:{node["host"]}'] = node

    def on_service_state_change(self, zeroconf, service_type, name,
                                state_change):

        if state_change is ServiceStateChange.Added or state_change is \
                ServiceStateChange.Updated:
            info = zeroconf.get_service_info(service_type, name)
            if info and info.properties:
                for key, value in info.properties.items():
                    if key == b"type" and value == self.identifier:
                        host = info._properties[b"host"].decode("utf-8")
                        port = info._properties[b"port"].decode("utf-8")
                        name = info._properties[b"name"].decode("utf-8")
                        node = {"host": host, "port": port, "name": name}
                        if state_change is ServiceStateChange.Added:
                            self.on_new_node(node)
                        else:
                            self.on_node_update(node)

    def start(self):
        self.zero = Zeroconf()
        self.browser = ServiceBrowser(self.zero, "_http._tcp.local.",
                                      handlers=[self.on_service_state_change])
        self.running = True

    def stop(self):
        if self.zero:
            self.zero.close()
        self.zero = None
        self.browser = None
        self.running = False