open-standards/wai/wai-quantum-py/python/wai_quantum/_api.py
Transaction Science 2593c1f945
Some checks failed
WAI breaking changes and the conformance runner / WAI breaking · does this change touch WAI (push) Successful in 7s
wai-quantum wasmQ / does this change touch it (push) Successful in 10s
wai-quantum energy-landscape jobs / does this change touch it (push) Successful in 28s
WAI breaking changes and the conformance runner / WAI breaking · wire, record and API against main · linux (push) Failing after 54s
wai-quantum wasmQ / the reference modules rebuild to their modules of record · linux (push) Successful in 40s
WAI breaking changes and the conformance runner / WAI breaking and conform · their tests, and the independent verifiers through wai conform · linux (push) Failing after 1m16s
joule-moq-remote host agent / joule-moq-remote · does this change touch it (push) Has been cancelled
joule-moq-remote host agent / joule-moq-remote, every target (linux-arm64) (push) Has been cancelled
joule-moq-remote host agent / joule-moq-remote, every target (macos) (push) Has been cancelled
joule-moq-room room service / joule-moq-room · does this change touch it (push) Has been cancelled
joule-moq-room room service / joule-moq-room, every target (push) Has been cancelled
joule-moq-web browser client / joule-moq-web · does this change touch it (push) Has been cancelled
joule-moq-web browser client / joule-moq-web · browser end-to-end (macos) (push) Has been cancelled
wai-app and Create / wai-app · does this change touch it (push) Has been cancelled
wai-app and Create / wai-app · tests, clippy, the line language's verifier (linux) (push) Has been cancelled
wai-app and Create / wai-app · Create, browser end-to-end (macos) (push) Has been cancelled
wai-audit collector and player integration / does this change touch it (push) Has been cancelled
wai-audit collector and player integration / collector · linux (push) Has been cancelled
wai-audit collector and player integration / collector, player integration and the browser end to end · macos (push) Has been cancelled
repository checks / lockfiles, advisories, tests in CI, workflow hygiene, spec and section references, public text (push) Has been cancelled
WAI operation catalogue / does this change touch it (push) Has been cancelled
WAI operation catalogue / the catalogue and every door · linux (push) Has been cancelled
wai-fleet service / does this change touch it (push) Has been cancelled
wai-fleet service / service · linux (push) Has been cancelled
wai-fleet service / service · macos (push) Has been cancelled
wai-ledger service / does this change touch it (push) Has been cancelled
wai-ledger service / service · linux (push) Has been cancelled
wai-ledger service / service · macos (push) Has been cancelled
wai-quantum for JavaScript / does this change touch it (push) Has been cancelled
wai-quantum for JavaScript / javascript · linux (push) Has been cancelled
wai-quantum studio and desktop / does this change touch it (push) Has been cancelled
wai-quantum studio and desktop / studio · browser (push) Has been cancelled
wai-quantum studio and desktop / desktop app · macos (push) Has been cancelled
wai-quantum studio and desktop / desktop app · linux (push) Has been cancelled
wai-quantum studio and desktop / desktop app · windows (push) Has been cancelled
wai-sink-android native sink library / does this change touch it (push) Has been cancelled
wai-sink-android native sink library / the JNI layer and the platform-free half on the host JVM · linux-arm64 (push) Has been cancelled
wai-sink-android native sink library / every ABI, the AAR, and the instrumented tests in a headless emulator · macos (push) Has been cancelled
wai-sink-swift native sink package / does this change touch it (push) Has been cancelled
wai-sink-swift native sink package / libwai slices, desktop tests, simulator tests · macos (push) Has been cancelled
WAI Studio / does this change touch it (push) Has been cancelled
WAI Studio / the catalogue's actions, the unit tests and the core · linux (push) Has been cancelled
WAI Studio / Studio in Chromium and Firefox · macos (push) Has been cancelled
WAI Studio / Studio in WebKit · linux (push) Has been cancelled
wai-quantum wasmQ / profile, engines, reference modules, adversarial, vectors · macos (push) Successful in 2m13s
wai-quantum energy-landscape jobs / the module under wasm32-wasip2 (wasmtime) (push) Successful in 2m33s
wai-quantum wasmQ / interpreter and vectors under wasm32-wasip2 (wasmtime) (push) Successful in 2m38s
wai-quantum energy-landscape jobs / the same bits · macos (push) Successful in 4m11s
wai-quantum wasmQ / profile, engines, reference modules, adversarial, vectors · linux (push) Successful in 4m51s
wai-quantum energy-landscape jobs / native against WebAssembly, over many seeds (push) Successful in 6m22s
wai-quantum energy-landscape jobs / the same bits · linux-arm64 (push) Successful in 7m45s
wai-quantum energy-landscape jobs / reader, kernels, answers and the stated result · linux, release (push) Successful in 9m14s
wai-quantum command line / does this change touch it (push) Has been cancelled
wai-quantum command line / command line · linux (push) Has been cancelled
wai-quantum command line / command line · macos (push) Has been cancelled
wai-quantum contract / does this change touch it (push) Has been cancelled
wai-quantum contract / API compatibility · linux (push) Has been cancelled
wai-quantum contract / schemas and the committed tool lists · linux (push) Has been cancelled
wai-quantum contract / the policy is linked, and the deprecations page reads (push) Has been cancelled
wai-quantum docs, course and locales / does this change touch it (push) Has been cancelled
wai-quantum docs, course and locales / every command and snippet in the docs · linux (push) Has been cancelled
wai-quantum docs, course and locales / the course and the command line's languages · linux (push) Has been cancelled
wai-quantum docs, course and locales / the Studio's words in every locale · linux (push) Has been cancelled
wai-quantum editors / does this change touch it (push) Has been cancelled
wai-quantum editors / editors · linux (push) Has been cancelled
wai-quantum editors / editors · macos (push) Has been cancelled
wai-quantum resource estimation / does this change touch it (push) Has been cancelled
wai-quantum resource estimation / estimation and calibration · linux, release (push) Has been cancelled
wai-quantum other formats / does this change touch it (push) Has been cancelled
wai-quantum other formats / readers, writers and the set · linux (push) Has been cancelled
wai-quantum other formats / the set under wasm32-wasip2 (wasmtime) (push) Has been cancelled
wai-quantum OpenQASM reader / does this change touch it (push) Has been cancelled
wai-quantum OpenQASM reader / reader and corpus · linux (push) Has been cancelled
wai-quantum OpenQASM reader / corpus under wasm32-wasip2 (wasmtime) (push) Has been cancelled
wai-quantum pulse and dynamics / does this change touch it (push) Has been cancelled
wai-quantum pulse and dynamics / reader, simulator and pulse jobs · linux, release (push) Has been cancelled
wai-quantum pulse and dynamics / the same bits · linux-arm64 (push) Has been cancelled
wai-quantum pulse and dynamics / the same bits · macos (push) Has been cancelled
wai-quantum pulse and dynamics / reader, simulator and pulse jobs under wasm32-wasip2 (wasmtime) (push) Has been cancelled
wai-quantum for Python / does this change touch it (push) Has been cancelled
wai-quantum for Python / python · linux (push) Has been cancelled
wai-quantum for Python / python · macos (push) Has been cancelled
wai-quantum error correction and mitigation / does this change touch it (push) Has been cancelled
wai-quantum error correction and mitigation / codes, decoders, texts and mitigation · linux, release (push) Has been cancelled
wai-quantum extension points / does this change touch it (push) Has been cancelled
wai-quantum extension points / the registry and the example · linux (push) Has been cancelled
wai-quantum extension points / the registry and the example · linux-arm64 (push) Has been cancelled
wai-quantum extension points / the registry under wasm32-wasip2 (wasmtime) (push) Has been cancelled
wai-quantum service / does this change touch it (push) Has been cancelled
wai-quantum service / service · linux (push) Has been cancelled
wai-quantum service / service · macos (push) Has been cancelled
wai-quantum service / service energy · linux packages (push) Has been cancelled
wai-quantum variational and algorithm library / does this change touch it (push) Has been cancelled
wai-quantum variational and algorithm library / parameters, gradients, optimisers, VQE, QAOA and builders · linux, release (push) Has been cancelled
wai-quantum variational and algorithm library / parameters, gradients, optimisers, VQE, QAOA and builders · linux-arm64, release (push) Has been cancelled
wai-quantum verdicts and repairs / does this change touch it (push) Has been cancelled
wai-quantum verdicts and repairs / verdicts, repairs and the intent set · linux (push) Has been cancelled
wai-quantum verdicts and repairs / the intent set and the verdict under wasm32-wasip2 (wasmtime) (push) Has been cancelled
wai-quantum views in agent hosts / does this change touch it (push) Has been cancelled
wai-quantum views in agent hosts / views · the page and its engine (push) Has been cancelled
wai-quantum views in agent hosts / views · a scripted host (push) Has been cancelled
wai-quantum-landscape-fpga reference design / does this change touch it (push) Has been cancelled
wai-quantum-landscape-fpga reference design / the member against the model of the logic, and the logic against the model · linux, release (push) Has been cancelled
wai-quantum-landscape-fpga reference design / the board's code on the board's processor's architecture · linux-arm64, release (push) Has been cancelled
wai: wasmQ, an OpenQASM 3 program and the WebAssembly its externs call, run the same everywhere, metered and receipted
Re-landed from PR #659 (branch wai/quantum-wasmq, head 39fdb517a) as one
commit on main, under the land-it model. Five review rounds; the last had
no high or critical finding, and its mediums were fixed on the branch.

