open-standards/wai/wai-quantum-py/python/wai_quantum/_data.py

"""The data each widget draws: plain values built from the engine's answers.

Kept apart from the widgets so that a job's static card (``Job._repr_html_``)
needs no widget library.
"""

from __future__ import annotations

from typing import Any, Dict, List, Mapping, Optional, Sequence, Union

from . import _api, _engine
from ._job import Job
from .keys import DEMO_SEED, Key

__all__ = ["circuit_data", "result_data", "receipt_data", "bloch_data"]

_DEMO_KEY = Key(DEMO_SEED).public_key


def circuit_data(program: _api.Program, *, hands: Optional[Sequence[str]] = None, history: Optional[Mapping[str, Any]] = None, version: Optional[str] = None) -> Dict[str, Any]:
    """A circuit's drawing: the engine's SVG, its text drawing and layout,
    and, when known, whose hand each gate is in (``hands``, one per
    operation, or read from a program ``history`` at ``version``) with the
    key's counts."""
    src = _api._source(program)
    if history is not None:
        if hands is not None:
            raise ValueError("give hands or a history, not both")
        blame = _api.history("blame", {"history": history, "version": version or _head(history)})
        if blame.get("unread"):
            raise ValueError(f"the history's version does not read: {blame['unread']}")
        hands = blame.get("gates")
    hands = list(hands) if hands is not None else None
    key: List[Dict[str, Any]] = []
    if hands:
        for name, label in _engine.HANDS:
            n = sum(1 for h in hands if h == name)
            if n:
                key.append({"hand": name, "label": label, "gates": n})
    return {
        "svg": _api.draw(src, format="svg", hands=hands),
        "text": _api.draw(src),
        "layout": _api.layout(src),
        "hands": hands,
        "key": key,
    }


def _head(history: Mapping[str, Any]) -> str:
    log = _api.history("log", {"history": history})
    heads = [v["id"] for v in log.get("versions", []) if v.get("head")]
    if len(heads) != 1:
        raise ValueError(f"the history has {len(heads)} heads: give version=")
    return heads[0]


def _seal_check(job: Mapping[str, Any]) -> Dict[str, Any]:
    v = _api.verify(job)
    r = (job.get("result") or {}).get("receipt") or {}
    return {
        "valid": v.valid,
        "signer": v.signer,
        "key": v.key,
        "digest": r.get("digest"),
        "signature": r.get("signature"),
        "alg": r.get("alg"),
        "canon": r.get("canon"),
        "demo": v.key == _DEMO_KEY,
        "binds": "the request and the result as given",
    }


def result_data(job: Mapping[str, Any]) -> Dict[str, Any]:
    """A result's histogram and figures: its counts (the most frequent
    first), the shots, its expectation values and amplitudes, the method, the
    accuracy class and what it says, and the receipt's check when it has
    one."""
    j = job if isinstance(job, Job) else Job(job)
    r = j.result
    acc = r.get("accuracy")
    note = None
    if isinstance(acc, dict):
        note = acc.get("note")
    data: Dict[str, Any] = {
        "method": j.method,
        "accuracy": j.accuracy,
        "accuracy_note": note,
        "why": j.get("why"),
        "counts": j.counts,
        "shots": j.shots,
        "distinct_outcomes": r.get("distinct_outcomes", len(j.counts)),
        "omitted_shots": r.get("omitted_shots", 0),
        "expectations": r.get("expectations") or [],
        "amplitudes": r.get("amplitudes") or [],
        "receipt": _seal_check(j) if j.receipt else None,
    }
    return data


def receipt_data(job: Mapping[str, Any]) -> Dict[str, Any]:
    """A receipt as the reader is shown it: the signer, the key, the digest
    and the signature, and whether it seals the request and the result,
    checked here by the engine."""
    if not (job.get("result") or {}).get("receipt"):
        raise ValueError("the result carries no receipt")
    return _seal_check(job)


def bloch_data(program: Union[_api.Program, Mapping[str, Any]], **kw: Any) -> Dict[str, Any]:
    """Each qubit's Bloch vector from the engine (``wai_quantum.bloch``), or
    from what ``bloch`` returned. The job itself is kept out of the data."""
    if isinstance(program, Mapping) and "qubits" in program and "qasm" not in program:
        b = program
    else:
        b = _api.bloch(program, **kw)
    return {"qubits": [dict(q) for q in b["qubits"]], "method": b["method"], "accuracy": b["accuracy"]}