diff --git a/.github/workflows/docs-executor.yml b/.github/workflows/docs-executor.yml new file mode 100644 index 0000000000..74fdfdf76f --- /dev/null +++ b/.github/workflows/docs-executor.yml @@ -0,0 +1,34 @@ +--- +name: Docs Upload (executor) + +on: + push: + branches: [ "executor_preview" ] + +jobs: + doc_publish: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.10' + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -U virtualenv setuptools wheel tox + sudo apt-get install -y graphviz pandoc + - name: Build docs + run: tox -e docs -- --tag executor + - name: Upload artifact + uses: actions/upload-artifact@v4 + with: + name: html_docs_executor + path: docs/_build/html + - name: Bypass Jekyll Processing # Necessary for setting the correct css path + run: touch docs/_build/html/.nojekyll + - name: Deploy + uses: JamesIves/github-pages-deploy-action@v4 + with: + folder: docs/_build/html/ diff --git a/.gitignore b/.gitignore index e1825b5c40..501fbb22f0 100644 --- a/.gitignore +++ b/.gitignore @@ -76,6 +76,7 @@ target/ # Jupyter Notebook .ipynb_checkpoints +.notebooks # pyenv .python-version diff --git a/.whitesource b/.whitesource new file mode 100644 index 0000000000..1dc084bacc --- /dev/null +++ b/.whitesource @@ -0,0 +1,3 @@ +{ + "settingsInheritedFrom": "ibm-q-research/whitesource-config@main" +} \ No newline at end of file diff --git a/README.md b/README.md index 1aa1ed9033..34d448eb56 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,5 @@ +*** THIS IS A PROTOTYPE FORK TO TEST A DEMO *** + # Qiskit Runtime IBM Client [![License](https://img.shields.io/github/license/Qiskit/qiskit-ibm-runtime.svg?style=popout-square)](https://opensource.org/licenses/Apache-2.0) [![CI](https://github.com/Qiskit/qiskit-ibm-runtime/actions/workflows/ci.yml/badge.svg)](https://github.com/Qiskit/qiskit-ibm-runtime/actions/workflows/ci.yml) diff --git a/docs/apidocs/index.rst b/docs/apidocs/index.rst index 526639f093..e2327cc22c 100644 --- a/docs/apidocs/index.rst +++ b/docs/apidocs/index.rst @@ -8,8 +8,10 @@ :maxdepth: 1 runtime_service + quantum_program noise_learner noise_learner_result + noise_learner_v3 options transpiler qiskit_ibm_runtime.transpiler.passes.scheduling diff --git a/docs/apidocs/noise_learner_v3.rst b/docs/apidocs/noise_learner_v3.rst new file mode 100644 index 0000000000..0318881618 --- /dev/null +++ b/docs/apidocs/noise_learner_v3.rst @@ -0,0 +1,4 @@ +.. automodule:: qiskit_ibm_runtime.noise_learner_v3 + :no-members: + :no-inherited-members: + :no-special-members: diff --git a/docs/apidocs/quantum_program.rst b/docs/apidocs/quantum_program.rst new file mode 100644 index 0000000000..a372e3aa34 --- /dev/null +++ b/docs/apidocs/quantum_program.rst @@ -0,0 +1,4 @@ +.. automodule:: qiskit_ibm_runtime.quantum_program + :no-members: + :no-inherited-members: + :no-special-members: diff --git a/docs/conf.py b/docs/conf.py index ed9846eb58..5556781595 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -15,6 +15,7 @@ import os import re import sys +import typing sys.path.insert(0, os.path.abspath(".")) @@ -64,6 +65,20 @@ link_str = f" https://github.com/Qiskit/qiskit-ibm-runtime/blob/stable/{vers[0]}.{vers[1]}/docs/" nbsphinx_prolog += link_str + "{{ docname }}" + +# ---------------------------------------------------------------------------------- +# Patch 'Self' for Python < 3.11 if missing +# ---------------------------------------------------------------------------------- + +if not hasattr(typing, "Self"): + try: + from typing_extensions import Self + except ImportError: + class Self: + """Dummy fallback for 'Self' for older python versions.""" + pass + typing.Self = Self + # ---------------------------------------------------------------------------------- # Intersphinx # ---------------------------------------------------------------------------------- @@ -137,6 +152,16 @@ html_sourcelink_suffix = "" +# ----------------------------------------------------------------------------- +# `executor`-specific options +# ----------------------------------------------------------------------------- +if "executor" in tags: + extensions += ["qiskit_sphinx_theme"] + html_theme = "qiskit-ecosystem" + html_theme_options = { + "sidebar_qiskit_ecosystem_member": False, + } + # ---------------------------------------------------------------------------------- # Source code links # ---------------------------------------------------------------------------------- diff --git a/docs/guides/executor_basic.rst b/docs/guides/executor_basic.rst new file mode 100644 index 0000000000..6b43bf64d5 --- /dev/null +++ b/docs/guides/executor_basic.rst @@ -0,0 +1,253 @@ +The Executor: A quick-start guide +================================= + +This guide provides a basic overview of the :class:`~.Executor`, a runtime program that allows +executing :class:`~.QuantumProgram`\s on IBM backends. At the end of this guide, you will +know how to: + +* Initialize a :class:`~.QuantumProgram` with your workload. +* Run :class:`~.QuantumProgram`\s on IBM backends using the :class:`~.Executor`. +* Interpret the outputs of the :class:`~.Executor`. + +In the remainder of the guide, we consider a circuit that generates a three-qubit GHZ state, rotates +the qubits around the Pauli-Z axis, and measures the qubits in the computational basis. We show how +to add this circuit to a :class:`~.QuantumProgram`, optionally randomizing its content with twirling +gates, and how to execute the program via the :class:`~.Executor`. + +.. code-block:: python + + from qiskit.circuit import Parameter, QuantumCircuit + + # A circuit of the type considered in this guide + circuit = QuantumCircuit(3) + circuit.h(0) + circuit.h(1) + circuit.cz(0, 1) + circuit.h(1) + circuit.h(2) + circuit.cz(1, 2) + circuit.h(2) + circuit.rz(Parameter("theta"), 0) + circuit.rz(Parameter("phi"), 1) + circuit.rz(Parameter("lam"), 2) + circuit.measure_all() + +Let us choose a backend to run our executor jobs with: + +.. code-block:: python + + from qiskit_ibm_runtime import QiskitRuntimeService + + service = QiskitRuntimeService() + backend = service.least_busy(operational=True, simulator=False) + +We can now begin by taking a look at the inputs to the :class:`~.Executor`, the :class:`~.QuantumProgram`\s. + +The inputs to the Executor: Quantum Programs +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +A :class:`~.QuantumProgram` is an iterable of +:class:`~.qiskit_ibm_runtime.quantum_program.QuantumProgramItem`\s. Each of these items represents a +different task for the :class:`~.Executor` to perform. Typically, each item owns: + +* a :class:`~qiskit.circuit.QuantumCircuit` with static, non-parametrized gates; +* or a parametrized :class:`~qiskit.circuit.QuantumCircuit`, together with an array of parameter values; +* or a parametrized :class:`~qiskit.circuit.QuantumCircuit`, together with a + :class:`~samplomatic.samplex.Samplex` to generate randomize arrays of parameter values. + +Let us take a closer look at each of these items and how to add them to a :class:`~.QuantumProgram`\. + +In the cell below, we initialize a :class:`~.QuantumProgram` and specify that we wish to perform ``1024`` +shots for every configuration of each item in the program. Next, we append a version of our target circuit with set parameters, +transpiled according to the backend's ISA. + +.. code-block:: python + + from qiskit.transpiler import generate_preset_pass_manager + from qiskit_ibm_runtime.quantum_program import QuantumProgram + + # Initialize an empty program + program = QuantumProgram(shots=1024) + + # Initialize circuit to generate and measure GHZ state + circuit = QuantumCircuit(3) + circuit.h(0) + circuit.h(1) + circuit.cz(0, 1) + circuit.h(1) + circuit.h(2) + circuit.cz(1, 2) + circuit.h(2) + circuit.rz(0.1, 0) + circuit.rz(0.2, 1) + circuit.rz(0.3, 2) + circuit.measure_all() + + # Transpile the circuit + preset_pass_manager = generate_preset_pass_manager(backend=backend, optimization_level=0) + isa_circuit = preset_pass_manager.run(circuit) + + # Append the circuit to the program + program.append(isa_circuit) + +We proceed to append a second item that contains a parametrized :class:`~qiskit.circuit.QuantumCircuit` +and an array containing ``10`` sets of parameter values. This amounts to a circuit task requiring a total +of ``10240`` shots (namely ``1024`` per set of parameter values). + +.. code-block:: python + + from qiskit.circuit import Parameter + import numpy as np + + # Initialize circuit to generate a GHZ state, rotate it around the Pauli-Z + # axis, and measure it + circuit = QuantumCircuit(3) + circuit.h(0) + circuit.h(1) + circuit.cz(0, 1) + circuit.h(1) + circuit.h(2) + circuit.cz(1, 2) + circuit.h(2) + circuit.rz(Parameter("theta"), 0) + circuit.rz(Parameter("phi"), 1) + circuit.rz(Parameter("lam"), 2) + circuit.measure_all() + + # Transpile the circuit + isa_circuit = preset_pass_manager.run(circuit) + + # Append the circuit and the parameter value to the program + program.append( + isa_circuit, + circuit_arguments=np.random.rand(10, 3), # 10 sets of parameter values + ) + +Finally, in the next cell we append a parametrized :class:`~qiskit.circuit.QuantumCircuit` and a +:class:`~samplomatic.samplex.Samplex`, which is responsible for generating randomized sets of +parameters for the given circuit. As part of the :class:`~samplomatic.samplex.Samplex` arguments, +we provide ``10`` sets of parameters for the parametric gates in the original circuit. +Additionally, we use the ``shape`` request argument to request an extension of the implicit shape +defined by the :class:`~samplomatic.samplex.Samplex` arguments. In particular, by setting ``shape`` +to ``(2, 14, 10)`` we request to randomize each of the ``10`` sets of parameters ``28`` times, and +to arrange the randomized parameter sets in an array of be arranged in an array of shape +``(2, 14, 10)``. + + We refer the reader to :mod:`~samplomatic` and its documentation for more details on the + :class:`~samplomatic.samplex.Samplex` and its arguments. + +.. code-block:: python + + from samplomatic import build + from samplomatic.transpiler import generate_boxing_pass_manager + + # Initialize circuit to generate a GHZ state, rotate it around the Pauli-Z + # axis, and measure it + circuit = QuantumCircuit(3) + circuit.h(0) + circuit.h(1) + circuit.cz(0, 1) + circuit.h(1) + circuit.h(2) + circuit.cz(1, 2) + circuit.h(2) + circuit.rz(Parameter("theta"), 0) + circuit.rz(Parameter("phi"), 1) + circuit.rz(Parameter("lam"), 2) + circuit.measure_all() + + # Transpile the circuit, additionally grouping gates and measurements into annotated boxes + preset_pass_manager = generate_preset_pass_manager(backend=backend, optimization_level=0) + preset_pass_manager.post_scheduling = generate_boxing_pass_manager( + enable_gates=True, + enable_measures=True, + ) + boxed_circuit = preset_pass_manager.run(circuit) + + # Build the template and the samplex + template, samplex = build(boxed_circuit) + + # Append the template and samplex as a samplex item + program.append( + template, + samplex=samplex, + samplex_arguments={ + # the arguments required by the samplex.sample method + "parameter_values": np.random.rand(10, 3), + }, + shape=(2, 14, 10), + ) + +Now that we have populated our :class:`~.QuantumProgram`, we can proceed with execution. + +Running an Executor job +~~~~~~~~~~~~~~~~~~~~~~~ + +In the cell below we initialize an :class:`~.Executor` and leave the default options: + + .. code-block:: python + + from qiskit_ibm_runtime import Executor + + executor = Executor(backend) + +Next, we use the :meth:`~.Executor.run` method to submit the job. + + .. code-block:: python + + job = executor.run(program) + + # Retrieve the result + result = job.result() + +Here, ``result`` is of type :class:`~.qiskit_ibm_runtime.quantum_program.QuantumProgramResult`. +We now take a closer look at this result object. + +The outputs of the Executor +~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +:class:`~.qiskit_ibm_runtime.quantum_program.QuantumProgramResult` is an iterable. It contains one +item per circuit task, and the items are in the same order as the items in the program. Every one of +these items is a dictionary from strings to an ``np.ndarray``. with elements of type ``bool``. Let us +take a look at the three items in ``result`` to understand the meaning of their key-value pairs. + +The first item in ``result`` contains the results of running the first task in the program, namely +the circuit with static gates. It contains a single key, ``'meas'``, corresponding to the name of the +classical register in the input circuit. The ``'meas'`` key is mapped to the results collected for this +classical registers, stored in an ``np.ndarray`` of shape ``(1024, 3)``. The first axis +is over shots, the second is over bits in the classical register. + + .. code-block:: python + + # Access the results of the classical register of task #0 + result_0 = result[0]["meas"] + print(f"Result shape: {result_0.shape}") + +The second item contains the results of running the second task in the program, namely +the circuit with parametrized gates. Again, it contains a single key, ``'meas'``, mapped to a +``np.ndarray`` of shape ``(1024, 10, 3)``. The central axis is over parameter sets, while the first +and last are again over shots and bits respectively. + + .. code-block:: python + + # Access the results of the classical register of task #1 + result_1 = result[1]["meas"] + print(f"Result shape: {result_1.shape}") + +Finally, the third item in ``result`` contains the results of running the third task in the program. This item +contains multiple key. In more detail, in addition to the ``'meas'`` key (mapped to the array of results for +that classical register), it contains ``'measurement_flips.meas'``, namely the bit-flip corrections to undo +the measurement twirling for the ``'meas'`` register. + + .. code-block:: python + + # Access the results of the classical register of task #2 + result_2 = result[2]["meas"] + print(f"Result shape: {result_2.shape}") + + # Access the bit-flip corrections + flips_2 = result[2]["measurement_flips.meas"] + print(f"Result shape: {result_0.shape}") + + # Undo the bit flips via classical XOR + unflipped_result_2 = result_2 ^ flips_2 \ No newline at end of file diff --git a/docs/guides/index.rst b/docs/guides/index.rst new file mode 100644 index 0000000000..9dce39e762 --- /dev/null +++ b/docs/guides/index.rst @@ -0,0 +1,8 @@ + +Guides +====== + +.. toctree:: + :maxdepth: 2 + + executor_basic diff --git a/docs/index.rst b/docs/index.rst index 0154b0573f..3ca87a1db0 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -2,13 +2,26 @@ Qiskit Runtime |release| API Docs Preview ######################################### -Qiskit Runtime docs live at http://quantum.cloud.ibm.com/docs and come from https://github.com/Qiskit/documentation. -This site is only used to generate our API docs, which then get migrated to -https://github.com/Qiskit/documentation. +.. only:: executor + + This is the documentation for the experimental ``executor_preview`` branch of Qiskit Runtime. + + .. warning:: + + This is the documentation for a feature that is in beta and might not be stable. Please refer + to the `API Reference page `_ + for documentation for stable releases. + +.. only:: not executor + + Qiskit Runtime docs live at http://quantum.cloud.ibm.com/docs and come from https://github.com/Qiskit/documentation. + This site is only used to generate our API docs, which then get migrated to + https://github.com/Qiskit/documentation. .. toctree:: :hidden: Documentation home API Reference + Guides Release Notes diff --git a/pyproject.toml b/pyproject.toml index dfdab61388..4df9739f61 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,5 +1,5 @@ [build-system] -requires = ["setuptools >= 40.6.0", "setuptools_scm[toml]>=6.2"] +requires = ["setuptools >= 40.6.0", "setuptools_scm[toml]~=8.0"] build-backend = "setuptools.build_meta" [tool.black] @@ -48,7 +48,7 @@ zip-safe = false [tool.setuptools_scm] root = "." -write_to = "qiskit_ibm_runtime/VERSION.txt" +version_file = "qiskit_ibm_runtime/VERSION.txt" version_scheme = "release-branch-semver" fallback_version = "0.45.0" @@ -104,7 +104,7 @@ ignore = [ [project] name = "qiskit-ibm-runtime" -dynamic = [ "version" ] +version = "0.44.0b2+executor.preview" description = "IBM Quantum client for Qiskit Runtime." readme = {file = "README.md", content-type = "text/markdown"} authors = [ @@ -138,9 +138,12 @@ dependencies = [ "urllib3>=1.21.1", "python-dateutil>=2.8.0", "ibm-platform-services>=0.22.6", + "ibm-quantum-schemas @ git+https://github.com/Qiskit/ibm-quantum-schemas.git@d579621311e62b5c342e804a605e4e4e96d88811#egg=ibm-quantum-schemas", "pydantic>=2.5.0", "qiskit>=2.0.0", - "packaging" + "packaging", + "pybase64>=1.0", + "samplomatic>=0.13.0" ] [project.entry-points."qiskit.transpiler.translation"] @@ -184,6 +187,7 @@ documentation = [ "jupyter-sphinx", "sphinxcontrib-katex==0.9.9", "packaging", + "qiskit-sphinx-theme~=2.0.0", ] visualization = ["plotly>=5.23.0"] diff --git a/qiskit_ibm_runtime/__init__.py b/qiskit_ibm_runtime/__init__.py index ef70971d7a..4d66e0b0d7 100644 --- a/qiskit_ibm_runtime/__init__.py +++ b/qiskit_ibm_runtime/__init__.py @@ -189,6 +189,7 @@ EstimatorV2 Sampler SamplerV2 + Executor Session Batch IBMBackend @@ -209,6 +210,8 @@ from .session import Session # pylint: disable=cyclic-import from .batch import Batch # pylint: disable=cyclic-import +from .quantum_program import QuantumProgram + from .exceptions import * from .utils.utils import setup_logger from .version import __version__ @@ -217,6 +220,7 @@ EstimatorV2, EstimatorV2 as Estimator, ) +from .executor import Executor from .sampler import SamplerV2, SamplerV2 as Sampler # pylint: disable=reimported from .options import ( # pylint: disable=reimported EstimatorOptions, diff --git a/qiskit_ibm_runtime/constants.py b/qiskit_ibm_runtime/constants.py index 94cadd849e..9f169eccee 100644 --- a/qiskit_ibm_runtime/constants.py +++ b/qiskit_ibm_runtime/constants.py @@ -18,6 +18,7 @@ from .utils.noise_learner_result_decoder import NoiseLearnerResultDecoder from .utils.estimator_result_decoder import EstimatorResultDecoder from .utils.sampler_result_decoder import SamplerResultDecoder +from .utils.executor_result_decoder import ExecutorResultDecoder from .utils.runner_result import RunnerResult @@ -38,7 +39,18 @@ DEFAULT_DECODERS = { "sampler": [ResultDecoder, SamplerResultDecoder], "estimator": [ResultDecoder, EstimatorResultDecoder], + "executor": ExecutorResultDecoder, "noise-learner": NoiseLearnerResultDecoder, "circuit-runner": RunnerResult, "qasm3-runner": RunnerResult, } + +DEFAULT_POST_SELECTION_SUFFIX = "_ps" +""" +The default suffix to append to the names of the classical registers used for post selection. +""" + +DEFAULT_SPECTATOR_CREG_NAME = "spec" +""" +The default name of the classical register used for measuring spectator qubits. +""" diff --git a/qiskit_ibm_runtime/executor.py b/qiskit_ibm_runtime/executor.py new file mode 100644 index 0000000000..bc1f7b5024 --- /dev/null +++ b/qiskit_ibm_runtime/executor.py @@ -0,0 +1,144 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Executor""" + +from __future__ import annotations + +from dataclasses import asdict +import logging + +from ibm_quantum_schemas.models.executor.version_0_1.models import ( + QuantumProgramResultModel, +) +from ibm_quantum_schemas.models.base_params_model import BaseParamsModel + +from .ibm_backend import IBMBackend +from .session import Session # pylint: disable=cyclic-import +from .batch import Batch # pylint: disable=cyclic-import +from .options.executor_options import ExecutorOptions +from .qiskit_runtime_service import QiskitRuntimeService +from .quantum_program import QuantumProgram +from .quantum_program.converters import quantum_program_result_from_0_1, quantum_program_to_0_1 +from .runtime_job_v2 import RuntimeJobV2 +from .runtime_options import RuntimeOptions +from .utils.default_session import get_cm_session + +logger = logging.getLogger() + + +class _Decoder: + @classmethod + def decode(cls, data: str): # type: ignore[no-untyped-def] + """Decode raw json to result type.""" + obj = QuantumProgramResultModel.model_validate_json(data) + return quantum_program_result_from_0_1(obj) + + +class Executor: + """Executor for :class:`~.QuantumProgram`\\s.""" + + _PROGRAM_ID = "executor" + _DECODER = _Decoder + + def __init__(self, mode: IBMBackend | Session | Batch | None): + + self._session: Session | None = None + self._backend: IBMBackend + self._service: QiskitRuntimeService + + self._options = ExecutorOptions() + + if isinstance(mode, (Session, Batch)): + self._session = mode + self._backend = self._session._backend + self._service = self._session.service + + elif open_session := get_cm_session(): + if open_session != mode: + if open_session._backend != mode: + raise ValueError( + "The backend passed in to the primitive is different from the session " + "backend. Please check which backend you intend to use or leave the mode " + "parameter empty to use the session backend." + ) + logger.warning( + "A backend was passed in as the mode but a session context manager " + "is open so this job will run inside this session/batch " + "instead of in job mode." + ) + self._session = open_session + self._backend = self._session._backend + self._service = self._session.service + + elif isinstance(mode, IBMBackend): + self._backend = mode + self._service = self._backend.service + + else: + raise ValueError( + "A backend or session/batch must be specified, or a session/batch must be open." + ) + + @property + def options(self) -> ExecutorOptions: + """The options of this executor.""" + return self._options + + def _runtime_options(self) -> RuntimeOptions: + return RuntimeOptions( + backend=self._backend.name, + image=self.options.environment.image, + job_tags=self.options.environment.job_tags, + log_level=self.options.environment.log_level, + private=self.options.environment.private, + max_execution_time=self.options.environment.max_execution_time, + ) + + def _run(self, params: BaseParamsModel) -> RuntimeJobV2: + runtime_options = self._runtime_options() + + if self._session: + run = self._session._run + else: + run = self._service._run + runtime_options.instance = self._backend._instance + + if get_cm_session(): + logger.warning( + "Even though a session/batch context manager is open this job will run in job " + "mode because the %s primitive was initialized outside the context manager. " + "Move the %s initialization inside the context manager to run in a " + "session/batch.", + self._PROGRAM_ID, + self._PROGRAM_ID, + ) + + inputs = params.model_dump(mode="json") + + return run( + program_id=self._PROGRAM_ID, + options=asdict(runtime_options), + inputs=inputs, + result_decoder=_Decoder, + ) + + def run(self, program: QuantumProgram) -> RuntimeJobV2: + """Run a quantum program. + + Args: + program: The program to run. + + Returns: + A job. + """ + return self._run(quantum_program_to_0_1(program, self.options)) diff --git a/qiskit_ibm_runtime/ibm_backend.py b/qiskit_ibm_runtime/ibm_backend.py index e4fcf82cdf..0df312c9c3 100644 --- a/qiskit_ibm_runtime/ibm_backend.py +++ b/qiskit_ibm_runtime/ibm_backend.py @@ -12,43 +12,58 @@ """Module for interfacing with an IBM Quantum Backend.""" +from __future__ import annotations + import logging from typing import Any from datetime import datetime as python_datetime from copy import deepcopy -from packaging.version import Version -from qiskit import QuantumCircuit, __version__ as qiskit_version +from packaging.version import Version +from qiskit import QuantumCircuit +from qiskit import __version__ as qiskit_version from qiskit.providers.backend import BackendV2 as Backend from qiskit.providers.options import Options from qiskit.transpiler.target import Target -from .models import ( - BackendStatus, - BackendProperties, - GateConfig, - QasmBackendConfiguration, +from ibm_quantum_schemas.models.executor.version_0_1.models import ( + QuantumProgramResultModel, ) from . import qiskit_runtime_service # pylint: disable=unused-import,cyclic-import - from .api.clients import RuntimeClient from .exceptions import ( IBMBackendApiProtocolError, IBMBackendError, ) +from .models import ( + BackendProperties, + BackendStatus, + GateConfig, + QasmBackendConfiguration, +) +from .options.executor_options import ExecutorOptions +from .quantum_program import QuantumProgram +from .quantum_program.converters import quantum_program_to_0_1, quantum_program_result_from_0_1 +from .runtime_job_v2 import RuntimeJobV2 +from .utils import local_to_utc from .utils.backend_converter import convert_to_target - from .utils.backend_decoder import ( - properties_from_server_data, configuration_from_server_data, + properties_from_server_data, ) -from .utils import local_to_utc + if Version(qiskit_version).major >= 2: - from qiskit.result import MeasLevel, MeasReturnType + from qiskit.result import ( # pylint: disable=ungrouped-imports + MeasLevel, + MeasReturnType, + ) else: - from qiskit.qobj.utils import MeasLevel, MeasReturnType # pylint: disable=import-error + from qiskit.qobj.utils import ( # pylint: disable=import-error + MeasLevel, + MeasReturnType, + ) logger = logging.getLogger(__name__) @@ -153,7 +168,7 @@ class IBMBackend(Backend): def __init__( self, configuration: QasmBackendConfiguration, - service: "qiskit_runtime_service.QiskitRuntimeService", + service: qiskit_runtime_service.QiskitRuntimeService, api_client: RuntimeClient, instance: str | None = None, calibration_id: str | None = None, @@ -220,6 +235,58 @@ def __getattr__(self, name: str) -> Any: "'{}' object has no attribute '{}'".format(self.__class__.__name__, name) ) + def submit( + self, program: QuantumProgram, options: ExecutorOptions | None = None + ) -> RuntimeJobV2: + """Submit a quantum program for execution. + + Args: + program: The program to execute. + options: Execution options. + + Returns: + A job. + """ + options = options or ExecutorOptions() + program_id = "executor" + model = quantum_program_to_0_1(program, options) + + params = model.model_dump() + params["version"] = 2 # TODO: this is a work-around for the dispatch while we use 'execute' + response = self._service._active_api_client._api.program_run( + program_id=program_id, + backend_name=self.name, + image=options.environment.image, + log_level=options.environment.log_level, + session_id=None, + job_tags=options.environment.job_tags, + max_execution_time=None, + start_session=False, + session_time=None, + params=params, + ) + + class Decoder: + """Decoder.""" + + @classmethod + def decode(cls, data: str): # type: ignore[no-untyped-def] + """Decode.""" + obj = QuantumProgramResultModel.model_validate_json(data) + return quantum_program_result_from_0_1(obj) + + return RuntimeJobV2( + backend=self, + api_client=self._service._active_api_client, + job_id=response["id"], + program_id=program_id, + result_decoder=Decoder, # type: ignore[arg-type] + image=options.environment.image, + service=self._service, + version=model.schema_version, + private=False, + ) + def _convert_to_target(self, refresh: bool = False) -> None: """Converts backend configuration and properties to Target object""" if refresh or not self._target: @@ -254,7 +321,7 @@ def calibration_id(self) -> str | None: return self._calibration_id @property - def service(self) -> "qiskit_runtime_service.QiskitRuntimeService": + def service(self) -> qiskit_runtime_service.QiskitRuntimeService: """Return the ``service`` object Returns: @@ -437,7 +504,7 @@ def configuration( def __repr__(self) -> str: return "<{}('{}')>".format(self.__class__.__name__, self.name) - def __call__(self) -> "IBMBackend": + def __call__(self) -> IBMBackend: # For backward compatibility only, can be removed later. return self @@ -475,7 +542,7 @@ def check_faulty(self, circuit: QuantumCircuit) -> None: f"{instr} operating on a faulty edge {qubit_indices}" ) - def __deepcopy__(self, _memo: dict = None) -> "IBMBackend": + def __deepcopy__(self, _memo: dict = None) -> IBMBackend: cpy = IBMBackend( configuration=deepcopy(self.configuration()), service=self._service, @@ -516,7 +583,7 @@ class IBMRetiredBackend(IBMBackend): def __init__( self, configuration: QasmBackendConfiguration, - service: "qiskit_runtime_service.QiskitRuntimeService", + service: qiskit_runtime_service.QiskitRuntimeService, api_client: RuntimeClient | None = None, ) -> None: """IBMRetiredBackend constructor. @@ -553,7 +620,7 @@ def from_name( cls, backend_name: str, api: RuntimeClient | None = None, - ) -> "IBMRetiredBackend": + ) -> IBMRetiredBackend: """Return a retired backend from its name.""" configuration = QasmBackendConfiguration( backend_name=backend_name, diff --git a/qiskit_ibm_runtime/noise_learner_v3/__init__.py b/qiskit_ibm_runtime/noise_learner_v3/__init__.py new file mode 100644 index 0000000000..65c8a263ed --- /dev/null +++ b/qiskit_ibm_runtime/noise_learner_v3/__init__.py @@ -0,0 +1,39 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025 +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +""" +============================================================= +Noise learner V3 (:mod:`qiskit_ibm_runtime.noise_learner_v3`) +============================================================= + +.. currentmodule:: qiskit_ibm_runtime.noise_learner_v3 + +The tools to characterize the noise processes affecting the instructions in noisy +quantum circuits. + +Classes +======= + +.. autosummary:: + :toctree: ../stubs/ + + NoiseLearnerV3 + NoiseLearnerV3Result + NoiseLearnerV3Results + +""" + +from .noise_learner_v3 import NoiseLearnerV3 +from .noise_learner_v3_result import ( # type: ignore[attr-defined] + NoiseLearnerV3Result, + NoiseLearnerV3Results, +) diff --git a/qiskit_ibm_runtime/noise_learner_v3/converters/__init__.py b/qiskit_ibm_runtime/noise_learner_v3/converters/__init__.py new file mode 100644 index 0000000000..60e93d8a95 --- /dev/null +++ b/qiskit_ibm_runtime/noise_learner_v3/converters/__init__.py @@ -0,0 +1,11 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. diff --git a/qiskit_ibm_runtime/noise_learner_v3/converters/version_0_1.py b/qiskit_ibm_runtime/noise_learner_v3/converters/version_0_1.py new file mode 100644 index 0000000000..059b39b991 --- /dev/null +++ b/qiskit_ibm_runtime/noise_learner_v3/converters/version_0_1.py @@ -0,0 +1,102 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Transport conversion functions""" + +from __future__ import annotations + +from collections.abc import Iterable + +from ibm_quantum_schemas.models.noise_learner_v3.version_0_1.models import ( + NoiseLearnerV3ResultModel, + NoiseLearnerV3ResultsModel, + ParamsModel, +) +from ibm_quantum_schemas.models.qpy_model import QpyModelV13ToV16 +from ibm_quantum_schemas.models.tensor_model import F64TensorModel +from qiskit.circuit import CircuitInstruction, QuantumCircuit +from qiskit.quantum_info import QubitSparsePauliList + +from ...options import NoiseLearnerV3Options +from ..noise_learner_v3_result import ( # type: ignore[attr-defined] + NoiseLearnerV3Result, + NoiseLearnerV3Results, +) + + +def noise_learner_v3_inputs_to_0_1( + instructions: Iterable[CircuitInstruction], + options: NoiseLearnerV3Options, +) -> ParamsModel: + """Convert noise learner V3 inputs a V0.1 model.""" + qubits = list({qubit for instr in instructions for qubit in instr.qubits}) + clbits = list({clbit for instr in instructions for clbit in instr.clbits}) + + circuit = QuantumCircuit(list(qubits), list(clbits)) + for instr in instructions: + circuit.append(instr, instr.qubits, instr.clbits) + + return ParamsModel( + instructions=QpyModelV13ToV16.from_quantum_circuit(circuit, qpy_version=16), + options=options.to_options_model(), + ) + + +def noise_learner_v3_inputs_from_0_1( + model: ParamsModel, +) -> tuple[list[CircuitInstruction], NoiseLearnerV3Options]: + """Convert a V0.1 model to noise learner V3 inputs.""" + instructions = list(model.instructions.to_quantum_circuit()) + options = NoiseLearnerV3Options( + **{key: val for key, val in model.options.model_dump().items() if val} + ) + return instructions, options + + +def noise_learner_v3_result_to_0_1( + result: NoiseLearnerV3Results, +) -> NoiseLearnerV3ResultsModel: + """Convert noise learner v3 results to a V0.1 model.""" + return NoiseLearnerV3ResultsModel( + data=[ + NoiseLearnerV3ResultModel( + generators_sparse=[gen.to_sparse_list() for gen in datum._generators], + num_qubits=datum._generators[0].num_qubits, + rates=F64TensorModel.from_numpy(datum._rates), + rates_std=F64TensorModel.from_numpy(datum._rates_std), + metadata=datum.metadata, + ) + for datum in result.data + ], + ) + + +def noise_learner_v3_result_from_0_1( + model: NoiseLearnerV3ResultsModel, +) -> NoiseLearnerV3Results: + """Convert a V0.1 model to noise learner v3 results""" + return NoiseLearnerV3Results( + data=[ + NoiseLearnerV3Result.from_generators( + generators=[ + QubitSparsePauliList.from_sparse_list( + [tuple(term) for term in sparse_list], datum["num_qubits"] + ) + for sparse_list in datum["generators_sparse"] + ], + rates=F64TensorModel(**datum["rates"]).to_numpy(), + rates_std=F64TensorModel(**datum["rates_std"]).to_numpy(), + metadata=datum["metadata"], + ) + for datum in model["data"] + ] + ) diff --git a/qiskit_ibm_runtime/noise_learner_v3/find_learning_protocol.py b/qiskit_ibm_runtime/noise_learner_v3/find_learning_protocol.py new file mode 100644 index 0000000000..99d5c1a32a --- /dev/null +++ b/qiskit_ibm_runtime/noise_learner_v3/find_learning_protocol.py @@ -0,0 +1,77 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +# pylint: disable=inconsistent-return-statements + +"""Noise learner program.""" + +from __future__ import annotations + + +from qiskit.circuit import BoxOp +from qiskit.exceptions import QiskitError +from qiskit.quantum_info import Clifford +from samplomatic.utils import undress_box + +from qiskit_ibm_runtime.exceptions import IBMInputValueError + +from .learning_protocol import LearningProtocol + + +def find_learning_protocol(instruction: BoxOp) -> LearningProtocol | None: + """Find which of the supported learning protocols is suitable to learn the noise of ``instruction``. + + Args: + instruction: The instruction to learn the noise of. + + Returns: + The supported protocol that can learn the noise of this instruction, or ``None`` if none of the + protocols are suitable. + + Raises: + IBMInputValueError: If ``instruction`` does not contain a box. + """ + if (name := instruction.operation.name) != "box": + raise IBMInputValueError(f"Expected a 'box' but found '{name}'.") + + undressed_box = undress_box(instruction.operation) + + if len(undressed_box.body) == 0: + return LearningProtocol.PAULI_LINDBLAD + + # Check if the undressed box contains a layer + active_qubits = [ + qubit for op in undressed_box.body for qubit in op.qubits if op.name != "barrier" + ] + is_layer = len(active_qubits) == len(set(active_qubits)) + + # Check if the undressed box only contains two-qubit Clifford gates + has_only_2q_clifford_gates = all( + (op.is_standard_gate() and op.operation.num_qubits == 2) or op.name == "barrier" + for op in undressed_box.body + ) + if has_only_2q_clifford_gates: + try: + Clifford(undressed_box.body) + except QiskitError: + has_only_2q_clifford_gates = False + + # Check if the undressed box only contains measurements + has_only_meas = all(op.name in ["measure", "barrier"] for op in undressed_box.body) + + if is_layer and has_only_2q_clifford_gates: + return LearningProtocol.PAULI_LINDBLAD + + if is_layer and has_only_meas and len(instruction.qubits) == len(active_qubits): + return LearningProtocol.TREX + + return None diff --git a/qiskit_ibm_runtime/noise_learner_v3/learning_protocol.py b/qiskit_ibm_runtime/noise_learner_v3/learning_protocol.py new file mode 100644 index 0000000000..9f342d62fd --- /dev/null +++ b/qiskit_ibm_runtime/noise_learner_v3/learning_protocol.py @@ -0,0 +1,33 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Learning protocols.""" + +from enum import Enum +from typing import Literal + + +class LearningProtocol(str, Enum): + """The supported learning protocols.""" + + PAULI_LINDBLAD = "pauli_lindblad" + """Pauli Lindblad learning from arXiv:2201.09866.""" + + TREX = "trex" + """Readout learning protocol.""" + + +LearningProtocolLiteral = LearningProtocol | Literal["pauli_lindblad", "trex"] +"""The supported learning protocols. + * ``pauli_lindblad``: Pauli Lindblad learning from arXiv:2201.09866.. + * ``trex``: Readout learning protocol. +""" diff --git a/qiskit_ibm_runtime/noise_learner_v3/noise_learner_v3.py b/qiskit_ibm_runtime/noise_learner_v3/noise_learner_v3.py new file mode 100644 index 0000000000..0bb68a7975 --- /dev/null +++ b/qiskit_ibm_runtime/noise_learner_v3/noise_learner_v3.py @@ -0,0 +1,189 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Noise learner program.""" + +from __future__ import annotations + +import logging +from collections.abc import Iterable + +from qiskit.circuit import CircuitInstruction +from qiskit.providers import BackendV2 + +from qiskit_ibm_runtime.options.utils import UnsetType + +from ..base_primitive import _get_mode_service_backend +from ..batch import Batch +from ..fake_provider.local_service import QiskitRuntimeLocalService +from ..options.noise_learner_v3_options import NoiseLearnerV3Options +from ..qiskit_runtime_service import QiskitRuntimeService +from ..runtime_job_v2 import RuntimeJobV2 + +# pylint: disable=unused-import,cyclic-import +from ..session import Session +from ..utils.default_session import get_cm_session +from ..utils.utils import is_simulator +from .converters.version_0_1 import noise_learner_v3_inputs_to_0_1 +from .noise_learner_v3_decoders import NoiseLearnerV3ResultDecoder +from .validation import validate_instruction, validate_options + +logger = logging.getLogger(__name__) + + +class NoiseLearnerV3: + """Class for executing noise learning experiments. + + The noise learner allows characterizing the noise processes affecting target instructions, based on + the Pauli-Lindblad noise model described in [1]. The instructions provided to the :meth:`~run` + method must contain a twirled-annotated :class:`~.qiskit.circuit.BoxOp` containing ISA operations. + The result of a noise learner job contains a list of :class:`.NoiseLearnerV3Result` objects, one for + each given instruction. + + Args: + mode: The execution mode used to make the primitive query. It can be: + + * A :class:`Backend` if you are using job mode. + * A :class:`Session` if you are using session execution mode. + * A :class:`Batch` if you are using batch execution mode. + + Refer to the + `Qiskit Runtime documentation `__ + for more information about the execution modes. + + options: The desired options. + + References: + 1. E. van den Berg, Z. Minev, A. Kandala, K. Temme, *Probabilistic error + cancellation with sparse Pauli–Lindblad models on noisy quantum processors*, + Nature Physics volume 19, pages 1116–1121 (2023). + `arXiv:2201.09866 [quant-ph] `_ + """ + + _PROGRAM_ID = "noise-learner" + _DECODER = NoiseLearnerV3ResultDecoder + + def __init__( + self, + mode: BackendV2 | Session | Batch | None = None, + options: NoiseLearnerV3Options | None = None, + ): + self._session: BackendV2 | None = None + self._backend: BackendV2 + self._service: QiskitRuntimeService + + self._options = options or NoiseLearnerV3Options() + if ( + isinstance(self._options.experimental, UnsetType) + or self._options.experimental.get("image") is None + ): + self._options.experimental = {} + + if isinstance(mode, (Session, Batch)): + self._session = mode + self._backend = self._session._backend + self._service = self._session.service + elif open_session := get_cm_session(): + if open_session != mode: + if open_session._backend != mode: + raise ValueError( + "The backend passed in to the primitive is different from the session " + "backend. Please check which backend you intend to use or leave the mode " + "parameter empty to use the session backend." + ) + logger.warning( + "A backend was passed in as the mode but a session context manager " + "is open so this job will run inside this session/batch " + "instead of in job mode." + ) + self._session = open_session + self._backend = self._session._backend + self._service = self._session.service + elif isinstance(mode, BackendV2): + self._backend = mode + self._service = self._backend.service + else: + raise ValueError( + "A backend or session/batch must be specified, or a session/batch must be open." + ) + self._mode, self._service, self._backend = _get_mode_service_backend( # type: ignore[assignment] + mode + ) + + if isinstance(self._service, QiskitRuntimeLocalService): # type: ignore[unreachable] + raise ValueError("``NoiseLearner`` not currently supported in local mode.") + + @property + def options(self) -> NoiseLearnerV3Options: + """The options in this noise learner.""" + return self._options + + def run(self, instructions: Iterable[CircuitInstruction]) -> RuntimeJobV2: + """Submit a request to the noise learner program. + + Args: + instructions: The instructions to learn the noise of. + + Returns: + The submitted job. + + Raises: + IBMInputValueError: If an instruction does not contain a box. + IBMInputValueError: If an instruction contains a box without twirl annotation. + IBMInputValueError: If an instruction contains unphysical qubits, i.e., qubits that do not + belong to the "physical" register ``QuantumRegister(backend.num_qubits, 'q')`` for the + backend in use. + IBMInputValueError: If an instruction a box with non-ISA gates. + IBMInputValueError: If an instruction cannot be learned by any of the supported learning + protocols. + """ + if self._backend: + target = getattr(self._backend, "target", None) + if target and not is_simulator(self._backend): + for instruction in instructions: + validate_instruction(instruction, target) + + configuration = getattr(self._backend, "configuration", None) + if configuration and not is_simulator(self._backend): + validate_options(self.options, configuration()) + + inputs = noise_learner_v3_inputs_to_0_1(instructions, self.options).model_dump() + inputs["version"] = 3 # TODO: this is a work-around for the dispatch + runtime_options = self.options.to_runtime_options() + runtime_options["backend"] = self._backend.name + + if self._session: + run = self._session._run + else: + run = self._service._run + runtime_options["instance"] = self._backend._instance + + if get_cm_session(): + logger.warning( + "Even though a session/batch context manager is open this job will run in job " + "mode because the %s primitive was initialized outside the context manager. " + "Move the %s initialization inside the context manager to run in a " + "session/batch.", + self._PROGRAM_ID, + self._PROGRAM_ID, + ) + + return run( + program_id=self._PROGRAM_ID, + options=runtime_options, + inputs=inputs, + result_decoder=self._DECODER, + ) + + def backend(self) -> BackendV2: + """Return the backend the primitive query will be run on.""" + return self._backend diff --git a/qiskit_ibm_runtime/noise_learner_v3/noise_learner_v3_decoders.py b/qiskit_ibm_runtime/noise_learner_v3/noise_learner_v3_decoders.py new file mode 100644 index 0000000000..e10853025d --- /dev/null +++ b/qiskit_ibm_runtime/noise_learner_v3/noise_learner_v3_decoders.py @@ -0,0 +1,46 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Noise learner program.""" + +from __future__ import annotations + +import logging + +# pylint: disable=unused-import,cyclic-import +from ..utils.result_decoder import ResultDecoder +from .converters.version_0_1 import noise_learner_v3_result_from_0_1 + +logger = logging.getLogger(__name__) + +AVAILABLE_DECODERS = {"v0.1": noise_learner_v3_result_from_0_1} + + +class NoiseLearnerV3ResultDecoder(ResultDecoder): + """Decoder for noise learner V3.""" + + @classmethod + def decode(cls, raw_result: str): # type: ignore[no-untyped-def] + """Decode raw json to result type.""" + decoded: dict[str, str] = super().decode(raw_result) + + try: + schema_version = decoded["schema_version"] + except KeyError: + raise ValueError("Missing schema version.") + + try: + decoder = AVAILABLE_DECODERS[schema_version] + except KeyError: + raise ValueError(f"No decoder found for schema version {schema_version}.") + + return decoder(decoded) diff --git a/qiskit_ibm_runtime/noise_learner_v3/noise_learner_v3_result.py b/qiskit_ibm_runtime/noise_learner_v3/noise_learner_v3_result.py new file mode 100644 index 0000000000..0ae7ffe56e --- /dev/null +++ b/qiskit_ibm_runtime/noise_learner_v3/noise_learner_v3_result.py @@ -0,0 +1,198 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Noise learner V3 results.""" + +from __future__ import annotations + +from typing import Union +from collections.abc import Iterable, Sequence +from numpy.typing import NDArray + +import numpy as np +from qiskit.circuit import BoxOp, CircuitInstruction +from qiskit.quantum_info import PauliLindbladMap, QubitSparsePauliList + +from samplomatic import InjectNoise +from samplomatic.utils import get_annotation + +MetadataLeafTypes = int | str | float +MetadataValue = Union[MetadataLeafTypes, "Metadata", list["MetadataValue"]] +Metadata = dict[str, MetadataValue] + + +class NoiseLearnerV3Result: + r"""The results of a noise learner experiment for a single instruction, in Pauli Lindblad format. + + An error channel Pauli Lindblad :math:`E` acting on a state :math:`\rho` can be expressed in Pauli + Lindblad format as :math:`E(\rho) = e^{\sum_j r_j D_{P_j}}(\rho)`, :math:`P_j` are Pauli operators + (or "generators") and :math:`r_j` are floats (or "rates") [1]. The equivalent Pauli error channel + can be constructed as a composition of single-Pauli channel terms + + .. math:: + + E = e^{\sum_j r_j D_{P_j}} = \prod_j e^{r_j D_{P_j}} + = prod_j \left( (1 - p_j) S_I + p_j S_{P_j} \right) + + where :math:`p_j = \frac12 - \frac12 e^{-2 r_j}`. + + Some strategies for learning noise channels, such as the Pauli Lindblad learning protocol in + Ref. [1], produce degenerate terms, meaning that they learn products of rates as opposed to + individual rates. + + References: + 1. E. van den Berg, Z. Minev, A. Kandala, K. Temme, *Probabilistic error + cancellation with sparse Pauli–Lindblad models on noisy quantum processors*, + Nature Physics volume 19, pages 1116–1121 (2023). + `arXiv:2201.09866 [quant-ph] `_ + """ + + def __init__(self) -> None: + self._generators: list[QubitSparsePauliList] = [] + self._rates: NDArray[np.float64] = np.array([]) + self._rates_std: NDArray[np.float64] = np.array([]) + self.metadata: MetadataValue = {} + + @classmethod + def from_generators( + cls, + generators: Iterable[QubitSparsePauliList], + rates: Iterable[float], + rates_std: Iterable[float] | None = None, + metadata: Metadata | None = None, + ) -> NoiseLearnerV3Result: + """ + Construct from a collection of generators and rates. + + Args: + generators: The generators describing the noise channel in the Pauli Lindblad format. This + is a list of :class:`~qiskit.quantum_info.QubitSparsePauliList` objects, as opposed to + a list of :class:`~qiskit.quantum_info.QubitSparsePauli`, in order to capture + degeneracies present within the model. + rates: The rates of the individual generators. The ``i``-th element in this list represents + the rate of all the Paulis in the ``i``-th generator. + rates_std: The standard deviation associated to the rates of the generators. If ``None``, + it sets all the standard deviations to ``0``. + metadata: A dictionary of metadata. + """ + obj = cls() + obj._generators = list(generators) + obj._rates = np.array(rates, dtype=np.float64) + obj._rates_std = np.array( + [0] * len(obj._generators) if rates_std is None else rates_std, dtype=np.float64 + ) + obj.metadata = metadata or {} + + if len({len(obj._generators), len(obj._rates), len(obj._rates_std)}) != 1: + raise ValueError("'generators', 'rates', and 'rates_std' must be of the same length.") + + if len({generator.num_qubits for generator in obj._generators}) != 1: + raise ValueError("All the generators must have the same number of qubits.") + + return obj + + def to_pauli_lindblad_map(self) -> PauliLindbladMap: + """Transform this result to a Pauli Lindblad map. + + The Pauli terms in the generators are indexed in physical qubit order, that is, the order + of the qubits in the outer-most circuit. + """ + coefficients = [ + repeated_rate + for generator, rate in zip(self._generators, self._rates) + for repeated_rate in [rate] * len(generator) + ] + paulis = QubitSparsePauliList.from_qubit_sparse_paulis( + pauli for generator in self._generators for pauli in generator + ) + + return PauliLindbladMap.from_components(coefficients, paulis) + + def __len__(self) -> int: + return len(self._generators) + + def __repr__(self) -> str: + return f"NoiseLearnerV3Result(<{len(self)}> generators)" + + +class NoiseLearnerV3Results: + """The results of a noise learner experiment. + + Args: + data: The data in this result object. + metadata: A dictionary of metadata. + """ + + def __init__(self, data: Iterable[NoiseLearnerV3Result], metadata: Metadata | None = None): + self.data = list(data) + self.metadata = metadata or None + + def to_dict( + self, + instructions: Sequence[CircuitInstruction], + require_refs: bool = True, + ) -> dict[int, PauliLindbladMap]: + """Convert to a dictionary from :attr:`InjectNoise.ref` to :class:`PauliLindbladMap` objects. + This function iterates over a sequence of instructions, extracts the ``ref`` value from the + inject noise annotation of each instruction, and returns a dictionary mapping those refs + to the corresponding noise data (in :class:`PauliLindbladMap` format) stored in this + :class:`NoiseLearnerV3Results` object. + + Args: + instructions: The instructions to get the refs from. + require_refs: Whether to raise if some of the instructions do not own an inject noise + annotation. If ``False``, all the instructions that do not contain an inject noise + annotations are simply skipped when constructing the returned dictionary. + + Raise: + ValueError: If ``instructions`` contains a number of elements that is not equal to the + item in this :class:`NoiseLearnerV3Results` object. + ValueError: If some of the instructions do not contain a box. + ValueError: If multiple instructions have the same ``ref``. + ValueError: If some of the instructions have no inject noise annotation and ``require_refs`` + if ``True``. + """ + if len(instructions) != len(self.data): + raise ValueError( + f"Expected {len(self.data)} instructions but found {len(instructions)}." + ) + + noise_source = {} + num_instr = 0 + for instr, datum in zip(instructions, self.data): + if not isinstance(instr.operation, BoxOp): + raise ValueError("Found an instruction that does not contain a box.") + if annotation := get_annotation(instr.operation, InjectNoise): + num_instr += 1 + noise_source[annotation.ref] = datum.to_pauli_lindblad_map() + elif require_refs: + raise ValueError( + "Found an instruction without an inject noise annotation. " + "Consider setting 'require_refs' to ``False``." + ) + + if num_instr != len(noise_source): + raise ValueError("Found multiple instructions with the same ``ref``.") + + return noise_source + + def __getitem__(self, idx: int) -> NoiseLearnerV3Result: + return self.data[idx] + + def __iter__(self) -> Iterable[NoiseLearnerV3Result]: + return iter(self.data) + + def __len__(self) -> int: + return len(self.data) + + def __repr__(self) -> str: + return f"NoiseLearnerV3Results(<{len(self.data)}> data)" diff --git a/qiskit_ibm_runtime/noise_learner_v3/validation.py b/qiskit_ibm_runtime/noise_learner_v3/validation.py new file mode 100644 index 0000000000..f5c2b8d475 --- /dev/null +++ b/qiskit_ibm_runtime/noise_learner_v3/validation.py @@ -0,0 +1,191 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Noise learner program.""" + +from __future__ import annotations + +import numpy as np +from qiskit.circuit import ( + CircuitInstruction, + ControlFlowOp, + ParameterExpression, + QuantumRegister, + BoxOp, +) +from qiskit.transpiler import Target +from samplomatic.annotations import Twirl +from samplomatic.utils import get_annotation + +from .find_learning_protocol import find_learning_protocol +from ..models.backend_configuration import BackendConfiguration + +from ..exceptions import IBMInputValueError +from ..options import NoiseLearnerV3Options +from ..options.post_selection_options import DEFAULT_X_PULSE_TYPE + + +def validate_options(options: NoiseLearnerV3Options, configuration: BackendConfiguration) -> None: + """Validates the options of a noise learner job.""" + if options.post_selection.enable is True: # type: ignore[union-attr] + x_pulse_type = ( + options.post_selection.x_pulse_type or DEFAULT_X_PULSE_TYPE # type: ignore[union-attr] + ) + if x_pulse_type not in (basis_gates := configuration.basis_gates): + raise ValueError( + f"Cannot apply Post Selection with X-pulse type '{x_pulse_type}' on a backend with " + f"basis gates {basis_gates}." + ) + + +def validate_instruction(instruction: CircuitInstruction, target: Target) -> None: + """Validates that an instruction is valid for the noise learner. + + Args: + instruction: The instruction to validate. + target: The target to validate against. + + Raises: + IBMInputValueError: If ``instruction`` does not contain a box. + IBMInputValueError: If the box in ``instruction`` does not contain a ``Twirl`` annotation. + IBMInputValueError: If ``instruction`` contains unphysical qubits. + IBMInputValueError: If the box in ``instruction`` contains non-ISA gates. + IBMInputValueError: If ``instruction`` cannot be learned by any of the supported learning + protocols. + """ + if reason := _contains_twirled_box(instruction): + raise IBMInputValueError(reason) + + if reason := _contains_physical_qubits(instruction, target): + raise IBMInputValueError(reason) + + if reason := _is_isa_instruction(instruction, target): + raise IBMInputValueError(reason) + + if not find_learning_protocol(instruction): + raise IBMInputValueError( + "Found an instruction that cannot be learned by any of the supported " + "learning protocols." + ) + + +def _contains_twirled_box(instruction: CircuitInstruction) -> str: + """Check that an instruction contains a box with a twirl annotation. + + Args: + instruction: The instruction to validate. + + Returns: + An error message if ``instruction`` does not contain a twirled-annotated box, or an empty + string otherwise. + """ + if (name := instruction.operation.name) != "box": + return f"Expected a 'box' but found '{name}'." + + if not get_annotation(instruction.operation, Twirl): + return "Found a box without a ``Twirl`` annotation." + + return "" + + +def _contains_physical_qubits(instruction: CircuitInstruction, target: Target) -> str: + """Check that ``instruction`` acts on physical qubits. + + Args: + instruction: The instruction to validate. + target: The target to validate against. + + Returns: + An error message if ``instruction`` doesn't contain physical qubits, an empty string otherwise. + """ + qreg = QuantumRegister(target.num_qubits, "q") + if unphysical_qubits := [qubit for qubit in instruction.qubits if qubit not in qreg]: + return ( + f"Every qubit must be part of {qreg}, but the following qubits " + f"are not part of it: {unphysical_qubits}." + ) + return "" + + +def _is_isa_instruction(instruction: CircuitInstruction, target: Target) -> str: + """Check that a box instruction contains an ISA circuit. + + Assumes but does not check that: + * ``instruction`` contains a box. + * ``instruction`` contains physical qubits. + + Args: + instruction: The instruction to validate. + target: The target to validate against. + + Returns: + An error message if ``instruction`` is not ISA, or an empty string otherwise. + """ + if instruction.operation.num_qubits > target.num_qubits: + return ( + f"The instruction has {instruction.num_qubits} qubits " + f"but the target system requires {target.num_qubits} qubits." + ) + + # A map from the instruction qubits to indexes + qreg = QuantumRegister(target.num_qubits, "q") + qubit_map = {qubit: idx for idx, qubit in enumerate(qreg)} + + # A map from the box qubits to indexes + box_qubit_map = { + box_qubit: qubit_map[instruction_qubit] + for instruction_qubit, box_qubit in zip( + instruction.qubits, instruction.operation.body.qubits + ) + } + + for op in instruction.operation.body: + if ( + not target.instruction_supported( + name := op.name, + qargs := tuple(box_qubit_map[box_qubit] for box_qubit in op.qubits), + ) + and op.name != "barrier" + ): + return f"The instruction {op.name} on qubits {qargs} is not supported by the target system." + + # rzz gate is calibrated only for the range [0, pi/2]. + # We allow an angle value of a bit more than pi/2, to compensate floating point rounding + # errors (beyond pi/2 does not trigger an error down the stack, only may become less + # accurate). + if name == "rzz" and (reason := _validate_rzz_angle(op.params[0])): + return reason + + if isinstance(op, (ControlFlowOp, BoxOp)): + return f"The instruction {op.name} on qubits {qargs} is not supported by the noise learner." + + return "" + + +def _validate_rzz_angle(angle: float) -> str: + """Verify that all rzz angles are in the range ``[0, pi/2]``. + + We allow an angle value of a bit more than pi/2, to compensate floating point rounding + errors. + + Args: + angle: An angle to be checked + + Returns: + An empty string if the angle is valid, otherwise an error message. + """ + if not isinstance(angle, ParameterExpression) and (angle < 0.0 or angle > np.pi / 2 + 1e-10): + return ( + f"'rzz' is supported only for angles in the range ``[0, pi/2]``, but an angle " + f"({angle}) outside of this range has been requested." + ) + return "" diff --git a/qiskit_ibm_runtime/options/__init__.py b/qiskit_ibm_runtime/options/__init__.py index a93e6f6755..a07a0bcc0d 100644 --- a/qiskit_ibm_runtime/options/__init__.py +++ b/qiskit_ibm_runtime/options/__init__.py @@ -61,6 +61,7 @@ EstimatorOptions SamplerOptions + ExecutorOptions Suboptions @@ -71,12 +72,14 @@ :nosignatures: NoiseLearnerOptions + NoiseLearnerV3Options DynamicalDecouplingOptions ResilienceOptionsV2 LayerNoiseLearningOptions MeasureNoiseLearningOptions PecOptions ZneOptions + PostSelectionOptions TwirlingOptions ExecutionOptionsV2 SamplerExecutionOptionsV2 @@ -85,18 +88,21 @@ """ +from .dynamical_decoupling_options import DynamicalDecouplingOptions from .environment_options import EnvironmentOptions +from .estimator_options import EstimatorOptions from .execution_options import ExecutionOptionsV2 +from .executor_options import ExecutorOptions +from .layer_noise_learning_options import LayerNoiseLearningOptions +from .measure_noise_learning_options import MeasureNoiseLearningOptions from .noise_learner_options import NoiseLearnerOptions +from .noise_learner_v3_options import NoiseLearnerV3Options from .options import OptionsV2 -from .simulator_options import SimulatorOptions +from .pec_options import PecOptions +from .post_selection_options import PostSelectionOptions from .resilience_options import ResilienceOptionsV2 -from .twirling_options import TwirlingOptions -from .estimator_options import EstimatorOptions +from .sampler_execution_options import SamplerExecutionOptionsV2 from .sampler_options import SamplerOptions -from .dynamical_decoupling_options import DynamicalDecouplingOptions -from .layer_noise_learning_options import LayerNoiseLearningOptions -from .measure_noise_learning_options import MeasureNoiseLearningOptions -from .pec_options import PecOptions +from .simulator_options import SimulatorOptions +from .twirling_options import TwirlingOptions from .zne_options import ZneOptions -from .sampler_execution_options import SamplerExecutionOptionsV2 diff --git a/qiskit_ibm_runtime/options/executor_options.py b/qiskit_ibm_runtime/options/executor_options.py new file mode 100644 index 0000000000..d4513bbdca --- /dev/null +++ b/qiskit_ibm_runtime/options/executor_options.py @@ -0,0 +1,88 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Executor options.""" + +from __future__ import annotations + + +from pydantic.dataclasses import dataclass +from pydantic import Field + +from .environment_options import LogLevelType + + +@dataclass +class ExecutionOptions: + """Low-level execution options.""" + + init_qubits: bool = True + r"""Whether to reset the qubits to the ground state for each shot. + """ + + rep_delay: float | None = None + r"""The repetition delay. This is the delay between a measurement and + the subsequent quantum circuit. This is only supported on backends that have + ``backend.dynamic_reprate_enabled=True``. It must be from the + range supplied by ``backend.rep_delay_range``. + Default is given by ``backend.default_rep_delay``. + """ + + +@dataclass +class EnvironmentOptions: + """Options related to the execution environment.""" + + log_level: LogLevelType = "WARNING" + r"""logging level to set in the execution environment. The valid + log levels are: ``DEBUG``, ``INFO``, ``WARNING``, ``ERROR``, and ``CRITICAL``. + """ + + job_tags: list[str] = Field(default_factory=list) + r"""Tags to be assigned to the job. + + The tags can subsequently be used as a filter in the + :meth:`qiskit_ibm_runtime.qiskit_runtime_service.jobs()` function call. + """ + + private: bool = False + r"""Boolean that indicates whether the job is marked as private. + + When set to true, + input parameters are not returned, and the results can only be read once. + After the job is completed, input parameters are deleted from the service. + After the results are read, these are also deleted from the service. + When set to false, the input parameters and results follow the + standard retention behavior of the API. + """ + + max_execution_time: int | None = None + """Maximum execution time in seconds. + + This value bounds system execution time (not wall clock time). System execution time is the + amount of time that the system is dedicated to processing your job. If a job exceeds + this time limit, it is forcibly cancelled. + """ + + image: str | None = None + r"""Runtime image used for this job.""" + + +@dataclass +class ExecutorOptions: + """Options for the executor.""" + + environment: EnvironmentOptions = Field(default_factory=EnvironmentOptions) + """Options related to the execution environment.""" + + execution: ExecutionOptions = Field(default_factory=ExecutionOptions) + """Low-level execution options.""" diff --git a/qiskit_ibm_runtime/options/noise_learner_v3_options.py b/qiskit_ibm_runtime/options/noise_learner_v3_options.py new file mode 100644 index 0000000000..a0f7c2b310 --- /dev/null +++ b/qiskit_ibm_runtime/options/noise_learner_v3_options.py @@ -0,0 +1,220 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""NoiseLearnerV3Options options.""" + +from __future__ import annotations + +import copy +from dataclasses import asdict, fields +from typing import Any +from collections.abc import Callable + +from pydantic import Field, ValidationInfo, field_validator +from qiskit.transpiler import CouplingMap + +from ibm_quantum_schemas.models.noise_learner_v3.version_0_1.models import ( + OptionsModel, +) + +from ..runtime_options import RuntimeOptions +from .environment_options import EnvironmentOptions +from .options import BaseOptions +from .post_selection_options import PostSelectionOptions +from .simulator_options import SimulatorOptions +from .utils import ( + Dict, + Unset, + UnsetType, + make_constraint_validator, + merge_options_v2, + primitive_dataclass, + remove_dict_unset_values, + remove_empty_dict, + skip_unset_validation, +) + + +@primitive_dataclass +class NoiseLearnerV3Options(BaseOptions): + """ + Options for :class:`.NoiseLearnerV3`. + """ + + shots_per_randomization: int = 128 + r"""The total number of shots to use per randomized learning circuit.""" + + num_randomizations: int = 32 + r"""The number of random circuits to use per learning circuit configuration. + + For TREX experiments, a configuration is a measurement basis. + + For Pauli Lindblad experiments, a configuration is a measurement basis and depth setting. + For example, if your experiment has six depths, then setting this value to ``32`` will result + in a total of ``32 * 9 * 6`` circuits that need to be executed (where ``9`` is the number + of circuits that need to be implemented to measure all the required observables, see the + note in the docstring for :class:`~.NoiseLearnerOptions` for mode details), at + :attr:`~shots_per_randomization` each. + """ + + layer_pair_depths: list[int] = (0, 1, 2, 4, 16, 32) # type: ignore[assignment] + r"""The circuit depths (measured in number of pairs) to use in Pauli Lindblad experiments. + + Pairs are used as the unit because we exploit the order-2 nature of our entangling gates in + the noise learning implementation. For example, a value of ``3`` corresponds to 6 repetitions + of the layer of interest. + + .. note:: + This field is ignored by TREX experiments. + """ + + post_selection: PostSelectionOptions | Dict = Field(default_factory=PostSelectionOptions) + r"""Options for post selecting the results of noise learning circuits. + """ + + experimental: UnsetType | dict = Unset + r"""Experimental options. + + These options are subject to change without notification, and stability is not guaranteed. + """ + + _ge0 = make_constraint_validator( + "num_randomizations", "shots_per_randomization", ge=1 # type: ignore[arg-type] + ) + + @field_validator("layer_pair_depths", mode="after") + @classmethod + @skip_unset_validation + def _nonnegative_list(cls, value: list[int], info: ValidationInfo) -> list[int]: + if any(i < 0 for i in value): + raise ValueError(f"`{cls.__name__}.{info.field_name}` option value must all be >= 0.") + return value + + def to_options_model(self) -> OptionsModel: + """Turn these options into an ``OptionsModel`` object. + + Filters out every irrelevant field and replaces ``Unset``\\s with ``None``\\s. + """ + options_dict = asdict(self) + + filtered_options = {} + for key in OptionsModel.model_fields: # pylint: disable=not-an-iterable + filtered_options[key] = options_dict.get(key) + + remove_dict_unset_values(filtered_options) + return OptionsModel(**filtered_options) + + def to_runtime_options(self) -> dict: + """Turn these options into a dictionary of runtime options object. + + Filters out every irrelevant field (i.e., those that are not fields of :class:`.RuntimeOptions`) + and replaces ``Unset``\\s with ``None``\\s. + """ + options_dict = asdict(self) + environment = options_dict.get("environment") + + filtered_options = {"max_execution_time": options_dict.get("max_execution_time", None)} + for fld in fields(RuntimeOptions): + if fld.name in environment: + filtered_options[fld.name] = environment[fld.name] + + if "image" in options_dict: + filtered_options["image"] = options_dict["image"] + elif "image" in options_dict.get("experimental", {}): + filtered_options["image"] = options_dict["experimental"]["image"] + + remove_dict_unset_values(filtered_options) + return filtered_options + + def get_callback(self) -> Callable | None: + """Get the callback.""" + options_dict = asdict(self) + remove_dict_unset_values(options_dict) + return options_dict.get("environment", {}).get("callback", None) + + # The following code is copy/pasted from OptionsV2. + # Reason not to use OptionsV2: As stated in the docstring, it is meant for v2 primitives, and + # NoiseLearnerV3 is neither a primitive nor a v2. + # Reason not to implement OptionsV3: I don't feel like committing to an API for it. + + # Options not really related to primitives. + max_execution_time: UnsetType | int = Unset + environment: EnvironmentOptions | Dict = Field(default_factory=EnvironmentOptions) + simulator: SimulatorOptions | Dict = Field(default_factory=SimulatorOptions) + + def update(self, **kwargs: Any) -> None: + """Update the options.""" + + def _set_attr(_merged: dict) -> None: + for key, val in _merged.items(): + if not key.startswith("_"): + setattr(self, key, val) + + merged = merge_options_v2(self, kwargs) + _set_attr(merged) + + @staticmethod + def _get_program_inputs(options: dict) -> dict: + """Convert the input options to program compatible inputs. + + Returns: + Inputs acceptable by primitives. + """ + + def _set_if_exists(name: str, _inputs: dict, _options: dict) -> None: + if name in _options: + _inputs[name] = _options[name] + + options_copy = copy.deepcopy(options) + output_options: dict[str, Any] = {} + sim_options = options_copy.get("simulator", {}) + coupling_map = sim_options.get("coupling_map", Unset) + # TODO: We can just move this to json encoder + if isinstance(coupling_map, CouplingMap): + sim_options["coupling_map"] = list(map(list, coupling_map.get_edges())) + + for fld in [ + "default_precision", + "default_shots", + "seed_estimator", + "dynamical_decoupling", + "resilience", + "twirling", + "simulator", + "execution", + ]: + _set_if_exists(fld, output_options, options_copy) + + # Add arbitrary experimental options + experimental = options_copy.get("experimental", None) + if isinstance(experimental, dict): + new_keys = {} + for key in list(experimental.keys()): + if key not in output_options: + new_keys[key] = experimental.pop(key) + output_options = merge_options_v2(output_options, experimental) + if new_keys: + output_options["experimental"] = new_keys + + # Remove image + output_options.get("experimental", {}).pop("image", None) + + remove_dict_unset_values(output_options) + remove_empty_dict(output_options) + + inputs = { + "options": output_options, + } + if options_copy.get("resilience_level", Unset) != Unset: + inputs["resilience_level"] = options_copy["resilience_level"] + + return inputs diff --git a/qiskit_ibm_runtime/options/post_selection_options.py b/qiskit_ibm_runtime/options/post_selection_options.py new file mode 100644 index 0000000000..22e0fff821 --- /dev/null +++ b/qiskit_ibm_runtime/options/post_selection_options.py @@ -0,0 +1,71 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Post selection options.""" + +from typing import Literal + +from .options import BaseOptions +from .utils import primitive_dataclass + +DEFAULT_X_PULSE_TYPE = "xslow" +"""The default for :meth:`.PostSelectionOptions.x_pulse_type`.""" + + +@primitive_dataclass +class PostSelectionOptions(BaseOptions): + """ + Options for post selecting results. + """ + + enable: bool = False + r"""Whether to enable Post Selection when performing learning experiments. + + If ``True``, Post Selection is applied to all the learning circuits. In particular, the following + steps are undertaken: + + * Using the passes in + :mod:`qiskit_addon_utils.noise_management.post_selection.transpiler.passes`, the learning + circuits are modified by adding measurements on the spectator qubits, as well as + post selection measurements. + * The results of each individual learning circuits are post selected by discarding the shots + where one or more bits failed to flip, as explained in the docstring of + :meth:`qiskit_addon_utils.noise_management.post_selection.PostSelector.compute_mask`. + + If ``False``, all the other Post Selection options will be ignored. + """ + + x_pulse_type: Literal["xslow", "rx"] = "xslow" + r"""The type of the X-pulse used for the post selection measurements.""" + + strategy: Literal["node", "edge"] = "node" + r"""The strategy used to decide if a shot should be kept or discarded. + + The available startegies are: + + * ``'node'``: Discard every shot where one or more bits failed to flip. Keep every other shot. + * ``'edge'``: Discard every shot where there exists a pair of neighbouring qubits for which both of + the bits failed to flip. Keep every other shot. + + See the dosctrings of :class:`.PostSelector` and :meth:`.PostSelector.compute_mask` for more details. + + Defaults to ``node``. + """ + + @staticmethod + def _get_program_inputs(options: dict) -> dict: + """Convert the input options to program compatible inputs. + + Returns: + Inputs acceptable by primitives. + """ + raise NotImplementedError() diff --git a/qiskit_ibm_runtime/quantum_program/__init__.py b/qiskit_ibm_runtime/quantum_program/__init__.py new file mode 100644 index 0000000000..53a4033f57 --- /dev/null +++ b/qiskit_ibm_runtime/quantum_program/__init__.py @@ -0,0 +1,42 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +""" +============================================================ +Quantum Programs (:mod:`qiskit_ibm_runtime.quantum_program`) +============================================================ + +.. currentmodule:: qiskit_ibm_runtime.quantum_program + +Overview +======== + +A quantum program consists of a list of ordered elements, each of which contains a single +circuit and an array of associated parameter values. Executing a quantum program will +sample the outcome of each circuit for the specified number of ``shots`` for each set of +circuit arguments provided. + + +Classes +======= + +.. autosummary:: + :toctree: ../stubs/ + :nosignatures: + + QuantumProgram + QuantumProgramItem + QuantumProgramResult +""" + +from .quantum_program import QuantumProgram, QuantumProgramItem +from .quantum_program_result import QuantumProgramResult diff --git a/qiskit_ibm_runtime/quantum_program/converters.py b/qiskit_ibm_runtime/quantum_program/converters.py new file mode 100644 index 0000000000..ee1b6f32a9 --- /dev/null +++ b/qiskit_ibm_runtime/quantum_program/converters.py @@ -0,0 +1,92 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Transport conversion functions""" + +from __future__ import annotations + +from dataclasses import asdict + +import numpy as np +from samplomatic.tensor_interface import TensorSpecification, PauliLindbladMapSpecification + +from ibm_quantum_schemas.models.executor.version_0_1.models import ( + ParamsModel, + CircuitItemModel, + SamplexItemModel, + QuantumProgramModel, + QuantumProgramResultModel, +) +from ibm_quantum_schemas.models.pauli_lindblad_map_model import PauliLindbladMapModel +from ibm_quantum_schemas.models.samplex_model import SamplexModelSSV1 +from ibm_quantum_schemas.models.tensor_model import F64TensorModel, TensorModel +from ibm_quantum_schemas.models.qpy_model import QpyModelV13ToV16 + + +from .quantum_program import QuantumProgram, CircuitItem, SamplexItem +from .quantum_program_result import QuantumProgramResult, ChunkPart, ChunkSpan, Metadata +from ..options.executor_options import ExecutorOptions + + +def quantum_program_to_0_1(program: QuantumProgram, options: ExecutorOptions) -> ParamsModel: + """Convert a :class:`~.QuantumProgram` to a V0.1 model.""" + model_items = [] + for item in program.items: + chunk_size = "auto" if item.chunk_size is None else item.chunk_size + if isinstance(item, CircuitItem): + model_item = CircuitItemModel( + circuit=QpyModelV13ToV16.from_quantum_circuit(item.circuit, qpy_version=16), + circuit_arguments=F64TensorModel.from_numpy(item.circuit_arguments), + chunk_size=chunk_size, + ) + elif isinstance(item, SamplexItem): + arguments = {} + for spec in item.samplex_arguments.specs: + if spec.name in item.samplex_arguments: + name, value = spec.name, item.samplex_arguments[spec.name] + if isinstance(spec, TensorSpecification) or isinstance(value, np.ndarray): + arguments[name] = TensorModel.from_numpy(value) + elif isinstance(spec, PauliLindbladMapSpecification): + arguments[name] = PauliLindbladMapModel.from_pauli_lindblad_map(value) + else: + arguments[name] = value + model_item = SamplexItemModel( + circuit=QpyModelV13ToV16.from_quantum_circuit(item.circuit, qpy_version=16), + samplex=SamplexModelSSV1.from_samplex(item.samplex, ssv=1), + samplex_arguments=arguments, + shape=item.shape, + chunk_size=chunk_size, + ) + else: + raise ValueError(f"Item {item} is not valid.") + model_items.append(model_item) + + return ParamsModel( + quantum_program=QuantumProgramModel(shots=program.shots, items=model_items), + options=asdict(options.execution), # type: ignore[call-overload] + ) + + +def quantum_program_result_from_0_1(model: QuantumProgramResultModel) -> QuantumProgramResult: + """Convert a V0.1 model to a :class:`QuantumProgramResult`.""" + metadata = Metadata( + chunk_timing=[ + ChunkSpan( + span.start, span.stop, [ChunkPart(part.idx_item, part.size) for part in span.parts] + ) + for span in model.metadata.chunk_timing + ] + ) + return QuantumProgramResult( + data=[{name: val.to_numpy() for name, val in item.results.items()} for item in model.data], + metadata=metadata, + ) diff --git a/qiskit_ibm_runtime/quantum_program/quantum_program.py b/qiskit_ibm_runtime/quantum_program/quantum_program.py new file mode 100644 index 0000000000..48c54ee61d --- /dev/null +++ b/qiskit_ibm_runtime/quantum_program/quantum_program.py @@ -0,0 +1,283 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""QuantumProgram""" + +from __future__ import annotations + +import abc +import math +from typing import TYPE_CHECKING, Any +from collections.abc import Iterable + +import numpy as np +from qiskit.circuit import QuantumCircuit +from qiskit.quantum_info import PauliLindbladMap +from samplomatic.samplex import Samplex + + +if TYPE_CHECKING: + from ..ibm_backend import IBMBackend + + +def _desc_arr(arr: Any) -> str: + if hasattr(arr, "shape") and hasattr(arr, "dtype"): + return f"<{arr.shape}, {arr.dtype}>" + return f"<{type(arr).__name__}>" + + +class QuantumProgramItem(abc.ABC): + """An item of a :class:`QuantumProgram`. + + Args: + circuit: The circuit to be executed. + chunk_size: The maximum number of bound circuits in each shot loop execution, or + ``None`` to use a server-side heuristic to optimize speed. When not executing + in a session, the server-side heuristic is always used and this value is ignored. + ignored. + """ + + def __init__(self, circuit: QuantumCircuit, chunk_size: int | None = None): + if not isinstance(circuit, QuantumCircuit): + raise ValueError(f"Expected {repr(circuit)} to be a QuantumCircuit.") + + self.circuit = circuit + self.chunk_size = chunk_size + + @property + @abc.abstractmethod + def shape(self) -> tuple[int]: + """The shape of this item when broadcasted over all arguments.""" + + def size(self) -> int: + """The total number elements in this item; the product of the entries of :attr:`~.shape`.""" + return math.prod(self.shape) + + +class CircuitItem(QuantumProgramItem): + """An item of a :class:`QuantumProgram` containing a circuit and its arguments. + + Args: + circuit: The circuit to be executed. + circuit_arguments: Arguments for the parameters of the circuit. + chunk_size: The maximum number of bound circuits in each shot loop execution, or + ``None`` to use a server-side heuristic to optimize speed. When not executing + in a session, the server-side heuristic is always used and this value is ignored. + """ + + def __init__( + self, + circuit: QuantumCircuit, + *, + circuit_arguments: np.ndarray | None = None, + chunk_size: int | None = None, + ): + if circuit_arguments is None: + if circuit.num_parameters: + raise ValueError( + f"{repr(circuit)} is parametric, but no 'circuit_arguments' were supplied." + ) + circuit_arguments = [] + + circuit_arguments = np.array(circuit_arguments, dtype=float) + + if circuit_arguments.shape[-1] != circuit.num_parameters: + raise ValueError( + "Expected the last axis of 'circuit_arguments' to have size " + f"{circuit.num_parameters} in order to match the number of parameters of the " + f"circuit, but found shape {circuit_arguments.shape} instead." + ) + + super().__init__(circuit=circuit, chunk_size=chunk_size) + self.circuit_arguments = circuit_arguments + + @property + def shape(self) -> tuple[int]: + return self.circuit_arguments.shape[:-1] + + def __repr__(self) -> str: + circuit = f"" + + if not self.circuit_arguments.size: + circuit_args = "" + else: + circuit_args = f", circuit_arguments={_desc_arr(self.circuit_arguments)}" + + chunk_size = "" if self.chunk_size is None else f", chunk_size={self.chunk_size}" + + return f"QuantumProgramSimpleItem({circuit}{circuit_args}{chunk_size})" + + +class SamplexItem(QuantumProgramItem): + """An item of a :class:`QuantumProgram` containing a circuit and samplex to feed it arguments. + + Args: + circuit: The circuit to be executed. + samplex: A samplex to draw random parameters for the circuit. + samplex_arguments: A map from argument names to argument values for the samplex. + shape: A shape tuple to extend the implicit shape defined by ``samplex_arguments``. + Non-trivial axes introduced by this extension enumerate randomizations. + chunk_size: The maximum number of bound circuits in each shot loop execution, or + ``None`` to use a server-side heuristic to optimize speed. When not executing + in a session, the server-side heuristic is always used and this value is + ignored. + """ + + def __init__( + self, + circuit: QuantumCircuit, + samplex: Samplex, + *, + samplex_arguments: dict[str, np.ndarray | PauliLindbladMap] | None = None, + shape: tuple[int, ...] | None = None, + chunk_size: int | None = None, + ): + if not isinstance(circuit, QuantumCircuit): + raise ValueError(f"Expected {repr(circuit)} to be a QuantumCircuit.") + + # Calling bind() here will do all Samplex validation + inputs = samplex.inputs().make_broadcastable().bind(**samplex_arguments) + + if not inputs.fully_bound: + raise ValueError( + "The following required samplex arguments are missing:\n" + f"{inputs.describe(prefix=' * ', include_bound=False)}" + ) + + try: + shape = np.broadcast_shapes(shape or (), inputs.shape) + except ValueError as exc: + raise ValueError( + f"The provided shape {shape} must be broadcastable with the shape implicit in " + f"the sample_arguments, which is {inputs.shape}." + ) from exc + + super().__init__(circuit=circuit, chunk_size=chunk_size) + self._shape = np.broadcast_shapes(shape, inputs.shape) + self.samplex = samplex + self.samplex_arguments = inputs + + @property + def shape(self) -> tuple[int]: + return self._shape + + def __repr__(self) -> str: + circuit = f"" + + samplex = f", " if self.samplex is not None else "" + + if not self.samplex_arguments: + samplex_args = "" + else: + content = ", ".join( + f"'{name}'={_desc_arr(val)}" for name, val in self.samplex_arguments.items() + ) + samplex_args = f", samplex_arguments={{{content}}}" + + shape = f", shape={self.shape}" + chunk_size = "" if self.chunk_size is None else f", chunk_size={self.chunk_size}" + + return f"QuantumProgramSamplexItem({circuit}{samplex}{samplex_args}{shape}{chunk_size})" + + +class QuantumProgram: + """A quantum runtime executable. + + A quantum program consists of a list of ordered elements, each of which contains a single + circuit and an array of associated parameter values. Executing a quantum program will + sample the outcome of each circuit for the specified number of ``shots`` for each set of + circuit arguments provided. + + Args: + shots: The number of shots for each circuit execution. + items: Items that comprise the program. + noise_maps: Noise maps to use with samplex items. + """ + + def __init__( + self, + shots: int, + items: Iterable[QuantumProgramItem] | None = None, + noise_maps: dict[str, PauliLindbladMap] | None = None, + ): + self.shots = shots + self.items: list[QuantumProgramItem] = list(items or []) + self.noise_maps = noise_maps or {} + + def append( + self, + circuit: QuantumCircuit, + *, + samplex: Samplex | None = None, + circuit_arguments: np.ndarray | None = None, + samplex_arguments: dict[str, np.ndarray] | None = None, + shape: tuple[int, ...] | None = None, + chunk_size: int | None = None, + ) -> None: + """Append a new :class:`QuantumProgramItem` to this program. + + Args: + circuit: The circuit of this item. + samplex: An (optional) samplex to draw random parameters for the circuit. + circuit_arguments: Arguments for the parameters of the circuit. A real array where the + last dimension matches the number of parameters in the circuit. Circuit execution + will be broadcasted over the leading axes. + samplex_arguments: A map from argument names to argument values for the samplex. If this + value is provided, a samplex must be present, and ``circuit_arguments`` must not be + supplied. + shape: A shape tuple to extend the implicit shape defined by ``samplex_arguments``. + Non-trivial axes introduced by this extension enumerate randomizations. If this + value is provided, a samplex must be present, and ``circuit_arguments`` must not be + supplied. + chunk_size: The maximum number of bound circuits in each shot loop execution, or + ``None`` to use a server-side heuristic to optimize speed. When not executing + in a session, the server-side heuristic is always used and this value is ignored. + """ + if samplex is None: + if samplex_arguments is not None: + raise ValueError("'samplex_arguments' cannot be supplied when no samplex is given.") + if shape is not None: + raise ValueError("'shape' cannot be supplied when no samplex is given.") + self.items.append( + CircuitItem( + circuit, + circuit_arguments=circuit_arguments, + chunk_size=chunk_size, + ) + ) + else: + if circuit_arguments is not None: + raise ValueError("'circuit_arguments' cannot be supplied when a samplex is given.") + # add the noise maps first so that samplex_arguments has the ability to overwrite them + arguments = {"pauli_lindblad_maps": self.noise_maps} + arguments.update(samplex_arguments or {}) + self.items.append( + SamplexItem( + circuit, + samplex, + samplex_arguments=arguments, + shape=shape, + chunk_size=chunk_size, + ) + ) + + def validate(self, backend: IBMBackend) -> None: + """Validate this quantum program against the given backend.""" + + def __repr__(self) -> str: + if not self.items: + return f"QuantumProgram(shots={self.shots})" + return "\n".join( + [f"QuantumProgram(shots={self.shots}, items=["] + + [f" {repr(item)}," for item in self.items] + + ["])"] + ) diff --git a/qiskit_ibm_runtime/quantum_program/quantum_program_decoders.py b/qiskit_ibm_runtime/quantum_program/quantum_program_decoders.py new file mode 100644 index 0000000000..6558b375bf --- /dev/null +++ b/qiskit_ibm_runtime/quantum_program/quantum_program_decoders.py @@ -0,0 +1,50 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Decoders for quantum programs.""" + +from __future__ import annotations + +import logging + +from ibm_quantum_schemas.models.executor.version_0_1.models import ( + QuantumProgramResultModel, +) + +# pylint: disable=unused-import,cyclic-import +from ..utils.result_decoder import ResultDecoder +from .converters import quantum_program_result_from_0_1 + +logger = logging.getLogger(__name__) + +AVAILABLE_DECODERS = {"v0.1": quantum_program_result_from_0_1} + + +class QuantumProgramResultDecoder(ResultDecoder): + """Decoder for quantum program results.""" + + @classmethod + def decode(cls, raw_result: str): # type: ignore[no-untyped-def] + """Decode raw json to result type.""" + decoded: dict[str, str] = super().decode(raw_result) + + try: + schema_version = decoded["schema_version"] + except KeyError: + raise ValueError("Missing schema version.") + + try: + decoder = AVAILABLE_DECODERS[schema_version] + except KeyError: + raise ValueError(f"No decoder found for schema version {schema_version}.") + + return decoder(QuantumProgramResultModel(**decoded)) diff --git a/qiskit_ibm_runtime/quantum_program/quantum_program_result.py b/qiskit_ibm_runtime/quantum_program/quantum_program_result.py new file mode 100644 index 0000000000..45262f8506 --- /dev/null +++ b/qiskit_ibm_runtime/quantum_program/quantum_program_result.py @@ -0,0 +1,88 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""QuantumProgramResult""" + +from __future__ import annotations + +from collections.abc import Iterator +from dataclasses import dataclass, field +import datetime + +import numpy as np + + +@dataclass +class ChunkPart: + """A description of the contents of a single part of an execution chunk.""" + + idx_item: int + """The index of an item in a quantum program.""" + + size: int + """The number of elements from the quantum program item that were executed. + + For example, if a quantum program item has shape ``(10, 5)``, then it has a total of ``50`` + elements, so that if this ``size`` is ``10``, it constitutes 20% of the total work for the item. + """ + + +@dataclass +class ChunkSpan: + """Timing information about a single chunk of execution. + + .. note:: + + This span may include some amount of non-circuit time. + """ + + start: datetime.datetime + """The start time of the execution chunk in UTC.""" + + stop: datetime.datetime + """The stop time of the execution chunk in UTC.""" + + parts: list[ChunkPart] + """A description of which parts of a quantum program are contained in this chunk.""" + + +@dataclass +class Metadata: + """Metadata about the execution of a quantum program run through the runtime executor.""" + + chunk_timing: list[ChunkSpan] = field(default_factory=list) + """Timing information about all executed chunks of a quantum program.""" + + +class QuantumProgramResult: + """A container to store results from executing a :class:`QuantumProgram`. + + Args: + data: A list of dictionaries with array-valued data. + metadata: A dictionary of metadata. + """ + + def __init__(self, data: list[dict[str, np.ndarray]], metadata: Metadata | None = None): + self._data = data + self.metadata = metadata or Metadata() + + def __iter__(self) -> Iterator[dict[str, np.ndarray]]: + yield from self._data + + def __getitem__(self, idx: int) -> dict[str, np.ndarray]: + return self._data[idx] + + def __len__(self) -> int: + return len(self._data) + + def __repr__(self) -> str: + return f"{type(self).__name__}(<{len(self)} results>)" diff --git a/qiskit_ibm_runtime/utils/executor_result_decoder.py b/qiskit_ibm_runtime/utils/executor_result_decoder.py new file mode 100644 index 0000000000..7e30f7ce23 --- /dev/null +++ b/qiskit_ibm_runtime/utils/executor_result_decoder.py @@ -0,0 +1,29 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2024. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""NoiseLearner result decoder.""" + +from .result_decoder import ResultDecoder + + +class ExecutorResultDecoder(ResultDecoder): + """Class used to decode noise learner results""" + + @classmethod + def decode(cls, raw_result: str): # type: ignore # pylint: disable=arguments-differ + """Convert the result to QuantumProgramResult.""" + # pylint: disable=import-outside-toplevel + from qiskit_ibm_runtime.quantum_program.quantum_program_decoders import ( + QuantumProgramResultDecoder, + ) + + return QuantumProgramResultDecoder().decode(raw_result) diff --git a/qiskit_ibm_runtime/utils/noise_learner_result_decoder.py b/qiskit_ibm_runtime/utils/noise_learner_result_decoder.py index ca3be68c18..1ee48276f3 100644 --- a/qiskit_ibm_runtime/utils/noise_learner_result_decoder.py +++ b/qiskit_ibm_runtime/utils/noise_learner_result_decoder.py @@ -13,7 +13,7 @@ """NoiseLearner result decoder.""" -from .noise_learner_result import PauliLindbladError, LayerError, NoiseLearnerResult +from .noise_learner_result import LayerError, NoiseLearnerResult, PauliLindbladError from .result_decoder import ResultDecoder @@ -23,6 +23,15 @@ class NoiseLearnerResultDecoder(ResultDecoder): @classmethod def decode(cls, raw_result: str) -> NoiseLearnerResult: """Convert the result to NoiseLearnerResult.""" + if "schema_version" in raw_result: + # pylint: disable=import-outside-toplevel + from qiskit_ibm_runtime.noise_learner_v3.noise_learner_v3_decoders import ( + NoiseLearnerV3ResultDecoder, + ) + + return NoiseLearnerV3ResultDecoder().decode(raw_result) + + # Decode for legacy noise learner decoded: dict = super().decode(raw_result) data = [] diff --git a/qiskit_ibm_runtime/visualization/draw_zne.py b/qiskit_ibm_runtime/visualization/draw_zne.py index a29beeabff..ddfeceb475 100644 --- a/qiskit_ibm_runtime/visualization/draw_zne.py +++ b/qiskit_ibm_runtime/visualization/draw_zne.py @@ -18,7 +18,6 @@ from typing import TYPE_CHECKING from collections.abc import Sequence import numpy as np - from .utils import plotly_module from ..utils.estimator_pub_result import EstimatorPubResult diff --git a/test/smoke/test_primitives.py b/test/smoke/test_primitives.py index ce5147e295..6a1c4b8f26 100644 --- a/test/smoke/test_primitives.py +++ b/test/smoke/test_primitives.py @@ -12,15 +12,23 @@ """Testing simple primitive jobs for smoke tests.""" -from qiskit import QuantumCircuit -from qiskit.primitives import PrimitiveResult -from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager +from qiskit import QuantumCircuit, QuantumRegister from qiskit.circuit.library import real_amplitudes +from qiskit.primitives import PrimitiveResult from qiskit.quantum_info import SparsePauliOp -from qiskit_ibm_runtime import SamplerV2, EstimatorV2 +from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager +from samplomatic.transpiler import generate_boxing_pass_manager + +from qiskit_ibm_runtime import EstimatorV2, SamplerV2 from qiskit_ibm_runtime.noise_learner import NoiseLearner -from qiskit_ibm_runtime.utils.noise_learner_result import NoiseLearnerResult +from qiskit_ibm_runtime.noise_learner_v3 import ( + NoiseLearnerV3, + NoiseLearnerV3Result, + NoiseLearnerV3Results, +) from qiskit_ibm_runtime.options import NoiseLearnerOptions +from qiskit_ibm_runtime.utils.noise_learner_result import NoiseLearnerResult + from ..ibm_test_case import IBMIntegrationTestCase @@ -30,15 +38,17 @@ class TestSmokePrimitives(IBMIntegrationTestCase): def setUp(self): super().setUp() self._backend = self.service.backend(self.dependencies.qpu) - pm = generate_preset_pass_manager(optimization_level=1, target=self._backend.target) + self.pm = generate_preset_pass_manager(optimization_level=1, target=self._backend.target) + self.boxing_pm = generate_boxing_pass_manager() + # bell circuit bell = QuantumCircuit(2, name="Bell") bell.h(0) bell.cx(0, 1) bell.measure_all() - self._isa_bell = pm.run(bell) + self._isa_bell = self.pm.run(bell) # estimator circuit - self._psi1 = pm.run(real_amplitudes(num_qubits=2, reps=2)) + self._psi1 = self.pm.run(real_amplitudes(num_qubits=2, reps=2)) # noise learner circuit c1 = QuantumCircuit(2) c1.ecr(0, 1) @@ -75,3 +85,22 @@ def test_noise_learner(self): learner = NoiseLearner(mode=self._backend, options=options) job = learner.run(self._circuits) self.assertIsInstance(job.result(), NoiseLearnerResult) + + def test_noise_learner_v3(self): + """Test noise learner V3 job.""" + circuit = QuantumCircuit(QuantumRegister(3, "q")) + circuit.cz(0, 1) + circuit.cz(1, 2) + circuit.measure_all() + + isa_circuit = self.pm.run(circuit) + boxed_circuit = self.boxing_pm.run(isa_circuit) + + learner = NoiseLearnerV3(mode=self._backend) + instructions = [instr for instr in boxed_circuit if instr.operation.name == "box"] + job = learner.run(instructions) + results = job.result() + + self.assertIsInstance(results, NoiseLearnerV3Results) + self.assertEqual(len(results), 3) + self.assertTrue(all(isinstance(res, NoiseLearnerV3Result) for res in results)) diff --git a/test/unit/noise_learner_v3/__init__.py b/test/unit/noise_learner_v3/__init__.py new file mode 100644 index 0000000000..60e93d8a95 --- /dev/null +++ b/test/unit/noise_learner_v3/__init__.py @@ -0,0 +1,11 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. diff --git a/test/unit/noise_learner_v3/test_converters.py b/test/unit/noise_learner_v3/test_converters.py new file mode 100644 index 0000000000..2e4224836d --- /dev/null +++ b/test/unit/noise_learner_v3/test_converters.py @@ -0,0 +1,138 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Tests the converters for the noise learner v3 model.""" + +import numpy as np +from pydantic import ValidationError +from qiskit.circuit import QuantumCircuit +from qiskit.quantum_info import QubitSparsePauliList + +from qiskit_ibm_runtime.noise_learner_v3.converters.version_0_1 import ( + noise_learner_v3_inputs_from_0_1, + noise_learner_v3_inputs_to_0_1, + noise_learner_v3_result_from_0_1, + noise_learner_v3_result_to_0_1, +) +from qiskit_ibm_runtime.noise_learner_v3.noise_learner_v3_result import ( # type: ignore[attr-defined] + NoiseLearnerV3Result, + NoiseLearnerV3Results, +) +from qiskit_ibm_runtime.options import NoiseLearnerV3Options + +from ...ibm_test_case import IBMTestCase + + +class TestConverters(IBMTestCase): + """Tests the converters for the noise learner v3 model.""" + + def test_converting_inputs(self): + """Tests converting inputs.""" + circuit = QuantumCircuit(3) + with circuit.box(): + circuit.noop(1) + circuit.cx(0, 2) + with circuit.box(): + circuit.noop(2) + circuit.cx(0, 1) + + instructions = [circuit[0], circuit[1]] + + options = NoiseLearnerV3Options() + options.layer_pair_depths = [2, 4, 10] + options.post_selection.enable = True + options.post_selection.strategy = "edge" + options.post_selection.x_pulse_type = "xslow" + + encoded = noise_learner_v3_inputs_to_0_1(instructions, options) + decoded = noise_learner_v3_inputs_from_0_1(encoded) + + assert decoded == (instructions, options) + + def test_converting_results(self): + """Tests converting results.""" + generators = [ + QubitSparsePauliList.from_list(["IX", "XX"]), + QubitSparsePauliList.from_list(["XI"]), + ] + rates = [0.1, 0.2] + rates_std = [0.01, 0.02] + + metadatum0 = { + "learning_protocol": "trex", + "post_selection": {"fraction_kept": 1}, + } + result0 = NoiseLearnerV3Result.from_generators(generators, rates, rates_std, metadatum0) + + metadatum1 = { + "learning_protocol": "lindblad", + "post_selection": {"fraction_kept": {0: 1, 4: 1}}, + } + result1 = NoiseLearnerV3Result.from_generators(generators, rates, metadata=metadatum1) + results = NoiseLearnerV3Results([result0, result1]) + + encoded = noise_learner_v3_result_to_0_1(results).model_dump() + decoded = noise_learner_v3_result_from_0_1(encoded) + for datum_in, datum_out in zip(results.data, decoded.data): + assert datum_in._generators == datum_out._generators + assert np.allclose(datum_in._rates, datum_out._rates) + assert np.allclose(datum_in._rates_std, datum_out._rates_std) + assert datum_in.metadata == datum_out.metadata + + def test_converting_invalid_results(self): + """Test that converting results raises when results are invalid.""" + + generators = [ + QubitSparsePauliList.from_list(["IX", "XX"]), + QubitSparsePauliList.from_list(["XI"]), + ] + rates = [0.1, 0.2] + + metadata = { + "input_options": { + "shots_per_randomization": 3, + "num_randomizations": 8, + "layer_pair_depths": [0, 2, 6], + "post_selection": { + "enable": True, + "strategy": "edge", + "x_pulse_type": "xslow", + }, + } + } + + for metadatum in [ + {"learning_protocol": "trex", "post_selection": {"strategy": "edge"}}, + {"learning_protocol": "trex", "post_selection": {"fraction_kept": 1.2}}, + {"learning_protocol": "trex", "post_selection": {"fraction_kept": -0.3}}, + ]: + result = NoiseLearnerV3Result.from_generators(generators, rates, metadata=metadatum) + results = NoiseLearnerV3Results([result], metadata) + with self.assertRaisesRegex( + ValidationError, + "1 validation error for NoiseLearnerV3ResultModel", + ): + noise_learner_v3_result_to_0_1(results).model_dump() + + for metadatum in [ + { + "learning_protocol": "trex", + "post_selection": {"fraction_kept": {0: 0.1, 2: 0.3}}, + }, + {"learning_protocol": "lindblad", "post_selection": {"fraction_kept": 0.3}}, + ]: + result = NoiseLearnerV3Result.from_generators(generators, rates, metadata=metadatum) + results = NoiseLearnerV3Results([result], metadata) + with self.assertRaisesRegex( + ValidationError, "1 validation error for NoiseLearnerV3ResultModel" + ): + noise_learner_v3_result_to_0_1(results).model_dump() diff --git a/test/unit/noise_learner_v3/test_find_learning_protocol.py b/test/unit/noise_learner_v3/test_find_learning_protocol.py new file mode 100644 index 0000000000..45e129a2ad --- /dev/null +++ b/test/unit/noise_learner_v3/test_find_learning_protocol.py @@ -0,0 +1,84 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Tests the `find_learning_protocol` function.""" + +from qiskit.circuit import QuantumCircuit +from samplomatic import Twirl + +from qiskit_ibm_runtime.noise_learner_v3.find_learning_protocol import ( + find_learning_protocol, +) + +from ...ibm_test_case import IBMTestCase + + +class TestFindLearningProtocol(IBMTestCase): + """Tests the `find_learning_protocol` function.""" + + def test_gate_instructions(self): + """Test gate instructions.""" + circuit = QuantumCircuit(4) + with circuit.box([Twirl()]): + circuit.x(0) + circuit.x(0) + circuit.cx(0, 1) + circuit.cx(2, 3) + with circuit.box([Twirl()]): + circuit.noop(2) + circuit.cx(0, 1) + with circuit.box([Twirl()]): + circuit.cx(0, 1) + circuit.cx(1, 2) + with circuit.box([Twirl()]): + circuit.rzz(0.01, 0, 1) + + protocols = [find_learning_protocol(instr) for instr in circuit] + self.assertEqual(protocols, ["pauli_lindblad"] * 2 + [None] * 2) + + def test_measure_instructions(self): + """Test measure instructions.""" + circuit = QuantumCircuit(4, 4) + with circuit.box([Twirl()]): + circuit.measure(range(4), range(4)) + with circuit.box([Twirl()]): + circuit.measure(range(2), range(2)) + with circuit.box([Twirl()]): + circuit.noop(range(4)) + circuit.measure(range(2), range(2)) + with circuit.box([Twirl()]): + circuit.measure(range(2), range(2)) + circuit.measure(range(2), range(2)) + + protocols = [find_learning_protocol(instr) for instr in circuit] + self.assertEqual(protocols, ["trex"] * 2 + [None] * 2) + + def test_empty_instructions(self): + """Test empty instructions.""" + circuit = QuantumCircuit(4) + with circuit.box([Twirl()]): + circuit.x(0) + with circuit.box([Twirl()]): + circuit.noop(2) + + protocols = [find_learning_protocol(instr) for instr in circuit] + self.assertEqual(protocols, ["pauli_lindblad"] * 2) + + def test_mixed_instructions(self): + """Test instructions with gates and measurements.""" + circuit = QuantumCircuit(4, 2) + with circuit.box([Twirl()]): + circuit.cx(0, 1) + circuit.measure([2, 3], [0, 1]) + + protocols = [find_learning_protocol(instr) for instr in circuit] + self.assertEqual(protocols, [None]) diff --git a/test/unit/noise_learner_v3/test_result.py b/test/unit/noise_learner_v3/test_result.py new file mode 100644 index 0000000000..c8205b68c1 --- /dev/null +++ b/test/unit/noise_learner_v3/test_result.py @@ -0,0 +1,236 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Tests the classes `NoiseLearnerV3Result` and `NoiseLearnerV3Results`.""" + +import numpy as np + +from samplomatic import InjectNoise, Twirl + +from qiskit import QuantumCircuit +from qiskit.quantum_info import QubitSparsePauliList, PauliLindbladMap + +from qiskit_ibm_runtime.noise_learner_v3.noise_learner_v3_result import ( # type: ignore[attr-defined] + NoiseLearnerV3Result, + NoiseLearnerV3Results, +) + +from ...ibm_test_case import IBMTestCase + + +class TestNoiseLearnerV3Result(IBMTestCase): + """Tests the ``NoiseLearnerV3Result`` class.""" + + def test_from_generators_valid_input(self): + """Test ``NoiseLearnerV3Result.from_generators``.""" + generators = [ + QubitSparsePauliList.from_label(pauli1 + pauli0) + for pauli1 in "IXYZ" + for pauli0 in "IXYZ" + ][1:] + rates = np.arange(0, 0.3, 0.02) + rates_std = np.arange(0, 0.15, 0.01) + metadata = {"learning_protocol": "lindblad"} + result = NoiseLearnerV3Result.from_generators(generators, rates, rates_std, metadata) + self.assertEqual(generators, result._generators) + self.assertTrue(np.array_equal(np.array(rates), result._rates)) + self.assertTrue(np.array_equal(rates_std, result._rates_std)) + self.assertEqual(metadata, result.metadata) + self.assertEqual(len(result), 15) + + def test_from_generators_different_lengths(self): + """Test that ``NoiseLearnerV3Result.from_generators`` raises if the specified generators + and rates have different lengths""" + generators = [ + QubitSparsePauliList.from_label(pauli1 + pauli0) + for pauli1 in "IXYZ" + for pauli0 in "IXYZ" + ][1:] + rates = np.arange(0, 0.2, 0.02) + with self.assertRaisesRegex(ValueError, "must be of the same length"): + NoiseLearnerV3Result.from_generators(generators, rates) + + def test_from_generators_different_num_qubits(self): + """Test that ``NoiseLearnerV3Result.from_generators`` raises if the specified generators + have different numbers of qubits.""" + generators = [ + QubitSparsePauliList.from_label(pauli1 + pauli0) + for pauli1 in "IXYZ" + for pauli0 in "IXYZ" + ][1:] + generators[4] = QubitSparsePauliList.from_label("XII") + rates = np.arange(0, 0.3, 0.02) + with self.assertRaisesRegex(ValueError, "number of qubits"): + NoiseLearnerV3Result.from_generators(generators, rates) + + def test_to_pauli_lindblad_map(self): + """Test ``NoiseLearnerV3Result.to_pauli_lindblad_map``.""" + generators = [ + QubitSparsePauliList.from_list(list_) + for list_ in [ + ["IX", "ZX"], + ["IY", "ZY"], + ["IZ"], + ["XI", "XZ"], + ["XX", "YY"], + ["XY", "YX"], + ["YI", "YZ"], + ["ZI"], + ["ZZ"], + ] + ] + rates = np.arange(0, 0.18, 0.02) + result = NoiseLearnerV3Result.from_generators(generators, rates) + flatenned_generators = QubitSparsePauliList.from_list( + [pauli1 + pauli0 for pauli1 in "IXYZ" for pauli0 in "IXYZ"][1:] + ) + flatenned_rates = [ + 0, + 0.02, + 0.04, + 0.06, + 0.08, + 0.1, + 0.06, + 0.12, + 0.1, + 0.08, + 0.12, + 0.14, + 0, + 0.02, + 0.16, + ] + self.assertEqual( + result.to_pauli_lindblad_map().simplify(), + PauliLindbladMap.from_components(flatenned_rates, flatenned_generators).simplify(), + ) + + +class TestNoiseLearnerV3Results(IBMTestCase): + """Tests the ``NoiseLearnerV3Results`` class.""" + + def setUp(self): + super().setUp() + self.generators = [ + QubitSparsePauliList.from_label(pauli1 + pauli0) + for pauli1 in "IXYZ" + for pauli0 in "IXYZ" + ][1:] + self.rates = [np.linspace(0, i * 0.1, 15) for i in range(3)] + self.results = [ + NoiseLearnerV3Result.from_generators(self.generators, rates) for rates in self.rates + ] + self.pauli_lindblad_maps = [result.to_pauli_lindblad_map() for result in self.results] + self.inject_noise_annotations = [InjectNoise(ref) for ref in ["hi", "bye"]] + + def test_properties_of_iterable(self): + """Test elementary methods of ``NoiseLearnerV3Results``: ``__init__``, ``__len__``, + ``__get_item__``.""" + results = NoiseLearnerV3Results(self.results, metadata := {"this is": "metadata"}) + self.assertEqual(results.data, self.results, metadata) + self.assertEqual(results[1], self.results[1]) + self.assertEqual(len(results), 3) + + def test_to_dict_valid_input_require_refs_true(self): + """Test ``NoiseLearnerV3Results.to_dict`` when ``require_refs`` is ``True``.""" + circuit = QuantumCircuit(2) + with circuit.box(annotations=[Twirl(), self.inject_noise_annotations[0]]): + circuit.cx(0, 1) + with circuit.box(annotations=[self.inject_noise_annotations[1]]): + circuit.cx(0, 1) + + returned_dict = NoiseLearnerV3Results(self.results[:2]).to_dict(circuit.data, True) + self.assertDictEqual( + { + annotation.ref: pauli_lindblad_map + for annotation, pauli_lindblad_map in zip( + self.inject_noise_annotations[:2], self.pauli_lindblad_maps[:2] + ) + }, + returned_dict, + ) + + def test_to_dict_valid_input_require_refs_false(self): + """Test ``NoiseLearnerV3Results.to_dict`` when ``require_refs`` is ``True``.""" + circuit = QuantumCircuit(2) + with circuit.box(annotations=[Twirl(), self.inject_noise_annotations[0]]): + circuit.cx(0, 1) + with circuit.box(annotations=[Twirl()]): + circuit.cx(0, 1) + with circuit.box(annotations=[self.inject_noise_annotations[1]]): + circuit.cx(0, 1) + + returned_dict = NoiseLearnerV3Results(self.results).to_dict(circuit.data, False) + self.assertDictEqual( + { + annotation.ref: pauli_lindblad_map + for annotation, pauli_lindblad_map in zip( + self.inject_noise_annotations, + [self.pauli_lindblad_maps[0], self.pauli_lindblad_maps[2]], + ) + }, + returned_dict, + ) + + def test_to_dict_wrong_num_of_instructions(self): + """Test that ``NoiseLearnerV3Results.to_dict`` raises if the number of instructions + does not match the number of the results.""" + circuit = QuantumCircuit(2) + with circuit.box(annotations=[Twirl(), self.inject_noise_annotations[0]]): + circuit.cx(0, 1) + with circuit.box(annotations=[self.inject_noise_annotations[1]]): + circuit.cx(0, 1) + + with self.assertRaisesRegex(ValueError, "Expected 3 instructions but found 2"): + NoiseLearnerV3Results(self.results).to_dict(circuit.data, True) + + def test_to_dict_invalid_for_require_refs_true(self): + """Test that ``NoiseLearnerV3Results.to_dict`` raises if an instruction does not contain + the ``InjectNoise`` annotation, when ``requires_ref`` is ``True``.""" + circuit = QuantumCircuit(2) + with circuit.box(annotations=[Twirl(), self.inject_noise_annotations[0]]): + circuit.cx(0, 1) + with circuit.box(annotations=[Twirl()]): + circuit.cx(0, 1) + with circuit.box(annotations=[self.inject_noise_annotations[1]]): + circuit.cx(0, 1) + + with self.assertRaisesRegex(ValueError, "without an inject noise"): + NoiseLearnerV3Results(self.results).to_dict(circuit.data, True) + + def test_to_dict_unboxed_instruction(self): + """Test that ``NoiseLearnerV3Results.to_dict`` raises if there is an instruction not in a + box.""" + circuit = QuantumCircuit(2) + with circuit.box(annotations=[Twirl(), self.inject_noise_annotations[0]]): + circuit.cx(0, 1) + circuit.cx(0, 1) + with circuit.box(annotations=[self.inject_noise_annotations[1]]): + circuit.cx(0, 1) + + with self.assertRaisesRegex(ValueError, "contain a box"): + NoiseLearnerV3Results(self.results).to_dict(circuit.data) + + def test_to_dict_ref_used_twice(self): + """Test that ``NoiseLearnerV3Results.to_dict`` raises if an annotation reference is + repeated.""" + circuit = QuantumCircuit(2) + with circuit.box(annotations=[Twirl(), self.inject_noise_annotations[0]]): + circuit.cx(0, 1) + with circuit.box(annotations=[Twirl(), self.inject_noise_annotations[0]]): + circuit.cx(0, 1) + with circuit.box(annotations=[self.inject_noise_annotations[1]]): + circuit.cx(0, 1) + + with self.assertRaisesRegex(ValueError, "multiple instructions with the same ``ref``"): + NoiseLearnerV3Results(self.results).to_dict(circuit.data) diff --git a/test/unit/noise_learner_v3/test_validation.py b/test/unit/noise_learner_v3/test_validation.py new file mode 100644 index 0000000000..fcc32b13e3 --- /dev/null +++ b/test/unit/noise_learner_v3/test_validation.py @@ -0,0 +1,122 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2025. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Tests the noise learner v3 validation.""" + +from qiskit import QuantumCircuit + +from samplomatic import Twirl + +from qiskit_ibm_runtime.noise_learner_v3.validation import validate_options, validate_instruction +from qiskit_ibm_runtime.options import NoiseLearnerV3Options +from qiskit_ibm_runtime.fake_provider.backends import FakeAlgiers, FakeFractionalBackend +from qiskit_ibm_runtime.exceptions import IBMInputValueError + +from ...ibm_test_case import IBMTestCase + + +class TestValidation(IBMTestCase): + """Tests the noise learner v3 validation.""" + + def test_validate_options(self): + """Test the validation of NLV3 options.""" + configuration = FakeFractionalBackend().configuration() + + options = NoiseLearnerV3Options() + options.post_selection = {"enable": True, "x_pulse_type": "rx"} + validate_options(options=options, configuration=configuration) + + options.post_selection = {"enable": False, "x_pulse_type": "xslow"} + validate_options(options=options, configuration=configuration) + + options.post_selection = {"enable": True, "x_pulse_type": "xslow"} + with self.assertRaisesRegex(ValueError, "xslow"): + validate_options(options=options, configuration=configuration) + + def test_validate_valid_instructions(self): + """Test instruction validation for valid instructions.""" + target = FakeAlgiers().target + circuit = QuantumCircuit(target.num_qubits) + with circuit.box(annotations=[Twirl()]): + circuit.cx(0, 1) + with circuit.box(annotations=[Twirl()]): + circuit.measure_all() + + validate_instruction(circuit.data[0], target) + validate_instruction(circuit.data[1], target) + + def test_validate_instruction_bad_box(self): + """Test that instruction validation raises when the box is badly annotated.""" + target = FakeAlgiers().target + circuit = QuantumCircuit(target.num_qubits) + with circuit.box(annotations=[]): + circuit.noop(1) + + with self.assertRaisesRegex( + IBMInputValueError, "Found a box without a ``Twirl`` annotation" + ): + validate_instruction(circuit.data[0], target) + + def test_validate_instruction_no_box(self): + """Test that instruction validation raises when there is no box.""" + target = FakeAlgiers().target + circuit = QuantumCircuit(target.num_qubits) + circuit.cx(0, 1) + + with self.assertRaisesRegex(IBMInputValueError, "Expected a 'box' but found 'cx'"): + validate_instruction(circuit.data[0], target) + + def test_validate_instruction_isa_basis_gate(self): + """Test that instruction validation raises for an operation that's not a basis gate.""" + target = FakeAlgiers().target + circuit = QuantumCircuit(target.num_qubits) + with circuit.box(annotations=[Twirl()]): + circuit.cz(0, 1) + + with self.assertRaisesRegex(IBMInputValueError, "instruction cz"): + validate_instruction(circuit.data[0], target) + + def test_validate_instruction_isa_connectivity(self): + """Test that instruction validation raises for 2Q gates that violate the coupling map.""" + target = FakeAlgiers().target + block = QuantumCircuit(2) + block.cx(0, 1) + circuit = QuantumCircuit(target.num_qubits) + circuit.box(block, annotations=[Twirl()], qubits=[0, 13], clbits=[]) + + with self.assertRaisesRegex(IBMInputValueError, r"instruction cx on qubits \(0, 13\)"): + validate_instruction(circuit.data[0], target) + + def test_validate_instruction_cannot_be_learned(self): + """Test that instruction validation raises when the instruction doesn't match any + learning protocol.""" + target = FakeAlgiers().target + circuit = QuantumCircuit(target.num_qubits) + with circuit.box(annotations=[Twirl()]): + circuit.cx(0, 1) + circuit.measure_all() + + with self.assertRaisesRegex(IBMInputValueError, "cannot be learned"): + validate_instruction(circuit.data[0], target) + + def test_validate_instruction_unphysical(self): + """Test that instruction validation raises when the qubits don't belong to the expected + register.""" + target = FakeAlgiers().target + circuit = QuantumCircuit(2) + with circuit.box(annotations=[Twirl()]): + circuit.cx(0, 1) + + with self.assertRaisesRegex( + IBMInputValueError, "Every qubit must be part of QuantumRegister" + ): + validate_instruction(circuit.data[0], target) diff --git a/test/unit/quantum_program/__init__.py b/test/unit/quantum_program/__init__.py new file mode 100644 index 0000000000..3a47f12735 --- /dev/null +++ b/test/unit/quantum_program/__init__.py @@ -0,0 +1,11 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2026. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. diff --git a/test/unit/quantum_program/test_circuit_item.py b/test/unit/quantum_program/test_circuit_item.py new file mode 100644 index 0000000000..7283e747c5 --- /dev/null +++ b/test/unit/quantum_program/test_circuit_item.py @@ -0,0 +1,75 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2026. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Tests the ``CircuitItem`` class.""" + +import numpy as np + +from qiskit.circuit import QuantumCircuit, Parameter + +from qiskit_ibm_runtime.quantum_program.quantum_program import CircuitItem + + +from ...ibm_test_case import IBMTestCase + + +class TestCircuitItem(IBMTestCase): + """Tests the ``CircuitItem`` class.""" + + def test_circuit_item(self): + """Test ``CircuitItem`` for a valid input.""" + circuit = QuantumCircuit(1) + circuit.rx(Parameter("p"), 0) + + circuit_arguments = np.array([[3], [4], [5]]) + expected_shape = (3,) + chunk_size = 6 + + circuit_item = CircuitItem( + circuit, circuit_arguments=circuit_arguments, chunk_size=chunk_size + ) + self.assertEqual(circuit_item.circuit, circuit) + self.assertTrue(np.array_equal(circuit_item.circuit_arguments, circuit_arguments)) + self.assertEqual(circuit_item.chunk_size, chunk_size) + self.assertEqual(circuit_item.shape, expected_shape) + + def test_circuit_item_no_params(self): + """Test ``CircuitItem`` when there are no parameters.""" + circuit = QuantumCircuit(1) + + expected_circuit_arguments = np.array([]) + expected_shape = () + + circuit_item = CircuitItem(circuit) + self.assertEqual(circuit_item.circuit, circuit) + self.assertTrue(np.array_equal(circuit_item.circuit_arguments, expected_circuit_arguments)) + self.assertEqual(circuit_item.chunk_size, None) + self.assertEqual(circuit_item.shape, expected_shape) + + def test_circuit_item_num_params_doesnt_match_circuit_arguments(self): + """Test that ``CircuitItem`` raises an error if the number of circuit parameters + doesn't match the shape of the circuit arguments.""" + circuit = QuantumCircuit(1) + circuit.rx(Parameter("p"), 0) + + circuit_arguments = np.array([[3, 10], [4, 11], [5, 12]]) + with self.assertRaisesRegex(ValueError, "match the number of parameters"): + CircuitItem(circuit, circuit_arguments=circuit_arguments) + + def test_circuit_item_no_circuit_arguments_for_parametric_circuit(self): + """Test that ``CircuitItem`` raises an error if the circuit has parameters + but the ``circuit_arguments`` parameter is unset.""" + circuit = QuantumCircuit(1) + circuit.rx(Parameter("p"), 0) + + with self.assertRaisesRegex(ValueError, "no 'circuit_arguments'"): + CircuitItem(circuit) diff --git a/test/unit/quantum_program/test_converters.py b/test/unit/quantum_program/test_converters.py new file mode 100644 index 0000000000..38bacfa9f0 --- /dev/null +++ b/test/unit/quantum_program/test_converters.py @@ -0,0 +1,191 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2026. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Tests the quantum program converters.""" + +from datetime import datetime +import numpy as np + +from samplomatic import Twirl, InjectNoise, build + +from ibm_quantum_schemas.models.executor.version_0_1.models import ( + QuantumProgramResultModel, + QuantumProgramResultItemModel, + ChunkPart, + ChunkSpan, + MetadataModel, +) +from ibm_quantum_schemas.models.tensor_model import TensorModel + +from qiskit.circuit import QuantumCircuit, Parameter +from qiskit.quantum_info import PauliLindbladMap + +from qiskit_ibm_runtime.quantum_program import QuantumProgram +from qiskit_ibm_runtime.quantum_program.converters import ( + quantum_program_to_0_1, + quantum_program_result_from_0_1, +) +from qiskit_ibm_runtime.options.executor_options import ExecutorOptions, ExecutionOptions + +from ...ibm_test_case import IBMTestCase + + +class TestQuantumProgramConverters(IBMTestCase): + """Tests the quantum program converters.""" + + def test_quantum_program_to_0_1(self): + """Test the function ``quantum_program_to_0_1``""" + shots = 100 + + noise_models = [ + PauliLindbladMap.from_list([("IX", 0.04), ("XX", 0.05)]), + PauliLindbladMap.from_list([("XI", 0.02), ("IZ", 0.035)]), + ] + + quantum_program = QuantumProgram( + shots=shots, + noise_maps={f"pl{i}": noise_model for i, noise_model in enumerate(noise_models)}, + ) + + circuit1 = QuantumCircuit(1) + circuit1.rx(Parameter("p"), 0) + + circuit_arguments = np.array([[3], [4], [5]]) + quantum_program.append(circuit1, circuit_arguments=circuit_arguments, chunk_size=6) + + circuit2 = QuantumCircuit(2) + with circuit2.box(annotations=[Twirl(), InjectNoise(ref="pl0")]): + circuit2.rx(Parameter("p"), 0) + circuit2.cx(0, 1) + with circuit2.box(annotations=[Twirl(), InjectNoise(ref="pl1")]): + circuit2.measure_all() + + template_circuit, samplex = build(circuit2) + parameter_values = np.array([[[1], [2]], [[3], [4]], [[5], [6]]]) + quantum_program.append( + template_circuit, + samplex=samplex, + samplex_arguments={"parameter_values": parameter_values}, + shape=(4, 3, 2), + chunk_size=7, + ) + + options = ExecutorOptions(execution=ExecutionOptions(init_qubits=False)) + + params_model = quantum_program_to_0_1(quantum_program, options) + + self.assertEqual(params_model.schema_version, "v0.1") + self.assertEqual(params_model.options.init_qubits, False) + self.assertEqual(params_model.options.rep_delay, None) + + quantum_program_model = params_model.quantum_program + self.assertEqual(quantum_program_model.shots, shots) + + circuit_item_model = quantum_program_model.items[0] + self.assertEqual(circuit_item_model.item_type, "circuit") + self.assertEqual(circuit_item_model.circuit.to_quantum_circuit(), circuit1) + self.assertTrue( + np.array_equal(circuit_item_model.circuit_arguments.to_numpy(), circuit_arguments) + ) + self.assertEqual(circuit_item_model.chunk_size, 6) + + samplex_item_model = quantum_program_model.items[1] + self.assertEqual(samplex_item_model.item_type, "samplex") + self.assertEqual(samplex_item_model.circuit.to_quantum_circuit(), template_circuit) + self.assertEqual(samplex_item_model.shape, [4, 3, 2]) + self.assertEqual(samplex_item_model.chunk_size, 7) + + samplex_decoded = samplex_item_model.samplex.to_samplex() + samplex_decoded.finalize() + self.assertEqual(samplex_decoded, samplex) + + samplex_arguments_model = samplex_item_model.samplex_arguments + self.assertTrue( + np.array_equal(samplex_arguments_model["parameter_values"].to_numpy(), parameter_values) + ) + for i, noise_model in enumerate(noise_models): + self.assertEqual( + samplex_arguments_model[f"pauli_lindblad_maps.pl{i}"].to_pauli_lindblad_map(), + noise_model, + ) + + def test_quantum_program_to_0_1_no_argument(self): + """Test the function ``quantum_program_to_0_1`` when there are no circuit arguments, samplex + arguments, and chunk size""" + quantum_program = QuantumProgram(100) + + circuit1 = QuantumCircuit(1) + quantum_program.append(circuit1) + + circuit2 = QuantumCircuit(2) + with circuit2.box(annotations=[Twirl()]): + circuit2.cx(0, 1) + with circuit2.box(annotations=[Twirl()]): + circuit2.measure_all() + + template_circuit, samplex = build(circuit2) + quantum_program.append( + template_circuit, + samplex=samplex, + ) + + params_model = quantum_program_to_0_1(quantum_program, ExecutorOptions()) + quantum_program_model = params_model.quantum_program + + circuit_item_model = quantum_program_model.items[0] + self.assertEqual(circuit_item_model.circuit_arguments.to_numpy().size, 0) + self.assertEqual(circuit_item_model.chunk_size, "auto") + + samplex_item_model = quantum_program_model.items[1] + self.assertEqual(samplex_item_model.shape, []) + self.assertEqual(samplex_item_model.chunk_size, "auto") + self.assertEqual(samplex_item_model.samplex_arguments, {}) + + def test_quantum_program_result_from_0_1(self): + """Test the function ``quantum_program_result_from_0_1``""" + meas1 = np.array([[False], [True], [True]]) + meas2 = np.array([[True, True], [True, False], [False, False]]) + meas_flips = np.array([[False, False]]) + chunk_start = datetime(2025, 12, 30, 14, 10) + chunk_stop = datetime(2025, 12, 30, 14, 15) + + chunk_model = ChunkSpan( + start=chunk_start, + stop=chunk_stop, + parts=[ChunkPart(idx_item=0, size=1), ChunkPart(idx_item=1, size=1)], + ) + metadata_model = MetadataModel(chunk_timing=[chunk_model]) + result1_model = QuantumProgramResultItemModel( + results={"meas": TensorModel.from_numpy(meas1)}, metadata=None + ) + result2_model = QuantumProgramResultItemModel( + results={ + "meas": TensorModel.from_numpy(meas2), + "measurement_flips.meas": TensorModel.from_numpy(meas_flips), + }, + metadata=None, + ) + result_model = QuantumProgramResultModel( + data=[result1_model, result2_model], metadata=metadata_model + ) + + result = quantum_program_result_from_0_1(result_model) + + self.assertTrue(np.array_equal(result[0]["meas"], meas1)) + self.assertTrue(np.array_equal(result[1]["meas"], meas2)) + self.assertTrue(np.array_equal(result[1]["measurement_flips.meas"], meas_flips)) + self.assertEqual(result.metadata.chunk_timing[0].start, chunk_start) + self.assertEqual(result.metadata.chunk_timing[0].stop, chunk_stop) + self.assertEqual(result.metadata.chunk_timing[0].parts[0].idx_item, 0) + self.assertEqual(result.metadata.chunk_timing[0].parts[0].size, 1) + self.assertEqual(result.metadata.chunk_timing[0].parts[1].idx_item, 1) + self.assertEqual(result.metadata.chunk_timing[0].parts[1].size, 1) diff --git a/test/unit/quantum_program/test_result.py b/test/unit/quantum_program/test_result.py new file mode 100644 index 0000000000..9b2d0c6b93 --- /dev/null +++ b/test/unit/quantum_program/test_result.py @@ -0,0 +1,43 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2026. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Tests the class ``QuantumProgramResult``.""" + +import numpy as np + +from qiskit_ibm_runtime.quantum_program.quantum_program_result import QuantumProgramResult + +from ...ibm_test_case import IBMTestCase + + +class TestQuantumProgramResult(IBMTestCase): + """Tests the ``QuantumProgramResult`` class.""" + + def test_quantum_program_result(self): + """Tests the ``QuantumProgramResult`` class.""" + meas1 = np.array([[False], [True], [True]]) + meas2 = np.array([[True, True], [True, False], [False, False]]) + meas_flips = np.array([[False, False]]) + + result1 = {"meas": meas1} + result2 = {"meas": meas2, "measurement_flips.meas": meas_flips} + result = QuantumProgramResult([result1, result2]) + + # test __len__ + self.assertEqual(len(result), 2) + + # test __iter__ + for res, expected_res in zip(result, [result1, result2]): + self.assertDictEqual(res, expected_res) + + # test __getitem__ + self.assertEqual([result[0], result[1]], [result1, result2])