Python SDK reference

Everything below is importable straight from the top-level package:

from epochix import parse, parse_string, compare, export, visualize, LiveReporter

Parsing

epochix.sdk.parse.parse(path, *, task=None, primary_metric=None, run_id=None, run_name=None, db=':memory:', locale='en')

Parse a training log file and return the completed :class:~epochix.models.Run.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the log file.	required
task	str | TaskType | None	Force a task type. Auto-detected when None.	None
primary_metric	str | None	Override the primary metric key.	None
run_id	str | None	Use a specific run ID. Auto-generated when None.	None
run_name	str | None	Human-readable name for the run.	None
db	str	SQLite database path. Defaults to :memory: (not persisted).	':memory:'
locale	str	Language code for narrative templates.	'en'

Returns:

Type	Description
Run	Finished run object with final_grade and story_summary.

Example

::

from epochix import parse

run = parse("train.log", task="biometric")
print(run.final_grade)   # Grade.A_PLUS
print(run.story_summary)

Source code in src/epochix/sdk/parse.py

def parse(
    path: str | Path,
    *,
    task: str | TaskType | None = None,
    primary_metric: str | None = None,
    run_id: str | None = None,
    run_name: str | None = None,
    db: str = ":memory:",
    locale: str = "en",
) -> Run:
    """Parse a training log file and return the completed :class:`~epochix.models.Run`.

    Parameters
    ----------
    path:
        Path to the log file.
    task:
        Force a task type.  Auto-detected when ``None``.
    primary_metric:
        Override the primary metric key.
    run_id:
        Use a specific run ID.  Auto-generated when ``None``.
    run_name:
        Human-readable name for the run.
    db:
        SQLite database path.  Defaults to ``:memory:`` (not persisted).
    locale:
        Language code for narrative templates.

    Returns
    -------
    Run
        Finished run object with ``final_grade`` and ``story_summary``.

    Example
    -------
    ::

        from epochix import parse

        run = parse("train.log", task="biometric")
        print(run.final_grade)   # Grade.A_PLUS
        print(run.story_summary)
    """
    log_path = Path(path)
    if not log_path.exists():
        raise FileNotFoundError(f"Log file not found: {log_path}")

    effective_task: TaskType | None = TaskType(task) if isinstance(task, str) else task

    try:
        _id = run_id or _gen_id()
    except Exception:  # noqa: BLE001
        import uuid

        _id = str(uuid.uuid4())

    return asyncio.run(
        _parse_async(
            log_path=log_path,
            run_id=_id,
            run_name=run_name,
            task=effective_task,
            primary_metric=primary_metric,
            db=db,
            locale=locale,
        )
    )

epochix.sdk.parse.parse_string(log_text, *, task=None, primary_metric=None, run_id=None, run_name=None, db=':memory:', locale='en')

Parse a log string (useful when the log is already in memory).

Writes log_text to a temporary file and delegates to :func:parse.

Source code in src/epochix/sdk/parse.py

def parse_string(
    log_text: str,
    *,
    task: str | TaskType | None = None,
    primary_metric: str | None = None,
    run_id: str | None = None,
    run_name: str | None = None,
    db: str = ":memory:",
    locale: str = "en",
) -> Run:
    """Parse a log string (useful when the log is already in memory).

    Writes *log_text* to a temporary file and delegates to :func:`parse`.
    """
    import tempfile

    with tempfile.NamedTemporaryFile(mode="w", suffix=".log", delete=False, encoding="utf-8") as fh:
        fh.write(log_text)
        tmp = fh.name

    try:
        return parse(
            tmp,
            task=task,
            primary_metric=primary_metric,
            run_id=run_id,
            run_name=run_name,
            db=db,
            locale=locale,
        )
    finally:
        Path(tmp).unlink(missing_ok=True)

Live reporting

epochix.sdk.live_reporter.LiveReporter

Push metrics from the training loop directly to epochix.

Does NOT require log parsing — the caller supplies key-value metrics. Under the hood the reporter formats them as a log line, feeds them through the SDK receiver, and runs the full pipeline in a background asyncio event loop on a dedicated thread.

Thread-safety: :meth:log and :meth:finish are safe to call from any thread (including the PyTorch DataLoader worker threads).

Source code in src/epochix/sdk/live_reporter.py

