Source code for pyquil.api._compiler

##############################################################################
# Copyright 2016-2018 Rigetti Computing
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.
##############################################################################
import logging
import threading
from contextlib import contextmanager
from typing import Dict, Optional, cast, List, Iterator

import httpx
from qcs_api_client.client import QCSClientConfiguration
from qcs_api_client.models.translate_native_quil_to_encrypted_binary_request import (
    TranslateNativeQuilToEncryptedBinaryRequest,
)
from qcs_api_client.operations.sync import (
    translate_native_quil_to_encrypted_binary,
    get_quilt_calibrations,
)
from qcs_api_client.types import UNSET
from rpcq.messages import ParameterSpec

from pyquil.api._abstract_compiler import AbstractCompiler, QuantumExecutable, EncryptedProgram
from pyquil.api._qcs_client import qcs_client
from pyquil.api._rewrite_arithmetic import rewrite_arithmetic
from pyquil.parser import parse_program, parse
from pyquil.quantum_processor import AbstractQuantumProcessor
from pyquil.quil import Program
from pyquil.quilatom import MemoryReference, ExpressionDesignator
from pyquil.quilbase import Declare, Gate

_log = logging.getLogger(__name__)


class QPUCompilerNotRunning(Exception):
    pass


def parse_mref(val: str) -> MemoryReference:
    """ Parse a memory reference from its string representation. """
    val = val.strip()
    try:
        if val[-1] == "]":
            name, offset = val.split("[")
            return MemoryReference(name, int(offset[:-1]))
        else:
            return MemoryReference(val)
    except Exception:
        raise ValueError(f"Unable to parse memory reference {val}.")


def _collect_memory_descriptors(program: Program) -> Dict[str, ParameterSpec]:
    """Collect Declare instructions that are important for building the patch table.

    :return: A dictionary of variable names to specs about the declared region.
    """

    return {
        instr.name: ParameterSpec(type=instr.memory_type, length=instr.memory_size)
        for instr in program
        if isinstance(instr, Declare)
    }


[docs]class QPUCompiler(AbstractCompiler): """ Client to communicate with the compiler and translation service. """ _calibration_program_lock: threading.Lock def __init__( self, *, quantum_processor_id: str, quantum_processor: AbstractQuantumProcessor, timeout: float = 10.0, client_configuration: Optional[QCSClientConfiguration] = None, ) -> None: """ Instantiate a new QPU compiler client. :param quantum_processor_id: Processor to target. :param quantum_processor: Quantum processor to use as compilation target. :param timeout: Time limit for requests, in seconds. :param client_configuration: Optional client configuration. If none is provided, a default one will be loaded. """ super().__init__( quantum_processor=quantum_processor, timeout=timeout, client_configuration=client_configuration, ) self.quantum_processor_id = quantum_processor_id self._calibration_program: Optional[Program] = None self._calibration_program_lock = threading.Lock()
[docs] def native_quil_to_executable(self, nq_program: Program) -> QuantumExecutable: arithmetic_response = rewrite_arithmetic(nq_program) request = TranslateNativeQuilToEncryptedBinaryRequest( quil=arithmetic_response.quil, num_shots=nq_program.num_shots ) with self._qcs_client() as qcs_client: # type: httpx.Client response = translate_native_quil_to_encrypted_binary( client=qcs_client, quantum_processor_id=self.quantum_processor_id, json_body=request, ).parsed ro_sources = cast(List[List[str]], [] if response.ro_sources == UNSET else response.ro_sources) 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]) return EncryptedProgram( program=response.program, memory_descriptors=_collect_memory_descriptors(nq_program), ro_sources={parse_mref(mref): source for mref, source in ro_sources}, recalculation_table={ mref: to_expression(rule) for mref, rule in arithmetic_response.recalculation_table.items() }, _memory=nq_program._memory.copy(), )
def _fetch_calibration_program(self) -> Program: with self._qcs_client() as qcs_client: # type: httpx.Client response = get_quilt_calibrations(client=qcs_client, quantum_processor_id=self.quantum_processor_id).parsed return parse_program(response.quilt)
[docs] def get_calibration_program(self, force_refresh: bool = False) -> Program: """ Get the Quil-T calibration program associated with the underlying QPU. This will return a cached copy of the calibration program if present. Otherwise (or if forcing a refresh), it will fetch and store the calibration program from the QCS API. A calibration program contains a number of DEFCAL, DEFWAVEFORM, and DEFFRAME instructions. In sum, those instructions describe how a Quil-T program should be translated into analog instructions for execution on hardware. :param force_refresh: Whether or not to fetch a new calibration program before returning. :returns: A Program object containing the calibration definitions.""" if force_refresh or self._calibration_program is None: try: self._calibration_program = self._fetch_calibration_program() except Exception as ex: raise RuntimeError("Could not fetch calibrations") from ex assert self._calibration_program is not None return self._calibration_program
[docs] def reset(self) -> None: """ Reset the state of the QPUCompiler. """ super().reset() self._calibration_program = None
@contextmanager def _qcs_client(self) -> Iterator[httpx.Client]: with qcs_client( client_configuration=self._client_configuration, request_timeout=self._timeout ) as client: # type: httpx.Client yield client
[docs]class QVMCompiler(AbstractCompiler): """ Client to communicate with the compiler. """ def __init__( self, *, quantum_processor: AbstractQuantumProcessor, timeout: float = 10.0, client_configuration: Optional[QCSClientConfiguration] = None, ) -> None: """ Client to communicate with compiler. :param quantum_processor: Quantum processor to use as compilation target. :param timeout: Number of seconds to wait for a response from the client. :param client_configuration: Optional client configuration. If none is provided, a default one will be loaded. """ super().__init__( quantum_processor=quantum_processor, timeout=timeout, client_configuration=client_configuration, )
[docs] def native_quil_to_executable(self, nq_program: Program) -> QuantumExecutable: return nq_program