system_view

system_view

Live system health from daemon profiler and controllers.

Wraps daemon IPC calls for system snapshots, status, rate limits, backpressure levels, and learning patterns.

Classes

DaemonSystemView

DaemonSystemView(client)

Live system health from the daemon profiler and controllers.

Parameters

client: Connected DaemonClient for IPC calls.

Source code in src/marianne/dashboard/services/system_view.py

def __init__(self, client: DaemonClient) -> None:
    self._client = client

Functions

get_snapshot async

get_snapshot()

Get latest SystemSnapshot from daemon.

Returns None if the daemon is down or the call fails.

Source code in src/marianne/dashboard/services/system_view.py

async def get_snapshot(self) -> dict[str, Any] | None:
    """Get latest SystemSnapshot from daemon.

    Returns ``None`` if the daemon is down or the call fails.
    """
    try:
        return cast("dict[str, Any] | None", await self._client.call("daemon.top"))
    except Exception:
        _logger.debug("get_snapshot_failed", exc_info=True)
        return None

get_daemon_status async

get_daemon_status()

Get daemon status (pid, uptime, running_jobs, memory).

Returns None if the daemon is unreachable.

Source code in src/marianne/dashboard/services/system_view.py

async def get_daemon_status(self) -> dict[str, Any] | None:
    """Get daemon status (pid, uptime, running_jobs, memory).

    Returns ``None`` if the daemon is unreachable.
    """
    try:
        result = await self._client.status()
        return result.model_dump()
    except Exception:
        _logger.debug("get_daemon_status_failed", exc_info=True)
        return None

rate_limit_state async

rate_limit_state()

Current rate limit state per backend.

Source code in src/marianne/dashboard/services/system_view.py

async def rate_limit_state(self) -> dict[str, Any]:
    """Current rate limit state per backend."""
    try:
        return await self._client.rate_limits()
    except Exception:
        _logger.debug("rate_limit_state_failed", exc_info=True)
        return {"backends": {}, "active_limits": 0}

pressure_level async

pressure_level()

Backpressure level from latest snapshot.

Source code in src/marianne/dashboard/services/system_view.py

async def pressure_level(self) -> dict[str, Any]:
    """Backpressure level from latest snapshot."""
    snap = await self.get_snapshot()
    if snap is None:
        return {"level": "unknown", "color": "gray"}
    level = snap.get("pressure_level", "NONE")
    colors = {
        "NONE": "green",
        "LOW": "yellow",
        "MEDIUM": "amber",
        "HIGH": "orange",
        "CRITICAL": "red",
    }
    return {"level": level, "color": colors.get(level, "gray")}

learning_patterns async

learning_patterns(limit=20)

Recent learning insights from the daemon.

Source code in src/marianne/dashboard/services/system_view.py

async def learning_patterns(self, limit: int = 20) -> list[dict[str, Any]]:
    """Recent learning insights from the daemon."""
    try:
        result = await self._client.learning_patterns(limit)
        return cast("list[dict[str, Any]]", result.get("patterns", []))
    except Exception:
        _logger.debug("learning_patterns_failed", exc_info=True)
        return []

Functions