Source code for pyquil.experiment._main

# Copyright 2016-2019 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
#    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.
Schema definition of a Experiment, which is a collection of ExperimentSetting
objects and a main program body (or ansatz). This schema is widely useful for defining
and executing many common types of algorithms / applications, including state and process
tomography, and the variational quantum eigensolver.
import json
import logging
import warnings
from json import JSONEncoder
from typing import (

from pyquil.experiment._calibration import CalibrationMethod
from pyquil.experiment._memory import (
from pyquil.experiment._program import (
from pyquil.experiment._result import ExperimentResult
from pyquil.experiment._setting import ExperimentSetting, _OneQState, TensorProductState
from pyquil.experiment._symmetrization import SymmetrizationLevel
from pyquil.gates import RESET
from pyquil.paulis import PauliTerm, is_identity
from pyquil.quil import Program
from pyquil.quilbase import Reset, ResetQubit

log = logging.getLogger(__name__)

def _abbrev_program(program: Program, max_len: int = 10) -> str:
    """Create an abbreviated string representation of a Program.

    This will join all instructions onto a single line joined by '; '. If the number of
    instructions exceeds ``max_len``, some will be excluded from the string representation.
    program_lines = program.out().splitlines()
    if max_len is not None and len(program_lines) > max_len:
        first_n = max_len // 2
        last_n = max_len - first_n
        excluded = len(program_lines) - max_len
        program_lines = program_lines[:first_n] + [f"... {excluded} instrs not shown ..."] + program_lines[-last_n:]

    return "   " + "\n   ".join(program_lines)

def _remove_reset_from_program(program: Program) -> Program:
    Trim the RESET from a program because in measure_observables it is re-added.

    :param program: Program to remove RESET(s) from.
    :return: Trimmed Program.
    p = program.copy_everything_except_instructions()

    for inst in program:
        if not isinstance(inst, (Reset, ResetQubit)):

    return p

[docs]class Experiment: """ A tomography-like experiment. Many near-term quantum algorithms involve: - some limited state preparation - enacting a quantum process (like in tomography) or preparing a variational ansatz state (like in VQE) - measuring observables of the state. Where we typically use a large number of (state_prep, measure) pairs but keep the ansatz program consistent. This class stores the ansatz program as a :py:class:`~pyquil.Program` and maintains a list of :py:class:`ExperimentSetting` objects which each represent a (state_prep, measure) pair. Settings diagonalized by a shared tensor product basis (TPB) can (optionally) be estimated simultaneously. Therefore, this class is backed by a list of list of ExperimentSettings. Settings sharing an inner list will be estimated simultaneously. If you don't want this, provide a list of length-1-lists. As a convenience, if you pass a 1D list to the constructor will expand it to a list of length-1-lists. This class will not group settings for you. Please see :py:func:`group_experiments` for a function that will automatically process a Experiment to group Experiments sharing a TPB. :ivar settings: The collection of ExperimentSetting objects that define this experiment. :ivar program: The main program body of this experiment. Also determines the ``shots`` and ``reset`` instance variables. The ``shots`` instance variable is the number of shots to take per ExperimentSetting. The ``reset`` instance variable is whether to actively reset qubits instead of waiting several times the coherence length for qubits to decay to ``|0>`` naturally. Setting this to True is much faster but there is a ~1% error per qubit in the reset operation. Thermal noise from "traditional" reset is not routinely characterized but is of the same order. :ivar symmetrization: the level of readout symmetrization to perform for the estimation and optional calibration of each observable. The following integer levels, encapsulated in the ``SymmetrizationLevel`` integer enum, are currently supported: * -1 -- exhaustive symmetrization uses every possible combination of flips * 0 -- no symmetrization * 1 -- symmetrization using an orthogonal array (OA) with strength 1 * 2 -- symmetrization using an orthogonal array (OA) with strength 2 * 3 -- symmetrization using an orthogonal array (OA) with strength 3 Note that (default) exhaustive symmetrization requires a number of QPU calls exponential in the number of qubits in the union of the support of the observables in any group of settings in ``tomo_experiment``; the number of shots may need to be increased to accommodate this. see :py:func:`run_symmetrized_readout` in api._quantum_computer for more information. """ def __init__( self, settings: Union[List[ExperimentSetting], List[List[ExperimentSetting]]], program: Program, *, symmetrization: int = SymmetrizationLevel.EXHAUSTIVE, calibration: int = CalibrationMethod.PLUS_EIGENSTATE, ): if len(settings) == 0: s: List[List[ExperimentSetting]] = [] else: if isinstance(settings[0], ExperimentSetting): # convenience wrapping in lists of length 1 s = [[expt] for expt in cast(List[ExperimentSetting], settings)] else: s = cast(List[List[ExperimentSetting]], settings) self._settings = s self.program = program self.symmetrization = SymmetrizationLevel(symmetrization) if self.symmetrization != SymmetrizationLevel.EXHAUSTIVE: if type(calibration) == int and calibration != 0: warnings.warn( "Calibration is only supported for exhaustive symmetrization, " "thus setting self.calibration = 0 (CalibrationMethod.NONE)." ) self.calibration = CalibrationMethod.NONE else: self.calibration = CalibrationMethod(calibration) self.shots = self.program.num_shots if "RESET" in self.program.out(): self.reset = True self.program = _remove_reset_from_program(self.program) else: self.reset = False def __len__(self) -> int: return len(self._settings) def __getitem__(self, item: int) -> List[ExperimentSetting]: return self._settings[item] def __setitem__(self, key: int, value: List[ExperimentSetting]) -> None: self._settings[key] = value def __delitem__(self, key: int) -> None: self._settings.__delitem__(key) def __iter__(self) -> Generator[List[ExperimentSetting], None, None]: yield from self._settings def __reversed__(self) -> Generator[List[ExperimentSetting], None, None]: yield from reversed(self._settings) def __contains__(self, item: List[ExperimentSetting]) -> bool: return item in self._settings
[docs] def append(self, expts: Union[ExperimentSetting, List[ExperimentSetting]]) -> None: if not isinstance(expts, list): expts = [expts] self._settings.append(expts)
[docs] def count(self, expt: List[ExperimentSetting]) -> int: return self._settings.count(expt)
[docs] def index(self, expt: List[ExperimentSetting], start: int = 0, stop: int = 0) -> int: return self._settings.index(expt, start, stop)
[docs] def extend(self, expts: List[List[ExperimentSetting]]) -> None: self._settings.extend(expts)
[docs] def insert(self, index: int, expt: List[ExperimentSetting]) -> None: self._settings.insert(index, expt)
[docs] def pop(self, index: int = 0) -> List[ExperimentSetting]: return self._settings.pop(index)
[docs] def remove(self, expt: List[ExperimentSetting]) -> None: self._settings.remove(expt)
[docs] def reverse(self) -> None: self._settings.reverse()
[docs] def sort(self, key: Optional[Callable[[List[ExperimentSetting]], Any]] = None, reverse: bool = False) -> None: return self._settings.sort(key=key, reverse=reverse)
[docs] def setting_strings(self) -> Generator[str, None, None]: yield from ( "{i}: {st_str}".format(i=i, st_str=", ".join(str(setting) for setting in settings)) for i, settings in enumerate(self._settings) )
[docs] def settings_string(self, abbrev_after: Optional[int] = None) -> str: setting_strs = list(self.setting_strings()) if abbrev_after is not None and len(setting_strs) > abbrev_after: first_n = abbrev_after // 2 last_n = abbrev_after - first_n excluded = len(setting_strs) - abbrev_after setting_strs = setting_strs[:first_n] + [f"... {excluded} settings not shown ..."] + setting_strs[-last_n:] return " " + "\n ".join(setting_strs)
def __repr__(self) -> str: string = f"shots: {self.shots}\n" if self.reset: string += "active reset: enabled\n" else: string += "active reset: disabled\n" string += f"symmetrization: {self.symmetrization} ({})\n" string += f"calibration: {self.calibration} ({})\n" string += f"program:\n{_abbrev_program(self.program)}\n" string += f"settings:\n{self.settings_string(abbrev_after=20)}" return string
[docs] def serializable(self) -> Dict[str, Any]: return { "type": "Experiment", "settings": self._settings, "program": self.program.out(), "symmetrization": self.symmetrization, "shots": self.shots, "reset": self.reset, }
def __eq__(self, other: object) -> bool: if not isinstance(other, Experiment): return False return self.serializable() == other.serializable()
[docs] def get_meas_qubits(self) -> List[int]: """ Return the sorted list of qubits that are involved in the all the out_operators of the settings for this ``Experiment`` object. """ meas_qubits: Set[int] = set() for settings in self: assert len(settings) == 1 meas_qubits.update(cast(List[int], settings[0].out_operator.get_qubits())) return sorted(meas_qubits)
[docs] def get_meas_registers(self, qubits: Optional[Sequence[int]] = None) -> List[int]: """ Return the sorted list of memory registers corresponding to the list of qubits provided. If no qubits are provided, just returns the list of numbers from 0 to n-1 where n is the number of qubits resulting from the ``get_meas_qubits`` method. """ meas_qubits = self.get_meas_qubits() if qubits is None: return list(range(len(meas_qubits))) meas_registers = [] for q in qubits: meas_registers.append(meas_qubits.index(q)) return sorted(meas_registers)
[docs] def generate_experiment_program(self) -> Program: """ Generate a parameterized program containing the main body program along with some additions to support the various state preparation, measurement, and symmetrization specifications of this ``Experiment``. State preparation and measurement are achieved via ZXZXZ-decomposed single-qubit gates, where the angles of each ``RZ`` rotation are declared parameters that can be assigned at runtime. Symmetrization is achieved by putting an ``RX`` gate (also parameterized by a declared value) before each ``MEASURE`` operation. In addition, a ``RESET`` operation is prepended to the ``Program`` if the experiment has active qubit reset enabled. Finally, each qubit specified in the settings is measured, and the number of shots is added. :return: Parameterized ``Program`` that is capable of collecting statistics for every ``ExperimentSetting`` in this ``Experiment``. """ meas_qubits = self.get_meas_qubits() p = Program() if self.reset: if any(isinstance(instr, (Reset, ResetQubit)) for instr in self.program): raise ValueError("RESET already added to program") p += RESET() for settings in self: assert len(settings) == 1 if ("X" in str(settings[0].in_state)) or ("Y" in str(settings[0].in_state)): if "DECLARE preparation_alpha" in self.program.out(): raise ValueError('Memory "preparation_alpha" has been declared already.') if "DECLARE preparation_beta" in self.program.out(): raise ValueError('Memory "preparation_beta" has been declared already.') if "DECLARE preparation_gamma" in self.program.out(): raise ValueError('Memory "preparation_gamma" has been declared already.') p += parameterized_single_qubit_state_preparation(meas_qubits) break p += self.program for settings in self: assert len(settings) == 1 if ("X" in str(settings[0].out_operator)) or ("Y" in str(settings[0].out_operator)): if "DECLARE measurement_alpha" in self.program.out(): raise ValueError('Memory "measurement_alpha" has been declared already.') if "DECLARE measurement_beta" in self.program.out(): raise ValueError('Memory "measurement_beta" has been declared already.') if "DECLARE measurement_gamma" in self.program.out(): raise ValueError('Memory "measurement_gamma" has been declared already.') p += parameterized_single_qubit_measurement_basis(meas_qubits) break if self.symmetrization != 0: if "DECLARE symmetrization" in self.program.out(): raise ValueError('Memory "symmetrization" has been declared already.') p += parameterized_readout_symmetrization(meas_qubits) if "DECLARE ro" in self.program.out(): raise ValueError('Memory "ro" has already been declared for this program.') p += measure_qubits(meas_qubits) p.wrap_in_numshots_loop(self.shots) return p
[docs] def build_setting_memory_map(self, setting: ExperimentSetting) -> Dict[str, List[float]]: """ Build the memory map corresponding to the state preparation and measurement specifications encoded in the provided ``ExperimentSetting``, taking into account the full set of qubits that are present in the ``Experiment`` object. :return: Memory map for state prep and measurement. """ meas_qubits = self.get_meas_qubits() in_pt = PauliTerm.from_list([(op, meas_qubits.index(cast(int, q))) for q, op in setting._in_operator()]) out_pt = PauliTerm.from_list([(op, meas_qubits.index(cast(int, q))) for q, op in setting.out_operator]) preparation_map = pauli_term_to_preparation_memory_map(in_pt) measurement_map = pauli_term_to_measurement_memory_map(out_pt) return {**preparation_map, **measurement_map}
[docs] def build_symmetrization_memory_maps( self, qubits: Sequence[int], label: str = "symmetrization" ) -> List[Dict[str, List[float]]]: """ Build a list of memory maps to be used in a program that is trying to perform readout symmetrization via parametric compilation. For example, if we have the following program: RX(symmetrization[0]) 0 RX(symmetrization[1]) 1 MEASURE 0 ro[0] MEASURE 1 ro[1] We can perform exhaustive readout symmetrization on our two qubits by providing the four following memory maps, and then appropriately flipping the resultant bitstrings: {'symmetrization': [0.0, 0.0]} -> XOR results with [0,0] {'symmetrization': [0.0, pi]} -> XOR results with [0,1] {'symmetrization': [pi, 0.0]} -> XOR results with [1,0] {'symmetrization': [pi, pi]} -> XOR results with [1,1] :param qubits: List of qubits to symmetrize readout for. :param label: Name of the declared memory region. Defaults to "symmetrization". :return: List of memory maps that performs the desired level of symmetrization. """ num_meas_registers = len(self.get_meas_qubits()) symm_registers = self.get_meas_registers(qubits) if self.symmetrization == SymmetrizationLevel.NONE: return [{}] # TODO: add support for orthogonal arrays if self.symmetrization != SymmetrizationLevel.EXHAUSTIVE: raise ValueError("We only support exhaustive symmetrization for now.") import numpy as np import itertools assignments = itertools.product(np.array([0, np.pi]), repeat=len(symm_registers)) memory_maps = [] for a in assignments: zeros = np.zeros(num_meas_registers) for idx, r in enumerate(symm_registers): zeros[r] = a[idx] memory_maps.append({f"{label}": list(zeros)}) return memory_maps
[docs] def generate_calibration_experiment(self) -> "Experiment": """ Generate another ``Experiment`` object that can be used to calibrate the various multi-qubit observables involved in this ``Experiment``. This is achieved by preparing the plus-one (minus-one) eigenstate of each ``out_operator``, and measuring the resulting expectation value of the same ``out_operator``. Ideally, this would always give +1 (-1), but when symmetric readout error is present the effect is to scale the resultant expectations by some constant factor. Determining this scale factor is what we call *readout calibration*, and then the readout error in subsequent measurements can then be mitigated by simply dividing by the scale factor. :return: A new ``Experiment`` that can calibrate the readout error of all the observables involved in this experiment. """ if self.calibration != CalibrationMethod.PLUS_EIGENSTATE: raise ValueError('We currently only support the "plus eigenstate" calibration method.') calibration_settings = [] for settings in self: assert len(settings) == 1 calibration_settings.append( ExperimentSetting( in_state=_pauli_to_product_state(settings[0].out_operator), out_operator=settings[0].out_operator, additional_expectations=settings[0].additional_expectations, ) ) calibration_program = Program() if self.reset: calibration_program += RESET() calibration_program.wrap_in_numshots_loop(self.shots) if self.symmetrization != SymmetrizationLevel.EXHAUSTIVE: raise ValueError("We currently only support calibration for exhaustive symmetrization") return Experiment( settings=calibration_settings, program=calibration_program, symmetrization=SymmetrizationLevel.EXHAUSTIVE, calibration=CalibrationMethod.NONE, )
def _pauli_to_product_state(in_state: PauliTerm) -> TensorProductState: """ Convert a Pauli term to a TensorProductState. """ if is_identity(in_state): return TensorProductState() else: return TensorProductState( [ _OneQState(label=pauli_label, index=0, qubit=cast(int, qubit)) for qubit, pauli_label in in_state._ops.items() ] )
[docs]class OperatorEncoder(JSONEncoder):
[docs] def default(self, o: Any) -> Any: if isinstance(o, ExperimentSetting): return o.serializable() if isinstance(o, Experiment): return o.serializable() if isinstance(o, ExperimentResult): return o.serializable() return o
[docs]def to_json(fn: str, obj: Any) -> str: """ Convenience method to save pyquil.experiment objects as a JSON file. See :py:func:`read_json`. """ # Specify UTF-8 to guard against systems that default to an ASCII locale. with open(fn, "w", encoding="utf-8") as f: json.dump(obj, f, cls=OperatorEncoder, indent=2, ensure_ascii=False) return fn
def _operator_object_hook(obj: Mapping[str, Any]) -> Union[Mapping[str, Any], Experiment]: if "type" in obj and obj["type"] in ["Experiment", "TomographyExperiment"]: # I bet this doesn't work for grouped experiment settings settings = [[ExperimentSetting.from_str(s) for s in stt] for stt in obj["settings"]] p = Program(obj["program"]) p.wrap_in_numshots_loop(obj["shots"]) ex = Experiment(settings=settings, program=p, symmetrization=obj["symmetrization"]) ex.reset = obj["reset"] return ex return obj
[docs]def read_json(fn: str) -> Any: """ Convenience method to read pyquil.experiment objects from a JSON file. See :py:func:`to_json`. """ # Specify UTF-8 to guard against systems that default to an ASCII locale. with open(fn, encoding="utf-8") as f: return json.load(f, object_hook=_operator_object_hook)