Skip to content

System

tangram_system

log module-attribute 

log = getLogger(__name__)

plugin module-attribute 

plugin = Plugin(
    frontend_path="dist-frontend",
    into_frontend_config_function=transform_config,
)

SystemConfig dataclass

Bases: HasSidebarUiConfig

Source code in packages/tangram_system/src/tangram_system/__init__.py 
15
16
17
@dataclass
class SystemConfig(tangram_core.config.HasSidebarUiConfig):
    topbar_order: int = 0

topbar_order class-attribute instance-attribute 

topbar_order: int = 0

sidebar_order instance-attribute 

sidebar_order: int

__init__ 

__init__(topbar_order: int = 0) -> None

transform_config 

transform_config(
    config_dict: dict[str, Any],
) -> SystemConfig
Source code in packages/tangram_system/src/tangram_system/__init__.py 
20
21
22
23
def transform_config(config_dict: dict[str, Any]) -> SystemConfig:
    from pydantic import TypeAdapter

    return TypeAdapter(SystemConfig).validate_python(config_dict)

uptime 

uptime(counter: int) -> dict[str, str]
Source code in packages/tangram_system/src/tangram_system/__init__.py 
26
27
28
29
30
def uptime(counter: int) -> dict[str, str]:
    return {
        "el": "uptime",
        "value": f"{timedelta(seconds=counter)}",
    }

info_utc 

info_utc() -> dict[str, str | int]
Source code in packages/tangram_system/src/tangram_system/__init__.py 
33
34
35
36
37
def info_utc() -> dict[str, str | int]:
    return {
        "el": "info_utc",
        "value": 1000 * int(datetime.now(timezone.utc).timestamp()),
    }

cpu_load 

cpu_load() -> dict[str, str]
Source code in packages/tangram_system/src/tangram_system/__init__.py 
40
41
42
43
44
45
46
47
def cpu_load() -> dict[str, str]:
    try:
        load1, _load5, _load15 = psutil.getloadavg()
        cpu_count = psutil.cpu_count(logical=True) or 1
        load_percent = (load1 / cpu_count) * 100
        return {"el": "cpu_load", "value": f"{load_percent:.2f}%"}
    except Exception:
        return {"el": "cpu_load", "value": "Unavailable"}

ram_usage 

ram_usage() -> dict[str, str]
Source code in packages/tangram_system/src/tangram_system/__init__.py 
50
51
52
53
54
55
def ram_usage() -> dict[str, str]:
    try:
        mem = psutil.virtual_memory()
        return {"el": "ram_usage", "value": f"{mem.percent:.2f}%"}
    except Exception:
        return {"el": "ram_usage", "value": "Unavailable"}

server_events async 

server_events(redis_client: Redis) -> NoReturn
Source code in packages/tangram_system/src/tangram_system/__init__.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
async def server_events(redis_client: redis.Redis) -> NoReturn:
    counter = 0
    log.info("serving system events...")

    while True:
        await redis_client.publish(
            "to:system:update-node", orjson.dumps(uptime(counter))
        )
        await redis_client.publish("to:system:update-node", orjson.dumps(info_utc()))
        await redis_client.publish("to:system:update-node", orjson.dumps(cpu_load()))
        await redis_client.publish("to:system:update-node", orjson.dumps(ram_usage()))
        counter += 1

        await asyncio.sleep(1)

run_system async 

run_system(backend_state: BackendState) -> None
Source code in packages/tangram_system/src/tangram_system/__init__.py 
79
80
81
@plugin.register_service()
async def run_system(backend_state: tangram_core.BackendState) -> None:
    await server_events(backend_state.redis_client)