File size: 2,707 Bytes
ebeed67 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | """``CAMSTACK_EVENT {json}`` progress lines on stdout.
The hub reads the provider's log and parses these lines; it never needs to
understand anything else the container prints. Every event carries ``v`` (the
event format version, 1), ``ts`` (epoch ms) and ``type``. One event per line,
compact JSON, flushed immediately.
"""
from __future__ import annotations
import json
import sys
import time
from typing import Any, Callable, Optional, TextIO
PREFIX = "CAMSTACK_EVENT "
EVENT_VERSION = 1
STAGES = ("prepare", "replay", "train", "export", "eval", "package")
class Emitter:
def __init__(
self,
stream: Optional[TextIO] = None,
clock: Callable[[], float] = time.time,
) -> None:
self._stream = stream
self._clock = clock
def emit(self, event_type: str, **fields: Any) -> dict[str, Any]:
event: dict[str, Any] = {
"v": EVENT_VERSION,
"ts": int(self._clock() * 1000),
"type": event_type,
**fields,
}
stream = self._stream if self._stream is not None else sys.stdout
stream.write(PREFIX + json.dumps(event, separators=(",", ":"), sort_keys=False) + "\n")
stream.flush()
return event
def stage(self, stage: str) -> None:
if stage not in STAGES:
raise ValueError(f"unknown stage {stage!r}")
self.emit("stage", stage=stage)
def epoch(
self,
epoch: int,
epochs: int,
box_loss: Optional[float],
map50: Optional[float],
map50_95: Optional[float],
eta_sec: Optional[float],
) -> None:
self.emit(
"epoch",
epoch=epoch,
epochs=epochs,
boxLoss=box_loss,
map50=map50,
map50_95=map50_95,
etaSec=eta_sec,
)
def metric(self, name: str, value: float) -> None:
self.emit("metric", name=name, value=value)
def artifact(self, path: str, size: int, sha256: str) -> None:
self.emit("artifact", path=path, bytes=size, sha256=sha256)
def error(self, stage: str, code: str, message: str) -> None:
self.emit("error", stage=stage, code=code, message=message)
def done(self, status: str, result_sha256: Optional[str]) -> None:
self.emit("done", status=status, resultSha256=result_sha256)
def parse_line(line: str) -> Optional[dict[str, Any]]:
"""The event on a log line, or ``None`` for any other line."""
if not line.startswith(PREFIX):
return None
try:
value = json.loads(line[len(PREFIX):])
except json.JSONDecodeError:
return None
return value if isinstance(value, dict) else None
|