Writer API¶

The Writer is the Python integration surface for CEMI. It builds a run snapshot in memory and emits run_record events that the local gateway and workspace consume.

Creating a writer¶

Local (explicit)From environment (cemi start)

from cemi.writer import create_writer

writer = create_writer(project="demo", log_dir=".cemi")

Use this when you are running the script directly, not through cemi start.

from cemi.writer import create_writer_from_env

writer = create_writer_from_env()

Use this when your script is launched by cemi start. The CLI injects CEMI_PROJECT_ID, CEMI_RUN_ID, CEMI_SAVE_DIR, and CEMI_LOCAL_SERVER_URL into the child process environment.

Run lifecycle¶

Context manager (recommended)¶

with writer.run(name="mobilenet-baseline", tags={"model": "mobilenetv2"}):
    writer.log_parameter(key="quantization", value="ptq-int8")
    writer.log_metric(name="loss", value=0.42, step=1)
    writer.emit_run_record()

The context manager calls start_run() on entry and end_run(status="succeeded") on exit. If an exception is raised, it calls end_run(status="failed") before re-raising.

Explicit lifecycle¶

writer.start_run(name="mobilenet-baseline", tags={"model": "mobilenetv2"})

writer.log_parameter(key="quantization", value="ptq-int8")
writer.log_metric(name="loss", value=0.42, step=1)
writer.emit_run_record()

writer.end_run(status="succeeded")
writer.emit_run_record()

`start_run(...)`¶

Initializes a new run and resets all in-memory accumulators.

Argument	Type	Description
`name`	`str`	Human-readable run name
`tags`	`dict`	Key/value pairs attached to the run
`run_id`	`str`	Explicit run ID; defaults to `CEMI_RUN_ID` env var, then a generated UUID
`project`	`str`	Project name; defaults to the writer's project
`stage`	`str`	Stage label (e.g., `"benchmark"`, `"production"`)
`status`	`str`	Initial status; usually `"running"`

Returns the resolved run_id.

`emit_run_record()`¶

Writes one complete snapshot to the configured sink. Every call appends a new line to save_dir/runs/<run_id>.jsonl. The gateway uses the last valid line as the current view of the run.

Call this at intermediate points to make progress visible in the workspace, and again after end_run() to write the final state.

`end_run(...)`¶

Marks the run complete and records end timestamps for the next emitted snapshot.

Argument	Type	Description
`status`	`str`	Final status: `"succeeded"`, `"failed"`, or `"cancelled"`
`ended_at`	`str`	ISO timestamp override (optional)

Parameters¶

Parameters are configuration values and static metadata. They appear in the Runs table and run detail view.

writer.log_parameter(key="batch_size", value=32)
writer.log_parameter(key="optimizer", value="adamw")
writer.log_parameter(key="quantization", value="ptq-int8")

Metrics¶

Time-series metrics¶

Use log_metric() for values that should appear as charts over steps or time.

writer.log_metric(name="loss", value=0.42, step=1)

writer.log_metric(
    name="latency_p99_ms",
    value=18.4,
    unit="ms",
    role="performance",
    aggregation="p99",
    direction="lower_is_better",
    step=1,
)

Argument	Type	Values	Description
`name`	`str`		Metric name
`value`	`float`		Observed value
`step`	`int`		Training step or iteration index (optional)
`timestamp_ms`	`int`		Unix timestamp in milliseconds (optional; defaults to now)
`unit`	`str`		Unit label (e.g., `"ms"`, `"MB"`, `"ips"`)
`role`	`str`	`quality`, `performance`, `resource`, `cost`, `custom`	Semantic category for UI grouping and contract gate matching
`aggregation`	`str`	`raw`, `mean`, `min`, `max`, `p50`, `p90`, `p95`, `p99`, `sum`, `count`, `last`	How this point was computed
`direction`	`str`	`higher_is_better`, `lower_is_better`, `none`	Optimization direction (used by Compare and contracts)
`tags`	`dict`		Scenario or sub-group tags (e.g., `{"scenario": "interactive"}`)

Summary metrics¶

Use summary metrics for aggregate values — the single number you want to appear in the Runs table and feed into contract gates.

writer.log_summary_metrics({
    "final_accuracy": 0.934,
    "model_size_mb": 4.2,
    "peak_ram_mb": 480,
})

# Single value
writer.log_summary_metric("throughput_ips", 1250.0)

For typed summaries (with role, aggregation, direction) stored in metrics.summary:

writer.log_summary(
    name="latency_p99_ms",
    value=18.4,
    aggregation="p99",
    role="performance",
    unit="ms",
    direction="lower_is_better",
)

Scalars (table-only)¶

Use log_scalar() for values that should appear in run tables but are not meant to drive chart widgets.

writer.log_scalar("memory_usage_mb", 512, unit="MB")
writer.log_scalar("throughput_p99", 1200)

