[QI2-1076] Backend construction (#11)

pyproject.toml

@@ -19,9 +19,10 @@ classifiers = [
 [tool.poetry.dependencies]
 python = "^3.9"
 qiskit = "^1.2.0"
-qi-compute-api-client = "^0.33.0"
+qi-compute-api-client = "^0.35.0"
 pydantic = "^2.9.1"
 requests = "^2.32.3"
+pytest-asyncio = "^0.24.0"
 
 [tool.poetry.group.dev.dependencies]
 black = "^24.4.2"

qiskit_quantuminspire/api/pagination.py

@@ -0,0 +1,36 @@
+from typing import Any, Awaitable, Callable, Generic, List, Optional, TypeVar, cast
+
+from pydantic import BaseModel, Field
+from typing_extensions import Annotated
+
+PageType = TypeVar("PageType")
+ItemType = TypeVar("ItemType")
+
+
+class PageInterface(BaseModel, Generic[ItemType]):
+    """The page models in the generated API client don't inherit from a common base class, so we have to trick the
+    typing system a bit with this fake base class."""
+
+    items: List[ItemType]
+    total: Optional[Annotated[int, Field(strict=True, ge=0)]]
+    page: Optional[Annotated[int, Field(strict=True, ge=1)]]
+    size: Optional[Annotated[int, Field(strict=True, ge=1)]]
+    pages: Optional[Annotated[int, Field(strict=True, ge=0)]] = None
+
+
+class PageReader(Generic[PageType, ItemType]):
+    """Helper class for reading fastapi-pagination style pages returned by the compute_api_client."""
+
+    async def get_all(self, api_call: Callable[..., Awaitable[PageType]], **kwargs: Any) -> List[ItemType]:
+        """Get all items from an API call that supports paging."""
+        items: List[ItemType] = []
+        page = 1
+
+        while True:
+            response = cast(PageInterface[ItemType], await api_call(page=page, **kwargs))
+
+            items.extend(response.items)
+            page += 1
+            if response.pages is None or page > response.pages:
+                break
+        return items

qiskit_quantuminspire/qi_backend.py

@@ -1,30 +1,111 @@
+import logging
+import math
 from typing import Any, List, Union
 
-from compute_api_client import BackendType, Metadata
-from qiskit.circuit import QuantumCircuit
+from compute_api_client import BackendType
+from qiskit.circuit import Instruction, Measure, QuantumCircuit
+from qiskit.circuit.library import (
+    CCXGate,
+    CPhaseGate,
+    CXGate,
+    IGate,
+    RXGate,
+    RYGate,
+    SdgGate,
+    TdgGate,
+    get_standard_gate_name_mapping,
+)
+from qiskit.circuit.parameter import Parameter
 from qiskit.providers import BackendV2 as Backend
 from qiskit.providers.options import Options
-from qiskit.transpiler import Target
+from qiskit.transpiler import CouplingMap, Target
 
 from qiskit_quantuminspire.qi_jobs import QIJob
+from qiskit_quantuminspire.utils import is_coupling_map_complete
+
+# Used for parameterizing Qiskit gates in the gate mapping
+_THETA = Parameter("ϴ")
+
+# Custom gate mapping for gates whose name do not match between cQASM and Qiskit
+_CQASM_QISKIT_GATE_MAPPING: dict[str, Instruction] = {
+    "i": IGate(),
+    "x90": RXGate(math.pi / 2),
+    "mx90": RXGate(-math.pi / 2),
+    "y90": RYGate(math.pi / 2),
+    "my90": RYGate(-math.pi / 2),
+    "toffoli": CCXGate(),
+    "sdag": SdgGate(),
+    "tdag": TdgGate(),
+    "cr": CPhaseGate(_THETA),
+    "cnot": CXGate(),
+    "measure_z": Measure(),
+}
+
+_IGNORED_GATES: list[str] = [
+    # Prep not viewed as separate gates in Qiskit
+    "prep_x",
+    "prep_y",
+    "prep_z",
+    # Measure x and y not natively supported https://github.com/Qiskit/qiskit/issues/3967
+    "measure_x",
+    "measure_y",
+    "measure_all",
+    # May be supportable through parameterized CPhaseGate.
+    # For now, direct usage of CPhaseGate is required
+    "crk",
+]
+
+_ALL_SUPPORTED_GATES: list[str] = list(get_standard_gate_name_mapping().keys()) + list(
+    _CQASM_QISKIT_GATE_MAPPING.keys()
+)
 
 
 # Ignore type checking for QIBackend due to missing Qiskit type stubs,
 # which causes the base class 'Backend' to be treated as 'Any'.
 class QIBackend(Backend):  # type: ignore[misc]
     """A wrapper class for QuantumInspire backendtypes to integrate with Qiskit's Backend interface."""
 
-    def __init__(self, backend_type: BackendType, metadata: Metadata, **kwargs: Any):
+    _max_shots: int
+
+    def __init__(self, backend_type: BackendType, **kwargs: Any):
         super().__init__(name=backend_type.name, description=backend_type.description, **kwargs)
-        self._target = Target(num_qubits=metadata.data["nqubits"])
+
+        self._max_shots = backend_type.max_number_of_shots
+
+        native_gates = [gate.lower() for gate in backend_type.gateset]
+        available_gates = [gate for gate in native_gates if gate in _ALL_SUPPORTED_GATES]
+        unknown_gates = set(native_gates) - set(_ALL_SUPPORTED_GATES) - set(_IGNORED_GATES)
+        coupling_map = CouplingMap(backend_type.topology)
+        coupling_map_complete = is_coupling_map_complete(coupling_map)
+
+        if len(unknown_gates) > 0:
+            logging.warning(f"Ignoring unknown native gate(s) {unknown_gates} for backend {backend_type.name}")
+
+        if "toffoli" in available_gates and not coupling_map_complete:
+            available_gates.remove("toffoli")
+            logging.warning(
+                f"Native toffoli gate in backend {backend_type.name} not supported for non-complete topology"
+            )
+
+        self._target = Target().from_configuration(
+            basis_gates=available_gates,
+            num_qubits=backend_type.nqubits,
+            coupling_map=None if coupling_map_complete else coupling_map,
+            custom_name_mapping=_CQASM_QISKIT_GATE_MAPPING,
+        )
 
     @classmethod
     def _default_options(cls) -> Options:
         return Options(shots=1024, optimization_level=1)
 
+    @property
     def target(self) -> Target:
         return self._target
 
+    @property
+    def max_shots(self) -> int:
+        return self._max_shots
+
     @property
     def max_circuits(self) -> Union[int, None]:
         return None

qiskit_quantuminspire/qi_provider.py

@@ -1,10 +1,10 @@
 import asyncio
-from datetime import datetime, timezone
 from typing import Any, List, Union
 
-from compute_api_client import ApiClient, BackendType, BackendTypesApi, Metadata
+from compute_api_client import ApiClient, BackendType, BackendTypesApi, PageBackendType
 
 from qiskit_quantuminspire.api.client import config
+from qiskit_quantuminspire.api.pagination import PageReader
 from qiskit_quantuminspire.qi_backend import QIBackend
 
 
@@ -14,28 +14,23 @@ class QIProvider:
     def __init__(self) -> None:
         self._qiskit_backends = self._construct_backends()
 
-    def _fetch_qi_backend_metadata(self, backend_type: BackendType) -> Metadata:
-        """Fetch backend metadata using api client."""
-        return Metadata(id=1, backend_id=1, created_on=datetime.now(timezone.utc), data={"nqubits": 6})
-
     async def _fetch_qi_backend_types(self) -> List[BackendType]:
         """Fetch backend types from CJM using api client.
 
         (Implemented without paging only for demonstration purposes, should get a proper implementation)
         """
         async with ApiClient(config()) as client:
+            page_reader = PageReader[PageBackendType, BackendType]()
             backend_types_api = BackendTypesApi(client)
-            backend_type_list = await backend_types_api.read_backend_types_backend_types_get()
-            backend_types: List[BackendType] = backend_type_list.items
+            backend_types: List[BackendType] = await page_reader.get_all(
+                backend_types_api.read_backend_types_backend_types_get
+            )
         return backend_types
 
     def _construct_backends(self) -> List[QIBackend]:
         """Construct QIBackend using fetched backendtypes and metadata."""
         qi_backend_types = asyncio.run(self._fetch_qi_backend_types())
-        qi_backends = [
-            QIBackend(provider=self, backend_type=backend_type, metadata=self._fetch_qi_backend_metadata(backend_type))
-            for backend_type in qi_backend_types
-        ]
+        qi_backends = [QIBackend(provider=self, backend_type=backend_type) for backend_type in qi_backend_types]
         return qi_backends
 
     def backends(self, name: Union[str, None] = None, **kwargs: Any) -> List[QIBackend]:

qiskit_quantuminspire/utils.py

@@ -0,0 +1,12 @@
+from qiskit.transpiler import CouplingMap
+
+
+def is_coupling_map_complete(coupling_map: CouplingMap) -> bool:
+    """A complete digraph is a digraph in which there is a directed edge from every vertex to every other vertex."""
+    distance_matrix = coupling_map.distance_matrix
+
+    assert distance_matrix is not None
+
+    is_semicomplete = all(distance in [1, 0] for distance in distance_matrix.flatten())
+
+    return is_semicomplete and coupling_map.is_symmetric

tests/api/test_pagination.py

@@ -0,0 +1,50 @@
+from unittest.mock import AsyncMock
+
+import pytest
+from compute_api_client import BackendType, PageBackendType
+
+from qiskit_quantuminspire.api.pagination import PageReader
+from tests.helpers import create_backend_type
+
+
+@pytest.mark.asyncio
+async def test_pagination_get_all() -> None:
+    # Arrange
+    def returned_pages(page: int) -> PageBackendType:
+        pages = [
+            PageBackendType(
+                items=[
+                    create_backend_type(name="qi_backend_1"),
+                    create_backend_type(name="spin"),
+                ],
+                total=5,
+                page=1,
+                size=2,
+                pages=2,
+            ),
+            PageBackendType(
+                items=[
+                    create_backend_type(name="qi_backend2"),
+                    create_backend_type(name="spin6"),
+                    create_backend_type(name="spin7"),
+                ],
+                total=5,
+                page=2,
+                size=3,
+                pages=2,
+            ),
+        ]
+        return pages[page - 1]
+
+    api_call = AsyncMock(side_effect=returned_pages)
+
+    page_reader = PageReader[PageBackendType, BackendType]()
+
+    # Act
+    backends = await page_reader.get_all(api_call)
+
+    # Assert
+    actual_backend_names = [backend.name for backend in backends]
+    expected_backend_names = ["qi_backend_1", "spin", "qi_backend2", "spin6", "spin7"]
+
+    assert actual_backend_names == expected_backend_names

tests/helpers.py

@@ -0,0 +1,28 @@
+from compute_api_client import BackendStatus, BackendType
+
+
+def create_backend_type(
+    gateset: list[str] = [],
+    topology: list[list[int]] = [],
+    nqubits: int = 0,
+    default_number_of_shots: int = 1024,
+    max_number_of_shots: int = 2048,
+    name: str = "qi_backend",
+) -> BackendType:
+    """Helper for creating a backendtype with only the fields you care about."""
+    return BackendType(
+        name=name,
+        nqubits=nqubits,
+        gateset=gateset,
+        topology=topology,
+        id=1,
+        is_hardware=True,
+        image_id="qi_backend",
+        features=[],
+        default_compiler_config="",
+        status=BackendStatus.IDLE,
+        default_number_of_shots=default_number_of_shots,
+        max_number_of_shots=max_number_of_shots,
+        infrastructure="QCI",
+        description="A Quantum Inspire backend",
+    )