"""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": ""}``, ``{"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}