Skip to content

System

tangram_system

log module-attribute 

log = getLogger(__name__)

plugin module-attribute 

plugin = Plugin(frontend_path='dist-frontend')

uptime 

uptime(counter: int) -> dict[str, str]
Source code in packages/tangram_system/src/tangram_system/__init__.py 
14
15
16
17
18
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 
21
22
23
24
25
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 
28
29
30
31
32
33
34
35
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 
38
39
40
41
42
43
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 
46
47
48
49
50
51
52
53
54
55
56
57
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", json.dumps(uptime(counter)))
        await redis_client.publish("to:system:update-node", json.dumps(info_utc()))
        await redis_client.publish("to:system:update-node", json.dumps(cpu_load()))
        await redis_client.publish("to:system:update-node", json.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 
63
64
65
@plugin.register_service()
async def run_system(backend_state: tangram_core.BackendState) -> None:
    await server_events(backend_state.redis_client)