qnetbench.trace

The versioned JSONL event schema and its I/O. This package and qnetbench.api are the two frozen contracts of the suite.

Narrative guide: Traces; the versioned contract itself is in the spec.

Events

qnetbench.trace.events

The versioned trace event schema — the frozen wire contract.

A run is a stream of these, one JSON object per line. Metrics are computed from traces, never inline, so any third-party tool can consume a run without importing qnetbench. Bump SCHEMA_VERSION (semver) on any change to these models.

SCHEMA_VERSION module-attribute

SCHEMA_VERSION = '0.2.0'

Event module-attribute

Event = Annotated[
    RunHeader
    | EntanglementRequested
    | EntanglementDelivered
    | ContractViolation
    | ClassicalMessage
    | Measurement
    | QubitSent
    | AppOutcomeEvent,
    Field(discriminator="kind"),
]

RunHeader

Bases: _Event

kind class-attribute instance-attribute

kind: Literal['run_header'] = 'run_header'

schema_version class-attribute instance-attribute

schema_version: str = SCHEMA_VERSION

api_version instance-attribute

api_version: str

app instance-attribute

app: str

backend instance-attribute

backend: str

arbitration instance-attribute

arbitration: str

topology instance-attribute

topology: str

seed instance-attribute

seed: int

EntanglementRequested

Bases: _Event

kind class-attribute instance-attribute

kind: Literal['ent_requested'] = 'ent_requested'

req_id instance-attribute

req_id: int

src instance-attribute

src: str

dst instance-attribute

dst: str

n instance-attribute

n: int

demand instance-attribute

demand: Demand

EntanglementDelivered

Bases: _Event

kind class-attribute instance-attribute

kind: Literal['ent_delivered'] = 'ent_delivered'

req_id instance-attribute

req_id: int

actual_fidelity instance-attribute

actual_fidelity: float

latency instance-attribute

latency: float

pair_age instance-attribute

pair_age: float

ContractViolation

Bases: _Event

kind class-attribute instance-attribute

kind: Literal['contract_violation'] = 'contract_violation'

req_id instance-attribute

req_id: int

violation instance-attribute

violation: ViolationKind

ClassicalMessage

Bases: _Event

kind class-attribute instance-attribute

kind: Literal['classical_msg'] = 'classical_msg'

src instance-attribute

src: str

dst instance-attribute

dst: str

n_bytes instance-attribute

n_bytes: int

Measurement

Bases: _Event

kind class-attribute instance-attribute

kind: Literal['measurement'] = 'measurement'

node instance-attribute

node: str

basis instance-attribute

basis: str

result instance-attribute

result: int

QubitSent

Bases: _Event

A single qubit transmitted over a quantum channel (prepare-and-measure protocols such as BB84), rather than a shared entangled pair.

kind class-attribute instance-attribute

kind: Literal['qubit_sent'] = 'qubit_sent'

src instance-attribute

src: str

dst instance-attribute

dst: str

fidelity instance-attribute

fidelity: float

AppOutcomeEvent

Bases: _Event

kind class-attribute instance-attribute

kind: Literal['app_outcome'] = 'app_outcome'

role instance-attribute

role: str

node instance-attribute

node: str

success instance-attribute

success: bool

utility instance-attribute

utility: float

payload class-attribute instance-attribute

payload: dict[str, object] = Field(default_factory=dict)

I/O

qnetbench.trace.io

Read and write traces as JSONL. One event per line.

TraceWriter

TraceWriter(stream: TextIO)

Append events to a JSONL stream. Usable as a context manager.

Source code in qnetbench/trace/io.py

def __init__(self, stream: TextIO) -> None:
    self._stream = stream

write

write(event: Event) -> None

Source code in qnetbench/trace/io.py

def write(self, event: Event) -> None:
    self._stream.write(_ADAPTER.dump_json(event).decode("utf-8"))
    self._stream.write("\n")

write_trace

write_trace(
    path: str | Path, events: Iterable[Event]
) -> None

Source code in qnetbench/trace/io.py

def write_trace(path: str | Path, events: Iterable[Event]) -> None:
    with open(path, "w", encoding="utf-8") as fh:
        writer = TraceWriter(fh)
        for event in events:
            writer.write(event)

read_trace

read_trace(path: str | Path) -> Iterator[Event]

Stream events back from a JSONL trace file.

Source code in qnetbench/trace/io.py

def read_trace(path: str | Path) -> Iterator[Event]:
    """Stream events back from a JSONL trace file."""
    with open(path, encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if line:
                yield _ADAPTER.validate_json(line)

parse_event

parse_event(line: str) -> Event

Source code in qnetbench/trace/io.py

def parse_event(line: str) -> Event:
    return _ADAPTER.validate_json(line)