Implement direct-access API

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## Unreleased
+
+* Support direct-access mode on AQT devices (#164)
+
 ## qiskit-aqt-provider v1.5.0
 
 * Docs: add examples on setting run options in primitives (#156)

docs/apidoc/job.rst

@@ -6,6 +6,11 @@ AQTJob
 
 .. autoclass:: qiskit_aqt_provider.aqt_job.AQTJob
    :members:
+   :exclude-members: __init__
+
+.. autoclass:: qiskit_aqt_provider.aqt_job.AQTDirectAccessJob
+   :members:
+   :exclude-members: __init__, submit
 
 .. autoclass:: qiskit_aqt_provider.aqt_job.Progress
    :members:

docs/apidoc/resource.rst

@@ -7,24 +7,22 @@ AQTResource
 .. autoclass:: qiskit_aqt_provider.aqt_resource.AQTResource
    :members:
    :show-inheritance:
+   :exclude-members: submit, result, __init__
 
-.. autoclass:: qiskit_aqt_provider.aqt_resource.OfflineSimulatorResource
+.. autoclass:: qiskit_aqt_provider.aqt_resource.AQTDirectAccessResource
    :members:
    :show-inheritance:
+   :exclude-members: submit, result, __init__
 
-.. autopydantic_model:: qiskit_aqt_provider.api_models.ResourceId
-   :exclude-members: model_computed_fields
-   :model-show-json: False
-   :model-show-validator-members: False
-   :model-show-validator-summary: False
-   :model-show-field-summary: False
-   :member-order: bysource
+.. autoclass:: qiskit_aqt_provider.aqt_resource.OfflineSimulatorResource
+   :members:
    :show-inheritance:
+   :exclude-members: submit, result, __init__
 
 .. autoclass:: qiskit_aqt_provider.aqt_resource.UnknownOptionWarning
    :exclude-members: __init__, __new__
    :show-inheritance:
 
-.. autoclass:: qiskit_aqt_provider.api_models.UnknownJobError
+.. autoclass:: qiskit_aqt_provider.aqt_resource._ResourceBase
    :show-inheritance:
    :exclude-members: __init__, __new__

docs/index.rst

@@ -66,10 +66,10 @@ For more details see the :ref:`user guide <user-guide>`, a selection of `example
   :caption: Reference
   :hidden:
 
-  AQTProvider <apidoc/provider>
-  AQTResource <apidoc/resource>
-  AQTJob <apidoc/job>
-  AQTOptions <apidoc/options>
+  Provider <apidoc/provider>
+  Backends <apidoc/resource>
+  Job handles <apidoc/job>
+  Options <apidoc/options>
   Qiskit primitives <apidoc/primitives>
   Transpiler plugin <apidoc/transpiler_plugin>
 

qiskit_aqt_provider/api_models.py

@@ -58,7 +58,7 @@ def http_client(*, base_url: str, token: str) -> httpx.Client:
         token: access token for the remote service.
     """
     headers = {"Authorization": f"Bearer {token}", "User-Agent": USER_AGENT}
-    return httpx.Client(headers=headers, base_url=base_url, timeout=10.0)
+    return httpx.Client(headers=headers, base_url=base_url, timeout=10.0, follow_redirects=True)
 
 
 class Workspaces(pdt.RootModel[list[api_models.Workspace]]):

qiskit_aqt_provider/api_models_direct.py

@@ -0,0 +1,58 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright Alpine Quantum Technologies GmbH 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.
+
+"""API models specific to the direct access API."""
+
+import uuid
+from typing import Annotated, Literal, Union
+
+import pydantic as pdt
+from typing_extensions import Self
+
+
+class JobResultError(pdt.BaseModel):
+    """Failed job result payload."""
+
+    status: Literal["error"] = "error"
+
+
+class JobResultFinished(pdt.BaseModel):
+    """Successful job result payload."""
+
+    status: Literal["finished"] = "finished"
+    result: list[list[Annotated[int, pdt.Field(le=1, ge=0)]]]
+
+
+class JobResult(pdt.BaseModel):
+    """Result model on the direct access API."""
+
+    job_id: uuid.UUID
+    payload: Union[JobResultFinished, JobResultError] = pdt.Field(discriminator="status")
+
+    @classmethod
+    def create_error(cls, *, job_id: uuid.UUID) -> Self:
+        """Create an error result (for tests).
+
+        Args:
+            job_id: job identifier.
+        """
+        return cls(job_id=job_id, payload=JobResultError())
+
+    @classmethod
+    def create_finished(cls, *, job_id: uuid.UUID, result: list[list[int]]) -> Self:
+        """Create a success result (for tests).
+
+        Args:
+            job_id: job identifier.
+            result: mock measured samples.
+        """
+        return cls(job_id=job_id, payload=JobResultFinished(result=result))

qiskit_aqt_provider/aqt_job.py

@@ -33,11 +33,12 @@
 from typing_extensions import Self, TypeAlias, assert_never
 
 from qiskit_aqt_provider import api_models_generated, persistence
+from qiskit_aqt_provider.api_models_direct import JobResultError
 from qiskit_aqt_provider.aqt_options import AQTOptions
 from qiskit_aqt_provider.circuit_to_aqt import circuits_to_aqt_job
 
 if TYPE_CHECKING:  # pragma: no cover
-    from qiskit_aqt_provider.aqt_resource import AQTResource
+    from qiskit_aqt_provider.aqt_resource import AQTDirectAccessResource, AQTResource
 
 
 # Tags for the status of AQT API jobs
@@ -114,7 +115,7 @@ def __exit__(*args) -> None: ...
 
 
 class AQTJob(JobV1):
-    """Handle for quantum circuits jobs running on AQT backends.
+    """Handle for quantum circuits jobs running on AQT cloud backends.
 
     Jobs contain one or more quantum circuits that are executed with a common
     set of options (see :class:`AQTOptions <qiskit_aqt_provider.aqt_options.AQTOptions>`).
@@ -365,30 +366,10 @@ def callback(
         if isinstance(self.status_payload, JobFinished):
             for circuit_index, circuit in enumerate(self.circuits):
                 samples = self.status_payload.results[circuit_index]
-                meas_map = _build_memory_mapping(circuit)
-                data: dict[str, Any] = {
-                    "counts": _format_counts(samples, meas_map),
-                }
-
-                if self.options.memory:
-                    data["memory"] = [
-                        "".join(str(x) for x in reversed(states)) for states in samples
-                    ]
-
                 results.append(
-                    {
-                        "shots": self.options.shots,
-                        "success": True,
-                        "status": JobStatus.DONE,
-                        "data": data,
-                        "header": {
-                            "memory_slots": circuit.num_clbits,
-                            "creg_sizes": [[reg.name, reg.size] for reg in circuit.cregs],
-                            "qreg_sizes": [[reg.name, reg.size] for reg in circuit.qregs],
-                            "name": circuit.name,
-                            "metadata": circuit.metadata or {},
-                        },
-                    }
+                    _partial_qiskit_result_dict(
+                        samples, circuit, shots=self.options.shots, memory=self.options.memory
+                    )
                 )
 
         return Result.from_dict(
@@ -405,6 +386,135 @@ def callback(
         )
 
 
+class AQTDirectAccessJob(JobV1):
+    """Handle for quantum circuits jobs running on direct-access AQT backends.
+
+    Use :meth:`AQTDirectAccessResource.run <qiskit_aqt_provider.aqt_resource.AQTDirectAccessResource.run>`
+    to get a handle and evaluate circuits on a direct-access backend.
+    """
+
+    _backend: "AQTDirectAccessResource"
+
+    def __init__(
+        self,
+        backend: "AQTDirectAccessResource",
+        circuits: list[QuantumCircuit],
+        options: AQTOptions,
+    ):
+        """Initialize the :class:`AQTDirectAccessJob` instance.
+
+        Args:
+            backend: backend to run the job on.
+            circuits: list of circuits to execute.
+            options: overridden resource options for this job.
+        """
+        super().__init__(backend, "")
+
+        self.circuits = circuits
+        self.options = options
+        self.api_submit_payload = circuits_to_aqt_job(circuits, options.shots)
+
+        self._job_id = uuid.uuid4()
+        self._status = JobStatus.INITIALIZING
+
+    def submit(self) -> None:
+        """No-op on direct-access backends."""
+
+    def result(self) -> Result:
+        """Iteratively submit all circuits and block until full completion.
+
+        If an error occurs, the remaining circuits are not executed and the whole
+        job is marked as failed.
+
+        Returns:
+            The combined result of all circuit evaluations.
+        """
+        if self.options.with_progress_bar:
+            context: Union[tqdm[NoReturn], _MockProgressBar] = tqdm(total=len(self.circuits))
+        else:
+            context = _MockProgressBar(total=len(self.circuits))
+
+        result = {
+            "backend_name": self._backend.name,
+            "backend_version": self._backend.version,
+            "qobj_id": id(self.circuits),
+            "job_id": self.job_id(),
+            "success": True,
+            "results": [],
+        }
+
+        with context as progress_bar:
+            for circuit_index, circuit in enumerate(self.circuits):
+                api_circuit = self.api_submit_payload.payload.circuits[circuit_index]
+                job_id = self._backend.submit(api_circuit)
+                api_result = self._backend.result(job_id)
+
+                if isinstance(api_result.payload, JobResultError):
+                    break
+
+                result["results"].append(
+                    _partial_qiskit_result_dict(
+                        api_result.payload.result,
+                        circuit,
+                        shots=self.options.shots,
+                        memory=self.options.memory,
+                    )
+                )
+
+                progress_bar.update(1)
+            else:  # no circuits in the job, or all executed successfully
+                self._status = JobStatus.DONE
+                return Result.from_dict(result)
+
+        self._status = JobStatus.ERROR
+        result["success"] = False
+        return Result.from_dict(result)
+
+    def status(self) -> JobStatus:
+        """Query the job's status.
+
+        Returns:
+            Aggregated job status for all the circuits in this job.
+        """
+        return self._status
+
+
+def _partial_qiskit_result_dict(
+    samples: list[list[int]], circuit: QuantumCircuit, *, shots: int, memory: bool
+) -> dict[str, Any]:
+    """Build the Qiskit result dict for a single circuit evaluation.
+
+    Args:
+        samples: measurement outcome of the circuit evaluation.
+        circuit: the evaluated circuit.
+        shots: number of repetitions of the circuit evaluation.
+        memory: whether to fill the classical memory dump field with the measurement results.
+
+    Returns:
+        Dict, suitable for Qiskit's `Result.from_dict` factory.
+    """
+    meas_map = _build_memory_mapping(circuit)
+
+    data: dict[str, Any] = {"counts": _format_counts(samples, meas_map)}
+
+    if memory:
+        data["memory"] = ["".join(str(x) for x in reversed(states)) for states in samples]
+
+    return {
+        "shots": shots,
+        "success": True,
+        "status": JobStatus.DONE,
+        "data": data,
+        "header": {
+            "memory_slots": circuit.num_clbits,
+            "creg_sizes": [[reg.name, reg.size] for reg in circuit.cregs],
+            "qreg_sizes": [[reg.name, reg.size] for reg in circuit.qregs],
+            "name": circuit.name,
+            "metadata": circuit.metadata or {},
+        },
+    }
+
+
 def _build_memory_mapping(circuit: QuantumCircuit) -> dict[int, set[int]]:
     """Scan the circuit for measurement instructions and collect qubit to classical bits mappings.
 

qiskit_aqt_provider/aqt_options.py

@@ -24,7 +24,7 @@ class AQTOptions(pdt.BaseModel, Mapping[str, Any]):
 
     Options can be set on a backend globally or on a per-job basis. To update an option
     globally, set the corresponding attribute in the backend's
-    :attr:`options <qiskit_aqt_provider.aqt_resource.AQTResource.options>` attribute:
+    `options <qiskit_aqt_provider.aqt_resource._ResourceBase.options>` attribute:
 
     >>> import qiskit
     >>> from qiskit_aqt_provider import AQTProvider
@@ -42,7 +42,8 @@ class AQTOptions(pdt.BaseModel, Mapping[str, Any]):
     50
 
     Option overrides can also be applied on a per-job basis, as keyword arguments to
-    :meth:`AQTResource.run <qiskit_aqt_provider.aqt_resource.AQTResource.run>`:
+    :meth:`AQTResource.run <qiskit_aqt_provider.aqt_resource.AQTResource.run>` or
+    :meth:`AQTDirectAccessResource.run <qiskit_aqt_provider.aqt_resource.AQTDirectAccessResource.run>`:
 
     >>> backend.options.shots
     50