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

Finished?

Browse Source
This commit is contained in:
= 2024-08-05 20:15:56 -04:00
parent 78c05943d7
commit da718a6fc4
3 changed files with 108 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
drawing.py
View File

@ -1,11 +1,13 @@
# Built In Dependencies
import time import time
import math import math
# Internal Dependencies
from commands import Commands, send_command
# External Dependencies
import numpy as np 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 # This table represents the 3x3 grid of LEDs to be drawn for each fill ratio
lookup_table = np.array( lookup_table = np.array(
@ -113,7 +115,7 @@ def draw_battery(grid, battery_ratio, battery_plugged, fill_value, battery_low_t
return return
for i in range(7): for i in range(7):
pixels_col = pixels_base pixels_col = pixels_base
if i <= remainder: if i < remainder:
pixels_col += 1 pixels_col += 1
grid[i+1,33-pixels_col:33] = fill_value grid[i+1,33-pixels_col:33] = fill_value
if battery_plugged: if battery_plugged:
@ -123,7 +125,7 @@ def draw_battery(grid, battery_ratio, battery_plugged, fill_value, battery_low_t
grid[1:8,20:33][indices] = -grid[1:8,20:33][indices] grid[1:8,20:33][indices] = -grid[1:8,20:33][indices]
def draw_borders(grid, border_value): def draw_borders_left(grid, border_value):
# Fill in the borders # Fill in the borders
# Cpu vertical partitions # Cpu vertical partitions
grid[4, :16] = border_value grid[4, :16] = border_value
@ -140,23 +142,34 @@ def draw_borders(grid, border_value):
grid[8, :] = border_value # Right grid[8, :] = border_value # Right
grid[:, 33] = border_value # Bottom grid[:, 33] = border_value # Bottom
def draw_borders_right(grid, border_value):
# Fill in the borders
# Middle Partition borders
grid[:, 16] = border_value
grid[4, :] = border_value
# Outer Edge borders
grid[:, 0] = border_value # Top
grid[0, :] = border_value # Left
grid[8, :] = border_value # Right
grid[:, 33] = border_value # Bottom
def draw_bar(grid, bar_ratio, bar_value, bar_x_offset = 1,draw_at_bottom = True):
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 draw_at_bottom:
grid[bar_x_offset+i,33-pixels_col:33] = bar_value
else:
grid[bar_x_offset+i,1:1+pixels_col] = bar_value
def draw_to_LEDs(s, grid): def draw_to_LEDs(s, grid):
for i in range(grid.shape[0]): for i in range(grid.shape[0]):
params = bytearray([i]) + bytearray(grid[i, :].tolist()) params = bytearray([i]) + bytearray(grid[i, :].tolist())
send_command(s, Commands.StageCol, parameters=params) send_command(s, Commands.StageCol, parameters=params)
send_command(s, Commands.FlushCols) send_command(s, Commands.FlushCols)
if __name__ == "__main__":
# LED array is 34x9, and is indexed left to right top to bottom
port = "COM3"
with serial.Serial(port, 115200) as s:
while True:
grid = np.zeros((9,34), dtype = int)
draw_cpu(grid, [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8], 30)
draw_memory(grid, 0.3, 30)
draw_battery(grid, 0.75, True, 30)
draw_borders(grid, 10)
print(grid.T)
draw_to_LEDs(s, grid)
time.sleep(0.05)

94
led_system_monitor.py
View File

@ -1,13 +1,10 @@
# Built In Dependencies # Built In Dependencies
import sys
import glob
import time import time
import queue import queue
# Internal Dependencies # Internal Dependencies
from commands import Commands, send_command from drawing import draw_cpu, draw_memory, draw_battery, draw_borders_left, draw_to_LEDs, draw_bar, draw_borders_right
from drawing import draw_cpu, draw_memory, draw_battery, draw_borders, draw_to_LEDs from monitors import CPUMonitorThread, MemoryMonitorThread, BatteryMonitorThread, DiskMonitorThread, NetworkMonitorThread
from monitors import CPUMonitorThread, MemoryMonitorThread, BatteryMonitorThread
# External Dependencies # External Dependencies
try: try:
@ -42,6 +39,12 @@ if __name__ == "__main__":
# Left LED Matrix location: "1-4.2" # Left LED Matrix location: "1-4.2"
# Right LED Matrix location: "1-3.3" # Right LED Matrix location: "1-3.3"
# Set up monitors and serial for left LED Matrix
min_background_brightness = 8
max_background_brightness = 35
min_foreground_brightness = 30
max_foreground_brightness = 160
cpu_queue = queue.Queue(2) cpu_queue = queue.Queue(2)
cpu_monitor = CPUMonitorThread(cpu_queue) cpu_monitor = CPUMonitorThread(cpu_queue)
cpu_monitor.start() cpu_monitor.start()
@ -54,19 +57,34 @@ if __name__ == "__main__":
battery_monitor = BatteryMonitorThread(battery_queue) battery_monitor = BatteryMonitorThread(battery_queue)
battery_monitor.start() battery_monitor.start()
min_background_brightness = 8
max_background_brightness = 20
min_foreground_brightness = 30
max_foreground_brightness = 110
last_cpu_values = cpu_queue.get() last_cpu_values = cpu_queue.get()
last_memory_values = memory_queue.get() last_memory_values = memory_queue.get()
last_battery_values = battery_queue.get() last_battery_values = battery_queue.get()
s = init_device() s1 = init_device("1-4.2")
# Set up monitors and serial for right LED Matrix
disk_queue = queue.Queue(2)
disk_monitor = DiskMonitorThread(disk_queue)
disk_monitor.start()
network_queue = queue.Queue(2)
network_monitor = NetworkMonitorThread(network_queue)
network_monitor.start()
last_disk_read, last_disk_write = disk_queue.get()
last_network_upload, last_network_download = network_queue.get()
s2 = init_device("1-3.3")
while True: while True:
try: try:
screen_brightness = sbc.get_brightness()[0]
background_value = int(screen_brightness / 100 * (max_background_brightness - min_background_brightness) + min_background_brightness)
foreground_value = int(screen_brightness / 100 * (max_foreground_brightness - min_foreground_brightness) + min_foreground_brightness)
# Draw to left LED Matrix
if not cpu_queue.empty(): if not cpu_queue.empty():
last_cpu_values = cpu_queue.get() last_cpu_values = cpu_queue.get()
if not memory_queue.empty(): if not memory_queue.empty():
@ -74,37 +92,39 @@ if __name__ == "__main__":
if not battery_queue.empty(): if not battery_queue.empty():
last_battery_values = battery_queue.get() last_battery_values = battery_queue.get()
screen_brightness = sbc.get_brightness()[0]
background_value = int(screen_brightness / 100 * (max_background_brightness - min_background_brightness) + min_background_brightness)
foreground_value = int(screen_brightness / 100 * (max_foreground_brightness - min_foreground_brightness) + min_foreground_brightness)
grid = np.zeros((9,34), dtype = int) grid = np.zeros((9,34), dtype = int)
draw_cpu(grid, last_cpu_values, foreground_value) draw_cpu(grid, last_cpu_values, foreground_value)
draw_memory(grid, last_memory_values, foreground_value) draw_memory(grid, last_memory_values, foreground_value)
draw_battery(grid, last_battery_values[0], last_battery_values[1], foreground_value) draw_battery(grid, last_battery_values[0], last_battery_values[1], foreground_value)
draw_borders(grid, background_value) draw_borders_left(grid, background_value)
draw_to_LEDs(s, grid) draw_to_LEDs(s1, grid)
# Draw to right LED Matrix
if not disk_queue.empty():
last_disk_read, last_disk_write = disk_queue.get()
if not network_queue.empty():
last_network_upload, last_network_download = network_queue.get()
grid = np.zeros((9,34), dtype = int)
draw_bar(grid, last_disk_read, foreground_value, bar_x_offset=1, draw_at_bottom=False) # Read
draw_bar(grid, last_disk_write, foreground_value, bar_x_offset=1, draw_at_bottom=True) # Write
draw_bar(grid, last_network_upload, foreground_value, bar_x_offset=5, draw_at_bottom=False) # Upload
draw_bar(grid, last_network_download, foreground_value, bar_x_offset=5, draw_at_bottom=True) # Download
draw_borders_right(grid, background_value)
draw_to_LEDs(s2, grid)
except Exception as e: except Exception as e:
print(f"Error in main loop: {e}") print(f"Error in main loop: {e}")
s = init_device() try:
del s1
except:
pass
try:
del s2
except:
pass
time.sleep(1.0) time.sleep(1.0)
s1 = init_device("1-4.2")
s2 = init_device("1-3.3")
time.sleep(0.05) time.sleep(0.05)
# # 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)

63
monitors.py
View File

@ -1,12 +1,11 @@
# Built In Dependencies
import time import time
import psutil import psutil
import threading import threading
import time
import queue
class DiskMonitorThread(threading.Thread): class DiskMonitorThread(threading.Thread):
def __init__(self, output_queue, hysterisis_time = 5, update_interval = 0.25): def __init__(self, output_queue, hysterisis_time = 3, update_interval = 0.5):
super().__init__() super().__init__()
self.daemon = True self.daemon = True
self.read_usage_history = [] self.read_usage_history = []
@ -23,6 +22,8 @@ class DiskMonitorThread(threading.Thread):
try: try:
if not self.output_queue.full(): if not self.output_queue.full():
disk_io = psutil.disk_io_counters() disk_io = psutil.disk_io_counters()
else:
print("Disk monitor queue is full")
read_usage = disk_io.read_bytes read_usage = disk_io.read_bytes
write_usage = disk_io.write_bytes write_usage = disk_io.write_bytes
self.read_usage_history.append(read_usage) self.read_usage_history.append(read_usage)
@ -49,7 +50,7 @@ class DiskMonitorThread(threading.Thread):
time.sleep(self.update_interval) time.sleep(self.update_interval)
class NetworkMonitorThread(threading.Thread): class NetworkMonitorThread(threading.Thread):
def __init__(self, output_queue, hysterisis_time = 5, update_interval = 0.25): def __init__(self, output_queue, hysterisis_time = 3, update_interval = 0.5):
super().__init__() super().__init__()
self.daemon = True self.daemon = True
self.sent_usage_history = [] self.sent_usage_history = []
@ -66,6 +67,8 @@ class NetworkMonitorThread(threading.Thread):
try: try:
if not self.output_queue.full(): if not self.output_queue.full():
net_io = psutil.net_io_counters() net_io = psutil.net_io_counters()
else:
print("Network monitor queue is full")
sent_usage = net_io.bytes_sent sent_usage = net_io.bytes_sent
recv_usage = net_io.bytes_recv recv_usage = net_io.bytes_recv
self.sent_usage_history.append(sent_usage) self.sent_usage_history.append(sent_usage)
@ -92,7 +95,7 @@ class NetworkMonitorThread(threading.Thread):
time.sleep(self.update_interval) time.sleep(self.update_interval)
class CPUMonitorThread(threading.Thread): class CPUMonitorThread(threading.Thread):
def __init__(self, output_queue, hysterisis_time = 5, update_interval = 0.25): def __init__(self, output_queue, hysterisis_time = 3, update_interval = 0.5):
super().__init__() super().__init__()
self.daemon = True self.daemon = True
self.cpu_count = psutil.cpu_count() // 2 # 2 logical cores per physical core self.cpu_count = psutil.cpu_count() // 2 # 2 logical cores per physical core
@ -107,6 +110,8 @@ class CPUMonitorThread(threading.Thread):
try: try:
if not self.output_queue.full(): if not self.output_queue.full():
cpu_usage = psutil.cpu_percent(percpu=True) cpu_usage = psutil.cpu_percent(percpu=True)
else:
print("CPU monitor queue is full")
for i in range(self.cpu_count): for i in range(self.cpu_count):
useage = 2 * max(cpu_usage[2*i], cpu_usage[2*i+1]) # Combine logical cores useage = 2 * max(cpu_usage[2*i], cpu_usage[2*i+1]) # Combine logical cores
if useage > 100: if useage > 100:
@ -125,7 +130,7 @@ class CPUMonitorThread(threading.Thread):
time.sleep(self.update_interval) time.sleep(self.update_interval)
class MemoryMonitorThread(threading.Thread): class MemoryMonitorThread(threading.Thread):
def __init__(self, output_queue, hysterisis_time = 5, update_interval = 0.25): def __init__(self, output_queue, hysterisis_time = 5, update_interval = 1.0):
super().__init__() super().__init__()
self.daemon = True self.daemon = True
self.memory_usage_history = [] self.memory_usage_history = []
@ -139,6 +144,8 @@ class MemoryMonitorThread(threading.Thread):
try: try:
if not self.output_queue.full(): if not self.output_queue.full():
memory_usage = psutil.virtual_memory().percent / 100.0 memory_usage = psutil.virtual_memory().percent / 100.0
else:
print("Memory monitor queue is full")
self.memory_usage_history.append(memory_usage) self.memory_usage_history.append(memory_usage)
self.history_times.append(time.time()) self.history_times.append(time.time())
if len(self.memory_usage_history) > self.max_history_size: if len(self.memory_usage_history) > self.max_history_size:
@ -167,48 +174,8 @@ class BatteryMonitorThread(threading.Thread):
battery_percentage = battery.percent / 100.0 battery_percentage = battery.percent / 100.0
battery_plugged = battery.power_plugged battery_plugged = battery.power_plugged
self.output_queue.put((battery_percentage, battery_plugged)) self.output_queue.put((battery_percentage, battery_plugged))
else:
print("Battery monitor queue is full")
except Exception as e: except Exception as e:
print(f"Error in BatteryMonitorThread: {e}") print(f"Error in BatteryMonitorThread: {e}")
time.sleep(self.update_interval) time.sleep(self.update_interval)
if __name__ == "__main__":
disk_queue = queue.Queue()
network_queue = queue.Queue()
cpu_queue = queue.Queue()
memory_queue = queue.Queue()
battery_queue = queue.Queue()
disk_monitor = DiskMonitorThread(disk_queue)
network_monitor = NetworkMonitorThread(network_queue)
cpu_monitor = CPUMonitorThread(cpu_queue)
memory_monitor = MemoryMonitorThread(memory_queue)
battery_monitor = BatteryMonitorThread(battery_queue)
disk_monitor.start()
network_monitor.start()
cpu_monitor.start()
memory_monitor.start()
battery_monitor.start()
while True:
if not disk_queue.empty():
read_percent, write_percent = disk_queue.get()
print(f"Disk: Read {read_percent:.2%}, Write {write_percent:.2%}")
if not network_queue.empty():
sent_percent, recv_percent = network_queue.get()
print(f"Network: Sent {sent_percent:.2%}, Received {recv_percent:.2%}")
if not cpu_queue.empty():
cpu_percentages = cpu_queue.get()
print(f"CPU: {cpu_percentages}")
if not memory_queue.empty():
memory_usage = memory_queue.get()
print(f"Memory: {memory_usage:.2%}")
if not battery_queue.empty():
battery_percentage, battery_plugged = battery_queue.get()
print(f"Battery: {battery_percentage:.2%}, Plugged: {battery_plugged}")
time.sleep(0.5)