commit bea626051892326ece1886d03dd03e82b78fdffe Author: Leone, Mark A [LGS] Date: Mon Jun 23 20:37:30 2025 -0400 Initial import diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..1059bd4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +*.pyc +fwledmonitor.service diff --git a/.vscode/launch.json b/.vscode/launch.json new file mode 100644 index 0000000..948a236 --- /dev/null +++ b/.vscode/launch.json @@ -0,0 +1,16 @@ +{ + // Use IntelliSense to learn about possible attributes. + // Hover to view descriptions of existing attributes. + // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387 + "version": "0.2.0", + "configurations": [ + { + "name": "Python: Current File", + "type": "python", + "request": "launch", + "program": "${file}", + "console": "integratedTerminal", + "justMyCode": true + } + ] +} \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..60421ee --- /dev/null +++ b/README.md @@ -0,0 +1,65 @@ +# Framework 16 LED Matrix System Monitoring Application + +This software is intended for use on a Framework 16 laptop with LED Matrix Panels installed. It's a clone of the [LED System Monitor](https://code.karsttech.com/jeremy/FW_LED_System_Monitor.git) project, with certain modifications and extensions applied. + +## Original Capabilities +Display system performance characteristics in real-time +* Top Left: CPU utilization +* Bottom Left: Battery charge level and plug status + memory utilization +* Top Right: Disk Input/Output rates +* Bottom Right: Network Upload/Download rates + +## Enhanced Capabilities +* Pre-defined applications can be displayed on any quadrant + * Top or bottom of left or right panel + * Specified via program arguments +* Additional system performance applications + * Temperature sensor values (average of sensors on each of up to eight devices) + * Fan speeds (max 2 supported) +* Keyboard shortcut identifies apps running in each quadrant by displaying abbeviated name +* Plugin framework supports simplified development of addiitonal LED Panel applications +* Automatic detection of left and right LED panels +## Installation +* Install [PyEnv](https://github.com/pyenv/pyenv) +* Any other python virtual environment package may be used. Commands below work with PyEnv. +``` +cd led-matrix +pyenv install 3.11 +pyenv virtualenv 3.11 3.11 +pyenv activate +pip install -r requirements.txt +``` +## Run +``` +cd led-matrix +python led-sysyem-monitor.py [--help] [--top-left {cpu,net,disk,mem-bat,none,temp,fan}] [--bottom-left {cpu,net,disk,mem-bat,none,temp,fan}] [--top-right {cpu,net,disk,mem-bat,none,temp,fan}] + [--bottom-right {cpu,net,disk,mem-bat,none,temp,fan}] [--no-key-listener] [--disable-plugins] +python led-sysyem-monitor.py --help #For more verbose help info +``` +## Run as a Linux service +``` +cd led-matrix +./install_as_service.sh [...args] #program args to be applied when starting or restarting the service +sudo systemctl start|stop|restart|status fwledmonitor +``` +## Keyboard Shortcut +* Alt+I: displays app names in each quadrant while keys are pressed +* Disable key listener with `--no-key-listener` program arg +* To use the key listener, the app must have read permission on the keyboard device (e.g /dev/input/event7). The service runs as root, and therefore has the required access. If you want toi use the key listener while running the app directly, you need to add your user account to the `input` group and ensure there is a group read permission on the keyboard device. **NB:** Consider the implications of this. Any program running as a user in the `input` group will be able to capture your keystrokes. +## Plugin Development +* Add a file in the `plugins` dir with a name that matches the blob pattern `*_plugin.py` +* See `temp_fan_plugin.py` for an implementation example +## Notes +* To list your input devices, use the following python code after installing [Python-evdev](https://python-evdev.readthedocs.io/en/latest/index.html) +``` +>>> import evdev + +>>> devices = [evdev.InputDevice(path) for path in evdev.list_devices()] +>>> for device in devices: +>>> print(device.path, device.name, device.phys) +/dev/input/event1 Dell Dell USB Keyboard usb-0000:00:12.1-2/input0 +/dev/input/event0 Dell USB Optical Mouse usb-0000:00:12.0-2/input0 +``` +* The baseline reference for calculating the ratio used to display temperature and fan speed readings were arbitarily defined. See `TEMP_REF` and `MAX_FAN_SPEED` in `temp_fan_plugin.py`. +* To examine system performance measures manually and in detail, run `python ps-util-sensors.py` +* To use a different key combination for identifying the running apps, see `KEY_I` and `MODIFIER_KEYS` in `led_system_monitor.py` diff --git a/commands.py b/commands.py new file mode 100644 index 0000000..8eba206 --- /dev/null +++ b/commands.py @@ -0,0 +1,37 @@ +from enum import Enum + +# https://github.com/FrameworkComputer/inputmodule-rs/blob/main/commands.md + +# Display is 9x34 wide x tall +class Commands(): + Brightness = 0x00 + Pattern = 0x01 + Bootloader = 0x02 + Sleep = 0x03 + GetSleep = 0x03 + Animate = 0x04 + GetAnimate = 0x04 + Panic = 0x05 + DrawBW = 0x06 + StageCol = 0x07 + FlushCols = 0x08 + SetText = 0x09 + StartGame = 0x10 + GameCtrl = 0x11 + GameStatus = 0x12 + SetColor = 0x13 + DisplayOn = 0x14 + InvertScreen = 0x15 + SetPxCol = 0x16 + FlushFB = 0x17 + Version = 0x20 + + +def send_command(s, command_id, parameters = None, with_response=False): + message = bytearray([0x32, 0xAC, command_id]) + if parameters: + message.extend(parameters) + s.write(message) + if with_response: + res = s.read(1) + return res diff --git a/drawing.py b/drawing.py new file mode 100644 index 0000000..a3c221e --- /dev/null +++ b/drawing.py @@ -0,0 +1,255 @@ +# Built In Dependencies +import time +import math +import threading +import importlib.util +import sys +import os +import re + +# Internal Dependencies +from commands import Commands, send_command +from patterns import lightning_bolt_bot, lightning_bolt_top, lookup_table, id_patterns + +# External Dependencies +import numpy as np +import serial # pyserial +from serial.tools import list_ports + +# Correct table orientation for visual orientation when drawn +for i in range(lookup_table.shape[0]): + lookup_table[i] = lookup_table[i].T + + +def spiral_index(fill_ratio): + return int(round(fill_ratio * 9.999999 - 0.5)) + +## App Draw Functions ## + +# Takes up 15 rows, 7 columns, starting at y,1 +# For bottom segment, the 16th row will be empty +def draw_spiral_vals(grid, cpu_values, fill_value, y): + y += 1 + for i, v in enumerate(cpu_values): + column_number = i % 2 + row_number = i // 2 + fill_grid = lookup_table[spiral_index(v)] + grid[1+column_number*4:4+column_number*4, y+row_number*4:y+3+row_number*4] = fill_grid * fill_value + +# Takes up 2 rows, 7 columns, starting at y, 1 +def draw_memory(grid, memory_ratio, fill_value, y): + lit_pixels = 7 * 2 * memory_ratio + pixels_bottom = int(round(lit_pixels / 2)) + pixels_top = int(round((lit_pixels - 0.49) / 2)) + grid[1:1+pixels_top,y+1] = fill_value + grid[1:1+pixels_bottom,y+2] = fill_value + +# Takes up 12 (top segment) or 13 (bottom segment) rows, 7 columns, starting at y,1 +def draw_battery(grid, battery_ratio, battery_plugged, fill_value, y, + battery_low_thresh = 0.07, battery_low_flash_time = 2, charging_pulse_time = 3): + if y == 19: # Placement on bottom + bot = 33 + num_rows = 13 + lightning_bolt = lightning_bolt_bot + else: # Placement on top (y == 3) + bot = 16 + num_rows = 12 + lightning_bolt = lightning_bolt_top + bat_top = y + 1 + bat_bot = bat_top + num_rows + lit_pixels = int(round(num_rows * 7 * battery_ratio)) + pixels_base = lit_pixels // 7 + remainder = lit_pixels % 7 + if battery_ratio <= battery_low_thresh and not battery_plugged: + if time.time() % battery_low_flash_time * 2 < battery_low_flash_time: # This will flash the battery indicator if too low + return + for i in range(7): + pixels_col = pixels_base + if i < remainder: + pixels_col += 1 + grid[i+1,bot-pixels_col:bot] = fill_value + if battery_plugged: + pulse_amount = math.sin(time.time() / charging_pulse_time) + grid[1:8,bat_top:bat_bot][lightning_bolt] -= np.rint(fill_value + 10 * pulse_amount).astype(int) + indices = grid[1:8,bat_top:bat_bot] < 0 + grid[1:8,bat_top:bat_bot][indices] = -grid[1:8,bat_top:bat_bot][indices] + +# Takes up 16 (top segment) or 17 (bottom segment) rows, 3 columns, starting at y,1 +def draw_bar(grid, bar_ratio, bar_value, bar_x_offset = 1, y=0): + bar_width = 3 + bar_height = 16 + lit_pixels = int(round(bar_height * bar_width * bar_ratio)) + pixels_base = lit_pixels // bar_width + remainder = lit_pixels % bar_width + for i in range(bar_width): + pixels_col = pixels_base + if i < remainder: + pixels_col += 1 + if y == 16: + grid[bar_x_offset+i,33-pixels_col:33] = bar_value + else: + grid[bar_x_offset+i,1:1+pixels_col] = bar_value + +## Border Draw Functions ## + +# Draws a border around a 16 (top segment) or a 17 (bottom segment) +# x 9 grid, divided into a 2 x 4 grid. For the bottom segment, +# the last grid will have an extra row +def draw_8_x_8_grid(grid, border_value, y): + height = 16 if y == 0 else 17 + grid[:, y] = border_value # Top + grid[:, y+height] = border_value # Bottom + + grid[0, y:y+height] = border_value # Left + grid[8, y:y+height] = border_value # Right + grid[4, y:y+height] = border_value # Middle + + # Horizontal grid borders + grid[:, y+4] = border_value + grid[:, y+8] = border_value + grid[:, y+12] = border_value + +# Draws a border around a 16 (top segment) or a 17 (bottom segment) +# x 9 grid, split horizontally into two sections at the specified column +def draw_2_x_1_horiz_grid(grid, border_value, y, x_split_idx=4): + height = 16 if y == 0 else 17 + grid[:, y] = border_value # Top + grid[:, y+height] = border_value # Bottom + + grid[0, y:y+height] = border_value # Left + grid[8, y:y+height] = border_value # Right + grid[x_split_idx, y:y+height] = border_value # Middle + +# Draws a border around a 16 (top segment) or a 17 (bottom segment), +# split vertically into two sections at the specified row +def draw_1_x_2_vert_grid(grid, border_value, y, y_split_idx = 3): + height = 16 if y == 0 else 17 + grid[:, y] = border_value # Top + grid[:, y+height] = border_value # Bottom + grid[:, y+y_split_idx] = border_value # Middle + + grid[0, y:y+height] = border_value # Left + grid[8, y:y+height] = border_value # Right + +# Draws a border around the entire panel, split +# vertically into two equal segments +def draw_outline_border(grid, border_value): + grid[:, 0] = border_value # Top + grid[:, 16] = border_value # Middle + grid[:, 33] = border_value # Bottom + grid[0, :] = border_value # Left + grid[8, :] = border_value # Right + +# Maps an app arg value to abstract app and border draw functions +metrics_funcs = { + "cpu": { + "fn": draw_spiral_vals, + "border": draw_8_x_8_grid + }, + "disk": { + "fn": draw_bar, + "border": draw_2_x_1_horiz_grid + }, + "net": { + "fn": draw_bar, + "border": draw_2_x_1_horiz_grid + }, + "mem": { + "fn": draw_memory, + "border": draw_1_x_2_vert_grid + }, + "bat": { + "fn": draw_battery, + "border": draw_1_x_2_vert_grid + }, + #noop + "none": { + "fn": lambda *x: x, + "border": lambda *x: x + } +} + +# Draws the app for the specified arg value +def draw_app(app, *arguments, **kwargs): + metrics_funcs[app].get('fn')(*arguments, **kwargs) + +# Draws the border for the specified arg value +def draw_app_border(app, *arguments): + metrics_funcs[app].get('border')(*arguments) + +# Draw the IDs of apps currently assigned to the top and bottom of the left panel +def draw_ids_left(grid, top_left, bot_left, fill_value): + fill_grid_top = id_patterns[top_left] + fill_grid_bot = id_patterns[bot_left] + grid[1:8, 1:16] = fill_grid_top * fill_value + grid[1:8, 18:-1] = fill_grid_bot * fill_value + +# Draw the IDs of apps currently assigned to the top and bottom of the right panel +def draw_ids_right(grid, top_right, bot_right, fill_value): + fill_grid_top = id_patterns[top_right] + fill_grid_bot = id_patterns[bot_right] + grid[1:8, 1:16] = fill_grid_top * fill_value + grid[1:8, 18:-1] = fill_grid_bot * fill_value + +def draw_to_LEDs(s, grid): + for i in range(grid.shape[0]): + params = bytearray([i]) + bytearray(grid[i, :].tolist()) + send_command(s, Commands.StageCol, parameters=params) + send_command(s, Commands.FlushCols) + + +def init_device(location = "1-3.2"): + try: + # VID = 1234 + # PID = 5678 + device_list = list_ports.comports() + for device in device_list: + if device.location and device.location.startswith(location): + s = serial.Serial(device.device, 115200) + return s + except Exception as e: + print(e) + + +class DrawingThread(threading.Thread): + def __init__(self, port_location, input_queue): + super().__init__() + self.daemon = True + self.port_location = port_location + self.serial_port = init_device(self.port_location) + self.input_queue = input_queue + + def run(self): + while True: + try: + grid = self.input_queue.get() + draw_to_LEDs(self.serial_port, grid) + except Exception as e: + print(f"Error in DrawingThread: {e}") + del self.serial_port + time.sleep(1.0) + self.serial_port = init_device(self.port_location) + +############################################################### +### Load metrics functions from plugins ### +############################################################### +# Keep this at the end of the module to avoid circular imports +if not re.search(r"--disable-plugins|-dp", str(sys.argv)): + plugins_dir = './plugins/' + for file in os.listdir(plugins_dir): + if file.endswith('_plugin.py'): + module_name = re.sub(file, "_plugin.py", "") + file_path = plugins_dir + file + spec = importlib.util.spec_from_file_location(module_name, file_path) + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + spec.loader.exec_module(module) + + for k,v in module.metrics_funcs.items(): + metrics_funcs[k] = v + + from drawing import id_patterns + for k,v in module.id_patterns.items(): + id_patterns[k] = v + +################################################################ diff --git a/find_ports.py b/find_ports.py new file mode 100644 index 0000000..d2ebe7e --- /dev/null +++ b/find_ports.py @@ -0,0 +1,23 @@ +import serial +from serial.tools import list_ports + +def init_device(location = "1-4.2"): + try: + # VID = 1234 + # PID = 5678 + device_list = list_ports.comports() + for device in device_list: + if device.location and device.location.startswith(location): + # s = serial.Serial(device.device, 115200) + print(device) + print(device, device.location, device.device, device.manufacturer, device.device_path, device.product, device.interface, device.description) + except Exception as e: + print(e) + +init_device("1-3.") + +""" +Panels on right side +Left: 1-3.2, ACM0 +Right: 1-3.3, ACM1 +""" \ No newline at end of file diff --git a/install_as_service.sh b/install_as_service.sh new file mode 100755 index 0000000..f5d39db --- /dev/null +++ b/install_as_service.sh @@ -0,0 +1,24 @@ +#!/bin/bash +args="$*" +chmod +x run.sh +rm -f fwledmonitor.service || true +sed -i "s#led_system_monitor.py.*\$#led_system_monitor.py ${args}#" run.sh +cat <>./fwledmonitor.service +[Unit] +Description=Framework 16 LED System Monitor +After=network.service + +[Service] +Type=simple +Restart=always +WorkingDirectory=$PWD +ExecStart=sh -c "'$PWD/run.sh'" + +[Install] +WantedBy=default.target +EOF + +sudo systemctl stop fwledmonitor +sudo cp fwledmonitor.service /lib/systemd/system +sudo systemctl daemon-reload +sudo systemctl enable fwledmonitor diff --git a/led_system_monitor.py b/led_system_monitor.py new file mode 100644 index 0000000..3611d5c --- /dev/null +++ b/led_system_monitor.py @@ -0,0 +1,229 @@ +# Built In Dependencies +import time +import queue +import sys +import re +from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter +import importlib.util +import sys +import os + + +# Internal Dependencies +from drawing import draw_outline_border, draw_ids_left, draw_ids_right, draw_app, draw_app_border, DrawingThread +from monitors import CPUMonitor, MemoryMonitor, BatteryMonitor, DiskMonitor, NetworkMonitor, get_monitor_brightness + +# External Dependencies +import numpy as np +import evdev +from serial.tools import list_ports + + +KEY_I = ('KEY_I', 23) +MODIFIER_KEYS = [('KEY_RIGHTALT', 100), ('KEY_LEFTALT', 56)] + +def discover_led_devices(): + locations = [] + try: + device_list = list_ports.comports() + for device in device_list: + if 'LED Matrix Input Module' in str(device): + locations.append((device.location, device.device)) + #location is of form: -[-]… port is of form x.y:n.m + # Sort by y:n.m to get the devices in left-right order + return sorted(locations, key = lambda x: re.sub('^\d+\-\d+\.', '', x[0])) + except Exception as e: + print(f"An Exception occured while tring to locate LED Matrix devices. {e}") + +device = evdev.InputDevice('/dev/input/event7') + +def main(args): + led_devices = discover_led_devices() + if not len(led_devices): + print("No LED devices found") + sys.exit(0) + elif len(led_devices) == 1: + print(f"Only one LED device found ({led_devices[0]}). Right panel args will be ignored") + args.top_right = args.bottom_right = "none" + else: + print(f"Found LED devices: Left: {led_devices[0]}, Right: {led_devices[1]}") + locations = list(map(lambda x: x[0], led_devices)) + drawing_queues = [] + + # Track key presses to reveal metrics ID in each panel section + global alt_pressed + alt_pressed = False + global i_pressed + i_pressed = False + + # Set up monitors and brightness parameters + min_background_brightness = 12 + max_background_brightness = 35 + min_foreground_brightness = 24 + max_foreground_brightness = 160 + + cpu_monitor = CPUMonitor() + memory_monitor = MemoryMonitor() + battery_monitor = BatteryMonitor() + disk_monitor = DiskMonitor() + network_monitor = NetworkMonitor() + + # Setuop left panel drawing queue + left_drawing_queue = queue.Queue(2) + left_drawing_thread = DrawingThread(locations[0], left_drawing_queue) + left_drawing_thread.start() + drawing_queues.append(left_drawing_queue) + + # Setup right panel drawing queue (if present) + if len(locations) == 2: + right_drawing_queue = queue.Queue(2) + right_drawing_thread = DrawingThread(locations[1], right_drawing_queue) + right_drawing_thread.start() + drawing_queues.append(right_drawing_queue) + + def draw_cpu(arg, grid, foreground_value, idx): + last_cpu_values = cpu_monitor.get() + draw_app(arg, grid, last_cpu_values, foreground_value, idx) + + def draw_mem_bat(arg, grid, foreground_value, idx): + last_memory_values = memory_monitor.get() + last_battery_values = battery_monitor.get() + draw_app("mem", grid, last_memory_values, foreground_value, idx) + draw_app("bat", grid, last_battery_values[0], last_battery_values[1], foreground_value, idx+3) + + def draw_disk(arg, grid, foreground_value, idx): + last_disk_read, last_disk_write = disk_monitor.get() + draw_app(arg, grid, last_disk_read, foreground_value, bar_x_offset=1, y=idx) # Read + draw_app(arg, grid, last_disk_write, foreground_value, bar_x_offset=5, y=idx) # Write + + def draw_net(arg, grid, foreground_value, idx): + last_network_upload, last_network_download = network_monitor.get() + draw_app(arg, grid, last_network_upload, foreground_value, bar_x_offset=1, y=idx) + draw_app(arg, grid, last_network_download, foreground_value, bar_x_offset=5, y=idx) + + app_functions = { + "cpu": draw_cpu, + "mem-bat": draw_mem_bat, + "disk": draw_disk, + "net": draw_net, + "none": lambda *x: x # noop + } + + ################################################# + ### Load app functions from plugins ### + if not re.search(r"--disable-plugins|-dp", str(sys.argv)): + plugins_dir = './plugins/' + for file in os.listdir(plugins_dir): + if file.endswith('_plugin.py'): + module_name = re.sub(file, "_plugin.py", "") + file_path = plugins_dir + file + spec = importlib.util.spec_from_file_location(module_name, file_path) + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + spec.loader.exec_module(module) + + for obj in module.app_funcs: + app_functions[obj["name"]] = obj["fn"] + ################################################# + + while True: + try: + screen_brightness = get_monitor_brightness() + background_value = int(screen_brightness * (max_background_brightness - min_background_brightness) + min_background_brightness) + foreground_value = int(screen_brightness * (max_foreground_brightness - min_foreground_brightness) + min_foreground_brightness) + grid = np.zeros((9,34), dtype = int) + active_keys = device.active_keys(verbose=True) + if (MODIFIER_KEYS[0] in active_keys or MODIFIER_KEYS[1] in active_keys) and KEY_I in active_keys and not args.no_key_listener: + draw_outline_border(grid, background_value) + draw_ids_left(grid, args.top_left, args.bottom_left, foreground_value) + left_drawing_queue.put(grid) + grid = np.zeros((9,34), dtype = int) + draw_outline_border(grid, background_value) + draw_ids_right(grid, args.top_right, args.bottom_right, foreground_value) + right_drawing_queue.put(grid) + grid = np.zeros((9,34), dtype = int) + time.sleep(0.1) + continue + + # Draw by quadrants (i.e. to top and bottom of left and right panels) + for i, draw_queue in enumerate(drawing_queues): + if i == 0: + panel = 'left' + _args = [args.top_left, args.bottom_left] + else: + panel = 'right' + _args = [args.top_right, args.bottom_right] + grid = np.zeros((9,34), dtype = int) + for j, arg in enumerate(_args): + if j == 0: + idx = 0 + loc = 'top' + else: + idx = 16 + loc = 'bottom' + try: + func = app_functions[arg] + func(arg, grid, foreground_value, idx) + except KeyError: + print(app_functions.keys()) + print(f"Unrecognized display option {arg} for {loc} {panel}") + if arg == 'mem-bat': arg = 'mem' # Single border draw for mem and bat together + draw_app_border(arg, grid, background_value, idx) + draw_queue.put(grid) + + except KeyboardInterrupt: + break + except Exception as e: + import traceback + print(f"Error in main loop: {e}") + traceback.print_exc() + time.sleep(1.0) + time.sleep(0.1) + + print("Exiting") + +if __name__ == "__main__": + app_names = ["cpu", "net", "disk", "mem-bat", "none"] + ############################################################### + ### Load additional app names from plugins for arg parser ### + if not re.search(r"--disable-plugins|-dp", str(sys.argv)): + plugins_dir = './plugins/' + for file in os.listdir(plugins_dir): + if file.endswith('_plugin.py'): + module_name = re.sub(file, "_plugin.py", "") + file_path = plugins_dir + file + spec = importlib.util.spec_from_file_location(module_name, file_path) + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + spec.loader.exec_module(module) + + app_names += module.metrics_funcs.keys() + ################################################################# + parser = ArgumentParser(prog="FW LED System Monitor", add_help=False, + description="Displays system performance metrics in the Framework 16 LED Matrix input module", + formatter_class=ArgumentDefaultsHelpFormatter) + mode_group = parser.add_mutually_exclusive_group() + mode_group.add_argument("--help", "-h", action="help", + help="Show this help message and exit") + + addGroup = parser.add_argument_group(title = "Metrics Display Options") + addGroup.add_argument("--top-left", "-tl", type=str, default="cpu", choices=app_names, + help="Metrics to display in the top section of the left matrix panel") + addGroup.add_argument("--bottom-left", "-bl", type=str, default="mem-bat", choices=app_names, + help="Metrics to display in the bottom section of the left matrix panel") + addGroup.add_argument("--top-right", "-tr", type=str, default="disk", choices=app_names, + help="Metrics to display in the top section of the right matrix panel") + addGroup.add_argument("--bottom-right", "-br", type=str, default="disk", choices=app_names, + help="Metrics to display in the top section of the right matrix panel") + + addGroup.add_argument("--no-key-listener", "-nkl", action="store_true", help="Do not listen for key presses") + addGroup.add_argument("--disable-plugins", "-dp", action="store_true", help="Do not load any plugin code") + + args = parser.parse_args() + print(f"top left {args.top_left}") + print(f"bottom left {args.bottom_left}") + print(f"top right {args.top_right}") + print(f"bottom right {args.bottom_right}") + if args.no_key_listener: print("Key listener disabled") + + main(args) \ No newline at end of file diff --git a/monitors.py b/monitors.py new file mode 100644 index 0000000..c4fcb2b --- /dev/null +++ b/monitors.py @@ -0,0 +1,147 @@ +# Built In Dependencies +import time +import psutil +import os +from statistics import mean + +# Reference for fractional measure of sensor temps (in degrees Celcius) +TEMP_REF = 120 + +# Reference for fractional measure of fan speeds (in rpm) +MAX_FAN_SPEED = 6_000 + +if os.name == 'nt': + import wmi + +class DiskMonitor: + def __init__(self, hysterisis_time = 20): + self.read_usage_history = [0] + self.write_usage_history = [0] + self.history_times = [0] + self.highest_read_rate = 0.00001 + self.highest_write_rate = 0.00001 + self.max_history_size = hysterisis_time + + def get(self): + try: + disk_io = psutil.disk_io_counters() + read_usage = disk_io.read_bytes + write_usage = disk_io.write_bytes + self.read_usage_history.append(read_usage) + self.write_usage_history.append(write_usage) + self.history_times.append(time.time()) + if len(self.read_usage_history) > self.max_history_size: + self.read_usage_history = self.read_usage_history[-self.max_history_size:] + self.write_usage_history = self.write_usage_history[-self.max_history_size:] + self.history_times = self.history_times[-self.max_history_size:] + + read_diff = self.read_usage_history[-1] - self.read_usage_history[0] + write_diff = self.write_usage_history[-1] - self.write_usage_history[0] + time_diff = self.history_times[-1] - self.history_times[0] + read_rate = read_diff / time_diff + write_rate = write_diff / time_diff + self.highest_read_rate = max(self.highest_read_rate, read_rate) + self.highest_write_rate = max(self.highest_write_rate, write_rate) + read_percent = min(1.0, read_rate / self.highest_read_rate) + write_percent = min(1.0, write_rate / self.highest_write_rate) + return read_percent, write_percent + except Exception as e: + print(f"Error in DiskMonitor.get(): {e}") + return 0, 0 + +class NetworkMonitor: + def __init__(self, hysterisis_time = 20): + self.sent_usage_history = [0] + self.recv_usage_history = [0] + self.history_times = [0] + self.highest_sent_rate = 0.00001 + self.highest_recv_rate = 0.00001 + self.max_history_size = hysterisis_time + + def get(self): + try: + net_io = psutil.net_io_counters() + sent_usage = net_io.bytes_sent + recv_usage = net_io.bytes_recv + self.sent_usage_history.append(sent_usage) + self.recv_usage_history.append(recv_usage) + self.history_times.append(time.time()) + if len(self.sent_usage_history) > self.max_history_size: + self.sent_usage_history = self.sent_usage_history[-self.max_history_size:] + self.recv_usage_history = self.recv_usage_history[-self.max_history_size:] + self.history_times = self.history_times[-self.max_history_size:] + + sent_diff = self.sent_usage_history[-1] - self.sent_usage_history[0] + recv_diff = self.recv_usage_history[-1] - self.recv_usage_history[0] + time_diff = self.history_times[-1] - self.history_times[0] + sent_rate = sent_diff / time_diff + recv_rate = recv_diff / time_diff + self.highest_sent_rate = max(self.highest_sent_rate, sent_rate) + self.highest_recv_rate = max(self.highest_recv_rate, recv_rate) + sent_percent = min(1.0, sent_rate / self.highest_sent_rate) + recv_percent = min(1.0, recv_rate / self.highest_recv_rate) + return sent_percent, recv_percent + except Exception as e: + print(f"Error in NetworkMonitor.get(): {e}") + return 0, 0 + +class CPUMonitor: + def __init__(self, hysterisis_time = 10): + self.cpu_count = psutil.cpu_count() // 2 # 2 logical cores per physical core + self.cpu_usage_history = [[] for _ in range(self.cpu_count)] + self.history_times = [] + self.max_history_size = hysterisis_time + + def get(self): + try: + cpu_usage = psutil.cpu_percent(percpu=True) + for i in range(self.cpu_count): + useage = 2 * max(cpu_usage[2*i], cpu_usage[2*i+1]) # Combine logical cores + if useage > 100: + useage = 100 + self.cpu_usage_history[i].append(useage / 100.0) + self.history_times.append(time.time()) + if len(self.cpu_usage_history[0]) > self.max_history_size: + for i in range(self.cpu_count): + self.cpu_usage_history[i] = self.cpu_usage_history[i][-self.max_history_size:] + self.history_times = self.history_times[-self.max_history_size:] + cpu_percentages = [sum(core_history) / self.max_history_size for core_history in self.cpu_usage_history] + # Somehow cpu_percentages can have values greater than 1 so we clamp them + return cpu_percentages + except Exception as e: + print(f"Error in CPUMonitor.get(): {e}") + return [0] * self.cpu_count + +class MemoryMonitor: + @staticmethod + def get(): + return psutil.virtual_memory().percent / 100.0 + + +class BatteryMonitor: + @staticmethod + def get(): + battery = psutil.sensors_battery() + if battery is not None: + battery_percentage = battery.percent / 100.0 + if os.name == "nt": + battery_plugged = battery.power_plugged + else: + bat_status = open('/sys/class/power_supply/BAT1/status', 'r').read().strip() + battery_plugged = (bat_status != 'Discharging') + return battery_percentage, battery_plugged + +def get_monitor_brightness(): + try: + if os.name == 'nt': + return wmi.WMI(namespace='wmi').WmiMonitorBrightness()[0].CurrentBrightness / 100.0 + else: + try: # First try the dGPU brightness + return int(open('/sys/class/backlight/amdgpu_bl2/brightness', 'r').read()) / 255.0 + except: # If that doesn't work, try the iGPU brightness + return int(open('/sys/class/backlight/amdgpu_bl1/brightness', 'r').read()) / 255.0 + except Exception as e: + return 1.0 + +if __name__ == "__main__": + print(get_monitor_brightness()) \ No newline at end of file diff --git a/patterns.py b/patterns.py new file mode 100644 index 0000000..40a3106 --- /dev/null +++ b/patterns.py @@ -0,0 +1,163 @@ +import numpy as np +import re +import os +import sys +import importlib + + +lightning_bolt_bot = np.array( [[0,0,0,0,0,0,0], # 0 + [0,0,0,0,0,1,0], # 1 + [0,0,0,0,1,1,0], # 2 + [0,0,0,1,1,0,0], # 3 + [0,0,1,1,1,0,0], # 4 + [0,1,1,1,0,0,0], # 5 + [0,1,1,1,1,1,0], # 6 + [0,0,0,1,1,1,0], # 7 + [0,0,1,1,1,0,0], # 8 + [0,0,1,1,0,0,0], # 9 + [0,1,1,0,0,0,0], #10 + [0,1,0,0,0,0,0], #11 + [0,0,0,0,0,0,0]],#12 + dtype=bool).T + +# Cut out bottom row because top segment is one row shorter +lightning_bolt_top = np.array( [[0,0,0,0,0,0,0], # 0 + [0,0,0,0,0,1,0], # 1 + [0,0,0,0,1,1,0], # 2 + [0,0,0,1,1,0,0], # 3 + [0,0,1,1,1,0,0], # 4 + [0,1,1,1,0,0,0], # 5 + [0,1,1,1,1,1,0], # 6 + [0,0,0,1,1,1,0], # 7 + [0,0,1,1,1,0,0], # 8 + [0,0,1,1,0,0,0], # 9 + [0,1,1,0,0,0,0], #10 + [0,1,0,0,0,0,0]],#11 + dtype=bool).T + +# This table represents the 3x3 grid of LEDs to be drawn for each fill ratio +lookup_table = np.array( + [ + [ + [0, 0, 0], + [0, 0, 0], + [0, 0, 0] + ], + [ + [0, 0, 0], + [0, 1, 0], + [0, 0, 0] + ], + [ + [0, 1, 0], + [0, 1, 0], + [0, 0, 0] + ], + [ + [0, 1, 1], + [0, 1, 0], + [0, 0, 0] + ], + [ + [0, 1, 1], + [0, 1, 1], + [0, 0, 0] + ], + [ + [0, 1, 1], + [0, 1, 1], + [0, 0, 1] + ], + [ + [0, 1, 1], + [0, 1, 1], + [0, 1, 1] + ], + [ + [0, 1, 1], + [0, 1, 1], + [1, 1, 1] + ], + [ + [0, 1, 1], + [1, 1, 1], + [1, 1, 1] + ], + [ + [1, 1, 1], + [1, 1, 1], + [1, 1, 1] + ] + ] +) + +id_patterns = { + "cpu": np.array([ + [0, 0, 1, 1, 1, 0, 0], + [0, 0, 1, 0, 0, 0, 0], + [0, 0, 1, 0, 0, 0, 0], + [0, 0, 1, 1, 1, 0, 0], + [0, 0, 0, 0, 0, 0, 0], + [0, 0, 1, 1, 0, 0, 0], + [0, 0, 1, 0, 1, 0, 0], + [0, 0, 1, 1, 0, 0, 0], + [0, 0, 1, 0, 0, 0, 0], + [0, 0, 1, 0, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0], + [0, 0, 1, 0, 1, 0, 0], + [0, 0, 1, 0, 1, 0, 0], + [0, 0, 1, 0, 1, 0, 0], + [0, 0, 1, 1, 1, 0, 0] + ]).T, + "mem-bat": np.array([ + [0, 1, 1, 0, 1, 1, 0], + [0, 1, 0, 1, 0, 1, 0], + [0, 1, 0, 0, 0, 1, 0], + [0, 1, 0, 0, 0, 1, 0], + [0, 0, 0, 0, 0, 0, 0], + [0, 0, 1, 1, 0, 0, 0], + [0, 0, 1, 0, 1, 0, 0], + [0, 0, 1, 1, 0, 0, 0], + [0, 0, 1, 0, 1, 0, 0], + [0, 0, 1, 1, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0], + [0, 0, 1, 1, 1, 0, 0], + [0, 0, 0, 1, 0, 0, 0], + [0, 0, 0, 1, 0, 0, 0], + [0, 0, 0, 1, 0, 0, 0], + ]).T, + "disk": np.array([ + [0, 0, 1, 1, 0, 0, 0], + [0, 0, 1, 0, 1, 0, 0], + [0, 0, 1, 0, 1, 0, 0], + [0, 0, 1, 1, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0], + [0, 0, 1, 1, 1, 0, 0], + [0, 0, 1, 0, 0, 0, 0], + [0, 0, 1, 1, 1, 0, 0], + [0, 0, 0, 0, 1, 0, 0], + [0, 0, 1, 1, 1, 0, 0], + [0, 0, 0, 0, 0, 0, 0], + [0, 0, 1, 0, 0, 0, 0], + [0, 0, 1, 0, 1, 0, 0], + [0, 0, 1, 1, 0, 0, 0], + [0, 0, 1, 0, 1, 0, 0], + ]).T, + "net": np.array([ + [0, 0, 1, 0, 0, 1, 0], + [0, 0, 1, 1, 0, 1, 0], + [0, 0, 1, 0, 1, 1, 0], + [0, 0, 1, 0, 0, 1, 0], + [0, 0, 0, 0, 0, 0, 0], + [0, 0, 1, 1, 1, 0, 0], + [0, 0, 1, 0, 0, 0, 0], + [0, 0, 1, 1, 0, 0, 0], + [0, 0, 1, 0, 0, 0, 0], + [0, 0, 1, 1, 1, 0, 0], + [0, 0, 0, 0, 0, 0, 0], + [0, 0, 1, 1, 1, 0, 0], + [0, 0, 0, 1, 0, 0, 0], + [0, 0, 0, 1, 0, 0, 0], + [0, 0, 0, 1, 0, 0, 0], + ]).T +} \ No newline at end of file diff --git a/plugins/__init__.py b/plugins/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/plugins/temp_fan_plugin.py b/plugins/temp_fan_plugin.py new file mode 100644 index 0000000..d985cf7 --- /dev/null +++ b/plugins/temp_fan_plugin.py @@ -0,0 +1,124 @@ +from statistics import mean +import psutil +import numpy as np + +# Reference for fractional measure of sensor temps (in degrees Celcius) +TEMP_REF = 120 + +# Reference for fractional measure of fan speeds (in rpm) +MAX_FAN_SPEED = 6_000 + +#### Implement monitor functions #### + +class TemperatureMonitor: + @staticmethod + def get(): + temps = [] + sensors = psutil.sensors_temperatures() + for _, entries in sensors.items(): + temps.append(mean([entry.current for entry in entries if entry.current > 0])) + # We can handle up to eight temps on the matrix display + _temps = list(map(lambda x: x / TEMP_REF, temps)) + return list(map(lambda x: x / TEMP_REF, temps))[:8] + +class FanSpeedMonitor: + @staticmethod + def get(): + fans = psutil.sensors_fans() + speeds = [] + for _, entries in fans.items(): + for entry in entries: + speeds.append(entry.current) + # We can handle up to two fan speeds on the matrix display + return list(map(lambda x: x / MAX_FAN_SPEED, speeds))[:2] + +temperature_monitor = TemperatureMonitor() +fan_speed_monitor = FanSpeedMonitor() + +#### Implement high-level drawing functions to be called by app functions below #### + +import drawing +draw_app = getattr(drawing, 'draw_app') + +def draw_temps(arg, grid, foreground_value, idx): + temp_values = temperature_monitor.get() + draw_app(arg, grid, temp_values, foreground_value, idx) + +def draw_fans(arg, grid, foreground_value, idx): + fan_speeds = fan_speed_monitor.get() + draw_app(arg, grid, fan_speeds[0], foreground_value, bar_x_offset=1, y=idx) + draw_app(arg, grid, fan_speeds[1], foreground_value, bar_x_offset=5, y=idx) + + +draw_spiral_vals = getattr(drawing, 'draw_spiral_vals') +draw_8_x_8_grid = getattr(drawing, 'draw_8_x_8_grid') +draw_bar = getattr(drawing, 'draw_bar') +draw_2_x_1_horiz_grid = getattr(drawing, 'draw_2_x_1_horiz_grid') + +#### Implement low-level drawing functions #### +# These functions will be dynamically imported by drawing.py and led_system_monitor.py + +metrics_funcs = { + "temp": { + "fn": draw_spiral_vals, + "border": draw_8_x_8_grid + }, + "fan": { + "fn": draw_bar, + "border": draw_2_x_1_horiz_grid + } +} + +# Implement app functions that call your high-level draw functions +# These functions will be dynamically imported by led_system_monitor.py + +app_funcs = [ + { + "name": "temp", + "fn": draw_temps + }, + { + "name": "fan", + "fn": draw_fans + } +] + +# Provide id patterns that identify your apps +# These items will be dynamically imported by drawing.py + +id_patterns = { + "temp": np.array([ + [0, 0, 0, 0, 0, 0, 0], + [0, 0, 1, 1, 1, 0, 0], + [0, 0, 0, 1, 0, 0, 0], + [0, 0, 0, 1, 0, 0, 0], + [0, 0, 0, 1, 0, 0, 0], + [0, 1, 1, 0, 1, 1, 0], + [0, 1, 0, 1, 0, 1, 0], + [0, 1, 0, 0, 0, 1, 0], + [0, 1, 0, 0, 0, 1, 0], + [0, 0, 1, 1, 1, 0, 0], + [0, 0, 1, 0, 0, 1, 0], + [0, 0, 1, 1, 1, 0, 0], + [0, 0, 1, 0, 0, 0, 0], + [0, 0, 1, 0, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0], + ]).T, + "fan": np.array([ + [0, 0, 1, 1, 1, 1, 0], + [0, 0, 1, 0, 0, 0, 0], + [0, 0, 1, 1, 1, 0, 0], + [0, 0, 1, 0, 0, 0, 0], + [0, 0, 1, 0, 0, 0, 0], + [0, 0, 0, 1, 1, 0, 0], + [0, 0, 1, 0, 0, 1, 0], + [0, 0, 1, 1, 1, 1, 0], + [0, 0, 1, 0, 0, 1, 0], + [0, 0, 0, 0, 0, 0, 0], + [0, 0, 1, 0, 0, 1, 0], + [0, 0, 1, 1, 0, 1, 0], + [0, 0, 1, 0, 1, 1, 0], + [0, 0, 1, 0, 0, 1, 0], + [0, 0, 0, 0, 0, 0, 0], + ]).T +} \ No newline at end of file diff --git a/psutil-sensors.py b/psutil-sensors.py new file mode 100644 index 0000000..458612d --- /dev/null +++ b/psutil-sensors.py @@ -0,0 +1,23 @@ +import psutil +from statistics import mean + +print("Temperature sensors") +temps = psutil.sensors_temperatures() +for name, entries in temps.items(): + print(f"Temp sensor {name}") + for entry in entries: + print(f"\tCurrent: {entry.current}\tHigh: {entry.high}\tCritical: {entry.critical}") + currAvg = mean([entry.current for entry in entries if entry.current > 0]) + print(f"Avg: {currAvg}") + print() + +print("Fan speeds") + +fans = psutil.sensors_fans() +for name, entries in fans.items(): + print(f"Fan sensor {name}") + for entry in entries: + print(f"\tCurrent: {entry.current}") + +battery = psutil.sensors_battery() +print(battery) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100755 index 0000000..3e54dbe --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +pyserial +numpy +psutils +psutil +evdev \ No newline at end of file diff --git a/run.bat b/run.bat new file mode 100644 index 0000000..7374ad7 --- /dev/null +++ b/run.bat @@ -0,0 +1,3 @@ +py -3 -m pip install pyserial numpy wmi psutil + +py -3 led_system_monitor.py \ No newline at end of file diff --git a/run.sh b/run.sh new file mode 100755 index 0000000..3c4cf5e --- /dev/null +++ b/run.sh @@ -0,0 +1,3 @@ +sudo apt install -y python3-numpy python3-psutil python3-serial python3-evdev + +python3 ./led_system_monitor.py