class LiveReporter:
    """Push metrics from the training loop directly to epochix.

    Does NOT require log parsing — the caller supplies key-value metrics.
    Under the hood the reporter formats them as a log line, feeds them
    through the SDK receiver, and runs the full pipeline in a background
    asyncio event loop on a dedicated thread.

    Thread-safety: :meth:`log` and :meth:`finish` are safe to call from
    any thread (including the PyTorch DataLoader worker threads).
    """

    def __init__(
        self,
        *,
        task: str | TaskType | None = None,
        primary_metric: str | None = None,
        name: str | None = None,
        total_epochs: int | None = None,
        port: int = 7860,
        open_browser: bool = True,
        locale: str = "en",
        run_id: str | None = None,
        model: object | None = None,
        capture_activations: bool = False,
        activation_hz: float = 2.0,
        capture_gradients: bool = True,
    ) -> None:
        """
        Parameters
        ----------
        task:
            Task type (``"gaze"``, ``"detection"``, ``"classification"``, …).
            Leave ``None`` to auto-detect from the metrics you log.
        primary_metric:
            Which logged metric drives the grade / phase / narrative. Pass the
            **same name you use in** :meth:`log` — any spelling works, it is
            normalised internally (``"val_mae_cm"``, ``"mae"`` and
            ``"MAE"`` all resolve to the same metric). Optional: when omitted,
            the task's standard metric is used (gaze/regression → MAE,
            classification → val_accuracy, detection → mAP50, …), which is the
            recommended default — only set this to override.
        name:
            Human-readable run name shown in the dashboard.
        total_epochs:
            Total planned epochs, used for the progress bar. Optional.
        port:
            Local dashboard port. Use a free port if you run several at once.
        open_browser:
            Open the dashboard in a browser tab on start (default True).
        locale:
            Dashboard language (``"en"`` / ``"fa"`` / ``"fr"``).
        run_id:
            Explicit run id; auto-generated when omitted.
        model:
            The model being trained (PyTorch ``nn.Module`` or Keras ``Model``).
            When given, its **real** architecture — actual layer names, types
            and parameter counts — is shown in the dashboard's Network State
            panel. Omitted → the panel honestly reports no architecture rather
            than showing a placeholder.
        capture_activations:
            When True (and a ``model`` is given), register forward hooks that
            capture **real** per-layer activation magnitudes and dead-unit
            fractions during training, so the Network State panel animates from
            measured values instead of a schematic. Default False — zero
            overhead unless you opt in. Requires PyTorch or Keras.
        activation_hz:
            Cap on how often each layer's activation is sampled (Hz). ``.item()``
            forces a GPU sync, so this wall-clock throttle keeps the overhead
            negligible. Default 2 Hz.
        capture_gradients:
            When capturing, also register backward hooks for mean ``|gradient|``
            per layer (PyTorch only). Default True; ignored unless
            ``capture_activations`` is on.
        """
        from epochix.sdk.architecture import architecture_from_model

        self._task: TaskType | None = TaskType(task) if isinstance(task, str) else task
        self._primary_metric = primary_metric
        self._name = name
        self._total_epochs = total_epochs
        self._port = port
        self._open_browser = open_browser
        self._locale = locale
        self._run_id = run_id or self._generate_run_id()
        self._architecture = architecture_from_model(model)
        self._model = model
        self._capture_activations = capture_activations
        self._activation_hz = activation_hz
        self._capture_gradients = capture_gradients
        self._capturer: object | None = None
        self._emit_activations: object | None = None
        self._store: object | None = None
        self._seq = 0

        self._receiver: object | None = None  # SDKReceiver, lazy
        self._pipeline_task: object | None = None
        self._loop: asyncio.AbstractEventLoop | None = None
        self._thread: threading.Thread | None = None
        self._started = False

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    def log(self, **metrics: float) -> None:
        """Push one epoch of metrics.

        All keyword arguments are treated as metric key-value pairs.
        The special key ``epoch`` is used to track training progress.

        Example::

            reporter.log(epoch=5, train_loss=0.312, val_accuracy=0.871)
        """
        if not self._started:
            self._start()

        line = "  ".join(f"{k}={v}" for k, v in metrics.items())
        self._push(line)
        self._flush_activations()
        self._seq += 1

    def log_line(self, text: str) -> None:
        """Feed one raw log line — exactly as a training script printed it —
        to the parser.

        Use this when you are relaying somebody else's stdout (a subprocess, a
        notebook cell) and the metrics are already formatted in the line. It is
        the same path ``epochix --live`` takes, so every parser applies.

        Example::

            reporter.log_line("Epoch 3/10 train_loss=0.42 val_accuracy=0.88")
        """
        if not self._started:
            self._start()
        self._push(text)
        self._seq += 1

    def _push(self, line: str) -> None:
        if self._receiver is not None:
            from epochix.ingester.sdk_receiver import SDKReceiver

            assert isinstance(self._receiver, SDKReceiver)
            self._receiver.push_line(line)

    def finish(self) -> None:
        """Signal end of training and wait for the pipeline to flush."""
        if not self._started:
            return
        final_snapshot: dict[str, dict[str, float]] = {}
        if self._capturer is not None:
            from epochix.sdk.activations import ActivationCapturer

            assert isinstance(self._capturer, ActivationCapturer)
            final_snapshot = self._capturer.snapshot()
            self._capturer.remove()
            self._capturer = None
        if self._receiver is not None:
            from epochix.ingester.sdk_receiver import SDKReceiver

            assert isinstance(self._receiver, SDKReceiver)
            self._receiver.close()
        if self._thread is not None:
            self._thread.join(timeout=30)
        # The pipeline has now created the run row and stopped its loop, so a
        # final synchronous persist guarantees the last real snapshot lands in
        # run.config even for runs that finish faster than the loop started
        # (where the per-epoch live emits raced ahead of run creation).
        self._persist_final_activations(final_snapshot)
        self._started = False

    def _persist_final_activations(self, snapshot: dict[str, dict[str, float]]) -> None:
        if not snapshot or self._store is None:
            return
        from epochix.store.sqlite_store import RunStore

        if not isinstance(self._store, RunStore):
            return
        try:
            existing = self._store.get_run(self._run_id)
            if existing is None:
                return  # run never materialised (nothing logged) — nothing to attach to
            cfg = existing.config
            self._store.update_run_config(self._run_id, {**cfg, "activations": snapshot})
        except Exception:  # noqa: BLE001 — teardown telemetry must never raise
            pass

    def _flush_activations(self) -> None:
        """Snapshot the latest activations and hand them to the pipeline loop.

        Runs on the *training* thread (from :meth:`log`); the store/hub writes
        are scheduled onto the background event loop so all persistence and
        broadcasting stays single-threaded. A no-op until capture is on and the
        loop is up (early ``log`` calls before the thread's loop exists are
        simply skipped — the next epoch's snapshot supersedes them anyway).
        """
        if self._capturer is None or self._loop is None or self._emit_activations is None:
            return
        from epochix.sdk.activations import ActivationCapturer

        assert isinstance(self._capturer, ActivationCapturer)
        snapshot = self._capturer.snapshot()
        if not snapshot:
            return
        emit = self._emit_activations
        # Loop already closed (finish in progress) → nothing to flush.
        with contextlib.suppress(RuntimeError):
            self._loop.call_soon_threadsafe(emit, snapshot)  # type: ignore[arg-type]

    # ------------------------------------------------------------------
    # Context manager
    # ------------------------------------------------------------------

    def __enter__(self) -> LiveReporter:
        return self

    def __exit__(self, exc_type: object, exc_val: object, exc_tb: object) -> None:
        self.finish()

    # ------------------------------------------------------------------
    # Internal
    # ------------------------------------------------------------------

    def _start(self) -> None:
        """Lazy start: spin up the background event loop + pipeline."""
        self._started = True
        settings = get_settings()

        from epochix.ingester.sdk_receiver import SDKReceiver
        from epochix.pipeline import run_pipeline
        from epochix.server.app import create_app
        from epochix.server.hub import Hub
        from epochix.store.sqlite_store import RunStore

        receiver = SDKReceiver(run_id=self._run_id)
        self._receiver = receiver

        store = RunStore(db_path=settings.db)
        hub = Hub()
        _app = create_app(settings=settings)
        _app.state.store = store
        _app.state.hub = hub
        _app.state.engine_map = {}

        if self._capture_activations and self._model is not None:
            from epochix.sdk.activations import ActivationCapturer

            self._capturer = ActivationCapturer(
                self._model, hz=self._activation_hz, gradients=self._capture_gradients
            )
            self._store = store
            self._emit_activations = self._make_emit(store, hub, run_id=self._run_id)

        port = self._port
        run_id = self._run_id
        task = self._task
        name = self._name
        primary_metric = self._primary_metric
        total_epochs = self._total_epochs
        locale = self._locale
        open_browser = self._open_browser
        architecture = self._architecture

        async def _pipeline() -> None:
            import uvicorn

            config = uvicorn.Config(
                _app,
                host="127.0.0.1",
                port=port,
                log_level="warning",
                lifespan="off",
            )
            server = uvicorn.Server(config)
            server_task = asyncio.create_task(server.serve())
            await asyncio.sleep(0.5)
            if open_browser:
                import webbrowser

                webbrowser.open(f"http://127.0.0.1:{port}/v/{run_id}")
            try:
                await run_pipeline(
                    ingester=receiver,
                    run_id=run_id,
                    store=store,
                    hub=hub,
                    run_name=name,
                    task=task,
                    primary_metric=primary_metric,
                    total_epochs=total_epochs,
                    locale=locale,
                    architecture=architecture,
                )
            finally:
                server.should_exit = True
                await server_task

        def _thread_main() -> None:
            loop = asyncio.new_event_loop()
            self._loop = loop
            asyncio.set_event_loop(loop)
            loop.run_until_complete(_pipeline())
            loop.close()

        thread = threading.Thread(
            target=_thread_main, daemon=True, name=f"epochix-pipeline-{run_id[:8]}"
        )
        self._thread = thread
        thread.start()

    def _make_emit(self, store: object, hub: object, *, run_id: str) -> object:
        """Build the callback that persists + broadcasts an activation snapshot.

        The returned function runs on the background event loop (scheduled via
        ``call_soon_threadsafe``), so its store/hub access is single-threaded
        alongside the pipeline. Persisting the latest snapshot in
        ``run.config["activations"]`` means a dashboard opened mid/after-run
        still shows real values, not just live WS subscribers.
        """

        def emit(snapshot: dict[str, dict[str, float]]) -> None:
            try:
                existing = store.get_run(run_id)  # type: ignore[attr-defined]
                cfg = existing.config if existing else {}
                store.update_run_config(  # type: ignore[attr-defined]
                    run_id, {**cfg, "activations": snapshot}
                )
                hub.publish(  # type: ignore[attr-defined]
                    run_id,
                    hub.make_message(  # type: ignore[attr-defined]
                        msg_type="activations",
                        run_id=run_id,
                        seq=-1,
                        payload={"layers": snapshot},
                    ),
                )
            except Exception:  # noqa: BLE001 — telemetry must never break training
                pass

        return emit

    @staticmethod
    def _generate_run_id() -> str:
        try:
            from ulid import ULID

            return str(ULID())
        except ImportError:
            import uuid

            return str(uuid.uuid4())

