Skip to content

System

tangram_system

log module-attribute 

log = getLogger(__name__)

plugin module-attribute 

plugin = Plugin(
    frontend_path="dist-frontend", config_class=SystemConfig
)

SystemConfig dataclass

Bases: HasSidebarUiConfig

Source code in packages/tangram_system/src/tangram_system/__init__.py 
16
17
18
@dataclass
class SystemConfig(tangram_core.config.HasSidebarUiConfig):
    topbar_order: Annotated[int, ExposeField()] = 0

topbar_order class-attribute instance-attribute 

topbar_order: Annotated[int, ExposeField()] = 0

sidebar_order instance-attribute 

sidebar_order: int

__init__ 

__init__(
    topbar_order: Annotated[int, ExposeField()] = 0,
) -> None

uptime 

uptime(counter: int) -> dict[str, str]
Source code in packages/tangram_system/src/tangram_system/__init__.py 
21
22
23
24
25
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 
28
29
30
31
32
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 
35
36
37
38
39
40
41
42
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 
45
46
47
48
49
50
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 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
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 
75
76
77
@plugin.register_service()
async def run_system(backend_state: tangram_core.BackendState) -> None:
    await server_events(backend_state.redis_client)