# Bulk
writer.log_scalars({"peak_ram_mb": 480, "flash_mb": 128})

Context namespaces¶

The Writer supports structured context namespaces so runs carry richer comparison metadata. These are stored under payload.context in the run_record.

Case metadata¶

Use for test suite and scenario information:

writer.case.set(
    suite="ptq-benchmark",
    task="classification",
    scenario="offline",
    dataset="imagenet-val",
)

Policy metadata¶

Use for the optimization objective driving this run:

writer.policy.set(
    name="accuracy-first",
    objective_metric="final_accuracy",
    objective_direction="higher_is_better",
)

Device metadata¶

Use for the target hardware and runtime:

writer.device.set(
    board="stm32h7",
    runtime="onnxruntime",
    memory_budget="512MB",
    flash_budget="2MB",
    ram_budget="256MB",
)

Context values are also mirrored into compatibility parameters like case.dataset and device.board for UI display.

Tags and metadata¶

# Set all tags at once
writer.set_tags({"model": "mobilenetv2", "dataset": "cifar10"})

# Add or update a single tag
writer.add_tag("quantization", "ptq-int8")

# Add human-readable notes
writer.set_notes("Baseline run before quantization sweep.")

Run lineage¶

Link a run to a baseline or parent for comparison in the workspace:

writer.set_lineage(
    baseline_run_id="local-abc123",
    parent_run_id="local-def456",
)

Artifacts¶

Register an external URI¶

writer.add_artifact(
    kind="model",
    name="model.onnx",
    uri="https://example.com/model.onnx",
    media_type="application/octet-stream",
)

Copy a local file into the CEMI artifact store¶

writer.add_local_file_artifact(
    path="artifacts/model.onnx",
    kind="model",
)

This copies the file to:

.cemi/artifacts/<run_id>/model.onnx

And generates a gateway-backed URL:

http://127.0.0.1:3141/api/runs/<run_id>/artifacts/model.onnx

Artifact kinds: model, checkpoint, report, graph, dataset, other.

Do not attach secrets, credentials, private datasets, or files you would not want duplicated locally.

Contract results¶

Attach a verification result to a run snapshot directly from the Writer (for programmatic contract evaluation):

writer.log_contract_result({
    "contract_id": "accuracy-sla",
    "verdict": "pass",
    "evaluated_at_ms": 1730000000000,
    "gates": [
        {"id": "accuracy-gate", "verdict": "pass", "metric": "accuracy", "threshold": 0.9, "actual": 0.934}
    ],
})

This stores the result under payload.contract_result in the run record and surfaces a pass/fail badge in the workspace Runs table (v0.2.0+).

Benchmark helpers¶

The Writer includes convenience methods for MLPerf-style benchmark integrations. These keep common benchmarking metadata consistent without changing the underlying schema.

`log_benchmark_config(...)`¶

Logs benchmark configuration as parameters:

writer.log_benchmark_config(
    benchmark_task="image_classification",
    benchmark_scenario="offline",
)

`log_latency_sample(...)`¶

Records a single latency measurement as a time-series metric event:

writer.log_latency_sample(latency_ms=17.8, step=42, scenario="offline")

`log_operator_hotspot(...)`¶

Records an operator-level profiling entry:

writer.log_operator_hotspot(
    op_name="Conv2d",
    latency_ms=4.1,
    step=1,
)

Timestamp overrides¶

Override run timestamps directly (useful when wrapping existing benchmark harnesses):

writer.set_times(
    created_at="2026-04-13T10:00:00Z",
    started_at="2026-04-13T10:00:01Z",
    ended_at="2026-04-13T10:05:00Z",
)

Full example¶

from cemi.writer import create_writer

writer = create_writer(project="ptq-sweep", log_dir=".cemi")

with writer.run(name="mobilenetv2-ptq-int8", tags={"model": "mobilenetv2", "quant": "int8"}):
    # Context
    writer.device.set(board="stm32h7", runtime="onnxruntime", memory_budget="512MB")
    writer.case.set(suite="ptq-benchmark", task="classification", dataset="imagenet-val")
    writer.policy.set(name="accuracy-first", objective_metric="final_accuracy", objective_direction="higher_is_better")

    # Configuration
    writer.log_parameter(key="quantization_scheme", value="ptq-int8")
    writer.log_parameter(key="calibration_samples", value=1000)

    # Per-step metrics
    for step in range(10):
        writer.log_metric("loss", value=0.5 - step * 0.01, step=step)
        writer.log_metric("latency_ms", value=18.0 + step * 0.1, unit="ms", role="performance", direction="lower_is_better", step=step)

    # Summary
    writer.log_summary_metrics({
        "final_accuracy": 0.934,
        "model_size_mb": 4.2,
        "peak_ram_mb": 480,
    })

    # Artifact
    writer.add_local_file_artifact("outputs/model.onnx", kind="model")

    # Emit intermediate snapshot
    writer.emit_run_record()