__init__(*, task=None, primary_metric=None, name=None, total_epochs=None, port=7860, open_browser=True, locale='en', run_id=None, model=None, capture_activations=False, activation_hz=2.0, capture_gradients=True)

Parameters:

Name	Type	Description	Default
task	str | TaskType | None	Task type ("gaze", "detection", "classification", …). Leave None to auto-detect from the metrics you log.	None
primary_metric	str | None	Which logged metric drives the grade / phase / narrative. Pass the same name you use in :meth:log — any spelling works, it is normalised internally ("val_mae_cm", "mae" and "MAE" all resolve to the same metric). Optional: when omitted, the task's standard metric is used (gaze/regression → MAE, classification → val_accuracy, detection → mAP50, …), which is the recommended default — only set this to override.	None
name	str | None	Human-readable run name shown in the dashboard.	None
total_epochs	int | None	Total planned epochs, used for the progress bar. Optional.	None
port	int	Local dashboard port. Use a free port if you run several at once.	7860
open_browser	bool	Open the dashboard in a browser tab on start (default True).	True
locale	str	Dashboard language ("en" / "fa" / "fr").	'en'
run_id	str | None	Explicit run id; auto-generated when omitted.	None
model	object | None	The model being trained (PyTorch nn.Module or Keras Model). When given, its real architecture — actual layer names, types and parameter counts — is shown in the dashboard's Network State panel. Omitted → the panel honestly reports no architecture rather than showing a placeholder.	None
capture_activations	bool	When True (and a model is given), register forward hooks that capture real per-layer activation magnitudes and dead-unit fractions during training, so the Network State panel animates from measured values instead of a schematic. Default False — zero overhead unless you opt in. Requires PyTorch or Keras.	False
activation_hz	float	Cap on how often each layer's activation is sampled (Hz). .item() forces a GPU sync, so this wall-clock throttle keeps the overhead negligible. Default 2 Hz.	2.0
capture_gradients	bool	When capturing, also register backward hooks for mean |gradient| per layer (PyTorch only). Default True; ignored unless capture_activations is on.	True

