9255177207

Committed 27 May 2024 12:52PM UTC coverage: 99.736%. Remained the same

Build # 9255177207

Build Type

push

github

Committed by

web-flow

Commit Message

Prepare release 1.5.0 (#161)

Run Details

2267 of 2273 relevant lines covered (99.74%)

3.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.18

/qiskit_aqt_provider/aqt_provider.py

# This code is part of Qiskit.
#
# (C) Copyright IBM 2019, Alpine Quantum Technologies 2020
#
# 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.


import contextlib
import os
import re
import warnings
from collections import defaultdict
from collections.abc import Sequence
from dataclasses import dataclass
from operator import attrgetter
from pathlib import Path
from re import Pattern
from typing import (
    Final,
    Literal,
    Optional,
    Union,
    overload,
)

import dotenv
import httpx
from qiskit.providers.exceptions import QiskitBackendNotFoundError
from tabulate import tabulate
from typing_extensions import TypeAlias, override

from qiskit_aqt_provider import api_models

from .aqt_resource import AQTResource, OfflineSimulatorResource

StrPath: TypeAlias = Union[str, Path]


class NoTokenWarning(UserWarning):
    """Warning emitted when a provider is initialized with no access token."""


@dataclass(frozen=True)
class OfflineSimulator:
    """Description of an offline simulator."""

    id: str
    """Unique identifier of the simulator."""

    name: str
    """Free-text description of the simulator."""

    noisy: bool
    """Whether the simulator uses a noise model."""


OFFLINE_SIMULATORS: Final = [
    OfflineSimulator(id="offline_simulator_no_noise", name="Offline ideal simulator", noisy=False),
    OfflineSimulator(id="offline_simulator_noise", name="Offline noisy simulator", noisy=True),
]


class BackendsTable(Sequence[AQTResource]):
    """Pretty-printable collection of AQT backends.

    The :meth:`__str__` method returns a plain text table representation of the available backends.
    The :meth:`_repr_html_` method returns an HTML representation that is automatically used
    in IPython/Jupyter notebooks.
    """

    def __init__(self, backends: list[AQTResource]):
        """Initialize the table.

        Args:
            backends: list of available backends.
        """
        self.backends = backends
        self.headers = ["Workspace ID", "Resource ID", "Description", "Resource type"]

    @overload
    def __getitem__(self, index: int) -> AQTResource: ...

    @overload
    def __getitem__(self, index: slice) -> Sequence[AQTResource]: ...

    @override
    def __getitem__(self, index: Union[slice, int]) -> Union[AQTResource, Sequence[AQTResource]]:
        """Retrieve a backend by index."""
        return self.backends[index]

    @override
    def __len__(self) -> int:
        """Number of backends."""
        return len(self.backends)

    @override
    def __str__(self) -> str:
        """Plain-text table representation."""
        return tabulate(self.table(), headers=self.headers, tablefmt="fancy_grid")

    def _repr_html_(self) -> str:
        """HTML table representation (for IPython/Jupyter)."""
        return tabulate(self.table(), headers=self.headers, tablefmt="html")  # pragma: no cover

    def by_workspace(self) -> dict[str, list[AQTResource]]:
        """Backends grouped by workspace ID."""
        data: defaultdict[str, list[AQTResource]] = defaultdict(list)

        for backend in self:
            data[backend.resource_id.workspace_id].append(backend)

        return dict(data)

    def table(self) -> list[list[str]]:
        """Assemble the data for the printable table."""
        table = []
        for workspace_id, resources in self.by_workspace().items():
            for count, resource in enumerate(
                sorted(resources, key=attrgetter("resource_id.resource_id"))
            ):
                line = [
                    workspace_id,
                    resource.resource_id.resource_id,
                    resource.resource_id.resource_name,
                    resource.resource_id.resource_type,
                ]
                if count != 0:
                    # don't repeat the workspace id
                    line[0] = ""

                table.append(line)

        return table


class AQTProvider:
    """Provider for backends from Alpine Quantum Technologies (AQT)."""

    # Set AQT_PORTAL_URL environment variable to override
    DEFAULT_PORTAL_URL: Final = "https://arnica.aqt.eu"

    def __init__(
        self,
        access_token: Optional[str] = None,
        *,
        load_dotenv: bool = True,
        dotenv_path: Optional[StrPath] = None,
    ):
        """Initialize the AQT provider.

        The access token for the AQT cloud can be provided either through the
        ``access_token`` argument or the ``AQT_TOKEN`` environment variable.

        .. hint:: If no token is set (neither through the ``access_token`` argument nor
            through the ``AQT_TOKEN`` environment variable), the provider is initialized
            with access to the offline simulators only and :class:`NoTokenWarning` is
            emitted.

        The AQT cloud portal URL can be configured using the ``AQT_PORTAL_URL``
        environment variable.

        If ``load_dotenv`` is true, environment variables are loaded from a file,
        by default any ``.env`` file in the working directory or above it in the
        directory tree.
        The ``dotenv_path`` argument allows to pass a specific file to load environment
        variables from.

        Args:
            access_token: AQT cloud access token.
            load_dotenv: whether to load environment variables from a ``.env`` file.
            dotenv_path: path to the environment file. This implies ``load_dotenv``.
        """
        if load_dotenv or dotenv_path is not None:
            dotenv.load_dotenv(dotenv_path)

        portal_base_url = os.environ.get("AQT_PORTAL_URL", AQTProvider.DEFAULT_PORTAL_URL)
        self.portal_url = f"{portal_base_url}/api/v1"

        if access_token is None:
            self.access_token = os.environ.get("AQT_TOKEN", "")
        else:
            self.access_token = access_token

        if not self.access_token:
            warnings.warn(
                "No access token provided: access is restricted to the 'default' workspace.",
                NoTokenWarning,
            )

        self.name = "aqt_provider"

    @property
    def _http_client(self) -> httpx.Client:
        """HTTP client for communicating with the AQT cloud service."""
        return api_models.http_client(base_url=self.portal_url, token=self.access_token)

    def backends(
        self,
        name: Optional[Union[str, Pattern[str]]] = None,
        *,
        backend_type: Optional[Literal["device", "simulator", "offline_simulator"]] = None,
        workspace: Optional[Union[str, Pattern[str]]] = None,
    ) -> BackendsTable:
        """Search for backends matching given criteria.

        With no arguments, return all backends accessible with the configured
        access token.

        Filters can be either strings or regular expression patterns. Strings filter by
        exact match.

        Args:
            name: filter for the backend name.
            backend_type: if given, restrict the search to the given backend type.
            workspace: filter for the workspace ID.

        Returns:
            Collection of backends accessible with the given access token that match the
            given criteria.
        """
        if isinstance(name, str):
            name = re.compile(f"^{name}$")

        if isinstance(workspace, str):
            workspace = re.compile(f"^{workspace}$")

        remote_workspaces = api_models.Workspaces(root=[])

        if backend_type != "offline_simulator":
            with contextlib.suppress(httpx.HTTPError, httpx.NetworkError):
                with self._http_client as client:
                    resp = client.get("/workspaces")
                    resp.raise_for_status()

                remote_workspaces = api_models.Workspaces.model_validate(resp.json()).filter(
                    name_pattern=name,
                    backend_type=api_models.ResourceType(backend_type) if backend_type else None,
                    workspace_pattern=workspace,
                )

        backends: list[AQTResource] = []

        # add offline simulators in the default workspace
        if (not workspace or workspace.match("default")) and (
            not backend_type or backend_type == "offline_simulator"
        ):
            for simulator in OFFLINE_SIMULATORS:
                if name and not name.match(simulator.id):
                    continue
                backends.append(
                    OfflineSimulatorResource(
                        self,
                        resource_id=api_models.ResourceId(
                            workspace_id="default",
                            resource_id=simulator.id,
                            resource_name=simulator.name,
                            resource_type="offline_simulator",
                        ),
                        with_noise_model=simulator.noisy,
                    )
                )

        # add (filtered) remote resources
        for _workspace in remote_workspaces.root:
            for resource in _workspace.resources:
                backends.append(
                    AQTResource(
                        self,
                        resource_id=api_models.ResourceId(
                            workspace_id=_workspace.id,
                            resource_id=resource.id,
                            resource_name=resource.name,
                            resource_type=resource.type.value,
                        ),
                    )
                )

        return BackendsTable(backends)

    def get_backend(
        self,
        name: Optional[Union[str, Pattern[str]]] = None,
        *,
        backend_type: Optional[Literal["device", "simulator", "offline_simulator"]] = None,
        workspace: Optional[Union[str, Pattern[str]]] = None,
    ) -> AQTResource:
        """Return a single backend matching the specified filtering.

        Args:
            name: filter for the backend name.
            backend_type: if given, restrict the search to the given backend type.
            workspace: if given, restrict to matching workspace IDs.

        Returns:
            Backend: backend matching the filtering.

        Raises:
            QiskitBackendNotFoundError: if no backend could be found or
                more than one backend matches the filtering criteria.
        """
        # From: https://github.com/Qiskit/qiskit/blob/8e3218bc0798b0612edf446db130e95ac9404968/qiskit/providers/provider.py#L53
        # after ProviderV1 deprecation.
        # See: https://github.com/Qiskit/qiskit/pull/12145.
        backends = self.backends(name, backend_type=backend_type, workspace=workspace)
        if len(backends) > 1:
            raise QiskitBackendNotFoundError("More than one backend matches the criteria")
        if not backends:
            raise QiskitBackendNotFoundError("No backend matches the criteria")

        return backends[0]

1	# This code is part of Qiskit.
2	#
3	# (C) Copyright IBM 2019, Alpine Quantum Technologies 2020
4	#
5	# This code is licensed under the Apache License, Version 2.0. You may
6	# obtain a copy of this license in the LICENSE.txt file in the root directory
7	# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
8	#
9	# Any modifications or derivative works of this code must retain this
10	# copyright notice, and modified files need to carry a notice indicating
11	# that they have been altered from the originals.
12
13
14	import contextlib	4✔
15	import os	4✔
16	import re	4✔
17	import warnings	4✔
18	from collections import defaultdict	4✔
19	from collections.abc import Sequence	4✔
20	from dataclasses import dataclass	4✔
21	from operator import attrgetter	4✔
22	from pathlib import Path	4✔
23	from re import Pattern	4✔
24	from typing import (	4✔
25	Final,
26	Literal,
27	Optional,
28	Union,
29	overload,
30	)
31
32	import dotenv	4✔
33	import httpx	4✔
34	from qiskit.providers.exceptions import QiskitBackendNotFoundError	4✔
35	from tabulate import tabulate	4✔
36	from typing_extensions import TypeAlias, override	4✔
37
38	from qiskit_aqt_provider import api_models	4✔
39
40	from .aqt_resource import AQTResource, OfflineSimulatorResource	4✔
41
42	StrPath: TypeAlias = Union[str, Path]	4✔
43
44
45	class NoTokenWarning(UserWarning):	4✔
46	"""Warning emitted when a provider is initialized with no access token."""	4✔
47
48
49	@dataclass(frozen=True)	4✔
50	class OfflineSimulator:	4✔
51	"""Description of an offline simulator."""	4✔
52
53	id: str	4✔
54	"""Unique identifier of the simulator."""	3✔
55
56	name: str	4✔
57	"""Free-text description of the simulator."""	3✔
58
59	noisy: bool	4✔
60	"""Whether the simulator uses a noise model."""	4✔
61
62
63	OFFLINE_SIMULATORS: Final = [	4✔
64	OfflineSimulator(id="offline_simulator_no_noise", name="Offline ideal simulator", noisy=False),
65	OfflineSimulator(id="offline_simulator_noise", name="Offline noisy simulator", noisy=True),
66	]
67
68
69	class BackendsTable(Sequence[AQTResource]):	4✔
70	"""Pretty-printable collection of AQT backends.	4✔
71
72	The :meth:`__str__` method returns a plain text table representation of the available backends.
73	The :meth:`_repr_html_` method returns an HTML representation that is automatically used
74	in IPython/Jupyter notebooks.
75	"""
76
77	def __init__(self, backends: list[AQTResource]):	4✔
78	"""Initialize the table.
79
80	Args:
81	backends: list of available backends.
82	"""
83	self.backends = backends	4✔
84	self.headers = ["Workspace ID", "Resource ID", "Description", "Resource type"]	4✔
85
86	@overload	4✔
87	def __getitem__(self, index: int) -> AQTResource: ...	4✔
88
89	@overload	4✔
90	def __getitem__(self, index: slice) -> Sequence[AQTResource]: ...	4✔
91
92	@override	4✔
93	def __getitem__(self, index: Union[slice, int]) -> Union[AQTResource, Sequence[AQTResource]]:	4✔
94	"""Retrieve a backend by index."""
95	return self.backends[index]	4✔
96
97	@override	4✔
98	def __len__(self) -> int:	4✔
99	"""Number of backends."""
100	return len(self.backends)	4✔
101
102	@override	4✔
103	def __str__(self) -> str:	4✔
104	"""Plain-text table representation."""
105	return tabulate(self.table(), headers=self.headers, tablefmt="fancy_grid")	4✔
106
107	def _repr_html_(self) -> str:	4✔
108	"""HTML table representation (for IPython/Jupyter)."""
109	return tabulate(self.table(), headers=self.headers, tablefmt="html") # pragma: no cover
110
111	def by_workspace(self) -> dict[str, list[AQTResource]]:	4✔
112	"""Backends grouped by workspace ID."""
113	data: defaultdict[str, list[AQTResource]] = defaultdict(list)	4✔
114
115	for backend in self:	4✔
116	data[backend.resource_id.workspace_id].append(backend)	4✔
117
118	return dict(data)	4✔
119
120	def table(self) -> list[list[str]]:	4✔
121	"""Assemble the data for the printable table."""
122	table = []	4✔
123	for workspace_id, resources in self.by_workspace().items():	4✔
124	for count, resource in enumerate(	4✔
125	sorted(resources, key=attrgetter("resource_id.resource_id"))
126	):
127	line = [	4✔
128	workspace_id,
129	resource.resource_id.resource_id,
130	resource.resource_id.resource_name,
131	resource.resource_id.resource_type,
132	]
133	if count != 0:	4✔
134	# don't repeat the workspace id
135	line[0] = ""	4✔
136
137	table.append(line)	4✔
138
139	return table	4✔
140
141
142	class AQTProvider:	4✔
143	"""Provider for backends from Alpine Quantum Technologies (AQT)."""	4✔
144
145	# Set AQT_PORTAL_URL environment variable to override
146	DEFAULT_PORTAL_URL: Final = "https://arnica.aqt.eu"	4✔
147
148	def __init__(	4✔
149	self,
150	access_token: Optional[str] = None,
151	*,
152	load_dotenv: bool = True,
153	dotenv_path: Optional[StrPath] = None,
154	):
155	"""Initialize the AQT provider.
156
157	The access token for the AQT cloud can be provided either through the
158	``access_token`` argument or the ``AQT_TOKEN`` environment variable.
159
160	.. hint:: If no token is set (neither through the ``access_token`` argument nor
161	through the ``AQT_TOKEN`` environment variable), the provider is initialized
162	with access to the offline simulators only and :class:`NoTokenWarning` is
163	emitted.
164
165	The AQT cloud portal URL can be configured using the ``AQT_PORTAL_URL``
166	environment variable.
167
168	If ``load_dotenv`` is true, environment variables are loaded from a file,
169	by default any ``.env`` file in the working directory or above it in the
170	directory tree.
171	The ``dotenv_path`` argument allows to pass a specific file to load environment
172	variables from.
173
174	Args:
175	access_token: AQT cloud access token.
176	load_dotenv: whether to load environment variables from a ``.env`` file.
177	dotenv_path: path to the environment file. This implies ``load_dotenv``.
178	"""
179	if load_dotenv or dotenv_path is not None:	4✔
180	dotenv.load_dotenv(dotenv_path)	4✔
181
182	portal_base_url = os.environ.get("AQT_PORTAL_URL", AQTProvider.DEFAULT_PORTAL_URL)	4✔
183	self.portal_url = f"{portal_base_url}/api/v1"	4✔
184
185	if access_token is None:	4✔
186	self.access_token = os.environ.get("AQT_TOKEN", "")	4✔
187	else:
188	self.access_token = access_token	4✔
189
190	if not self.access_token:	4✔
191	warnings.warn(	4✔
192	"No access token provided: access is restricted to the 'default' workspace.",
193	NoTokenWarning,
194	)
195
196	self.name = "aqt_provider"	4✔
197
198	@property	4✔
199	def _http_client(self) -> httpx.Client:	4✔
200	"""HTTP client for communicating with the AQT cloud service."""
201	return api_models.http_client(base_url=self.portal_url, token=self.access_token)	4✔
202
203	def backends(	4✔
204	self,
205	name: Optional[Union[str, Pattern[str]]] = None,
206	*,
207	backend_type: Optional[Literal["device", "simulator", "offline_simulator"]] = None,
208	workspace: Optional[Union[str, Pattern[str]]] = None,
209	) -> BackendsTable:
210	"""Search for backends matching given criteria.
211
212	With no arguments, return all backends accessible with the configured
213	access token.
214
215	Filters can be either strings or regular expression patterns. Strings filter by
216	exact match.
217
218	Args:
219	name: filter for the backend name.
220	backend_type: if given, restrict the search to the given backend type.
221	workspace: filter for the workspace ID.
222
223	Returns:
224	Collection of backends accessible with the given access token that match the
225	given criteria.
226	"""
227	if isinstance(name, str):	4✔
228	name = re.compile(f"^{name}$")	4✔
229
230	if isinstance(workspace, str):	4✔
231	workspace = re.compile(f"^{workspace}$")	4✔
232
233	remote_workspaces = api_models.Workspaces(root=[])	4✔
234
235	if backend_type != "offline_simulator":	4✔
236	with contextlib.suppress(httpx.HTTPError, httpx.NetworkError):	4✔
237	with self._http_client as client:	4✔
238	resp = client.get("/workspaces")	4✔
239	resp.raise_for_status()	4✔
240
241	remote_workspaces = api_models.Workspaces.model_validate(resp.json()).filter(	4✔
242	name_pattern=name,
243	backend_type=api_models.ResourceType(backend_type) if backend_type else None,
244	workspace_pattern=workspace,
245	)
246
247	backends: list[AQTResource] = []	4✔
248
249	# add offline simulators in the default workspace
250	if (not workspace or workspace.match("default")) and (	4✔
251	not backend_type or backend_type == "offline_simulator"
252	):
253	for simulator in OFFLINE_SIMULATORS:	4✔
254	if name and not name.match(simulator.id):	4✔
255	continue	4✔
256	backends.append(	4✔
257	OfflineSimulatorResource(
258	self,
259	resource_id=api_models.ResourceId(
260	workspace_id="default",
261	resource_id=simulator.id,
262	resource_name=simulator.name,
263	resource_type="offline_simulator",
264	),
265	with_noise_model=simulator.noisy,
266	)
267	)
268
269	# add (filtered) remote resources
270	for _workspace in remote_workspaces.root:	4✔
271	for resource in _workspace.resources:	4✔
272	backends.append(	4✔
273	AQTResource(
274	self,
275	resource_id=api_models.ResourceId(
276	workspace_id=_workspace.id,
277	resource_id=resource.id,
278	resource_name=resource.name,
279	resource_type=resource.type.value,
280	),
281	)
282	)
283
284	return BackendsTable(backends)	4✔
285
286	def get_backend(	4✔
287	self,
288	name: Optional[Union[str, Pattern[str]]] = None,
289	*,
290	backend_type: Optional[Literal["device", "simulator", "offline_simulator"]] = None,
291	workspace: Optional[Union[str, Pattern[str]]] = None,
292	) -> AQTResource:
293	"""Return a single backend matching the specified filtering.
294
295	Args:
296	name: filter for the backend name.
297	backend_type: if given, restrict the search to the given backend type.
298	workspace: if given, restrict to matching workspace IDs.
299
300	Returns:
301	Backend: backend matching the filtering.
302
303	Raises:
304	QiskitBackendNotFoundError: if no backend could be found or
305	more than one backend matches the filtering criteria.
306	"""
307	# From: https://github.com/Qiskit/qiskit/blob/8e3218bc0798b0612edf446db130e95ac9404968/qiskit/providers/provider.py#L53
308	# after ProviderV1 deprecation.
309	# See: https://github.com/Qiskit/qiskit/pull/12145.
310	backends = self.backends(name, backend_type=backend_type, workspace=workspace)	4✔
311	if len(backends) > 1:	4✔
312	raise QiskitBackendNotFoundError("More than one backend matches the criteria")	×
313	if not backends:	4✔
314	raise QiskitBackendNotFoundError("No backend matches the criteria")	×
315
316	return backends[0]	4✔

qiskit-community / qiskit-aqt-provider / 9255177207

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous