Skip to content

System

tangram_system

log module-attribute 

log = getLogger(__name__)

plugin module-attribute 

plugin = Plugin(
    frontend_path="dist-frontend",
    config_class=SystemConfig,
    frontend_config_class=SystemFrontendConfig,
    into_frontend_config_function=into_frontend,
)

SystemConfig dataclass

Source code in packages/tangram_system/src/tangram_system/__init__.py 
16
17
18
@dataclass
class SystemConfig:
    topbar_order: int = 0

topbar_order class-attribute instance-attribute 

topbar_order: int = 0

__init__ 

__init__(topbar_order: int = 0) -> None

SystemFrontendConfig dataclass

Bases: HasTopbarUiConfig

Source code in packages/tangram_system/src/tangram_system/__init__.py 
21
22
23
@dataclass
class SystemFrontendConfig(tangram_core.config.HasTopbarUiConfig):
    topbar_order: Annotated[int, FrontendMutable()]

topbar_order instance-attribute 

topbar_order: Annotated[int, FrontendMutable()]

__init__ 

__init__(
    topbar_order: Annotated[int, FrontendMutable()],
) -> None

into_frontend 

into_frontend(config: SystemConfig) -> SystemFrontendConfig
Source code in packages/tangram_system/src/tangram_system/__init__.py 
26
27
def into_frontend(config: SystemConfig) -> SystemFrontendConfig:
    return SystemFrontendConfig(topbar_order=config.topbar_order)

uptime 

uptime(counter: int) -> dict[str, str]
Source code in packages/tangram_system/src/tangram_system/__init__.py 
30
31
32
33
34
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 
37
38
39
40
41
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 
44
45
46
47
48
49
50
51
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 
54
55
56
57
58
59
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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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 
86
87
88
@plugin.register_service()
async def run_system(backend_state: tangram_core.BackendState) -> None:
    await server_events(backend_state.redis_client)