Source code in src/epochix/sdk/live_reporter.py

def __init__(
    self,
    *,
    task: str | TaskType | None = None,
    primary_metric: str | None = None,
    name: str | None = None,
    total_epochs: int | None = None,
    port: int = 7860,
    open_browser: bool = True,
    locale: str = "en",
    run_id: str | None = None,
    model: object | None = None,
    capture_activations: bool = False,
    activation_hz: float = 2.0,
    capture_gradients: bool = True,
) -> None:
    """
    Parameters
    ----------
    task:
        Task type (``"gaze"``, ``"detection"``, ``"classification"``, …).
        Leave ``None`` to auto-detect from the metrics you log.
    primary_metric:
        Which logged metric drives the grade / phase / narrative. Pass the
        **same name you use in** :meth:`log` — any spelling works, it is
        normalised internally (``"val_mae_cm"``, ``"mae"`` and
        ``"MAE"`` all resolve to the same metric). Optional: when omitted,
        the task's standard metric is used (gaze/regression → MAE,
        classification → val_accuracy, detection → mAP50, …), which is the
        recommended default — only set this to override.
    name:
        Human-readable run name shown in the dashboard.
    total_epochs:
        Total planned epochs, used for the progress bar. Optional.
    port:
        Local dashboard port. Use a free port if you run several at once.
    open_browser:
        Open the dashboard in a browser tab on start (default True).
    locale:
        Dashboard language (``"en"`` / ``"fa"`` / ``"fr"``).
    run_id:
        Explicit run id; auto-generated when omitted.
    model:
        The model being trained (PyTorch ``nn.Module`` or Keras ``Model``).
        When given, its **real** architecture — actual layer names, types
        and parameter counts — is shown in the dashboard's Network State
        panel. Omitted → the panel honestly reports no architecture rather
        than showing a placeholder.
    capture_activations:
        When True (and a ``model`` is given), register forward hooks that
        capture **real** per-layer activation magnitudes and dead-unit
        fractions during training, so the Network State panel animates from
        measured values instead of a schematic. Default False — zero
        overhead unless you opt in. Requires PyTorch or Keras.
    activation_hz:
        Cap on how often each layer's activation is sampled (Hz). ``.item()``
        forces a GPU sync, so this wall-clock throttle keeps the overhead
        negligible. Default 2 Hz.
    capture_gradients:
        When capturing, also register backward hooks for mean ``|gradient|``
        per layer (PyTorch only). Default True; ignored unless
        ``capture_activations`` is on.
    """
    from epochix.sdk.architecture import architecture_from_model

    self._task: TaskType | None = TaskType(task) if isinstance(task, str) else task
    self._primary_metric = primary_metric
    self._name = name
    self._total_epochs = total_epochs
    self._port = port
    self._open_browser = open_browser
    self._locale = locale
    self._run_id = run_id or self._generate_run_id()
    self._architecture = architecture_from_model(model)
    self._model = model
    self._capture_activations = capture_activations
    self._activation_hz = activation_hz
    self._capture_gradients = capture_gradients
    self._capturer: object | None = None
    self._emit_activations: object | None = None
    self._store: object | None = None
    self._seq = 0

    self._receiver: object | None = None  # SDKReceiver, lazy
    self._pipeline_task: object | None = None
    self._loop: asyncio.AbstractEventLoop | None = None
    self._thread: threading.Thread | None = None
    self._started = False

