d9dabd9aa7
qec_code, qec_decode and qec_threshold call the engine's error correction in-process, each answer sealed in a job's form, and run() takes mitigate=. The tests decode every error of weight up to two on the distance-5 colour code to a trivial residual, check a certificate and a forged one, and hold each answer to the command line's. Co-Authored-By: Claude Opus 5.5 <[email protected]>
550 lines
23 KiB
Python
550 lines
23 KiB
Python
"""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 verify(job: Any) -> Verification:
|
||
"""Whether a job's receipt seals its request and its result, 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 a path."""
|
||
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,
|
||
) -> 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``). 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
|
||
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}
|