New: Experimental QCS Client

pyproject.toml

@@ -28,7 +28,7 @@ lark = "^0.11.1"
 rpcq = "^3.10.0"
 networkx = "^2.5"
 importlib-metadata = { version = "^3.7.3", python = "<3.8" }
-qcs-api-client = ">=0.20.13,<0.22.0"
+qcs-api-client = "0.8.0.dev1457087914"
 retry = "^0.9.2"
 
 # latex extra

pyquil/experimental/__init__.py

@@ -0,0 +1,2 @@
+from ._program import ExperimentalProgram
+from .api import get_experimental_qc

pyquil/experimental/_program.py

@@ -0,0 +1,14 @@
+from typing import Optional
+from pyquil.quil import InstructionDesignator, Program
+
+
+class ExperimentalProgram(Program):
+    """
+    An ExperimentalProgram is identical to a Program except that ``num_shots`` is optional.
+    """
+
+    num_shots: Optional[int]
+
+    def __init__(self, *instructions: InstructionDesignator):
+        super().__init__(*instructions)
+        self.num_shots = None

pyquil/experimental/api/__init__.py

@@ -0,0 +1,3 @@
+from ._compiler import ExperimentalQPUCompiler
+from ._qpu import ExperimentalQPU
+from ._quantum_computer import ExperimentalQuantumComputer, get_experimental_qc

pyquil/experimental/api/_compiler.py

@@ -0,0 +1,81 @@
+from dataclasses import dataclass
+from pyquil.quilbase import Gate
+from typing import Dict, Iterator, Optional, cast
+
+from pyquil.api._compiler import rewrite_arithmetic
+from pyquil.experimental._program import ExperimentalProgram
+from pyquil.quilatom import ExpressionDesignator
+from qcs_api_client.client._configuration.configuration import QCSClientConfiguration
+from qcs_api_client.grpc.services.translation import TranslationStub, TranslateQuilToEncryptedControllerJobRequest
+from qcs_api_client.grpc.models.controller import EncryptedControllerJob
+from rpcq.messages import ParameterAref
+from pyquil.parser import parse
+
+
+@dataclass
+class ExperimentalExecutable:
+    job: EncryptedControllerJob
+
+    recalculation_table: Dict[ParameterAref, ExpressionDesignator]
+    """A mapping from memory references to the original gate arithmetic."""
+
+
+class ExperimentalQPUCompiler:
+    quantum_processor_id: str
+    _client_configuration: QCSClientConfiguration
+    _service_stub_cache: Optional[TranslationStub]
+    _timeout: Optional[int] = None
+
+    def __init__(
+        self,
+        *,
+        client_configuration: Optional[QCSClientConfiguration] = None,
+        quantum_processor_id: str,
+        timeout: Optional[int] = None,
+    ):
+        self.quantum_processor_id = quantum_processor_id
+        self._client_configuration = client_configuration or QCSClientConfiguration.load()
+        self._timeout = timeout
+        self._service_stub_cache = None
+
+    async def quil_to_native_quil(self, program: ExperimentalProgram):
+        raise NotImplementedError("compilation of quil to native quil is not yet supported for ExperimentalProgram")
+
+    async def native_quil_to_executable(self, native_quil_program: ExperimentalProgram) -> EncryptedControllerJob:
+        """
+        Compile the provided native quil program to an executable suitable for use on the
+        experimental backend.
+        """
+
+        # TODO: Expand calibrations within the program, and then remove calibrations and unused frames and waveforms
+
+        arithmetic_response = rewrite_arithmetic(native_quil_program)
+
+        request = TranslateQuilToEncryptedControllerJobRequest(
+            quantum_processor_id=self.quantum_processor_id,
+            quil_program=arithmetic_response.quil,
+            num_shots_value=native_quil_program.num_shots,
+        )
+        job = await self._get_service_stub().translate_quil_to_encrypted_controller_job(request)
+
+        return ExperimentalExecutable(
+            job=job,
+            recalculation_table={
+                mref: _to_expression(rule) for mref, rule in arithmetic_response.recalculation_table.items()
+            },
+        )
+
+    def _get_service_stub(self) -> Iterator[TranslationStub]:
+        """
+        Return a service stub targeting the configured
+        """
+        if self._service_stub_cache is None:
+            self._service_stub_cache = TranslationStub(url=self._client_configuration.profile.grpc_api_url)
+        return self._service_stub_cache
+
+
+def _to_expression(rule: str) -> ExpressionDesignator:
+    # We can only parse complete lines of Quil, so we wrap the arithmetic expression
+    # in a valid Quil instruction to parse it.
+    # TODO: This hack should be replaced after #687
+    return cast(ExpressionDesignator, cast(Gate, parse(f"RZ({rule}) 0")[0]).params[0])

