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

site updated with php for static data

Browse Source
This commit is contained in:
Matt
2026-03-17 23:56:09 -07:00
parent 61421305ed
commit fd7e0aaf31
6 changed files with 193 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
files/api/Components.py
View File

@ -27,10 +27,13 @@ class Component:
# instantiate new component
############################################################
def __init__(self, name: str, comp_type: str, this_device="None"):
def __init__(self, name: str, comp_type: str, this_device="None", is_virtual = "True"):
self.name = name
self.type = comp_type
# this variable is set when the device can have multiples
# it indicates that the commands in the descriptor might need templating
self.this_device = this_device
self.is_virtual = is_virtual
print(f"This device - {self.this_device}")
# build the component descriptor dictionary
for component in component_class_tree:
@ -46,13 +49,17 @@ class Component:
# store static properties
self.multi_check = self.is_multi()
self.virt_ignore = self._descriptor.get('virt_ignore', [])
if self.is_virtual:
self.virt_ignore = []
self._properties: Dict[str, str] = {}
for key, command in descriptor.get('properties', {}).items():
if self.this_device != "None":
# this means this component type is a multi and the commands need templating for each device
formatted_command = command.format(this_device=self.this_device)
self._properties[key] = run_command(formatted_command, True)
else:
self._properties[key] = run_command(command, True)
self._properties[key] = run_command(command, zero_only = True)
print(self._properties[key])
# build the description string
self._description_template: str | None = descriptor.get("description")
self.description = self._description_template.format(**self._properties)
@ -82,10 +89,13 @@ class Component:
if this_metric is not None:
self._metrics[key] = this_metric
else:
self._metrics[key] = run_command(command, True)
self._metrics[key] = run_command(command, zero_only = True)
def get_property(self, type):
return self._properties[type]
def get_property(self, type = None):
if type == None:
return self._properties
else:
return self._properties[type]
def is_multi(self):
for component_type in component_types:
@ -97,9 +107,14 @@ class Component:
# redis data functions
########################################################
def get_properties_keys(self):
def get_properties_keys(self, component = None):
result = []
for name, value in self._properties.items():
component_properties = []
if component == None:
component_properties = self._properties.items()
else:
component_properties = self.get_property(component)
for name, value in component_properties:
this_property = {
"Source": self.name,
"Property": name,
@ -109,15 +124,21 @@ class Component:
result.append(this_property)
return result
def get_properties_strings(self):
def get_properties_strings(self, return_simple = False):
result = []
for name, value in self._properties.items():
this_property = {
component_properties = self._properties.items()
print(component_properties)
for name, value in component_properties:
simple_property = f"{name}: {value}"
complex_property = {
"Source": self.name,
"Property": f"{name}: {value}"
"Property": simple_property
}
if name not in self.virt_ignore:
result.append(this_property)
if name not in self.virt_ignore:
if return_simple:
result.append(simple_property)
else:
result.append(complex_property)
return result
def get_metrics_keys(self):
@ -167,7 +188,7 @@ class Component:
return result
# complex data type return
def get_properties(self, type = None):
def get_property_summary(self, type = None):
these_properties = []
if type == None:
for name, value in self._properties.items():
@ -212,7 +233,7 @@ class System:
static_key_variables = [
{"name": "Hostname", "command": "hostname"},
{"name": "Virtual Machine", "command": "echo $([[ \"$(systemd-detect-virt)\" == none ]] && echo False || echo True)"},
{"name": "Virtual Machine", "command": 'echo $( [ "$(systemd-detect-virt)" = none ] && echo False || echo True )', "req_check": "False"},
{"name": "CPU Architecture:", "command": "lscpu --json | jq -r '.lscpu[] | select(.field==\"Architecture:\") | .data'"},
{"name": "OS Kernel", "command": "uname -r"},
{"name": "OS Name", "command": "cat /etc/os-release | grep PRETTY | cut -d\\\" -f2"},
@ -244,13 +265,20 @@ class System:
# initialize system properties and metrics dicts
self._properties: Dict[str, str] = {}
self._metrics: Dict[str, str] = {}
self._virt_string = run_command('systemd-detect-virt', zero_only = True, req_check = False)
self._virt_ignore = self.virt_ignore
if self._virt_string == "none":
self._virt_ignore = []
# timekeeping for websocket
self.recent_check = int(time.time())
self.recent_check = int(time.time())
# load static keys
for static_key in self.static_key_variables:
if static_key["name"] not in self.virt_ignore:
if static_key["name"] not in self._virt_ignore:
command = static_key["command"]
result = run_command(command, True)
if "req_check" in static_key:
result = run_command(command, zero_only = True, req_check = static_key["req_check"])
else:
result = run_command(command, zero_only = True)
if debug_output:
print(f'Static key [{static_key["name"]}] - command [{command}] - output [{result}]')
self._properties[static_key["name"]] = result
@ -272,7 +300,7 @@ class System:
for live_key in self.dynamic_key_variables:
if live_key['command'] is not None:
command = live_key['command']
result = run_command(command, True)
result = run_command(command, zero_only = True)
self._metrics[live_key['name']] = result
if debug_output:
print(f'Command {live_key["name"]} - [{command}] Result - [{result}]')
@ -298,12 +326,12 @@ class System:
this_component_letter = letters[component_type_device_list.index(this_device)]
this_component_name = f"{component_name} {this_component_letter}"
print(f"{this_component_name} - {component_name} - {this_device}")
self.add_components(Component(this_component_name, component_name, this_device))
self.add_components(Component(name = this_component_name, comp_type = component_name, this_device = this_device))
else:
if debug_output:
print(f'Creating component {component["name"]}')
self.add_components(Component(component_name, component_name))
self.add_components(Component(name = component_name, comp_type = component_name, is_virtual = self.is_virtual()))
# Add a component to the system
def add_components(self, component: Component,):
@ -332,9 +360,17 @@ class System:
def get_component_count(self):
result = int(len(self.components))
return result
def get_property(self, property = None):
if property == None:
return None
else:
return self._properties.get(property, {})
def is_virtual(self):
virt_check = self._properties.get('virt_ignore', {}).items()
vm_check = self.get_property('Virtual Machine')
print(f'vm_check: {vm_check}')
return vm_check
def check_system_timer(self):
time_lapsed = time.time() - float(self.recent_check)
@ -353,14 +389,14 @@ class System:
result.append(system_property)
return result
def get_component_properties(self, human_readable = False):
def get_component_properties(self, human_readable = False, component = None):
result = []
for component in self.components:
if human_readable:
for metric in component.get_properties_strings():
for metric in component.get_properties_strings(component = component):
result.append(metric)
else:
for metric in component.get_properties_keys():
for metric in component.get_properties_keys(component = component):
result.append(metric)
return result
@ -486,9 +522,9 @@ class System:
############################################################
# subroutine to run a command, return stdout as array unless zero_only then return [0]
def run_command(cmd, zero_only=False):
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=True, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
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