finish()

Signal end of training and wait for the pipeline to flush.

Source code in src/epochix/sdk/live_reporter.py

def finish(self) -> None:
    """Signal end of training and wait for the pipeline to flush."""
    if not self._started:
        return
    final_snapshot: dict[str, dict[str, float]] = {}
    if self._capturer is not None:
        from epochix.sdk.activations import ActivationCapturer

        assert isinstance(self._capturer, ActivationCapturer)
        final_snapshot = self._capturer.snapshot()
        self._capturer.remove()
        self._capturer = None
    if self._receiver is not None:
        from epochix.ingester.sdk_receiver import SDKReceiver

        assert isinstance(self._receiver, SDKReceiver)
        self._receiver.close()
    if self._thread is not None:
        self._thread.join(timeout=30)
    # The pipeline has now created the run row and stopped its loop, so a
    # final synchronous persist guarantees the last real snapshot lands in
    # run.config even for runs that finish faster than the loop started
    # (where the per-epoch live emits raced ahead of run creation).
    self._persist_final_activations(final_snapshot)
    self._started = False

log(**metrics)

Push one epoch of metrics.

All keyword arguments are treated as metric key-value pairs. The special key epoch is used to track training progress.

Example::

reporter.log(epoch=5, train_loss=0.312, val_accuracy=0.871)