pyquil/experimental/api/_qpu.py

@@ -0,0 +1,166 @@
+from contextlib import asynccontextmanager, contextmanager
+from dataclasses import dataclass
+
+import httpx
+from pyquil.quilatom import ExpressionDesignator
+from sys import executable
+from typing import Dict, Iterator, Optional, cast
+
+import numpy as np
+from rpcq.messages import ParameterAref
+
+from pyquil.api import EngagementManager
+from pyquil.api._qpu import QPU
+from pyquil.experimental.api._compiler import ExperimentalExecutable
+from qcs_api_client.models.endpoint import Endpoint
+from qcs_api_client.operations.asyncio import get_default_endpoint
+from qcs_api_client.client._configuration.configuration import QCSClientConfiguration
+from qcs_api_client.client.client import build_async_client
+from qcs_api_client.grpc.services.controller import ControllerStub
+from qcs_api_client.grpc.models.controller import ControllerJobExecutionResultStatus, DataValue, ReadoutValues
+
+
+@dataclass
+class ExperimentalQPUExecuteResponse:
+    job_id: str
+    """The ID of the job sent for execution."""
+
+    executable: ExperimentalExecutable
+    """The executable sent for execution."""
+
+
+@dataclass
+class ExperimentalQPUExecutionResult:
+    memory_values: Dict[str, np.ndarray]
+    """
+    Data values stored in named memory locations at the completion of the program.
+    """
+
+    readout_values: Dict[str, np.ndarray]
+    """
+    Data captured from the control hardware during capture operations on reference frames.
+    """
+
+    success: bool
+    """
+    Whether or not execution succeeded.
+    """
+
+
+class ExperimentalQPU:
+    _client_configuration: QCSClientConfiguration
+    _service_stub_cache: Optional[ControllerStub]
+    _quantum_processor_id: str
+    _timeout: Optional[int]
+
+    def __init__(
+        self,
+        *,
+        quantum_processor_id: str,
+        timeout: Optional[int] = None,
+        client_configuration: Optional[QCSClientConfiguration] = None,
+    ) -> None:
+        self._quantum_processor_id = quantum_processor_id
+        self._client_configuration = client_configuration or QCSClientConfiguration.load()
+        self._timeout = timeout
+        self._service_stub_cache = None
+
+    async def execute(self, executable: ExperimentalExecutable) -> ExperimentalQPUExecuteResponse:
+        assert isinstance(
+            executable, ExperimentalExecutable
+        ), "ExperimentalQPU#execute requires an ExperimentalExecutable. Create one with ExperimentalQuantumComputer#compile"
+
+        response = await self._get_service_stub().execute_encrypted_controller_job(job=executable.job)
+
+        return ExperimentalQPUExecuteResponse(job_id=response.job_id, executable=executable)
+
+    async def get_result(self, execute_response: ExperimentalQPUExecuteResponse) -> ExperimentalQPUExecutionResult:
+        """
+        Retrieve results from execution on the QPU.
+        """
+        response = await self._get_service_stub().get_controller_job_results(job_id=execute_response.job_id)
+
+        return ExperimentalQPUExecutionResult(
+            memory_values=_transform_memory_values(
+                response.result.memory_values, recalculation_table=execute_response.executable.recalculation_table
+            ),
+            readout_values=_transform_readout_values(response.result.readout_values),
+            success=response.result.status == ControllerJobExecutionResultStatus.SUCCESS,
+        )
+
+    async def run(self, executable: ExperimentalExecutable) -> ExperimentalQPUExecutionResult:
+        return await self.get_result(execute_response=self.execute(executable=executable))
+
+    async def _get_execution_url(self) -> str:
+        """
+        Return the URL to which to send requests for execution.
+
+        Return a cached result if stored; otherwise, query the QCS API and store the result.
+        """
+        async with self._qcs_client() as client:
+            response = await get_default_endpoint(client, quantum_processor_id=self._quantum_processor_id)
+            response.raise_for_status()
+            body = cast(Endpoint, response.parsed)
+            url = body.addresses.grpc
+            assert url is not None, f"no gRPC address found for quantum processor {self._quantum_processor_id}"
+
+    @asynccontextmanager
+    async def _qcs_client(self) -> Iterator[httpx.Client]:
+        async with build_async_client(configuration=self._client_configuration) as client:
+            yield client
+
+    def _get_service_stub(self) -> Iterator[ControllerStub]:
+        if self._service_stub_cache is None:
+            url = self._get_execution_url()
+            self._service_stub_cache = ControllerStub(url=url)
+        return self._service_stub_cache
+
+
+def _transform_memory_values(
+    memory_values: Dict[str, "DataValue"], recalculation_table: Dict[ParameterAref, ExpressionDesignator]
+) -> Dict[str, np.ndarray]:
+    """
+    Transform the memory values returned from the experimental backend into more ergonomic values
+    for use in Python: numpy arrays.
+
+    TODO: Use recalculation table to map values to the user-provided names. See ``pyquil.api._qpu.QPU``.
+    """
+
+    result = {}
+
+    for key, value in memory_values.items():
+        if value.binary is not None:
+            bytes = np.frombuffer(value.binary.data, dtype=np.uint8)
+
+            # TODO: only `unpackbits` if the memory datatype is `BIT`; return bytes if `OCTET`
+            # TODO: ensure bit order (endianness) is correct; set `bitorder` if not
+            # TODO: use `count` kwarg to handle bit arrays that are not a multiple of 8 in length
+            result[key] = np.unpackbits(bytes)
+        elif value.integer is not None:
+            result[key] = np.asarray(value.integer.data, dtype=np.int64)
+        elif value.real is not None:
+            result[key] = np.asarray(value.real.data, dtype=np.float64)
+        else:
+            RuntimeError(f"memory values for {key} were unset; expected binary, integer, or real")
+
+    return result
+
+
+def _transform_readout_values(readout_values: Dict[str, "ReadoutValues"]) -> Dict[str, np.ndarray]:
+    """
+    Transform the readout values returned from the experimental backend into more ergonomic values
+    for use in Python: numpy arrays.
+    """
+    result = {}
+
+    for key, value in readout_values.items():
+        if value.integer_values is not None:
+            result[key] = np.asarray(value.integer_values.values, dtype=np.int32)
+        elif value.complex_values is not None:
+            result[key] = np.fromiter(
+                (v.real + v.imaginary * 1j for v in value.complex_values.values), dtype=np.complex64
+            )
+        else:
+            RuntimeError(f"readout values for {key} were unset; expected integer or complex data")
+
+    return result

