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

Added auto brightness adjustment

Browse Source
This commit is contained in:
= 2024-08-01 10:23:52 -04:00
parent 8537b30b49
commit 383cb15866
4 changed files with 93 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored Normal file
View File

@ -0,0 +1 @@
*.pyc

27
led_system_monitor.py
View File

@ -10,7 +10,18 @@ from drawing import make_cpu_grid, draw_to_LEDs
from monitors import CPUMonitorThread
# External Dependencies
import serial # pyserial
try:
import serial # pyserial
import numpy as np # This is used in a module and we import it here to fetch it if needed
import screen_brightness_control as sbc
except ImportError:
import pip
for dependency in ["numpy", "pyserial", "screen-brightness-control"]:
pip.main(['install', '--user', dependency])
import serial
import screen_brightness_control as sbc
# print(sbc.get_brightness())
def get_ports():
"""Returns a list of all available serial ports on the system.
@ -47,18 +58,26 @@ if __name__ == "__main__":
# print(get_ports())
port = "COM3"
cpu_queue = queue.Queue()
cpu_queue = queue.Queue(2)
cpu_monitor = CPUMonitorThread(cpu_queue)
cpu_monitor.start()
s = serial.Serial(port, 115200)
min_background_brightness = 10
max_background_brightness = 30
min_foreground_brightness = 30
max_foreground_brightness = 120
while True:
if not cpu_queue.empty():
cpu_values = cpu_queue.get()
grid = make_cpu_grid(cpu_values, 10, 30)
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 = make_cpu_grid(cpu_values, background_value, foreground_value)
draw_to_LEDs(s, grid)
time.sleep(0.1)
time.sleep(0.05)

132
monitors.py
View File

@ -20,28 +20,29 @@ class DiskMonitorThread(threading.Thread):
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 not self.output_queue.full():
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))
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)
@ -60,28 +61,29 @@ class NetworkMonitorThread(threading.Thread):
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 not self.output_queue.full():
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))
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)
@ -98,20 +100,21 @@ class CPUMonitorThread(threading.Thread):
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:
if not self.output_queue.full():
cpu_usage = psutil.cpu_percent(percpu=True)
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)
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):
@ -126,15 +129,16 @@ class MemoryMonitorThread(threading.Thread):
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)
if not self.output_queue.full():
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)

1
start.bat Normal file
View File

@ -0,0 +1 @@
py -3 led_system_monitor.py