From da718a6fc40409697cabc205b2d9a9351f9047d2 Mon Sep 17 00:00:00 2001 From: = <=> Date: Mon, 5 Aug 2024 20:15:56 -0400 Subject: [PATCH] Finished? --- drawing.py | 55 +++++++++++++++---------- led_system_monitor.py | 96 ++++++++++++++++++++++++++----------------- monitors.py | 65 ++++++++--------------------- 3 files changed, 108 insertions(+), 108 deletions(-) diff --git a/drawing.py b/drawing.py index c5cfe2b..493518e 100644 --- a/drawing.py +++ b/drawing.py @@ -1,11 +1,13 @@ +# Built In Dependencies import time import math +# Internal Dependencies +from commands import Commands, send_command + +# External Dependencies 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( @@ -113,7 +115,7 @@ def draw_battery(grid, battery_ratio, battery_plugged, fill_value, battery_low_t return for i in range(7): pixels_col = pixels_base - if i <= remainder: + if i < remainder: pixels_col += 1 grid[i+1,33-pixels_col:33] = fill_value 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] -def draw_borders(grid, border_value): +def draw_borders_left(grid, border_value): # Fill in the borders # Cpu vertical partitions grid[4, :16] = border_value @@ -140,23 +142,34 @@ def draw_borders(grid, border_value): grid[8, :] = border_value # Right 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): 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 - 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) \ No newline at end of file + send_command(s, Commands.FlushCols) \ No newline at end of file diff --git a/led_system_monitor.py b/led_system_monitor.py index 87dfec5..ac37720 100644 --- a/led_system_monitor.py +++ b/led_system_monitor.py @@ -1,13 +1,10 @@ # Built In Dependencies -import sys -import glob import time import queue # Internal Dependencies -from commands import Commands, send_command -from drawing import draw_cpu, draw_memory, draw_battery, draw_borders, draw_to_LEDs -from monitors import CPUMonitorThread, MemoryMonitorThread, BatteryMonitorThread +from drawing import draw_cpu, draw_memory, draw_battery, draw_borders_left, draw_to_LEDs, draw_bar, draw_borders_right +from monitors import CPUMonitorThread, MemoryMonitorThread, BatteryMonitorThread, DiskMonitorThread, NetworkMonitorThread # External Dependencies try: @@ -42,6 +39,12 @@ if __name__ == "__main__": # Left LED Matrix location: "1-4.2" # 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_monitor = CPUMonitorThread(cpu_queue) cpu_monitor.start() @@ -54,19 +57,34 @@ if __name__ == "__main__": battery_monitor = BatteryMonitorThread(battery_queue) 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_memory_values = memory_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: 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(): last_cpu_values = cpu_queue.get() if not memory_queue.empty(): @@ -74,37 +92,39 @@ if __name__ == "__main__": if not battery_queue.empty(): 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) draw_cpu(grid, last_cpu_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_borders(grid, background_value) - draw_to_LEDs(s, grid) + draw_borders_left(grid, background_value) + 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: 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(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) \ No newline at end of file + s1 = init_device("1-4.2") + s2 = init_device("1-3.3") + time.sleep(0.05) \ No newline at end of file diff --git a/monitors.py b/monitors.py index 2dba5d0..b83f83d 100644 --- a/monitors.py +++ b/monitors.py @@ -1,12 +1,11 @@ +# Built In Dependencies 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): + def __init__(self, output_queue, hysterisis_time = 3, update_interval = 0.5): super().__init__() self.daemon = True self.read_usage_history = [] @@ -23,6 +22,8 @@ class DiskMonitorThread(threading.Thread): try: if not self.output_queue.full(): disk_io = psutil.disk_io_counters() + else: + print("Disk monitor queue is full") read_usage = disk_io.read_bytes write_usage = disk_io.write_bytes self.read_usage_history.append(read_usage) @@ -49,7 +50,7 @@ class DiskMonitorThread(threading.Thread): time.sleep(self.update_interval) 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__() self.daemon = True self.sent_usage_history = [] @@ -66,6 +67,8 @@ class NetworkMonitorThread(threading.Thread): try: if not self.output_queue.full(): net_io = psutil.net_io_counters() + else: + print("Network monitor queue is full") sent_usage = net_io.bytes_sent recv_usage = net_io.bytes_recv self.sent_usage_history.append(sent_usage) @@ -92,7 +95,7 @@ class NetworkMonitorThread(threading.Thread): time.sleep(self.update_interval) 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__() self.daemon = True self.cpu_count = psutil.cpu_count() // 2 # 2 logical cores per physical core @@ -107,6 +110,8 @@ class CPUMonitorThread(threading.Thread): try: if not self.output_queue.full(): cpu_usage = psutil.cpu_percent(percpu=True) + else: + print("CPU monitor queue is full") 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: @@ -125,7 +130,7 @@ class CPUMonitorThread(threading.Thread): time.sleep(self.update_interval) 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__() self.daemon = True self.memory_usage_history = [] @@ -139,6 +144,8 @@ class MemoryMonitorThread(threading.Thread): try: if not self.output_queue.full(): memory_usage = psutil.virtual_memory().percent / 100.0 + else: + print("Memory monitor queue is full") self.memory_usage_history.append(memory_usage) self.history_times.append(time.time()) 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_plugged = battery.power_plugged self.output_queue.put((battery_percentage, battery_plugged)) + else: + print("Battery monitor queue is full") except Exception as e: print(f"Error in BatteryMonitorThread: {e}") - 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) \ No newline at end of file + time.sleep(self.update_interval) \ No newline at end of file