Source code for pennylane_ionq.device

# Copyright 2019-2021 Xanadu Quantum Technologies Inc.

# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at

#     http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
This module contains the device class for constructing IonQ devices for PennyLane.
"""

# pylint: disable=too-many-arguments

import inspect
import logging
import warnings
from time import sleep

import numpy as np

from pennylane import BooleanFn, pauli_decompose, SparseHamiltonian
from pennylane.devices import QubitDevice
from pennylane.measurements import MidMeasureMP
from pennylane.ops.op_math import Conditional, Exp, Sum, SProd
from pennylane.ops import Identity, PauliX, PauliY, PauliZ
from pennylane.ops.op_math.prod import Prod

from pennylane.ops.op_math.linear_combination import LinearCombination

from .api_client import Job, JobExecutionError, ResultsTypes
from .qasm3 import operations_to_qasm3
from .exceptions import (
    CircuitIndexNotSetException,
    ComplexEvolutionCoefficientsNotSupported,
    NotSupportedEvolutionInstance,
    OperatorNotSupportedInEvolutionGateGenerator,
)
from ._version import __version__

logger = logging.getLogger(__name__)
logger.addHandler(logging.NullHandler())

_qis_operation_map = {
    # native PennyLane operations also native to IonQ
    "PauliX": "x",
    "PauliY": "y",
    "PauliZ": "z",
    "Hadamard": "h",
    "CNOT": "cnot",
    # IonQ expresses multi-controlled gates with a list of controls
    "Toffoli": "x",
    "Evolution": "pauliexp",
    "SWAP": "swap",
    "RX": "rx",
    "RY": "ry",
    "RZ": "rz",
    "S": "s",
    "S.inv": "si",
    "T": "t",
    "T.inv": "ti",
    "SX": "v",
    "SX.inv": "vi",
    # Ising gates defined in this plugin for IonQ hardware
    "XX": "xx",
    "YY": "yy",
    "ZZ": "zz",
    # PennyLane native Ising gates (equivalent to plugin gates above)
    "IsingXX": "xx",
    "IsingYY": "yy",
    "IsingZZ": "zz",
}

_native_operation_map = {
    "GPI": "gpi",
    "GPI2": "gpi2",
    "MS": "ms",
}

_GATESET_OPS = {
    "native": _native_operation_map,
    "qis": _qis_operation_map,
}

PAULI_MAP = {"PauliX": "X", "PauliY": "Y", "PauliZ": "Z", "Identity": "I"}

NO_ANALYTIC_MSG = "The ionq device does not support analytic expectation values."

NO_RESULTS_MSG = "No results are available. Execute a circuit on the device first."

NO_CONDITIONAL_MSG = (
    "Classically controlled operations (qp.cond) are not supported by the "
    "IonQ device. Apply qp.defer_measurements to the circuit (or use "
    'mcm_method="deferred") to convert conditionals to controlled gates.'
)


def circuit_requires_qasm3(circuit):
    """Whether a circuit must be submitted as an ``ionq.qasm3.v1`` job.

    True for circuits containing a mid-circuit measurement (including
    measurements with reset); these are not expressible as a flat
    ``ionq.circuit.v1`` gate list and are submitted as an OpenQASM 3
    program instead.

    Args:
        circuit (QuantumTape or Iterable[Operation]): a tape, or its operations

    Returns:
        bool: whether the qasm3 submission path is required

    Raises:
        ValueError: if the circuit contains a classically controlled
            operation (``qp.cond``), which is not supported
    """
    operations = list(getattr(circuit, "operations", circuit))
    if any(isinstance(operation, Conditional) for operation in operations):
        raise ValueError(NO_CONDITIONAL_MSG)
    return any(isinstance(operation, MidMeasureMP) for operation in operations)


class IonQDevice(QubitDevice):
    r"""IonQ device for PennyLane.

    Args:
        wires (int or Iterable[Number, str]]): Number of wires to initialize the device with,
            or iterable that contains unique labels for the subsystems as numbers (i.e., ``[-1, 0, 2]``)
            or strings (``['ancilla', 'q1', 'q2']``).

            .. note::

                Custom wire labels (e.g., strings or non-consecutive integers) are used for user convenience only.
                They have no effect on the transpilation process or the final qubit layout on the hardware.

    Kwargs:
        target (str): the target device, either ``"simulator"`` or ``"qpu"``. Defaults to ``simulator``.
        gateset (str): the target gateset, either ``"qis"`` or ``"native"``. Defaults to ``qis``.
        shots (int, list[int], None): Number of circuit evaluations/random samples used to estimate
            expectation values of observables. Defaults to None.
            If a list of integers is passed, the circuit evaluations are batched over the list of shots.
        job_name (str | None): Optional job name. Defaults to None.
        api_key (str): The IonQ API key. If not provided, the environment
            variable ``IONQ_API_KEY`` is used.
        compilation (dict | None): Settings for compilation when creating a job. Defaults to None.
            Example: ``{"opt": 0, "precision": "1E-3"}``. See
            `IonQ API Job Creation <https://docs.ionq.com/api-reference/v0.4/jobs/create-job>`_ for details.
        error_mitigation (dict | None): settings for error mitigation when creating a job. Defaults to None.
            Not available on all backends. Set by default on some hardware systems. See
            `IonQ API Job Creation <https://docs.ionq.com/api-reference/v0.4/jobs/create-job>`_  and
            `IonQ Debiasing and Sharpening <https://ionq.com/resources/debiasing-and-sharpening>`_ for details.
            Valid keys include: ``debiasing`` (bool).
        sharpen (bool): whether to use sharpening when accessing the results of an executed job. Defaults to None
            (no value passed at job retrieval). Will generally return more accurate results if your expected output
            distribution has peaks. See `IonQ Debiasing and Sharpening
            <https://ionq.com/resources/debiasing-and-sharpening>`_ for details.
        noise_model (str): the noise model to use for simulation. Only applies when ``target="simulator"``.
            Valid values are ``"ideal"``, ``"harmony"``, ``"aria-1"``, ``"aria-2"``, ``"forte-1"``,
            and ``"forte-enterprise-1"``. Defaults to None (ideal simulation). See
            `IonQ Simulation with Noise Models <https://docs.ionq.com/guides/simulation-with-noise-models>`_
            for details.
        noise_seed (int): seed for the noise model random number generator, for reproducible noisy
            simulation results. Must be an integer between 1 and 2\ :sup:`31` - 1. Only used when ``noise_model`` is set.
            Defaults to None (random seed).
        memory (bool): if True, use the shotwise measurement outcomes returned by the IonQ API as the
            device samples, instead of generating samples locally from the returned probabilities.
            Shotwise results are not available for ideal simulation. Defaults to False.
        dry_run (bool): If True, the job will be submitted by the API client but not processed remotely.
            Useful for obtaining cost estimates. Defaults to False.
        metadata (dict | None): optional metadata to attach to the job. Defaults to None.
        timeout (float): Request timeout in seconds. Defaults to None, which uses the
            ``APIClient`` default.
        max_retries (int): Maximum number of retries for retriable HTTP errors. Defaults to
            None, which uses the ``APIClient`` default.
        retry_delay (float): Base delay in seconds between retries (uses exponential backoff).
            Defaults to None, which uses the ``APIClient`` default.
    """

    # pylint: disable=too-many-instance-attributes
    name = "IonQ PennyLane plugin"
    short_name = "ionq"
    pennylane_requires = ">=0.44.0"
    version = __version__
    author = "Xanadu Inc."

    _capabilities = {
        "model": "qubit",
        "tensor_observables": True,
        "inverse_operations": True,
        "supports_mid_measure": True,
    }

    # Note: unlike QubitDevice, IonQ does not support QubitUnitary,
    # and therefore does not support the Hermitian observable.
    observables = {"PauliX", "PauliY", "PauliZ", "Hadamard", "Identity", "Prod"}

    # pylint: disable=too-many-arguments
    NOISE_MODELS = {
        "ideal",
        "harmony",
        "aria-1",
        "aria-2",
        "forte-1",
        "forte-enterprise-1",
    }

    def __init__(
        self,
        wires,
        *,
        target="simulator",
        gateset="qis",
        shots=None,
        job_name=None,
        api_key=None,
        compilation=None,
        error_mitigation=None,
        sharpen=None,
        noise_model=None,
        noise_seed=None,
        memory=False,
        dry_run=False,
        metadata=None,
        timeout=None,
        max_retries=None,
        retry_delay=None,
    ):

        if noise_model is not None and noise_model not in self.NOISE_MODELS:
            raise ValueError(
                f"Invalid noise model '{noise_model}'. Valid options are: "
                f"{', '.join(sorted(self.NOISE_MODELS))}."
            )
        if noise_seed is not None and noise_model is None:
            raise ValueError("noise_seed requires noise_model to be set.")
        if noise_seed is not None:
            if (
                isinstance(noise_seed, bool)
                or not isinstance(noise_seed, int)
                or not 1 <= noise_seed <= 2**31 - 1
            ):
                raise ValueError(
                    f"noise_seed must be an integer between 1 and 2^31 - 1, got {noise_seed}."
                )
        if memory and target == "simulator" and noise_model in (None, "ideal"):
            warnings.warn(
                "Shotwise results are not available for ideal simulation; samples "
                "will be generated from the returned probabilities.",
                UserWarning,
            )
            memory = False

        super().__init__(wires=wires, shots=shots)
        self._current_circuit_index = None
        self.job_name = job_name
        self.target = target
        self.api_key = api_key
        self.gateset = gateset
        self.compilation = compilation
        self.error_mitigation = error_mitigation
        self.sharpen = sharpen
        self.noise_model = noise_model
        self.noise_seed = noise_seed
        self.memory = memory
        self.dry_run = dry_run
        self.metadata = metadata
        self._operation_map = _GATESET_OPS[gateset]
        self.histograms = []
        self.memory_results = None
        self._samples = None

        # API client configuration.
        self.timeout = timeout
        self.max_retries = max_retries
        self.retry_delay = retry_delay

        self.reset()

    def batch_transform(self, circuit):
        """Apply a batch transform for preprocessing a circuit prior to execution."""

        if not circuit.shots:
            raise ValueError(NO_ANALYTIC_MSG)
        return super().batch_transform(circuit)

    def reset(self, circuits_array_length=1):
        """Reset the device"""
        self._current_circuit_index = None
        self._samples = None
        self.histograms = []
        self.memory_results = None
        self.input = {
            "qubits": self.num_wires,
            "gateset": self.gateset,
        }
        if circuits_array_length > 1:
            self.input["circuits"] = [{"circuit": []} for _ in range(circuits_array_length)]
        else:
            self.input["circuit"] = []
        self.job = {
            "type": ("ionq.multi-circuit.v1" if circuits_array_length > 1 else "ionq.circuit.v1"),
            "input": self.input,
            "backend": self.target,
        }
        if self.shots is not None:
            self.job["shots"] = self.shots
        if self.job_name is not None:
            self.job["name"] = self.job_name
        if self.dry_run:
            self.job["dry_run"] = self.dry_run
        if self.noise_model is not None:
            noise = {"model": self.noise_model}
            if self.noise_seed is not None:
                noise["seed"] = self.noise_seed
            self.job["noise"] = noise
        if self.metadata is not None:
            self.job["metadata"] = self.metadata
        if self.compilation is not None:
            self.job["settings"] = {"compilation": self.compilation}
        if self.error_mitigation is not None:
            if "settings" not in self.job:
                self.job["settings"] = {}
            self.job["settings"]["error_mitigation"] = self.error_mitigation
        if self.job["backend"] == "qpu":
            self.job["backend"] = "qpu.aria-1"
            warnings.warn(
                "The ionq_qpu backend is deprecated. Defaulting to ionq_qpu.aria-1.",
                UserWarning,
                stacklevel=2,
            )

    def set_current_circuit_index(self, circuit_index):
        """Sets the index of the current circuit for which operations are applied upon.
        In case of multiple circuits being submitted via batch_execute method
        self._current_circuit_index tracks the index of the current circuit.
        """
        self._current_circuit_index = circuit_index

    def batch_execute(self, circuits, postselect_mode=None):
        """Execute a batch of quantum circuits on the device.

        The circuits are represented by tapes, and they are executed one-by-one using the
        device's ``execute`` method. The results are collected in a list.

        Args:
            circuits (list[~.tape.QuantumTape]): circuits to execute on the device
            postselect_mode (str | None): passed by PennyLane for devices that
                support mid-circuit measurements; postselection is not
                supported on IonQ hardware, so the argument has no effect and
                a ``UserWarning`` is raised if it is explicitly set

        Returns:
            list[array[float]]: list of measured value(s)
        """
        if postselect_mode is not None:
            warnings.warn(
                "'postselect_mode' is ignored: postselection is not supported " "on IonQ hardware.",
                UserWarning,
            )
        if logger.isEnabledFor(logging.DEBUG):
            logger.debug(  # pragma: no cover
                """Entry with args=(circuits=%s) called by=%s""",
                circuits,
                "::L".join(
                    str(i) for i in inspect.getouterframes(inspect.currentframe(), 2)[1][1:3]
                ),
            )

        requires_qasm3 = False
        if any(circuit_requires_qasm3(circuit) for circuit in circuits):
            if len(circuits) != 1:
                raise ValueError(
                    "Mid-circuit measurement and reset require a single-circuit "
                    "job; submit these circuits one at a time."
                )
            requires_qasm3 = True

        self.reset(circuits_array_length=len(circuits))

        # Use tape-level shots if device shots are not set
        tape_shots = circuits[0].shots.total_shots if circuits[0].shots else None
        if self.shots is None and tape_shots is not None:
            self.job["shots"] = tape_shots  # pylint: disable=access-member-before-definition

        if not requires_qasm3:
            for circuit_index, circuit in enumerate(circuits):
                self.check_validity(circuit.operations, circuit.observables)
                self.batch_apply(
                    circuit.operations,
                    rotations=self._get_diagonalizing_gates(circuit),
                    circuit_index=circuit_index,
                )
        else:
            # Only single-circuit submission is supported for qasm3
            circuit = circuits[0]
            self.check_validity(circuit.operations, circuit.observables)
            self._apply_qasm3(
                list(circuit.operations) + list(self._get_diagonalizing_gates(circuit))
            )

        self._submit_job()

        if self.dry_run:
            return [[] for _ in circuits]

        original_shots = self.shots  # pylint: disable=access-member-before-definition
        original_shot_vector = self._shot_vector  # pylint: disable=access-member-before-definition

        results = []
        for circuit_index, circuit in enumerate(circuits):
            self.set_current_circuit_index(circuit_index)
            if self.shots is None and circuit.shots:
                self._shot_vector = circuit.shots.shot_vector
                self.shots = circuit.shots.total_shots
            self._samples = self.generate_samples()

            # compute the required statistics
            if self._shot_vector is not None:
                result = self.shot_vec_statistics(circuit)
            else:
                result = self.statistics(circuit)
                single_measurement = len(circuit.measurements) == 1

                result = result[0] if single_measurement else tuple(result)

            self.set_current_circuit_index(None)
            self._samples = None
            results.append(result)

        self.shots = original_shots
        self._shot_vector = original_shot_vector

        # increment counter for number of executions of qubit device
        self._num_executions += 1

        if self.tracker.active:
            for circuit in circuits:
                tape_resources = circuit.specs["resources"]

                self.tracker.update(
                    executions=1,
                    shots=self._shots,
                    results=results,
                    resources=tape_resources,
                )

            self.tracker.update(batches=1, batch_len=len(circuits))
            self.tracker.record()

        return results

    def batch_apply(self, operations, circuit_index, **kwargs):
        "Apply circuit operations when submitting for execution a batch of circuits."

        rotations = kwargs.pop("rotations", [])

        if len(operations) == 0 and len(rotations) == 0:
            warnings.warn("Circuit is empty. Empty circuits return failures. Submitting anyway.")

        for operation in operations:
            self._apply_operation(operation, circuit_index)

        # diagonalize observables
        for operation in rotations:
            self._apply_operation(operation, circuit_index)

    @property
    def operations(self):
        """Get the supported set of operations.

        Returns:
            set[str]: the set of PennyLane operation names the device supports
        """
        return set(self._operation_map.keys())

    @property
    def stopping_condition(self):
        """.BooleanFn: Returns the stopping condition for the device.

        Extends the default to accept mid-circuit measurements, which are
        submitted via the qasm3 path. Classically controlled operations
        (``qp.cond``) are also accepted here so that they reach the device
        intact and are rejected with a clear error instead of a failed
        decomposition.
        """

        def accepts_obj(obj):
            if isinstance(obj, MidMeasureMP):
                return True
            if isinstance(obj, Conditional):
                return accepts_obj(obj.base)
            return self.supports_operation(obj.name)

        return BooleanFn(accepts_obj)

    def apply(self, operations, **kwargs):
        """Implementation of QubitDevice abstract method apply."""

        self.reset()
        rotations = kwargs.pop("rotations", [])

        if len(operations) == 0 and len(rotations) == 0:
            warnings.warn("Circuit is empty. Empty circuits return failures. Submitting anyway.")

        if circuit_requires_qasm3(operations):
            self._apply_qasm3(list(operations) + list(rotations))
        else:
            for operation in operations:
                self._apply_operation(operation)

            # diagonalize observables
            for operation in rotations:
                self._apply_operation(operation)

        self._submit_job()

    def _apply_qasm3(self, operations):
        """Turns the job prepared by ``reset`` into an ``ionq.qasm3.v1`` job.

        Job-level fields (shots, name, noise, settings, metadata, ...) are kept
        as assembled by ``reset``; only the type and input are replaced.
        """
        self.input = {
            "qubits": self.num_wires,
            "data": operations_to_qasm3(operations, wires=self.wires, gateset=self.gateset),
        }
        self.job["type"] = "ionq.qasm3.v1"
        self.job["input"] = self.input

        settings = dict(self.job.get("settings") or {})
        compilation = dict(settings.get("compilation") or {})
        settings["compilation"] = compilation
        self.job["settings"] = settings

    def _apply_operation(self, operation, circuit_index=0):
        """Applies operations to the internal device state.

        Args:
            operation (.Operation): operation to apply on the device
            circuit_index: index of the circuit to apply operation to
        """
        wires = self.map_wires(operation.wires).tolist()
        if operation.name == "Evolution":
            self._apply_evolution_operation(operation, circuit_index, wires)
        else:
            self._apply_simple_operation(operation, circuit_index, wires)

    def _append_gate(self, gate, circuit_index):
        """Appends a gate dict to the appropriate circuit in the input payload."""
        if "circuits" in self.input:
            self.input["circuits"][circuit_index]["circuit"].append(gate)
        else:
            self.input["circuit"].append(gate)

    def _apply_evolution_operation(self, operation, circuit_index, wires):
        """Applies Evolution operations to the internal device state.
        The number of steps argument for Evolution gate will be ignored even if provided because
        IonQ implements hardware-efficient approximate compilation schemes for pauliexp gates.
        """
        warnings.warn(
            "The 'num_steps' argument for the Evolution gate will be ignored. The API maps this "
            "gate to IonQ's 'pauliexp' gate, for which IonQ implements its own hardware-efficient "
            "approximate compilation schemes.",
            UserWarning,
        )
        name = operation.name
        terms, coefficients = self._decompose_evolution(operation, wires)
        terms, coefficients = self._remove_trivial_terms(terms, coefficients)
        if len(terms) > 0:
            gate = {"gate": self._operation_map[name]}
            gate["targets"] = wires
            gate["terms"] = terms
            # 1. Float conversion to prevent numpy types (np.float64) in the JSON payload.
            # 2. IonQ API expects positive time values for their `pauliexp` gate.
            gate["time"] = abs(float(operation.param))
            # 3. Add missing sign convention to coefficients by multiplying by -1.
            gate["coefficients"] = [
                np.sign(operation.param) * (-1) * float(v) for v in coefficients
            ]
            self._append_gate(gate, circuit_index)

    def _apply_simple_operation(self, operation, circuit_index, wires):
        """Applies regular operations (gates) to the internal device state."""
        name = operation.name
        params = operation.parameters
        gate = {"gate": self._operation_map[name]}
        if name == "Toffoli":
            gate["controls"] = wires[:2]
            gate["target"] = wires[2]
        elif len(wires) == 2:
            if name in {
                "SWAP",
                "XX",
                "YY",
                "ZZ",
                "IsingXX",
                "IsingYY",
                "IsingZZ",
                "MS",
            }:
                # these gates takes two targets
                gate["targets"] = wires
            else:
                gate["control"] = wires[0]
                gate["target"] = wires[1]
        else:
            gate["target"] = wires[0]

        if self.gateset == "native":
            if len(params) > 1:
                gate["phases"] = [float(v) for v in params[:2]]
                if len(params) > 2:
                    gate["angle"] = float(params[2])
            else:
                gate["phase"] = float(params[0])
        elif params:
            gate["rotation"] = float(params[0])
        self._append_gate(gate, circuit_index)

    def _remove_trivial_terms(self, terms, coefficients):
        """Removes all-identity (II..I) terms from the list of terms."""
        filtered = [(t, c) for t, c in zip(terms, coefficients) if "X" in t or "Y" in t or "Z" in t]
        if not filtered:
            return [], []
        terms, coefficients = zip(*filtered)
        return list(terms), list(coefficients)

    def _decompose_evolution(self, operation, wires: list[int]) -> tuple[list[str], list[float]]:
        """Decompose an Evolution gate's generator into IonQ Pauli terms and coefficients.

        Returns:
            tuple: (terms, coefficients) where terms is a list of Pauli strings
                and coefficients is a list of floats.
        """
        ops = None
        coefficients = None
        generator = operation.generator()

        if isinstance(generator, LinearCombination):
            ops = []
            coefficients = []
            coeffs = generator.coeffs.tolist()
            gen_ops = generator.ops
            for coeff, gen_op in zip(coeffs, gen_ops):
                scaled = coeff * gen_op
                if isinstance(gen_op, (Sum, Prod, PauliX, PauliY, PauliZ, Identity)):
                    c, o = scaled.terms()
                    coefficients.extend(c)
                    ops.extend(o)
                else:
                    op_wires = scaled.wires.tolist()
                    decomp = pauli_decompose(scaled.matrix(), wire_order=op_wires, pauli=False)
                    coefficients.extend(decomp.coeffs.tolist())
                    ops.extend(decomp.ops)
        elif isinstance(generator, SparseHamiltonian):
            decomp = pauli_decompose(generator.H.toarray(), wire_order=wires, pauli=False)
            ops = decomp.ops
            coefficients = decomp.coeffs.tolist()
        elif isinstance(generator, SProd):
            if isinstance(generator.base, (PauliX, PauliY, PauliZ, Identity)):
                ops = [generator.base]
                coefficients = [generator.scalar]
            elif isinstance(generator.base, (Sum, Prod)):
                base_coeffs, base_ops = generator.base.terms()
                coefficients = [generator.scalar * float(c) for c in base_coeffs]
                ops = base_ops
            elif isinstance(generator.base, Exp):
                decomp = pauli_decompose(generator.matrix(), wire_order=wires, pauli=False)
                ops = decomp.ops
                coefficients = decomp.coeffs.tolist()

        if ops is None:
            raise NotSupportedEvolutionInstance()

        if any(isinstance(c, complex) for c in coefficients):
            raise ComplexEvolutionCoefficientsNotSupported()

        terms = self._operations_to_ionq_pauli_names(ops, wires)
        return terms, coefficients

    def _operations_to_ionq_pauli_names(self, ops, wires) -> list[str]:
        """Converts a list of operations to a list of IonQ compatible Pauli matrix names."""

        def map_operand_to_term(operand):
            try:
                return PAULI_MAP[operand.name]
            except KeyError as exc:
                supported = ", ".join(PAULI_MAP.keys())
                raise KeyError(
                    f"Operand {operand.name} is not supported for Evolution gate. "
                    f"Supported operands: {supported}."
                ) from exc

        def join_terms(terms, wires):
            """Pennylane uses big-endian ordering, IonQ uses little-endian ordering."""
            big_endian_term = "".join(terms.get(wire, "I") for wire in wires)
            little_endian_term = big_endian_term[::-1]
            return little_endian_term

        ionq_terms = []
        for op in ops:
            terms = {}
            if isinstance(op, Prod):
                for operand in op.operands:
                    term_name = map_operand_to_term(operand)
                    term_wire = operand.wires[0]
                    terms[term_wire] = term_name
                ionq_terms.append(join_terms(terms, wires))
            elif isinstance(op, (PauliX, PauliY, PauliZ)):
                term_name = map_operand_to_term(op)
                term_wire = op.wires[0]
                terms[term_wire] = term_name
                ionq_terms.append(join_terms(terms, wires))
            elif isinstance(op, Identity):
                ionq_terms.append(join_terms(terms, wires))
            else:
                raise OperatorNotSupportedInEvolutionGateGenerator(
                    f"Unsupported operator in generator of Evolution gate: {op}"
                )
        return ionq_terms

    def _submit_job(self):
        job = Job(
            api_key=self.api_key,
            timeout=self.timeout,
            max_retries=self.max_retries,
            retry_delay=self.retry_delay,
        )

        # send job for execution
        job.manager.create(**self.job)

        if self.dry_run:
            return

        # retrieve results
        while not job.is_complete:
            sleep(0.01)
            job.reload()
            if job.is_failed:
                raise JobExecutionError("Job failed")

        params = {} if self.sharpen is None else {"sharpen": self.sharpen}

        if self.memory:
            # Shotwise outputs using memory
            if self.job.get("type") == "ionq.multi-circuit.v1":
                # Multi-circuit job shots:
                # ------------------------
                # Parent job only carries aggregated probabilities;
                # Results are keyed by child job ID. Each child job advertises
                # its own shotwise results URL so store these IDs.
                child_ids = job.data.value.keys()
                job_ids = list(child_ids)
            else:
                job_ids = [job.id.value]

            memory_results = []
            for job_id in job_ids:
                job.manager.get(resource_id=job_id, params=params, results_type=ResultsTypes.SHOTS)
                memory_results.append(self._shotwise_to_samples(job.data.value))
            self.memory_results = memory_results
        else:
            job.manager.get(
                resource_id=job.id.value, params=params, results_type=ResultsTypes.PROBS
            )

            # The returned job histogram is of the form
            # dict[str, float], and maps the computational basis
            # state (as a base-10 integer string) to the probability
            # as a floating point value between 0 and 1.
            # e.g., {"0": 0.413, "9": 0.111, "17": 0.476}
            some_inner_value = next(iter(job.data.value.values()))
            if isinstance(some_inner_value, dict):
                self.histograms = list(job.data.value.values())
            else:
                self.histograms = [job.data.value]

    def _shotwise_to_samples(self, raw_shots):
        """Convert IonQ API shotwise results into PennyLane binary samples.

        The API encodes each shot as a basis-state integer using little-endian
        ordering, like the histogram keys. ``states_to_binary`` places the least
        significant bit last, so reversing the bit axis yields rows in the
        big-endian wire ordering expected by PennyLane.

        Args:
            raw_shots (list[str]): one little-endian basis-state integer per shot.

        Returns:
            array[int]: binary samples of shape ``(shots, num_wires)``.
        """
        basis_states = np.fromiter((int(s) for s in raw_shots), dtype=np.int64)
        return QubitDevice.states_to_binary(basis_states, self.num_wires)[:, ::-1]

    def _current_circuit_results(self, results):
        """Select the entry of the current circuit from per-circuit results.

        Args:
            results (list): one results entry per submitted circuit, i.e.
                ``self.histograms`` or ``self.memory_results``.

        Returns:
            the entry belonging to ``self._current_circuit_index``

        Raises:
            ValueError: if no results are available
            CircuitIndexNotSetException: if several circuits were submitted but no
                circuit index is set
        """
        if not results:
            raise ValueError(NO_RESULTS_MSG)

        if self._current_circuit_index is None and len(results) > 1:
            raise CircuitIndexNotSetException()

        return results[self._current_circuit_index or 0]

    @property
    def prob(self):
        """array[float]: Array of computational basis state probabilities
        computed from the histogram returned by the API. Returns ``None`` if no
        histogram is available, i.e., if no job has been submitted or if the
        device uses shotwise results (``memory=True``).
        """
        histogram = self._current_circuit_results(self.histograms)

        # The IonQ API returns basis states using little-endian ordering.
        # Here, we rearrange the states to match the big-endian ordering
        # expected by PennyLane.
        basis_states = (int(bin(int(k))[2:].rjust(self.num_wires, "0")[::-1], 2) for k in histogram)
        idx = np.fromiter(basis_states, dtype=int)

        # convert the sparse probs into a probability array
        prob_array = np.zeros([2**self.num_wires])

        # histogram values don't always perfectly sum to exactly one
        histogram_values = histogram.values()
        norm = sum(histogram_values)
        prob_array[idx] = np.fromiter(histogram_values, float) / norm

        return prob_array

    def probability(self, wires=None, shot_range=None, bin_size=None):
        wires = wires or self.wires

        if self.memory or shot_range is not None or bin_size is not None:
            if self._samples is None:
                raise ValueError(NO_RESULTS_MSG)
            return self.estimate_probability(wires=wires, shot_range=shot_range, bin_size=bin_size)

        # Use the histogram returned by the API.
        return self.marginal_prob(self.prob, wires)


[docs] class SimulatorDevice(IonQDevice): r"""Simulator device for IonQ. Args: wires (int or Iterable[Number, str]]): Number of wires to initialize the device with, or iterable that contains unique labels for the subsystems as numbers (i.e., ``[-1, 0, 2]``) or strings (``['ancilla', 'q1', 'q2']``). .. note:: Custom wire labels (e.g., strings or non-consecutive integers) are used for user convenience only. They have no effect on the transpilation process or the final qubit layout on the hardware. gateset (str): the target gateset, either ``"qis"`` or ``"native"``. Defaults to ``qis``. shots (int, list[int], None): Number of circuit evaluations/random samples used to estimate expectation values of observables. If ``None``, the device calculates probability, expectation values, and variances analytically. If an integer, it specifies the number of samples to estimate these quantities. If a list of integers is passed, the circuit evaluations are batched over the list of shots. Defaults to None. api_key (str): The IonQ API key. If not provided, the environment variable ``IONQ_API_KEY`` is used. noise_model (str): the noise model to use for simulation. Valid values are ``"ideal"``, ``"harmony"``, ``"aria-1"``, ``"aria-2"``, ``"forte-1"``, and ``"forte-enterprise-1"``. Defaults to None (ideal simulation). See `IonQ Simulation with Noise Models <https://docs.ionq.com/guides/simulation-with-noise-models>`_ for details. noise_seed (int): seed for the noise model random number generator, for reproducible noisy simulation results. Must be an integer between 1 and 2\ :sup:`31` - 1. Only used when ``noise_model`` is set. Defaults to None (random seed). memory (bool): if True, use the shotwise measurement outcomes returned by the IonQ API as the device samples, instead of generating samples locally from the returned probabilities. Shotwise results are not available for ideal simulation. Defaults to False. metadata (dict | None): optional metadata to attach to the job. Defaults to None. timeout (float): Request timeout in seconds. Defaults to None, which uses the ``APIClient`` default. max_retries (int): Maximum number of retries for retriable HTTP errors. Defaults to None, which uses the ``APIClient`` default. retry_delay (float): Base delay in seconds between retries (uses exponential backoff). Defaults to None, which uses the ``APIClient`` default. """ name = "IonQ Simulator PennyLane plugin" short_name = "ionq.simulator" def __init__( self, wires, *, gateset="qis", shots=None, job_name=None, compilation=None, api_key=None, noise_model=None, noise_seed=None, memory=False, dry_run=False, metadata=None, timeout=None, max_retries=None, retry_delay=None, ): super().__init__( wires=wires, target="simulator", gateset=gateset, shots=shots, job_name=job_name, api_key=api_key, compilation=compilation, noise_model=noise_model, noise_seed=noise_seed, memory=memory, dry_run=dry_run, metadata=metadata, timeout=timeout, max_retries=max_retries, retry_delay=retry_delay, )
[docs] def generate_samples(self): """Generates samples from the simulator. If ``memory=True``, the shotwise results retrieved from the API are used as the samples. Otherwise, samples are generated by random sampling with the probabilities returned by the simulator. """ if self.memory: return self._current_circuit_results(self.memory_results) number_of_states = 2**self.num_wires samples = self.sample_basis_states(number_of_states, self.prob) return QubitDevice.states_to_binary(samples, self.num_wires)
[docs] class QPUDevice(IonQDevice): r"""QPU device for IonQ. Args: wires (int or Iterable[Number, str]]): Number of wires to initialize the device with, or iterable that contains unique labels for the subsystems as numbers (i.e., ``[-1, 0, 2]``) or strings (``['ancilla', 'q1', 'q2']``). .. note:: Custom wire labels (e.g., strings or non-consecutive integers) are used for user convenience only. They have no effect on the transpilation process or the final qubit layout on the hardware. gateset (str): the target gateset, either ``"qis"`` or ``"native"``. Defaults to ``qis``. backend (str): Optional specifier for an IonQ backend. Can be ``"aria-1"``, ``"aria-2"``, etc. Default to ``aria-1``. shots (int, list[int], None): Number of circuit evaluations/random samples used to estimate expectation values of observables. Defaults to None. If a list of integers is passed, the circuit evaluations are batched over the list of shots. api_key (str): The IonQ API key. If not provided, the environment variable ``IONQ_API_KEY`` is used. error_mitigation (dict | None): settings for error mitigation when creating a job. Defaults to None. Not available on all backends. Set by default on some hardware systems. See `IonQ API Job Creation <https://docs.ionq.com/api-reference/v0.4/jobs/create-job>`_ and `IonQ Debiasing and Sharpening <https://ionq.com/resources/debiasing-and-sharpening>`_ for details. Valid keys include: ``debiasing`` (bool). sharpen (bool): whether to use sharpening when accessing the results of an executed job. Defaults to None (no value passed at job retrieval). Will generally return more accurate results if your expected output distribution has peaks. See `IonQ Debiasing and Sharpening <https://ionq.com/resources/debiasing-and-sharpening>`_ for details. memory (bool): if True, use the shotwise measurement outcomes returned by the API as the device samples, instead of generating samples locally from the returned probabilities. Defaults to False. dry_run (bool): whether to run the job in dry run mode. Defaults to False. metadata (dict | None): optional metadata to attach to the job. Defaults to None. timeout (float): Request timeout in seconds. Defaults to None, which uses the ``APIClient`` default. max_retries (int): Maximum number of retries for retriable HTTP errors. Defaults to None, which uses the ``APIClient`` default. retry_delay (float): Base delay in seconds between retries (uses exponential backoff). Defaults to None, which uses the ``APIClient`` default. """ name = "IonQ QPU PennyLane plugin" short_name = "ionq.qpu" # pylint: disable=too-many-arguments def __init__( self, wires, *, gateset="qis", shots=None, job_name=None, backend="aria-1", compilation=None, error_mitigation=None, sharpen=None, api_key=None, memory=False, dry_run=False, metadata=None, timeout=None, max_retries=None, retry_delay=None, ): target = "qpu" self.backend = backend if self.backend is not None: target += "." + self.backend super().__init__( wires=wires, target=target, gateset=gateset, shots=shots, job_name=job_name, api_key=api_key, compilation=compilation, error_mitigation=error_mitigation, sharpen=sharpen, memory=memory, dry_run=dry_run, metadata=metadata, timeout=timeout, max_retries=max_retries, retry_delay=retry_delay, )
[docs] def generate_samples(self): """Generates samples from the qpu. If ``memory=True``, the shotwise results retrieved from the API are used as the samples, in the order the shots were measured. Otherwise, samples are generated from the returned probabilities, and their order is not indicative of the order in which the experiments were done, but is instead controlled by a random shuffle (and hence set by numpy random seed). """ if self.memory: return self._current_circuit_results(self.memory_results) number_of_states = 2**self.num_wires counts = np.rint( self.prob * self.shots, out=np.zeros(number_of_states, dtype=int), casting="unsafe", ) samples = np.repeat(np.arange(number_of_states), counts) np.random.shuffle(samples) return QubitDevice.states_to_binary(samples, self.num_wires)