Source code in src/epochix/sdk/live_reporter.py

def log(self, **metrics: float) -> None:
    """Push one epoch of metrics.

    All keyword arguments are treated as metric key-value pairs.
    The special key ``epoch`` is used to track training progress.

    Example::

        reporter.log(epoch=5, train_loss=0.312, val_accuracy=0.871)
    """
    if not self._started:
        self._start()

    line = "  ".join(f"{k}={v}" for k, v in metrics.items())
    self._push(line)
    self._flush_activations()
    self._seq += 1

log_line(text)

Feed one raw log line — exactly as a training script printed it — to the parser.

Use this when you are relaying somebody else's stdout (a subprocess, a notebook cell) and the metrics are already formatted in the line. It is the same path epochix --live takes, so every parser applies.

Example::

reporter.log_line("Epoch 3/10 train_loss=0.42 val_accuracy=0.88")

Source code in src/epochix/sdk/live_reporter.py

def log_line(self, text: str) -> None:
    """Feed one raw log line — exactly as a training script printed it —
    to the parser.

    Use this when you are relaying somebody else's stdout (a subprocess, a
    notebook cell) and the metrics are already formatted in the line. It is
    the same path ``epochix --live`` takes, so every parser applies.

    Example::

        reporter.log_line("Epoch 3/10 train_loss=0.42 val_accuracy=0.88")
    """
    if not self._started:
        self._start()
    self._push(text)
    self._seq += 1

Comparison

epochix.sdk.compare.compare(run_a, run_b, *, db=None)

Compare two runs and return a :class:RunDiff.

Parameters:

Name	Type	Description	Default
run_a	Run | str	Either a :class:~epochix.models.Run object, or a path to a JSON export file (produced by epochix export --format json).	required
run_b	Run | str	Either a :class:~epochix.models.Run object, or a path to a JSON export file (produced by epochix export --format json).	required
db	str | None	SQLite DB path (used when run_a/run_b is a run ID string).	None

Returns:

Type	Description
RunDiff

Source code in src/epochix/sdk/compare.py

def compare(
    run_a: Run | str,
    run_b: Run | str,
    *,
    db: str | None = None,
) -> RunDiff:
    """Compare two runs and return a :class:`RunDiff`.

    Parameters
    ----------
    run_a, run_b:
        Either a :class:`~epochix.models.Run` object, or a path to a
        JSON export file (produced by ``epochix export --format json``).
    db:
        SQLite DB path (used when *run_a*/*run_b* is a run ID string).

    Returns
    -------
    RunDiff
    """

    a = _resolve(run_a, db=db)
    b = _resolve(run_b, db=db)

    # Grade comparison
    from epochix.enums import Grade

    _GRADE_ORDER = list(Grade)
    ga_idx = _GRADE_ORDER.index(a.final_grade) if a.final_grade else len(_GRADE_ORDER)
    gb_idx = _GRADE_ORDER.index(b.final_grade) if b.final_grade else len(_GRADE_ORDER)

    if ga_idx < gb_idx:
        better = "a"
        ga_str = a.final_grade.value if a.final_grade else "?"
        gb_str = b.final_grade.value if b.final_grade else "?"
        grade_delta = f"{ga_str} vs {gb_str}"
    elif gb_idx < ga_idx:
        better = "b"
        gb_str = b.final_grade.value if b.final_grade else "?"
        ga_str = a.final_grade.value if a.final_grade else "?"
        grade_delta = f"{gb_str} vs {ga_str}"
    else:
        better = "tie"
        grade_delta = "equal"

    summary = (
        f"Run {'A' if better == 'a' else 'B'} achieves a better grade ({grade_delta})."
        if better != "tie"
        else "Both runs achieved the same grade."
    )

    return RunDiff(
        run_a=a,
        run_b=b,
        grade_delta=grade_delta,
        summary=summary,
        better=better,
    )

