123 lines
3.3 KiB
Python
123 lines
3.3 KiB
Python
"""Official server-health presentation and request state."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import time
|
|
from typing import Callable
|
|
|
|
from ...server_status import RealmServerStatus, ServerStatusError, fetch_server_status
|
|
|
|
|
|
SERVER_STATUS_STALE_SECONDS = 5 * 60
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ServerStatusPresentation:
|
|
"""Text and Tk style for one official realm-health result."""
|
|
|
|
text: str
|
|
style: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ServerStatusCheck:
|
|
"""One measured attempt to retrieve official realm health."""
|
|
|
|
statuses: tuple[RealmServerStatus, ...] | None
|
|
checked_at: float
|
|
duration_seconds: float
|
|
error: str | None = None
|
|
|
|
|
|
def perform_server_status_check(
|
|
*,
|
|
fetcher: Callable[[], tuple[RealmServerStatus, ...]] = fetch_server_status,
|
|
monotonic: Callable[[], float] = time.monotonic,
|
|
wall_clock: Callable[[], float] = time.time,
|
|
) -> ServerStatusCheck:
|
|
"""Measure one official status request without leaking its endpoint."""
|
|
|
|
started_at = monotonic()
|
|
try:
|
|
statuses = fetcher()
|
|
except ServerStatusError as error:
|
|
duration_seconds = max(0.0, monotonic() - started_at)
|
|
return ServerStatusCheck(
|
|
statuses=None,
|
|
checked_at=wall_clock(),
|
|
duration_seconds=duration_seconds,
|
|
error=str(error),
|
|
)
|
|
duration_seconds = max(0.0, monotonic() - started_at)
|
|
return ServerStatusCheck(
|
|
statuses=statuses,
|
|
checked_at=wall_clock(),
|
|
duration_seconds=duration_seconds,
|
|
)
|
|
|
|
|
|
def server_status_check_text(
|
|
check: ServerStatusCheck,
|
|
*,
|
|
timestamp: float | None = None,
|
|
) -> str:
|
|
"""Render fetch age, latency, and freshness for the hidden debug view."""
|
|
|
|
now = time.time() if timestamp is None else timestamp
|
|
age_seconds = max(0.0, now - check.checked_at)
|
|
whole_seconds = int(age_seconds)
|
|
if whole_seconds < 60:
|
|
age = f"{whole_seconds}s"
|
|
elif whole_seconds < 60 * 60:
|
|
age = f"{whole_seconds // 60}m"
|
|
elif whole_seconds < 24 * 60 * 60:
|
|
age = f"{whole_seconds // (60 * 60)}h"
|
|
else:
|
|
age = f"{whole_seconds // (24 * 60 * 60)}d"
|
|
latency_milliseconds = max(0, round(check.duration_seconds * 1000))
|
|
timing = f"checked {age} ago · {latency_milliseconds} ms"
|
|
if check.error is not None:
|
|
return f"failed · {timing} · {check.error}"
|
|
freshness = "stale" if age_seconds >= SERVER_STATUS_STALE_SECONDS else "fresh"
|
|
return f"{freshness} · {timing}"
|
|
|
|
|
|
def _server_component_text(name: str, online: bool | None) -> str:
|
|
if online is True:
|
|
state = "up"
|
|
elif online is False:
|
|
state = "down"
|
|
else:
|
|
state = "unknown"
|
|
return f"{name} {state}"
|
|
|
|
|
|
def server_status_presentation(
|
|
status: RealmServerStatus,
|
|
*,
|
|
show_debug_details: bool = False,
|
|
) -> ServerStatusPresentation:
|
|
if status.online is True:
|
|
text = f"● {status.name}"
|
|
style = "ServerOnline.TLabel"
|
|
elif status.online is False:
|
|
text = f"● {status.name} · offline"
|
|
style = "ServerOffline.TLabel"
|
|
else:
|
|
text = f"○ {status.name} · status unknown"
|
|
style = "ServerUnknown.TLabel"
|
|
if show_debug_details:
|
|
details: list[str] = []
|
|
if status.players_online is not None:
|
|
label = "player" if status.players_online == 1 else "players"
|
|
details.append(f"{status.players_online} {label}")
|
|
details.extend(
|
|
(
|
|
_server_component_text("login", status.login_online),
|
|
_server_component_text("game", status.game_online),
|
|
)
|
|
)
|
|
text = " · ".join((text, *details))
|
|
return ServerStatusPresentation(text=text, style=style)
|