Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2025-10-11 20:06:18 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

EOSdash: Enable EOS configuration by EOSdash. (#477)

Browse Source 
Improve config page to edit actual configuration used by EOS.
Add admin page to save the actual configuration to the configuration file.

Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com>
This commit is contained in:
Bobby Noelte
2025-03-27 21:53:01 +01:00
committed by GitHub
parent 61c5efc74f
commit 7aaf193682
18 changed files with 393 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

127
src/akkudoktoreos/server/dash/admin.py Normal file
View File

@@ -0,0 +1,127 @@
"""Admin UI components for EOS Dashboard.
This module provides functions to generate administrative UI components
for the EOS dashboard.
"""
from typing import Any, Optional, Union
import requests
from fasthtml.common import Div
from monsterui.foundations import stringify
from monsterui.franken import (
Button,
ButtonT,
Card,
Details,
DivHStacked,
DividerLine,
Grid,
P,
Summary,
UkIcon,
)
def AdminButton(*c: Any, cls: Optional[Union[str, tuple]] = None, **kwargs: Any) -> Button:
"""Creates a styled button for administrative actions.
Args:
*c (Any): Positional arguments representing the button's content.
cls (Optional[Union[str, tuple]]): Additional CSS classes for styling. Defaults to None.
**kwargs (Any): Additional keyword arguments passed to the `Button`.
Returns:
Button: A styled `Button` component for admin actions.
"""
new_cls = f"{ButtonT.primary}"
if cls:
new_cls += f" {stringify(cls)}"
kwargs["cls"] = new_cls
return Button(*c, submit=False, **kwargs)
def AdminConfig(eos_host: str, eos_port: Union[str, int], data: Optional[dict]) -> Card:
"""Creates a configuration management card with save-to-file functionality.
Args:
eos_host (str): The hostname of the EOS server.
eos_port (Union[str, int]): The port of the EOS server.
data (Optional[dict]): Incoming data containing action and category for processing.
Returns:
tuple[str, Card]: A tuple containing the configuration category label and the `Card` UI component.
"""
server = f"http://{eos_host}:{eos_port}"
category = "configuration"
status = (None,)
if data and data["category"] == category:
# This data is for us
if data["action"] == "save_to_file":
# Safe current configuration to file
try:
result = requests.put(f"{server}/v1/config/file")
result.raise_for_status()
config_file_path = result.json()["general"]["config_file_path"]
status = P(
f"Actual config saved to {config_file_path} on {server}",
cls="text-left",
)
except requests.exceptions.HTTPError as err:
detail = result.json()["detail"]
status = P(
f"Can not save actual config to file on {server}: {err}, {detail}",
cls="text-left",
)
return (
category,
Card(
Details(
Summary(
Grid(
DivHStacked(
UkIcon(icon="play"),
AdminButton(
"Save to file",
hx_post="/eosdash/admin",
hx_target="#page-content",
hx_swap="innerHTML",
hx_vals='{"category": "configuration", "action": "save_to_file"}',
),
),
status,
),
cls="list-none",
),
P(f"Safe actual configuration to config file on {server}."),
),
),
)
def Admin(eos_host: str, eos_port: Union[str, int], data: Optional[dict] = None) -> Div:
"""Generates the administrative dashboard layout.
This includes configuration management and other administrative tools.
Args:
eos_host (str): The hostname of the EOS server.
eos_port (Union[str, int]): The port of the EOS server.
data (Optional[dict], optional): Incoming data to trigger admin actions. Defaults to None.
Returns:
Div: A `Div` component containing the assembled admin interface.
"""
rows = []
last_category = ""
for category, admin in [
AdminConfig(eos_host, eos_port, data),
]:
if category != last_category:
rows.append(P(category))
rows.append(DividerLine())
last_category = category
rows.append(admin)
return Div(*rows, cls="space-y-4")

10
src/akkudoktoreos/server/dash/bokeh.py
View File

@@ -8,19 +8,19 @@ from bokeh.models import Plot
from monsterui.franken import H4, Card, NotStr, Script
BokehJS = [
Script(src="https://cdn.bokeh.org/bokeh/release/bokeh-3.6.3.min.js", crossorigin="anonymous"),
Script(src="https://cdn.bokeh.org/bokeh/release/bokeh-3.7.0.min.js", crossorigin="anonymous"),
Script(
src="https://cdn.bokeh.org/bokeh/release/bokeh-widgets-3.6.3.min.js",
src="https://cdn.bokeh.org/bokeh/release/bokeh-widgets-3.7.0.min.js",
crossorigin="anonymous",
),
Script(
src="https://cdn.bokeh.org/bokeh/release/bokeh-tables-3.6.3.min.js", crossorigin="anonymous"
src="https://cdn.bokeh.org/bokeh/release/bokeh-tables-3.7.0.min.js", crossorigin="anonymous"
),
Script(
src="https://cdn.bokeh.org/bokeh/release/bokeh-gl-3.6.3.min.js", crossorigin="anonymous"
src="https://cdn.bokeh.org/bokeh/release/bokeh-gl-3.7.0.min.js", crossorigin="anonymous"
),
Script(
src="https://cdn.bokeh.org/bokeh/release/bokeh-mathjax-3.6.3.min.js",
src="https://cdn.bokeh.org/bokeh/release/bokeh-mathjax-3.7.0.min.js",
crossorigin="anonymous",
),
]

52
src/akkudoktoreos/server/dash/components.py
View File

@@ -1,8 +1,6 @@
from typing import Any, Optional, Union
from fasthtml.common import H1, Div, Li
# from mdit_py_plugins import plugin1, plugin2
from monsterui.foundations import stringify
from monsterui.franken import (
Button,
@@ -13,6 +11,7 @@ from monsterui.franken import (
Details,
DivLAligned,
DivRAligned,
Form,
Grid,
Input,
P,
@@ -70,8 +69,22 @@ def ScrollArea(
def ConfigCard(
config_name: str, config_type: str, read_only: str, value: str, default: str, description: str
config_name: str,
config_type: str,
read_only: str,
value: str,
default: str,
description: str,
update_error: Optional[str],
update_value: Optional[str],
update_open: Optional[bool],
) -> Card:
"""Creates a styled configuration card."""
config_id = config_name.replace(".", "-")
if not update_value:
update_value = value
if not update_open:
update_open = False
return Card(
Details(
Summary(
@@ -85,24 +98,45 @@ def ConfigCard(
P(read_only),
),
),
Input(value=value) if read_only == "rw" else P(value),
P(value),
),
# cls="flex cursor-pointer list-none items-center gap-4",
cls="list-none",
),
Grid(
P(description),
P(config_type),
),
# Default
Grid(
DivRAligned(
P("default") if read_only == "rw" else P(""),
),
P(default) if read_only == "rw" else P(""),
DivRAligned(P("default")),
P(default),
)
if read_only == "rw"
else None,
# Set value
Grid(
DivRAligned(P("update")),
Grid(
Form(
Input(value=config_name, type="hidden", id="key"),
Input(value=update_value, type="text", id="value"),
hx_put="/eosdash/configuration",
hx_target="#page-content",
hx_swap="innerHTML",
),
),
)
if read_only == "rw"
else None,
# Last error
Grid(
DivRAligned(P("update error")),
P(update_error),
)
if update_error
else None,
cls="space-y-4 gap-4",
open=update_open,
),
cls="w-full",
)

124
src/akkudoktoreos/server/dash/configuration.py
View File

@@ -1,7 +1,8 @@
import json
from typing import Any, Dict, List, Optional, Sequence, TypeVar, Union
import requests
from monsterui.franken import Div, DividerLine, P, Table, Tbody, Td, Th, Thead, Tr
from monsterui.franken import Div, DividerLine, P
from pydantic.fields import ComputedFieldInfo, FieldInfo
from pydantic_core import PydanticUndefined
@@ -15,6 +16,10 @@ config_eos = get_config()
T = TypeVar("T")
# Latest configuration update results
# Dictionary of config names and associated dictionary with keys "value", "result", "error", "open".
config_update_latest: dict[str, dict[str, Optional[Union[str, bool]]]] = {}
def get_nested_value(
dictionary: Union[Dict[str, Any], List[Any]],
@@ -151,8 +156,8 @@ def configuration(values: dict) -> list[dict]:
config["type"] = (
type_description.replace("typing.", "")
.replace("pathlib.", "")
.replace("[", "[ ")
.replace("NoneType", "None")
.replace("<class 'float'>", "float")
)
configs.append(config)
found_basic = True
@@ -171,20 +176,16 @@ def configuration(values: dict) -> list[dict]:
return sorted(configs, key=lambda x: x["name"])
def get_configuration(eos_host: Optional[str], eos_port: Optional[Union[str, int]]) -> list[dict]:
def get_configuration(eos_host: str, eos_port: Union[str, int]) -> list[dict]:
"""Fetch and process configuration data from the specified EOS server.
Args:
eos_host (Optional[str]): The hostname of the server.
eos_port (Optional[Union[str, int]]): The port of the server.
eos_host (str): The hostname of the EOS server.
eos_port (Union[str, int]): The port of the EOS server.
Returns:
List[dict]: A list of processed configuration entries.
"""
if eos_host is None:
eos_host = config_eos.server.host
if eos_port is None:
eos_port = config_eos.server.port
server = f"http://{eos_host}:{eos_port}"
# Get current configuration from server
@@ -201,25 +202,37 @@ def get_configuration(eos_host: Optional[str], eos_port: Optional[Union[str, int
return configuration(config)
def Configuration(eos_host: Optional[str], eos_port: Optional[Union[str, int]]) -> Div:
def Configuration(
eos_host: str, eos_port: Union[str, int], configuration: Optional[list[dict]] = None
) -> Div:
"""Create a visual representation of the configuration.
Args:
eos_host (Optional[str]): The hostname of the EOS server.
eos_port (Optional[Union[str, int]]): The port of the EOS server.
eos_host (str): The hostname of the EOS server.
eos_port (Union[str, int]): The port of the EOS server.
configuration (Optional[list[dict]]): Optional configuration. If not provided it will be
retrievd from EOS.
Returns:
Table: A `monsterui.franken.Table` component displaying configuration details.
rows: Rows of configuration details.
"""
flds = "Name", "Type", "RO/RW", "Value", "Default", "Description"
if not configuration:
configuration = get_configuration(eos_host, eos_port)
rows = []
last_category = ""
for config in get_configuration(eos_host, eos_port):
for config in configuration:
category = config["name"].split(".")[0]
if category != last_category:
rows.append(P(category))
rows.append(DividerLine())
last_category = category
update_error = config_update_latest.get(config["name"], {}).get("error")
update_value = config_update_latest.get(config["name"], {}).get("value")
update_open = config_update_latest.get(config["name"], {}).get("open")
# Make mypy happy - should never trigger
assert isinstance(update_error, (str, type(None)))
assert isinstance(update_value, (str, type(None)))
assert isinstance(update_open, (bool, type(None)))
rows.append(
ConfigCard(
config["name"],
@@ -228,48 +241,59 @@ def Configuration(eos_host: Optional[str], eos_port: Optional[Union[str, int]])
config["value"],
config["default"],
config["description"],
update_error,
update_value,
update_open,
)
)
return Div(*rows, cls="space-y-4")
def ConfigurationOrg(eos_host: Optional[str], eos_port: Optional[Union[str, int]]) -> Table:
"""Create a visual representation of the configuration.
def ConfigKeyUpdate(eos_host: str, eos_port: Union[str, int], key: str, value: str) -> P:
"""Update configuration key and create a visual representation of the configuration.
Args:
eos_host (Optional[str]): The hostname of the EOS server.
eos_port (Optional[Union[str, int]]): The port of the EOS server.
eos_host (str): The hostname of the EOS server.
eos_port (Union[str, int]): The port of the EOS server.
key (str): configuration key in dot notation
value (str): configuration value as json string
Returns:
Table: A `monsterui.franken.Table` component displaying configuration details.
rows: Rows of configuration details.
"""
flds = "Name", "Type", "RO/RW", "Value", "Default", "Description"
rows = [
Tr(
Td(
config["name"],
cls="max-w-64 text-wrap break-all",
),
Td(
config["type"],
cls="max-w-48 text-wrap break-all",
),
Td(
config["read-only"],
cls="max-w-24 text-wrap break-all",
),
Td(
config["value"],
cls="max-w-md text-wrap break-all",
),
Td(config["default"], cls="max-w-48 text-wrap break-all"),
Td(
config["description"],
cls="max-w-prose text-wrap",
),
cls="",
)
for config in get_configuration(eos_host, eos_port)
]
head = Thead(*map(Th, flds), cls="text-left")
return Table(head, Tbody(*rows), cls="w-full uk-table uk-table-divider uk-table-striped")
server = f"http://{eos_host}:{eos_port}"
path = key.replace(".", "/")
try:
data = json.loads(value)
except:
if value in ("None", "none", "Null", "null"):
data = None
else:
data = value
error = None
result = None
try:
result = requests.put(f"{server}/v1/config/{path}", json=data)
result.raise_for_status()
except requests.exceptions.HTTPError as err:
if result:
detail = result.json()["detail"]
else:
detail = "No details"
error = f"Can not set {key} on {server}: {err}, {detail}"
# Mark all updates as closed
for k in config_update_latest:
config_update_latest[k]["open"] = False
# Remember this update as latest one
config_update_latest[key] = {
"error": error,
"result": result.json() if result else None,
"value": value,
"open": True,
}
if error or result is None:
# Reread configuration to be shure we display actual data
return Configuration(eos_host, eos_port)
# Use configuration already provided
return Configuration(eos_host, eos_port, configuration(result.json()))