Export

epochix.sdk.export.export(run, fmt='html', *, output=None, db=None)

Export a run to a file.

Parameters:

Name	Type	Description	Default
run	Run	A completed :class:~epochix.models.Run.	required
fmt	ExportFormat	Export format: "html", "pdf", "md", or "json".	'html'
output	str | Path | None	Output path. Defaults to <run_id>.<fmt> in the current directory.	None
db	str | None	SQLite DB containing the run (defaults to the configured DB).	None

Returns:

Type	Description
Path	Absolute path to the written file.

Source code in src/epochix/sdk/export.py

def export(
    run: Run,
    fmt: ExportFormat = "html",
    *,
    output: str | Path | None = None,
    db: str | None = None,
) -> Path:
    """Export a run to a file.

    Parameters
    ----------
    run:
        A completed :class:`~epochix.models.Run`.
    fmt:
        Export format: ``"html"``, ``"pdf"``, ``"md"``, or ``"json"``.
    output:
        Output path.  Defaults to ``<run_id>.<fmt>`` in the current directory.
    db:
        SQLite DB containing the run (defaults to the configured DB).

    Returns
    -------
    Path
        Absolute path to the written file.
    """
    from epochix.config import get_settings
    from epochix.store.sqlite_store import RunStore

    settings = get_settings()
    store = RunStore(db_path=db or settings.db)

    out = Path(output) if output else Path(f"{run.id}.{fmt}")

    if fmt == "json":
        from epochix.exporters.json_export import build_json

        out.write_text(build_json(run_id=run.id, store=store), encoding="utf-8")

    elif fmt == "md":
        from epochix.exporters.markdown_export import build_markdown

        md = build_markdown(run_id=run.id, store=store)
        out.write_text(md, encoding="utf-8")

    elif fmt == "html":
        from epochix.exporters.html_export import build_html

        html = build_html(run_id=run.id, store=store)
        out.write_text(html, encoding="utf-8")

    elif fmt == "pdf":
        from epochix.exporters.pdf_export import build_pdf

        pdf_bytes = build_pdf(run_id=run.id, store=store)
        out.write_bytes(pdf_bytes)

    return out.resolve()

Visualisation

epochix.sdk.visualize.visualize(run, *, port=7860, host='127.0.0.1', db=None, blocking=True)

Serve a finished run and open it in the browser.

Parameters:

Name	Type	Description	Default
run	Run	A :class:~epochix.models.Run previously returned by :func:~epochix.sdk.parse.parse.	required
port	int	Port for the embedded server.	7860
host	str	Bind address.	'127.0.0.1'
db	str | None	SQLite DB path that contains the run (defaults to the configured DB).	None
blocking	bool	If True (default), block until the user closes the server (Ctrl-C). Set to False to start the server in a background thread and return immediately.	True

Returns:

Type	Description
str	The URL where the run is being served.

Source code in src/epochix/sdk/visualize.py

def visualize(
    run: Run,
    *,
    port: int = 7860,
    host: str = "127.0.0.1",
    db: str | None = None,
    blocking: bool = True,
) -> str:
    """Serve a finished run and open it in the browser.

    Parameters
    ----------
    run:
        A :class:`~epochix.models.Run` previously returned by
        :func:`~epochix.sdk.parse.parse`.
    port:
        Port for the embedded server.
    host:
        Bind address.
    db:
        SQLite DB path that contains the run (defaults to the configured DB).
    blocking:
        If ``True`` (default), block until the user closes the server
        (Ctrl-C).  Set to ``False`` to start the server in a background
        thread and return immediately.

    Returns
    -------
    str
        The URL where the run is being served.
    """
    import uvicorn

    from epochix.config import get_settings
    from epochix.server.app import create_app
    from epochix.server.hub import Hub
    from epochix.store.sqlite_store import RunStore

    settings = get_settings()
    effective_db = db or settings.db
    store = RunStore(db_path=effective_db)
    hub = Hub()

    _app = create_app(settings=settings)
    _app.state.store = store
    _app.state.hub = hub
    _app.state.engine_map = {}

    url = f"http://{host}:{port}/v/{run.id}"
    webbrowser.open(url)

    if blocking:
        uvicorn.run(_app, host=host, port=port, log_level="warning")
    else:
        import threading

        thread = threading.Thread(
            target=uvicorn.run,
            args=(_app,),
            kwargs={"host": host, "port": port, "log_level": "warning"},
            daemon=True,
        )
        thread.start()

    return url

