matt/cosmoserver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

web dashboard in single container

Browse Source
This commit is contained in:
Matt
2026-04-05 01:50:41 -07:00
parent a89703c420
commit c6007d9c33
26 changed files with 460 additions and 365 deletions
Download Patch File Download Diff File Expand all files Collapse all files

166
files/api/Components.py
View File

@ -13,6 +13,7 @@ import ipaddress
from typing import Dict, Any, List
# Import Cosmos Settings
from Cosmos_Settings import *
from Helpers import *
# Global Class Vars
global_max_length = 500
@ -158,6 +159,7 @@ class Component:
# iterate over all properties to process descriptor
def _process_properties(self):
for key, command in self._descriptor.get('properties', {}).items():
# if this is a component that can be multi, then it might need to return a list
return_string = True
if key in self.multi_metrics:
return_string = False
@ -168,6 +170,8 @@ class Component:
self._properties[key] = result
# helper function to parse command key
# this is for substituting variables in descriptor commands when there are multi things
# think needing to grep eth0 or grep eth1 to get the same metric for a different component
def _parse_command(self, key: str, command: str | list[str], return_string = True):
result_command = command
log_data(log_output = f"_parse_command - {command}", log_level = "debug_output")
@ -187,6 +191,8 @@ class Component:
return result_command
# check if this property should show in the System Properties box
# these are intended to provide a summary of critical things that all computers have
# like memory and cpu and cores
def check_php_extra(self, property_name):
result = False
if property_name in self.php_extra_list:
@ -195,6 +201,7 @@ class Component:
########################################################
# keyed data functions
# various ways to query the component data
########################################################
def get_properties_keys(self, component = None):
@ -269,73 +276,6 @@ class Component:
return result[0]
else:
return result
########################################################
# various data functions
########################################################
# complex data type return
def get_metrics(self, type = None):
these_metrics = []
if type == None:
for name, value in self._metrics:
these_metrics.append({"Metric": name, "Data": value})
else:
for name, value in self._metrics:
if name == type:
these_metrics.append({"Metric": name, "Data": value})
result = {
"Source": self.name,
"Component Type": self.type,
"Metrics": these_metrics
}
return result
# complex data type return
def get_property_summary(self, type = None):
these_properties = []
if type == None:
for name, value in self._properties.items():
these_properties.append({"Property": name, "Value": value})
else:
for name, value in self._properties.items():
if type in name:
these_properties.append({"Property": name, "Value": value})
result = {
"Source": self.name,
"Component Type": self.type,
"Properties": these_properties
}
return result
# simple data value return
def get_property_value(self, type = None):
result = []
print(f"Component type: {type}")
for name, value in self._properties.items():
print(f"Component Property: {name}")
if type in name:
result.append(value)
if len(result) == 1:
return result[0]
else:
return result
# full data return
def get_description(self):
these_properties = []
for name, value in self._properties.items():
these_properties.append({"Property": name, "Value": value})
these_metrics = []
for name, value in self._metrics.items():
these_metrics.append({"Metric": name, "Data": value})
result = {
"Source": self.name,
"Type": self.type,
"Properties": these_properties,
"Metrics": these_metrics
}
return result
############################################################
############################################################
@ -396,7 +336,7 @@ class System:
# load static keys
for static_key in self.static_key_variables:
if static_key["name"] not in self._virt_ignore:
self.process_property(static_key = static_key)
self._process_property(static_key = static_key)
# initialize live keys
self.update_live_keys()
# initialze components
@ -422,8 +362,8 @@ class System:
# critical class functions
########################################################
# process static keys
def process_property(self, static_key):
# process static properties, done once at instantiation
def _process_property(self, static_key):
command = static_key["command"]
if "arch_check" in static_key:
arch_string = run_command("lscpu --json | jq -r '.lscpu[] | select(.field==\"Architecture:\") | .data'", zero_only = True)
@ -439,7 +379,7 @@ class System:
if result not in null_result:
self._properties[static_key["name"]] = result
# update only system dynamic keys
# update only system dynamic keys, done periodically
def update_live_keys(self):
for live_key in self.dynamic_key_variables:
if live_key['command'] is not None:
@ -457,25 +397,53 @@ class System:
# component creation helper
def create_component(self, component):
# this is the type, i.e. CPU MEM LAN etc
component_name = component["name"]
# if there can be multiples, i.e.LAN STOR GPU etc
multi_check = component["multi_check"]
# if multi, note that the command in device_list creates the list of things to pipe into this_device
if multi_check:
log_data(log_output = f"Creating one component of type {component_name} for each one found", log_level = "log_output")
component_type_device_list = get_device_list(component_name)
# build list of components of this type to initialize
component_type_device_list = self.get_device_list(component_name)
for this_device in component_type_device_list:
this_component_ID = component_type_device_list.index(this_device)
this_component_name = f"{component_name} {this_component_ID}"
log_data(log_output = f"{this_component_name} - {component_name} - {this_device}", log_level = "debug_output")
new_component = Component(name = this_component_name, comp_type = component_name, this_device = this_device, parent_system = self)
# add new component to system
self.components.append(new_component)
# if it's just a single thing like CPU RAM then just do one device
else:
log_data(log_output = f'Creating component {component["name"]}', log_level = "debug_output")
new_component = Component(name = component_name, comp_type = component_name, parent_system = self)
# add new component to system
self.components.append(new_component)
# helper function - component type list builder for multi check items
def get_device_list(self, device_type_name: str):
result = []
for component in component_class_tree:
# pre-set to 1 for true value by default
precheck_value = 1
# the precheck is critical, i.e there can be 0 or 1 or 2 GPU
if "precheck" in component:
precheck_command = component["precheck"]
precheck_value_output = run_command(precheck_command, zero_only = True)
precheck_value = int(precheck_value_output)
log_data(log_output = f"Precheck found - {precheck_command} - {precheck_value}", log_level = "log_output")
# build a list of devices if precheck passes, this can be a list of 1 like a single GPU system
if component["name"] == device_type_name and precheck_value != 0:
device_list_command = component["device_list"]
device_list_result = run_command(device_list_command)
result = device_list_result
return result
# return the IP of the interface with the gateway
# remember the IP is stored like this 192.168.1.0/24
# this data is pulled from the system properties class data
# this was built for the update inventory generation
# but is not the complete picture
def get_primary_ip(self):
primary_ip = None
interfaces = self.get_components(component_type = "LAN")
@ -577,6 +545,7 @@ class System:
return result
# helper function for system properties for dashboard rendering
def php_component_data(self):
result = []
for component in self.components:
@ -653,57 +622,6 @@ class System:
})
return result
############################################################
# Non-class Helper Functions
############################################################
# subroutine to run a command, return stdout as array unless zero_only then return [0]
def run_command(cmd, zero_only=False, use_shell=True, req_check = True):
# Run the command and capture the output
result = subprocess.run(cmd, shell=use_shell, check=req_check, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# Decode the byte output to a string
output = result.stdout.decode('utf-8')
# Split the output into lines and store it in an array
output_lines = [line for line in output.split('\n') if line]
# Return result
try:
return output_lines[0] if zero_only else output_lines
except:
return output_lines
# need to add a archticture checker for this
# i also want to make the loop cleaner
# i don't need to iterate over the component class tree
# to get what I want, i think
def get_device_list(device_type_name: str):
result = []
for component in component_class_tree:
precheck_value = 1
if "precheck" in component:
precheck_command = component["precheck"]
precheck_value_output = run_command(precheck_command, zero_only = True)
precheck_value = int(precheck_value_output)
log_data(log_output = f"Precheck found - {precheck_command} - {precheck_value}", log_level = "log_output")
if component["name"] == device_type_name and precheck_value != 0:
device_list_command = component["device_list"]
device_list_result = run_command(device_list_command)
result = device_list_result
return result
# subnet helper app
def is_ip_in_subnets(ip, subnet):
try:
ip_obj = ipaddress.IPv4Address(ip)
subnet_obj = ipaddress.IPv4Network(subnet)
if ip_obj in subnet_obj:
return True
return False
except ValueError as e:
# If the IP address is not valid, raise an error
return False