pyquil/experimental/api/_quantum_computer.py

@@ -0,0 +1,28 @@
+from typing import Optional
+from pyquil.api._quantum_computer import QuantumComputer
+from pyquil.experimental._program import ExperimentalProgram
+from pyquil.experimental.api._compiler import ExperimentalQPUCompiler, ExperimentalExecutable
+from pyquil.experimental.api._qpu import ExperimentalQPU, ExperimentalQPUExecutionResult
+
+
+class ExperimentalQuantumComputer:
+    _compiler: ExperimentalQPUCompiler
+    qam: ExperimentalQPU
+
+    def __init__(self, *, qam: ExperimentalQPU, compiler: ExperimentalQPUCompiler) -> None:
+        self.qam = qam
+        self.compiler = compiler
+
+    async def compile(self, program: ExperimentalProgram) -> ExperimentalExecutable:
+        return await self.compiler.native_quil_to_executable(native_quil_program=program)
+
+    async def run(self, executable: ExperimentalExecutable) -> ExperimentalQPUExecutionResult:
+        return await self.qam.run(executable)
+
+
+def get_experimental_qc(
+    quantum_processor_id: str, *, compilation_timeout: Optional[int] = None, execution_timeout: Optional[int] = None
+) -> ExperimentalQuantumComputer:
+    compiler = ExperimentalQPUCompiler(quantum_processor_id=quantum_processor_id, timeout=compilation_timeout)
+    qpu = ExperimentalQPU(quantum_processor_id=quantum_processor_id, timeout=execution_timeout)
+    return ExperimentalQuantumComputer(qam=qpu, compiler=compiler)