jeremy/FW_LED_System_Monitor
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Finished CPU monitor and drawing

Browse Source
This commit is contained in:
= 2024-08-01 02:47:31 -04:00
commit 8537b30b49
6 changed files with 409 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
.vscode/launch.json vendored Normal file
View File

@ -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
}
]
}

37
commands.py Normal file
View File

@ -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

101
drawing.py Normal file
View File

@ -0,0 +1,101 @@
import numpy as np
import serial
from commands import Commands, send_command
# 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]
]
]
)
# 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))
def make_cpu_grid(cpu_values, border_value, fill_value):
grid = np.zeros((9,34), dtype = int)
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, 1+row_number*4:4+row_number*4] = fill_grid * fill_value
# Fill in the borders
grid[0, :16] = border_value
grid[4, :16] = border_value
grid[8, :16] = border_value
grid[:, 0] = border_value
grid[:, 4] = border_value
grid[:, 8] = border_value
grid[:, 12] = border_value
grid[:, 16] = border_value
return grid
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)
if __name__ == "__main__":
# LED array is 34x9, and is indexed left to right top to bottom
grid = make_cpu_grid([0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8], 10, 30)
port = "COM3"
with serial.Serial(port, 115200) as s:
draw_to_LEDs(s, grid)

81
led_system_monitor.py Normal file
View File

@ -0,0 +1,81 @@
# Built In Dependencies
import sys
import glob
import time
import queue
# Internal Dependencies
from commands import Commands, send_command
from drawing import make_cpu_grid, draw_to_LEDs
from monitors import CPUMonitorThread
# External Dependencies
import serial # pyserial
def get_ports():
"""Returns a list of all available serial ports on the system.
Raises:
EnvironmentError: Will be returned if the platform is not Windows, Linux, Cygwin, or Darwin.
Returns:
[list(str)]: A list of valid serial ports on the system.
"""
if sys.platform.startswith('win'):
ports = ['COM%s' % (i+1) for i in range(256)]
elif sys.platform.startswith('linux') or sys.platform.startswith('cygwin'):
ports = reversed(glob.glob('/dev/ttyUSB*'))
elif sys.platform.startswith('darwin'):
ports = glob.glob('/dev/tty.*')
else:
raise EnvironmentError('Unsupported platform')
result = []
for port in ports:
try:
s = serial.Serial(port)
s.close()
result.append(port)
except (OSError, serial.SerialException):
pass
return result
if __name__ == "__main__":
# print(get_ports())
port = "COM3"
cpu_queue = queue.Queue()
cpu_monitor = CPUMonitorThread(cpu_queue)
cpu_monitor.start()
s = serial.Serial(port, 115200)
while True:
if not cpu_queue.empty():
cpu_values = cpu_queue.get()
grid = make_cpu_grid(cpu_values, 10, 30)
draw_to_LEDs(s, grid)
time.sleep(0.1)
# # print(send_command(port, Commands.Version, with_response=True))
# with serial.Serial(port, 115200) as s:
# for cval in range(16):
# for column_number in range(9):
# column_values = [cval] * 34
# params = bytearray([column_number]) + bytearray(column_values)
# send_command(s, Commands.StageCol, parameters=params)
# print(f"Flushing cval: {cval}")
# send_command(s, Commands.FlushCols)
# Columns are filled left to right top to bottom
# with serial.Serial(port, 115200) as s:
# column_number = 0
# column_values = [50] * 17 + [0] * 17
# params = bytearray([column_number]) + bytearray(column_values)
# send_command(s, Commands.StageCol, parameters=params)
# send_command(s, Commands.FlushCols)

BIN
ledmatrix_gui_windows.exe Normal file
View File

Binary file not shown.

174
monitors.py Normal file
View File

