#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Publish low-overhead Linux kiosk performance telemetry to MQTT.""" import glob import json import os import socket import time import psutil import paho.mqtt.client as mqtt # type: ignore[import-not-found] MQTT_HOST = os.environ["MQTT_HOST"] MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883")) MQTT_USER = os.environ.get("MQTT_USER") MQTT_PASSWORD = os.environ.get("MQTT_PASSWORD") KIOSK_ID = os.environ.get("KIOSK_ID", "kiosk") KIOSK_ENTITY_PREFIX = os.environ.get("KIOSK_ENTITY_PREFIX", f"{KIOSK_ID}_kiosk") KIOSK_NAME = os.environ.get("KIOSK_NAME", f"{KIOSK_ID.replace('-', ' ').title()} Kiosk") KIOSK_MANUFACTURER = os.environ.get("KIOSK_MANUFACTURER", "Unknown") KIOSK_MODEL = os.environ.get("KIOSK_MODEL", "Linux Touchscreen Kiosk") MQTT_TOPIC = os.environ.get("MQTT_TOPIC", f"{KIOSK_ID}/kiosk/performance") SAMPLE_SECONDS = max(5, int(os.environ.get("SAMPLE_SECONDS", "10"))) STATE_TOPIC = f"{MQTT_TOPIC}/state" AVAILABILITY_TOPIC = f"{MQTT_TOPIC}/availability" DISCOVERY_PREFIX = os.environ.get("MQTT_DISCOVERY_PREFIX", "homeassistant") DEVICE = { "identifiers": [KIOSK_ENTITY_PREFIX], "name": KIOSK_NAME, "manufacturer": KIOSK_MANUFACTURER, "model": KIOSK_MODEL, } METRICS = { "system_cpu_percent": ("System CPU", "%", None, "measurement", "mdi:cpu-64-bit"), "system_iowait_percent": ("CPU I/O Wait", "%", None, "measurement", "mdi:timer-sand"), "load_1m": ("Load 1 Minute", None, None, "measurement", "mdi:gauge"), "load_5m": ("Load 5 Minutes", None, None, "measurement", "mdi:gauge"), "memory_percent": ("Memory Used", "%", None, "measurement", "mdi:memory"), "cpu_temperature": ("CPU Temperature", "°C", "temperature", "measurement", "mdi:thermometer"), "cpu_frequency": ("CPU Frequency", "MHz", "frequency", "measurement", "mdi:sine-wave"), "fan_speed": ("Fan Speed", "rpm", None, "measurement", "mdi:fan"), "processor_fan_speed": ("Processor Fan Speed", "rpm", None, "measurement", "mdi:fan"), "motherboard_fan_speed": ("Motherboard Fan Speed", "rpm", None, "measurement", "mdi:fan"), "chrome_cpu_percent": ("Chrome CPU", "%", None, "measurement", "mdi:google-chrome"), "chrome_browser_cpu_percent": ("Chrome Browser CPU", "%", None, "measurement", "mdi:web"), "chrome_renderer_cpu_percent": ("Chrome Renderer CPU", "%", None, "measurement", "mdi:monitor-dashboard"), "chrome_gpu_cpu_percent": ("Chrome GPU Process CPU", "%", None, "measurement", "mdi:expansion-card"), "chrome_utility_cpu_percent": ("Chrome Utility CPU", "%", None, "measurement", "mdi:cog-transfer"), "chrome_memory": ("Chrome Memory", "MB", "data_size", "measurement", "mdi:memory"), "chrome_processes": ("Chrome Processes", None, None, "measurement", "mdi:format-list-numbered"), "gpu_busy_percent": ("GPU Busy", "%", None, "measurement", "mdi:expansion-card-variant"), "network_receive_rate": ("Network Receive Rate", "kB/s", "data_rate", "measurement", "mdi:download-network"), "network_transmit_rate": ("Network Transmit Rate", "kB/s", "data_rate", "measurement", "mdi:upload-network"), "disk_read_rate": ("Disk Read Rate", "kB/s", "data_rate", "measurement", "mdi:harddisk"), "disk_write_rate": ("Disk Write Rate", "kB/s", "data_rate", "measurement", "mdi:harddisk"), } def round_or_none(value, digits=1): return round(value, digits) if value is not None else None def cpu_temperature(): values = [] try: for entries in psutil.sensors_temperatures().values(): values.extend(entry.current for entry in entries if entry.current is not None) except (AttributeError, OSError): pass return max(values) if values else None def fan_speeds(): result: dict[str, float | None] = { "maximum": None, "processor": None, "motherboard": None, } try: for entries in psutil.sensors_fans().values(): for entry in entries: if entry.current is None: continue result["maximum"] = max(result["maximum"] or 0, entry.current) label = (entry.label or "").lower() if "processor" in label: result["processor"] = entry.current elif "motherboard" in label: result["motherboard"] = entry.current except (AttributeError, OSError): pass return result def gpu_busy_percent(): for pattern in ( "/sys/class/drm/card*/device/gpu_busy_percent", "/sys/class/drm/card*/gt_busy_percent", ): for path in glob.glob(pattern): try: with open(path, encoding="ascii") as source: return float(source.read().strip()) except (OSError, ValueError): continue return None def chrome_processes(): processes = [] for process in psutil.process_iter(["name", "cmdline"]): try: name = (process.info["name"] or "").lower() command = " ".join(process.info["cmdline"] or []).lower() if not any(token in name for token in ("chrome", "chromium")): continue if "--type=renderer" in command: group = "renderer" elif "--type=gpu-process" in command: group = "gpu" elif "--type=utility" in command: group = "utility" elif "--type=" not in command: group = "browser" else: group = "other" process.cpu_percent(None) processes.append((process, group)) except (psutil.AccessDenied, psutil.NoSuchProcess, psutil.ZombieProcess): continue return processes def publish_discovery(client): for key, (name, unit, device_class, state_class, icon) in METRICS.items(): config = { "name": name, "unique_id": f"{KIOSK_ENTITY_PREFIX}_{key}", "state_topic": STATE_TOPIC, "availability_topic": AVAILABILITY_TOPIC, "value_template": f"{{{{ value_json.{key} }}}}", "device": DEVICE, "icon": icon, } if unit: config["unit_of_measurement"] = unit if device_class: config["device_class"] = device_class if state_class: config["state_class"] = state_class topic = f"{DISCOVERY_PREFIX}/sensor/{KIOSK_ENTITY_PREFIX}/{key}/config" client.publish(topic, json.dumps(config, separators=(",", ":")), qos=1, retain=True) def on_connect(client, userdata, flags, reason_code, properties=None): """Restore all MQTT state after the initial connection or a reconnect.""" publish_discovery(client) client.publish(AVAILABILITY_TOPIC, "online", qos=1, retain=True) def collect_sample(): chrome = chrome_processes() net_before = psutil.net_io_counters() disk_before = psutil.disk_io_counters() cpu_times = psutil.cpu_times_percent(interval=SAMPLE_SECONDS, percpu=False) net_after = psutil.net_io_counters() disk_after = psutil.disk_io_counters() cpu_by_group = {group: 0.0 for group in ("browser", "renderer", "gpu", "utility", "other")} chrome_memory = 0 active_processes = 0 for process, group in chrome: try: cpu_by_group[group] += process.cpu_percent(None) chrome_memory += process.memory_info().rss active_processes += 1 except (psutil.AccessDenied, psutil.NoSuchProcess, psutil.ZombieProcess): continue load_1m, load_5m, _ = os.getloadavg() frequency = psutil.cpu_freq() fans = fan_speeds() interval = float(SAMPLE_SECONDS) disk_read_rate = None disk_write_rate = None if disk_before is not None and disk_after is not None: disk_read_rate = (disk_after.read_bytes - disk_before.read_bytes) / interval / 1024 disk_write_rate = (disk_after.write_bytes - disk_before.write_bytes) / interval / 1024 sample = { "timestamp": int(time.time()), "hostname": socket.gethostname(), "sample_seconds": SAMPLE_SECONDS, "system_cpu_percent": round_or_none(100.0 - cpu_times.idle), "system_iowait_percent": round_or_none(getattr(cpu_times, "iowait", 0.0)), "load_1m": round(load_1m, 2), "load_5m": round(load_5m, 2), "memory_percent": round_or_none(psutil.virtual_memory().percent), "cpu_temperature": round_or_none(cpu_temperature()), "cpu_frequency": round_or_none(frequency.current if frequency else None, 0), "fan_speed": round_or_none(fans["maximum"], 0), "processor_fan_speed": round_or_none(fans["processor"], 0), "motherboard_fan_speed": round_or_none(fans["motherboard"], 0), "chrome_browser_cpu_percent": round_or_none(cpu_by_group["browser"]), "chrome_renderer_cpu_percent": round_or_none(cpu_by_group["renderer"]), "chrome_gpu_cpu_percent": round_or_none(cpu_by_group["gpu"]), "chrome_utility_cpu_percent": round_or_none(cpu_by_group["utility"]), "chrome_cpu_percent": round_or_none(sum(cpu_by_group.values())), "chrome_memory": round(chrome_memory / 1048576, 1), "chrome_processes": active_processes, "gpu_busy_percent": round_or_none(gpu_busy_percent()), "network_receive_rate": round((net_after.bytes_recv - net_before.bytes_recv) / interval / 1024, 1), "network_transmit_rate": round((net_after.bytes_sent - net_before.bytes_sent) / interval / 1024, 1), "disk_read_rate": round_or_none(disk_read_rate), "disk_write_rate": round_or_none(disk_write_rate), } return sample def main(): client = mqtt.Client(client_id=f"{KIOSK_ID}-kiosk-telemetry") if MQTT_USER: client.username_pw_set(MQTT_USER, MQTT_PASSWORD) client.will_set(AVAILABILITY_TOPIC, "offline", qos=1, retain=True) client.on_connect = on_connect client.reconnect_delay_set(min_delay=1, max_delay=30) client.connect(MQTT_HOST, MQTT_PORT, keepalive=60) client.loop_start() try: while True: payload = json.dumps(collect_sample(), separators=(",", ":"), allow_nan=False) client.publish(STATE_TOPIC, payload, qos=0, retain=False) finally: client.publish(AVAILABILITY_TOPIC, "offline", qos=1, retain=True).wait_for_publish() client.disconnect() client.loop_stop() if __name__ == "__main__": main()