feat: ServerState data model with serialization

Add dataclasses for ServerState, LoadingInfo, and Metrics to represent
the server's runtime state. ServerState includes a to_dict() method for
serialization with timestamp.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Vlad Doloman
2026-06-22 18:42:45 +03:00
parent ecfd759d7f
commit 28400b3cae
2 changed files with 84 additions and 0 deletions

View File

@@ -24,3 +24,39 @@ def classify_line(line: str) -> tuple[str, Optional[int], str]:
if m: if m:
return ('child', int(m.group(1)), line[m.end():]) return ('child', int(m.group(1)), line[m.end():])
return ('router', None, line) return ('router', None, line)
# ── Data model ───────────────────────────────────────────────────────────────
@dataclass
class LoadingInfo:
stages: list
current: str
progress: float # 0.0 1.0
@dataclass
class Metrics:
prompt_speed: Optional[float] = None # tokens/sec during prompt eval
prompt_tokens: Optional[int] = None # total prompt tokens
gen_speed: Optional[float] = None # tokens/sec during generation
n_decoded: Optional[int] = None # tokens decoded so far
@dataclass
class ServerState:
state: str = "offline"
model: Optional[str] = None
loading: Optional[LoadingInfo] = None
metrics: Metrics = field(default_factory=Metrics)
request_count: int = 0
def to_dict(self) -> dict:
return {
"state": self.state,
"model": self.model,
"loading": asdict(self.loading) if self.loading else None,
"metrics": asdict(self.metrics),
"request_count": self.request_count,
"timestamp": datetime.now().isoformat(timespec="seconds"),
}

View File

@@ -45,3 +45,51 @@ def test_classify_router_proxy_line():
kind, port, content = classify_line(line) kind, port, content = classify_line(line)
assert kind == "router" assert kind == "router"
assert port is None assert port is None
# ── ServerState data model ───────────────────────────────────────────────────
from log_parser import ServerState, LoadingInfo, Metrics
def test_server_state_defaults():
s = ServerState()
assert s.state == "offline"
assert s.model is None
assert s.loading is None
assert s.request_count == 0
def test_server_state_to_dict_offline():
s = ServerState()
d = s.to_dict()
assert d["state"] == "offline"
assert d["model"] is None
assert d["loading"] is None
assert d["metrics"]["gen_speed"] is None
assert "timestamp" in d
def test_server_state_to_dict_loading():
s = ServerState(
state="loading",
model="Qwen3",
loading=LoadingInfo(stages=["text_model", "mmproj_model"], current="text_model", progress=0.5),
)
d = s.to_dict()
assert d["state"] == "loading"
assert d["loading"]["current"] == "text_model"
assert d["loading"]["progress"] == 0.5
def test_server_state_to_dict_generating():
s = ServerState(
state="generating",
model="Qwen3",
metrics=Metrics(gen_speed=72.2, n_decoded=218),
request_count=5,
)
d = s.to_dict()
assert d["metrics"]["gen_speed"] == 72.2
assert d["metrics"]["n_decoded"] == 218
assert d["request_count"] == 5