@ -0,0 +1,174 @@
import time
import psutil
import threading
import time
import queue
class DiskMonitorThread(threading.Thread):
def __init__(self, output_queue, hysterisis_time = 5, update_interval = 0.25):
super().__init__()
self.daemon = True
self.read_usage_history = []
self.write_usage_history = []
self.history_times = []
self.highest_read_rate = 0.00001
self.highest_write_rate = 0.00001
self.max_history_size = int(round(hysterisis_time / update_interval))
self.update_interval = update_interval
self.output_queue = output_queue
def run(self):
while True:
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:]
if len(self.read_usage_history) == 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)
self.output_queue.put((read_percent, write_percent))
time.sleep(self.update_interval)
class NetworkMonitorThread(threading.Thread):
def __init__(self, output_queue, hysterisis_time = 5, update_interval = 0.25):
super().__init__()
self.daemon = True
self.sent_usage_history = []
self.recv_usage_history = []
self.history_times = []
self.highest_sent_rate = 0.00001
self.highest_recv_rate = 0.00001
self.max_history_size = int(round(hysterisis_time / update_interval))
self.update_interval = update_interval
self.output_queue = output_queue
def run(self):
while True:
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:]
if len(self.sent_usage_history) == 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)
self.output_queue.put((sent_percent, recv_percent))
time.sleep(self.update_interval)
class CPUMonitorThread(threading.Thread):
def __init__(self, output_queue, hysterisis_time = 5, update_interval = 0.25):
super().__init__()
self.daemon = True
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 = int(round(hysterisis_time / update_interval))
self.update_interval = update_interval
self.output_queue = output_queue
def run(self):
while True:
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:]
if len(self.cpu_usage_history[0]) == self.max_history_size:
cpu_percentages = [sum(core_history) / self.max_history_size for core_history in self.cpu_usage_history]
self.output_queue.put(cpu_percentages)
time.sleep(self.update_interval)
class MemoryMonitorThread(threading.Thread):
def __init__(self, output_queue, hysterisis_time = 5, update_interval = 0.25):
super().__init__()
self.daemon = True
self.memory_usage_history = []
self.history_times = []
self.max_history_size = int(round(hysterisis_time / update_interval))
self.update_interval = update_interval
self.output_queue = output_queue
def run(self):
while True:
memory_usage = psutil.virtual_memory().percent / 100.0
self.memory_usage_history.append(memory_usage)
self.history_times.append(time.time())
if len(self.memory_usage_history) > self.max_history_size:
self.memory_usage_history = self.memory_usage_history[-self.max_history_size:]
self.history_times = self.history_times[-self.max_history_size:]
if len(self.memory_usage_history) == self.max_history_size:
avg_memory_usage = sum(self.memory_usage_history) / self.max_history_size
self.output_queue.put(avg_memory_usage)
time.sleep(self.update_interval)
if __name__ == "__main__":
disk_queue = queue.Queue()
network_queue = queue.Queue()
cpu_queue = queue.Queue()
memory_queue = queue.Queue()
disk_monitor = DiskMonitorThread(disk_queue)
network_monitor = NetworkMonitorThread(network_queue)
cpu_monitor = CPUMonitorThread(cpu_queue)
memory_monitor = MemoryMonitorThread(memory_queue)
disk_monitor.start()
network_monitor.start()
cpu_monitor.start()
memory_monitor.start()
while True:
if not disk_queue.empty():
read_percent, write_percent = disk_queue.get()
print(f"Disk Usage: Read {read_percent:.2%}, Write {write_percent:.2%}")
if not network_queue.empty():
sent_percent, recv_percent = network_queue.get()
print(f"Network Usage: Sent {sent_percent:.2%}, Received {recv_percent:.2%}")
if not cpu_queue.empty():
cpu_percentages = cpu_queue.get()
print(f"CPU Usage: {cpu_percentages}")
if not memory_queue.empty():
memory_usage = memory_queue.get()
print(f"Memory Usage: {memory_usage:.2%}")
time.sleep(0.5)