epochix.sdk.visualize.serve(port=7860, host='127.0.0.1', db=None, open_browser=True)

Start the epochix server without a specific run.

Returns the base URL. Blocks until the server is stopped.

Source code in src/epochix/sdk/visualize.py

def serve(
    port: int = 7860,
    host: str = "127.0.0.1",
    db: str | None = None,
    open_browser: bool = True,
) -> str:
    """Start the epochix server without a specific run.

    Returns the base URL.  Blocks until the server is stopped.
    """
    import uvicorn

    from epochix.config import get_settings
    from epochix.server.app import create_app

    settings = get_settings()
    _app = create_app(settings=settings)

    url = f"http://{host}:{port}"
    if open_browser:
        webbrowser.open(url)

    uvicorn.run(_app, host=host, port=port, log_level="warning")
    return url

Core models

epochix.models.Run

Bases: BaseModel

Source code in src/epochix/models.py

class Run(BaseModel):
    id: str
    name: str | None = None
    task_type: TaskType
    started_at: datetime
    finished_at: datetime | None = None
    primary_metric: str
    framework_detected: str | None = None
    parser_used: str
    total_epochs_est: int | None = None
    final_grade: Grade | None = None
    story_summary: str | None = None
    config: dict[str, Any] = Field(default_factory=dict)

epochix.models.MetricEvent

Bases: BaseModel

Source code in src/epochix/models.py

class MetricEvent(BaseModel):
    run_id: str
    seq: int
    timestamp: datetime
    epoch: float | None = None
    step: int | None = None
    canonical_key: str
    raw_key: str
    value: FiniteFloat
    unit: str | None = None
    task_hint: TaskType | None = None

epochix.models.StoryFrame

Bases: BaseModel

Source code in src/epochix/models.py

class StoryFrame(BaseModel):
    run_id: str
    seq: int
    epoch: float | None = None
    progress: float = Field(ge=0.0, le=1.0)
    phase: Phase
    grade: Grade
    primary_metric_value: FiniteFloat
    confidence: float = Field(ge=0.0, le=1.0)
    narrative: str
    metaphor_cards: list[MetaphorCard] = Field(default_factory=list)
    skill_dimensions: dict[str, float] = Field(default_factory=dict)
    milestones: list[Milestone] = Field(default_factory=list)
    warnings: list[Warning] = Field(default_factory=list)
    task_type: TaskType

Plugin interface

epochix.parsers.base.BaseParser

Bases: Protocol

Protocol every parser must satisfy.

Source code in src/epochix/parsers/base.py

@runtime_checkable
class BaseParser(Protocol):
    """Protocol every parser must satisfy."""

    name: str
    priority: int

    def sniff(self, sample_lines: list[str]) -> float:
        """Return confidence 0.0–1.0 that this parser owns the format.

        Called on the first 50 lines (or 5 s of live data). Must be fast and
        stateless — it may be called multiple times with different samples.
        """
        ...

    def parse_line(self, line: str, ctx: ParserContext) -> list[RawMetric]:
        """Parse a single line and return zero or more RawMetric objects."""
        ...

parse_line(line, ctx)

Parse a single line and return zero or more RawMetric objects.

Source code in src/epochix/parsers/base.py

def parse_line(self, line: str, ctx: ParserContext) -> list[RawMetric]:
    """Parse a single line and return zero or more RawMetric objects."""
    ...

sniff(sample_lines)

Return confidence 0.0–1.0 that this parser owns the format.

Called on the first 50 lines (or 5 s of live data). Must be fast and stateless — it may be called multiple times with different samples.

Source code in src/epochix/parsers/base.py

def sniff(self, sample_lines: list[str]) -> float:
    """Return confidence 0.0–1.0 that this parser owns the format.

    Called on the first 50 lines (or 5 s of live data). Must be fast and
    stateless — it may be called multiple times with different samples.
    """
    ...

epochix.parsers.registry.register_parser(cls)

Decorator: register a parser class into the global registry.

Source code in src/epochix/parsers/registry.py

def register_parser(cls: type) -> type:
    """Decorator: register a parser class into the global registry."""
    instance = cls()
    _registry.append(instance)
    _registry.sort(key=lambda p: -p.priority)
    return cls