A hybrid quantum program runs classical code between its gates. OpenQASM 3
declares such a function with `extern` and leaves its body and linking
outside the language. wasmQ (wai.quantum.wasmq) packs the program and the
WebAssembly that defines its externs into one content-addressed WAI
envelope; a run gives the same result bytes and fuel on every host and
seals the engine's job receipt.

- extensions/quantum-wasmq.md (normative) and quantum-wasmq.wit.
- wai/wai-wasmq: the profile (Lime1, nothing imported, bounded memory,
  refusal codes W001-W017); what a run asks of its host, bounded from the
  bundle alone before anything runs; fuel defined on the module as written;
  a weighted stack; two engines (the pinned wasmtime 50.0.1 and an
  interpreter of the crate's own that also runs on wasm32); the envelope,
  runner, receipts, transcript replay, reference programs and vectors.
- wai-quantum: engine::wasmq (built with quantum_receipt), the job form
  beside its receipts; the estimate verifier answers a request no host
  re-runs with a verdict carrying rerun_error, combined here with main's
  reaction clock (clock_of).
- wai/quantum-conformance/wasmq: envelopes, refusals and receipts.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-11-09 13:22:09 -05:00

594 lines
26 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""The engine's functions as a person calls them from Python.
Each one is the shared job engine, in this process: the same request gives
the same plan, the same result and the same receipt as the command line, the
service, the MCP servers, the Studio and the desktop app.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Union
from . import _engine
from ._job import Job, Verification
from .errors import EngineError
from .keys import Key, key_of
__all__ = [
"Program",
"job_request",
"run",
"plan",
"lint",
"draw",
"layout",
"critique",
"improve",
"shortcuts",
"equivalent",
"verify",
"verdict",
"repair",
"seal",
"canonical",
"digest",
"methods",
"accuracy_classes",
"bloch",
"history",
"estimate",
"verify_estimate",
"qec_code",
"qec_decode",
"qec_threshold",
"DEFAULT_MEMORY_GB",
]
#: The memory one job may take here, GiB: the command line's default, so a
#: job plans onto the same method on both.
DEFAULT_MEMORY_GB = 16
#: A program: OpenQASM 3 (or 2) text, or a path to a file of it.
Program = Union[str, "os.PathLike[str]"]
KeyLike = Union[Key, bytes, str, None]
def _source(program: Any) -> str:
if isinstance(program, os.PathLike):
return Path(program).read_text()
if isinstance(program, str):
return program
raise TypeError(f"a program is OpenQASM text or a path to a file of it, not {type(program).__name__}")
def job_request(
program: Union[Program, Mapping[str, Any]],
*,
shots: Optional[int] = None,
seed: Optional[int] = None,
method: Optional[str] = None,
observables: Optional[Sequence[Any]] = None,
amplitudes: Optional[Sequence[str]] = None,
noise: Optional[Mapping[str, float]] = None,
options: Optional[Mapping[str, Any]] = None,
intent: Any = None,
verdict: Optional[bool] = None,
repair: Any = None,
mitigate: Union[str, Mapping[str, Any], None] = None,
) -> Dict[str, Any]:
"""The service's request for a program (or a job dict) and what is asked
of it, with only the fields given, as every surface sends it.
``observables`` are Pauli strings such as ``"Z0 Z1"`` or
``{"pauli": "X3 Y4", "coeff": 0.5}``; ``amplitudes`` bit strings with
qubit n − 1 first; ``noise`` channels and their probabilities;
``options`` per method (``wai_quantum.methods()`` lists them).
``intent`` is what the program must do (``{"distribution": {...}}``,
``{"equivalent": "<a reference program>"}``, ``{"prepares": ...}``, or a
list of them): the result then carries the engine's verdict, sealed by
its receipt, as ``verdict=True`` asks with no intent. ``repair`` (True,
or the choices ``{"max_iterations", "dyadic_bits", "sizes"}``) repairs
the program first on the lines its diagnostics flag. ``mitigate`` (a
method's name, ``zne``, ``pea``, ``pec``, ``cdr``, ``vd`` or
``readout``, or ``{"method", ...}`` with its options) mitigates a noisy
run's values: the result's ``mitigation`` gives each value raw and
mitigated with its error bar, the method, and what the method assumes.
"""
if isinstance(program, Mapping):
req: Dict[str, Any] = dict(program)
else:
req = {"qasm": _source(program)}
given = {"shots": shots, "seed": seed, "method": method, "observables": observables, "amplitudes": amplitudes, "noise": noise, "options": options, "intent": intent, "verdict": verdict, "repair": repair, "mitigate": mitigate}
for k, v in given.items():
if v is None:
continue
if k in ("observables", "amplitudes"):
v = list(v)
elif k in ("noise", "options"):
v = dict(v)
elif k == "mitigate" and isinstance(v, Mapping):
v = dict(v)
req[k] = v
return req
def run(program: Union[Program, Mapping[str, Any]], *, key: KeyLike = None, memory_gb: int = DEFAULT_MEMORY_GB, **job: Any) -> Job:
"""Run a job here, as the service runs it: planned onto the strongest
method the program admits (or ``method=``), run, and the result sealed
with ``key`` (a ``wai_quantum.Key``, a 32-byte seed, 64 hex digits; by
default this user's key, the command line's).
``program`` is OpenQASM text, a path, or a job dict; the other keywords
are :func:`job_request`'s. Returns the job in the service's form.
"""
req = job_request(program, **job)
return Job(_engine.run(req, key_of(key).seed, memory_gb))
def plan(program: Union[Program, Mapping[str, Any]], *, memory_gb: int = DEFAULT_MEMORY_GB, **job: Any) -> Dict[str, Any]:
"""The method a job would run on and why, the memory and work it would
take, without running it."""
return _engine.plan(job_request(program, **job), memory_gb)
def lint(program: Program) -> Dict[str, Any]:
"""The program read as the engine reads it: ``{"ok": True}`` with its
qubits, gates, bits and measurements, or ``{"ok": False}`` with the
problem and its line. It never raises for a program that does not read:
saying so is its answer."""
return _engine.lint(_source(program))
def draw(program: Program, *, format: str = "text", width: int = 100, hands: Optional[Sequence[str]] = None) -> str:
"""The circuit drawn: ``format="text"`` (one wire per qubit, wrapped at
``width`` columns, as ``wai-quantum draw`` prints it) or ``"svg"`` (the
Studio's picture as a standalone SVG document, each gate marked with
whose hand it is in when ``hands`` gives one per operation)."""
src = _source(program)
if format == "text":
if hands is not None:
raise ValueError("the text drawing carries no authorship marks: use format=\"svg\"")
return _engine.draw_text(src, width)
if format == "svg":
return _engine.draw_svg(src, list(hands) if hands is not None else None)
raise ValueError(f"format is \"text\" or \"svg\", not {format!r}")
def layout(program: Program) -> Dict[str, Any]:
"""The drawing as data: the qubits, the columns (each a list of items:
their cells as ``[qubit, label]``, whether a line joins them, and for a
gate its operation's index) and the legend of conditions."""
return _engine.layout(_source(program))
def _answer(v: Dict[str, Any]) -> Dict[str, Any]:
if v.get("ok") is False:
raise EngineError(v.get("error") or "the program does not read", line=v.get("line"))
return v
def critique(program: Program) -> Dict[str, Any]:
"""What the engine can say about a design before it runs: its measures
(gates, depth, two-qubit gates, ...) and its smells, each with its lines
and whether a fix is offered. Raises ``EngineError`` when the program
does not read."""
return _answer(_engine.critique(_source(program)))
def improve(program: Program) -> Dict[str, Any]:
"""The variants the critique's fixes make, each checked equivalent to the
program before it is offered."""
return _answer(_engine.improve(_source(program)))
def shortcuts(program: Program, *, thorough: bool = False) -> Dict[str, Any]:
"""Every exact classical route the engine knows for a circuit, with its
cost on it, and the cheapest for each question. ``thorough`` also
searches a contraction tree (seconds to minutes on a large program)."""
return _answer(_engine.shortcuts(_source(program), thorough))
def equivalent(a: Program, b: Program) -> Dict[str, Any]:
"""Whether two programs apply the same gates up to a global phase:
``verdict`` (``equivalent``, ``different`` or ``unchecked``), ``how`` the
check decided it and ``why``."""
return _engine.equivalent(_source(a), _source(b))
def _job_of(job: Any) -> Mapping[str, Any]:
if isinstance(job, Mapping):
return job
if isinstance(job, str) and job.lstrip().startswith("{"):
return json.loads(job)
if isinstance(job, (str, os.PathLike)):
return json.loads(Path(job).read_text())
raise TypeError(f"a job is a dict, JSON text or a path to a JSON file, not {type(job).__name__}")
def _names_a_file(s: str) -> bool:
"""Whether a str names a file here: not text that fails as a name (too long, a NUL)."""
try:
return os.path.isfile(s)
except (OSError, ValueError):
return False
def _utf8(s: str) -> bytes:
"""A str's UTF-8 bytes; a lone surrogate, which no UTF-8 holds, as the bytes that are
not UTF-8, so the text is unreadable as a reader of its bytes finds it."""
return s.encode("utf-8", "surrogatepass")
def verify(job: Any) -> Verification:
"""Whether a job's receipt seals its request and its result, and every
rule of the provider protocol's section 8.4 holds, checked here, offline,
as every surface checks it: a job run here, by the command line, by the
service or in a browser tab. ``job`` is a dict (``{"request": ...,
"result": ...}``), JSON text or its bytes, or a path: a str is a path when
it names a file, and the job's text otherwise. Text and a path are read as
the protocol reads them, so a member named twice is found, which a dict can
no longer show. Every str and dict gets a verdict (text that does not read
is ``unreadable``), never an error. ``reasons`` names what is wrong."""
if isinstance(job, (bytes, bytearray)):
return Verification(_engine.verify_text(bytes(job)))
# A str is a path when it names a file, and otherwise the job's text, whatever it begins
# with: text that is no JSON, or no job, is a verdict like any other, never an error.
if isinstance(job, str) and not _names_a_file(job):
return Verification(_engine.verify_text(_utf8(job)))
if isinstance(job, (str, os.PathLike)):
return Verification(_engine.verify_text(Path(job).read_bytes()))
if isinstance(job, Mapping):
# A dict is judged as its text is, so that what a dict can show gets the verdict
# every surface gives: nesting deeper than a job's text holds is unreadable, a
# number past 64 bits is a number, read as its double, and a lone surrogate (which
# json.loads makes of its escape) is written as that escape, which reads as
# unreadable (section 8.2). A dict that is no JSON at all is unreadable too.
try:
text = json.dumps(job, ensure_ascii=True)
except (RecursionError, TypeError, ValueError):
return Verification({"valid": False, "reasons": ["unreadable"], "key": None, "form": None, "signer": None})
return Verification(_engine.verify_text(text.encode("ascii")))
return Verification(_engine.verify(_job_of(job)))
def verdict(job: Any, intent: Any = None, *, memory_gb: int = DEFAULT_MEMORY_GB) -> Dict[str, Any]:
"""How far the engine checked a job: its level (``refused``, ``parsed``,
``executed``, ``evidence``, ``proof``), whether its intents hold, each
intent's check, the critique's findings, the repairs made, and the level
in one plain ``statement``.
``job`` is a job (a dict, JSON text or a path: ``{"request", "result"}``,
or ``{"request", "error"}`` for one refused). With no ``intent``, the
result's own verdict, which its receipt seals (``sealed: True``); with
one, that intent checked here against the job's program, which says it
is not sealed. To have the verdict sealed, run with ``intent=``."""
return _engine.verdict(_job_of(job), intent, memory_gb)
def repair(
program: Program,
*,
proposed: Optional[Program] = None,
max_iterations: Optional[int] = None,
dyadic_bits: Optional[int] = None,
sizes: Optional[Mapping[str, int]] = None,
diagnostics: Optional[Sequence[Any]] = None,
rules: bool = True,
) -> Dict[str, Any]:
"""The program repaired on the lines its diagnostics flag alone:
``{"ok", "changed", "qasm", "repairs", "needs_person", "verdict"}``.
The engine makes the fixes that are mechanical (a missing ``;``, an
undeclared register's size, a measurement with no target, an unbounded
``while`` given ``max_iterations``, an angle the bit-exact method refuses
rounded to ``dyadic_bits``), each recorded with its code and every line
it changed; what needs a person comes with the reader's advice.
``proposed`` is a repair someone wrote: taken only if it changes no line
but those the diagnostics flag, otherwise ``EngineError`` with the code
``J021`` and nothing changed."""
o: Dict[str, Any] = {}
if proposed is not None:
o["proposed"] = _source(proposed)
for k, v in {"max_iterations": max_iterations, "dyadic_bits": dyadic_bits}.items():
if v is not None:
o[k] = v
if sizes is not None:
o["sizes"] = dict(sizes)
if diagnostics is not None:
o["diagnostics"] = list(diagnostics)
if not rules:
o["rules"] = False
return _engine.repair(_source(program), o)
def seal(request: Mapping[str, Any], result: Mapping[str, Any], *, key: KeyLike = None) -> Dict[str, Any]:
"""The receipt for ``result`` of ``request``, sealed with ``key``: the
Ed25519 signature over both in their RFC 8785 canonical JSON. A receipt
the result already carries is not part of what is sealed."""
return _engine.seal(request, result, key_of(key).seed)
def canonical(value: Any) -> bytes:
"""The RFC 8785 canonical JSON of ``value``: keys sorted, no whitespace,
each number as the double it is, written as ECMAScript writes it. The
bytes a receipt signs."""
return _engine.canonical(value)
def digest(request: Mapping[str, Any], result: Mapping[str, Any]) -> str:
"""The digest a receipt over ``result`` of ``request`` signs, as hex."""
return _engine.digest(request, result)
def methods(*, memory_gb: int = DEFAULT_MEMORY_GB) -> List[Dict[str, Any]]:
"""Each method, its accuracy, the circuits it takes, its options and the
most qubits it holds in a job of ``memory_gb`` GiB."""
return _engine.methods(memory_gb)
def accuracy_classes() -> Dict[str, str]:
"""The accuracy classes a result names, and what each promises."""
return _engine.accuracy_classes()
def history(op: str, args: Mapping[str, Any], *, key: KeyLike = None) -> Dict[str, Any]:
"""A program's history, attributed: who made each version and how, and
whose hand each line and gate is in (the engine's versions). ``op`` is
``start``, ``commit``, ``propose``, ``merge`` (each recorded version is
sealed with ``key``), ``changes``, ``apply``, ``blame``, ``attest``,
``log`` or ``check_job``; ``args`` as the Studio passes them."""
sealing = op in ("start", "commit", "propose", "merge")
return _engine.history(op, args, key_of(key).seed if sealing else None)
def _document(doc: Any) -> Any:
if isinstance(doc, Mapping):
return dict(doc)
if isinstance(doc, str) and doc.lstrip().startswith("{"):
return json.loads(doc)
if isinstance(doc, (str, os.PathLike)):
return json.loads(Path(doc).read_text())
raise TypeError(f"a document is a dict, JSON text or a path to a JSON file, not {type(doc).__name__}")
def estimate(
program: Optional[Program] = None,
*,
counts: Any = None,
routine: Any = None,
budget: Optional[float] = None,
qubit: Union[str, Mapping[str, Any], None] = None,
calibration: Optional[str] = None,
max_slowdown: Optional[int] = None,
export: Optional[str] = None,
name: Optional[str] = None,
key: KeyLike = None,
reaction_us: Optional[float] = None,
) -> Dict[str, Any]:
"""What a fault-tolerant machine would need to run ``program`` (or the
logical-counts document ``counts``, or the routine document
``routine``), as every surface estimates it.
The answer (``{"request", "result"}``) gives the program's logical
counts (T gates, Toffolis, rotations at their synthesised words' T
counts, measurements, depths, logical qubits with ancillas), the error
``budget`` (default 1e-3) and each rotation's share of it, the code
model's calibration (``calibration``: ``exact_matching``, the default,
``union_find`` or ``assumed``), the headline estimate on ``qubit`` (a
reference qubit's name, ``ns-e4`` by default, or a dict describing one)
and the frontier between physical qubits and runtime (slowed up to
``max_slowdown``). With ``reaction_us`` each Toffoli runs on the
reaction-limited clock, one tick of that reaction time in µs (no shorter
than a measurement on the qubit), instead of three logical steps. Every
estimate is sealed with ``key`` in the
resource-estimate receipt family; ``verify_estimate`` checks the seals
and re-runs it. ``export`` (``counts`` or ``routine``) adds the
program's counts as a document, named ``name``. A request the engine
cannot count raises ``EngineError`` with its code."""
given = [x is not None for x in (program, counts, routine)]
if sum(given) != 1:
raise TypeError("give one of program, counts and routine")
request: Dict[str, Any] = {}
if program is not None:
request["qasm"] = _source(program)
if counts is not None:
request["counts"] = _document(counts)
if routine is not None:
request["routine"] = _document(routine)
for k, v in (("budget", budget), ("qubit", qubit), ("calibration", calibration), ("max_slowdown", max_slowdown), ("export", export), ("name", name)):
if v is not None:
request[k] = dict(v) if isinstance(v, Mapping) else v
if reaction_us is not None:
request["clock"] = {"reaction_us": reaction_us}
return _engine.estimate(request, key_of(key).seed)
def verify_estimate(estimate: Any) -> Dict[str, Any]:
"""Whether an estimate holds, checked here: its seal, every estimate's
receipt (one key, the frontier's naming the headline's), and its request
re-run to the same receipts and figures. ``estimate`` is a dict, JSON
text or a path. ``{"valid", "signed", "receipts", "reproduced", "key",
"signer", "estimates"}``."""
return _engine.estimate_verify(_job_of(estimate))
def _qec(op: str, request: Dict[str, Any], key: KeyLike) -> Dict[str, Any]:
return _engine.qec(op, request, key_of(key).seed)
def _given(request: Dict[str, Any], **fields: Any) -> Dict[str, Any]:
for k, v in fields.items():
if v is None:
continue
if isinstance(v, Mapping):
v = dict(v)
elif isinstance(v, Sequence) and not isinstance(v, (str, bytes)):
v = list(v)
request[k] = v
return request
def qec_code(
code: Union[str, Mapping[str, Any]],
*,
certify: Optional[bool] = None,
bound: Optional[int] = None,
certificate: Any = None,
key: KeyLike = None,
) -> Dict[str, Any]:
"""An error-correcting code and its certified distance, as every surface
shows it.
``code`` is a name (``surface:D``, ``toric:L``, ``steane``, ``color:D``
for the triangular 6.6.6 colour code, ``bb:72``, ``bb:90``, ``bb:108``,
``bb:144`` or ``gross``, ``bb:288``) or ``{"family", "distance" | "size"
| "n"}``. The answer (``{"request", "result"}``, sealed with ``key``)
gives its checks (Z-type: they see X errors) and stabilizers, k counted
by rank, a layout to draw it, the decoders it takes, and its distance
certificate: in each sector the least weight of a logical found, a
witness of that weight, and every lighter support ruled out by an
exhaustive search, re-checkable node for node. ``bound`` is the heaviest
support searched (a search stopped there settles nothing past it);
``certify=False`` shows the code alone; ``certificate`` (a sealed answer,
the certificate alone, JSON text or a path) is checked against the code,
its witnesses and search run again."""
request: Dict[str, Any] = {"code": dict(code) if isinstance(code, Mapping) else code}
if certificate is not None:
doc = _document(certificate)
held = doc.get("result", {}).get("certificate") if isinstance(doc.get("result"), Mapping) else None
certificate = held if held is not None else doc
return _qec("code", _given(request, bound=bound, certify=certify, certificate=certificate), key)
def qec_decode(
code: Union[str, Mapping[str, Any], None] = None,
*,
syndrome: Optional[Sequence[int]] = None,
error: Optional[Sequence[int]] = None,
max_weight: Optional[int] = None,
memory: Union[Mapping[str, Any], Sequence[float], None] = None,
circuit: Optional[Program] = None,
model: Optional[Program] = None,
detectors: Optional[Sequence[int]] = None,
shots: Optional[int] = None,
decoder: Optional[str] = None,
window: Union[Mapping[str, int], Sequence[int], None] = None,
seed: Optional[int] = None,
key: KeyLike = None,
) -> Dict[str, Any]:
"""Decode with the engine's decoders, as every surface decodes.
At code capacity, give ``code`` and a ``syndrome`` (the violated checks)
or an ``error`` (the flipped qubits: its syndrome is decoded and its
residual judged, logical or not), with ``decoder`` ``matching``,
``union_find``, ``color`` (a colour code), ``relay_bp`` or ``search``
(the lightest error with the syndrome, up to ``max_weight``).
At circuit level, give ``memory`` (``{"distance", "rounds", "p"}`` or
``(distance, rounds, p)``: the rotated surface-code memory experiment),
``circuit`` (the circuit text, or a path to it) or ``model`` (detector
error model text, or a path), with the fired ``detectors`` (the answer
predicts which observables flipped) or ``shots`` (sampled with ``seed``
and decoded: the failures, the rate and its standard error), and
``decoder`` ``matching``, ``correlated``, ``union_find``, ``bp_osd`` or
``window`` (``window={"commit", "buffer"}`` rounds, or ``(commit,
buffer)``). The answer is sealed with ``key``."""
request: Dict[str, Any] = {}
if isinstance(memory, Sequence) and not isinstance(memory, (str, bytes)):
if len(memory) != 3:
raise TypeError("memory is {distance, rounds, p} or (distance, rounds, p)")
d, r, p = memory
memory = {"distance": int(d), "rounds": int(r), "p": float(p)}
if isinstance(window, Sequence) and not isinstance(window, (str, bytes)):
if len(window) != 2:
raise TypeError("window is {commit, buffer} or (commit, buffer)")
window = {"commit": int(window[0]), "buffer": int(window[1])}
_given(
request,
code=code,
syndrome=syndrome,
error=error,
max_weight=max_weight,
memory=memory,
circuit=None if circuit is None else _source(circuit),
model=None if model is None else _source(model),
detectors=detectors,
shots=shots,
window=window,
decoder=decoder,
seed=seed,
)
return _qec("decode", request, key)
def qec_threshold(
family: str = "surface",
*,
distances: Optional[Sequence[int]] = None,
p: Optional[Sequence[float]] = None,
shots: Optional[int] = None,
decoder: Optional[str] = None,
rounds: Optional[int] = None,
seed: Optional[int] = None,
key: KeyLike = None,
) -> Dict[str, Any]:
"""Logical failure rates over ``distances`` and physical rates ``p``, as
every surface measures them: ``family`` ``surface`` at circuit level
(``decoder`` ``matching``, ``correlated`` or ``union_find``; ``rounds``,
by default the distance; the code model fitted to the points), or
``color``, ``toric`` or ``surface_capacity`` at code capacity. Each point
gives its shots, failures, rate and standard error; ``below_threshold``
says, at each p, whether every larger distance failed less. Report rates
with their errors: a crossing measured on few shots is not a threshold.
The answer is sealed with ``key``."""
request: Dict[str, Any] = {"family": family}
_given(request, distances=distances, p=p, shots=shots, rounds=rounds, seed=seed, decoder=decoder)
return _qec("threshold", request, key)
def bloch(
program: Union[Program, Mapping[str, Any]],
*,
key: KeyLike = None,
memory_gb: int = DEFAULT_MEMORY_GB,
qubits: Optional[Iterable[int]] = None,
**job: Any,
) -> Dict[str, Any]:
"""Each qubit's reduced state as a Bloch vector, from the engine.
A qubit's reduced density matrix is ``(I + x X + y Y + z Z) / 2``, so its
vector is the expectation values of ``X``, ``Y`` and ``Z`` on it: one job
asks for them, planned and sealed as any other. Its length is 1 for a
pure reduced state and less when the qubit is mixed (entangled with the
others, or noisy). Returns ``{"qubits": [{"qubit", "x", "y", "z",
"length"}], "method", "accuracy", "job"}``; where the method bounds or
estimates the values, each vector carries the ``bound`` or ``stderr`` of
its three components.
"""
req = job_request(program, **job)
n = lint(req["qasm"])
if not n.get("ok"):
raise EngineError(n.get("error") or "the program does not read", line=n.get("line"))
which = list(range(int(n["qubits"]))) if qubits is None else [int(q) for q in qubits]
if 3 * len(which) > 256:
raise ValueError(f"{len(which)} qubits need {3 * len(which)} observables; a job takes at most 256 (give qubits=)")
req["observables"] = [f"{p}{q}" for q in which for p in "XYZ"]
done = run(req, key=key, memory_gb=memory_gb)
entries = done.result.get("expectations") or []
out = []
for i, q in enumerate(which):
x, y, z = (entries[3 * i + j] for j in range(3))
v: Dict[str, Any] = {"qubit": q, "x": x["value"], "y": y["value"], "z": z["value"]}
v["length"] = (v["x"] ** 2 + v["y"] ** 2 + v["z"] ** 2) ** 0.5
# What the method says of each component's error: a rigorous bound,
# or a standard error, as the expectation values carry them.
for field in ("bound", "stderr"):
errs = [e.get(field) for e in (x, y, z)]
if any(b is not None for b in errs):
v[field] = errs
out.append(v)
return {"qubits": out, "method": done.method, "accuracy": done.accuracy, "job": done}