Source code for c3.qiskit.c3_backend

import uuid
import time
import numpy as np
import logging
import warnings

from qiskit import qobj
from qiskit import QuantumCircuit
from qiskit.circuit import Instruction
from qiskit.exceptions import QiskitError
from qiskit.providers import BackendV1 as Backend
from qiskit.providers.models import QasmBackendConfiguration
from qiskit.result import Result
from qiskit.compiler import assemble
from qiskit.qobj.qasm_qobj import QasmQobjExperiment
from qiskit.qobj.pulse_qobj import PulseQobj
import tensorflow as tf

from c3.experiment import Experiment
from c3.c3objs import Quantity as Qty

from .c3_exceptions import C3QiskitError
from .c3_job import C3Job
from .c3_backend_utils import get_init_ground_state
from .c3_options import C3Options

from typing import Any, Dict, List, Tuple
from abc import ABC, abstractclassmethod, abstractmethod

logger = logging.getLogger(__name__)

[docs]class C3QasmSimulator(Backend, ABC): """An Abtract Base Class for C3 Qasm Simulators for Qiskit. This class CAN NOT be instantiated directly. Classes derived from this must compulsorily implement :: def __init__(self, configuration=None, provider=None, **fields): def _default_options(cls) -> None: def run_experiment(self, experiment: QasmQobjExperiment) -> Dict[str, Any]: Parameters ---------- Backend : qiskit.providers.BackendV1 The C3QasmSimulator is derived from BackendV1 ABC : abc.ABC Helper class for defining Abstract classes using ABCMeta """ @abstractclassmethod def _default_options(cls) -> C3Options: raise NotImplementedError("This must be implemented in the derived class")
[docs] def set_device_config(self, config_file: str) -> None: """Set the path to the config for the device Parameters ---------- config_file : str path to hjson file storing the configuration for all device parameters for simulation """ self._device_config = config_file self.c3_exp.load_quick_setup(self._device_config)
[docs] def set_c3_experiment(self, exp: Experiment) -> None: """Set user-provided c3 experiment object for backend Parameters ---------- exp : Experiment C3 experiment object """ self.c3_exp = exp
def _setup_c3_experiment(self): """Setup C3 Experiment object for simulation Parameters ---------- experiment : QasmQobjExperiment Qasm Experiment object from qiskit """ exp = self.c3_exp exp.enable_qasm() pmap = exp.pmap # TODO (Check) Assume all qubits have same Hilbert dims self._number_of_levels = pmap.model.dims[0] # Validate the dimension of initial statevector if set self._validate_initial_statevector() # TODO set simulator seed, check qiskit python qasm simulator # qiskit-terra/qiskit/providers/basicaer/ self.seed_simulator = 2441129
[docs] def get_labels(self, format: str = "qiskit") -> List[str]: """Return state labels for the system Parameters ---------- format : str, optional How to format the state labels, by default "qiskit" Returns ------- List[str] A list of state labels in hex if qiskit format and decimal if c3 format :: labels = ['0x1', ...] labels = ['(0, 0)', '(0, 1)', '(0, 2)', ...] Raises ------ C3QiskitError When an supported format is passed """ if format == "qiskit": labels = [ hex(i) for i in range( 0, pow( self._number_of_levels, self._number_of_qubits, ), ) ] elif format == "c3": labels = [ "".join(str(label)) for label in self.c3_exp.pmap.model.state_labels ] else: raise C3QiskitError("Incorrect format specifier for get_labels") return labels
[docs] def disable_flip_labels(self) -> None: """Disable flipping of labels State Labels are flipped before returning results to match Qiskit style qubit indexing convention This function allows disabling of the flip """ self._flip_labels = False
[docs] def locate_measurements(self, instructions_list: List[Dict]) -> List[int]: """Locate the indices of measurement operations in circuit Parameters ---------- instructions_list : List[Dict] Instructions List in Qasm style Returns ------- List[int] The indices where measurement operations occur """ meas_index: List[int] = [] meas_index = [ index for index, instruction in enumerate(instructions_list) if instruction["name"] == "measure" ] return meas_index
[docs] def sanitize_instructions( self, instructions: Instruction ) -> Tuple[List[Any], List[Any]]: """Convert from qiskit instruction object and Sanitize instructions by removing unsupported operations Parameters ---------- instructions : Instruction qasm as Qiskit Instruction object Returns ------- Tuple[List[Any], List[Any]] Sanitized instruction list Qasm style list of instruction represented as dicts Raises ------- UserWarning Warns user about unsupported operations in circuit """ instructions_list = [instruction.to_dict() for instruction in instructions] sanitized_instructions = [ instruction for instruction in instructions_list if instruction["name"] not in self.UNSUPPORTED_OPERATIONS ] if sanitized_instructions != instructions_list: warnings.warn( f"The following operations are not supported yet: {self.UNSUPPORTED_OPERATIONS}" ) return sanitized_instructions, instructions_list
[docs] def generate_shot_readout(self): """Generate shot style readout from population Returns ------- List[int] List of shots for each output state """ # TODO Further harmonize readout generation for perfect and physics simulation # TODO a sophisticated readout/measurement routine (w/ SPAM) return (np.round(self.pops_array * self._shots)).astype("int32").tolist()
[docs] def run(self, qobj: qobj.Qobj, **backend_options) -> C3Job: """Parse and run a Qobj Parameters ---------- qobj : Qobj The Qobj payload for the experiment backend_options : dict backend options Returns ------- C3Job An instance of the C3Job (derived from JobV1) with the result Raises ------ QiskitError Support for Pulse Jobs is not implemented Notes ----- backend_options: Is a dict of options for the backend. It may contain:: "initial_statevector": vector_like The ``initial_statevector`` option specifies a custom initial statevector for the simulator to be used instead of the all zero state. This size of this vector must be correct for the number of qubits in all experiments in the qobj. Example:: backend_options = { "initial_statevector": np.array([1, 0, 0, 1j]) / np.sqrt(2), } :: "params": list List of parameter values. Can be nested, if a parameter is matrix valued. :: "opt_map": list Corresponding identifiers for the parameter values. :: "shots": int Total number of measurement shots. :: "memory": bool Whether individual measurement readout is stored. """ if isinstance(qobj, (QuantumCircuit, list)): qobj = assemble(qobj, self, **backend_options) qobj_options = qobj.config elif isinstance(qobj, PulseQobj): raise QiskitError("Pulse jobs are not accepted") else: qobj_options = qobj.config self._set_options(qobj_config=qobj_options, backend_options=backend_options) job_id = str(uuid.uuid4()) job = C3Job(self, job_id, self._run_job(job_id, qobj)) return job
def _run_job(self, job_id, qobj): """Run experiments in qobj Parameters ---------- job_id : str unique id for the job qobj : Qobj job description Returns ------- Result Result object """ self._validate(qobj) result_list = [] self._shots = qobj.config.shots self._memory = getattr(qobj.config, "memory", False) self._qobj_config = qobj.config if not hasattr(self.c3_exp.pmap, "model"): raise C3QiskitError( "Experiment Object has not been correctly initialised. \nUse set_device_config() or set_c3_experiment()" ) start = time.time() self._setup_c3_experiment() # runtime options for parameter update override gate-based updated if self.options.get("params"): params = self.options.get("params") opt_map = self.options.get("opt_map") if not opt_map: raise KeyError( "Missing opt_map in options to run(), required for updating parameters" ) # Reset options to ensure this isn't reused in next call to self.options.params = None self.options.opt_map = None self.c3_exp.pmap.set_parameters(params, opt_map) self.c3_exp.compute_propagators() for experiment in qobj.experiments: result_list.append(self.run_experiment(experiment)) end = time.time() result = { "backend_name":, "backend_version": self._configuration.backend_version, "qobj_id": qobj.qobj_id, "job_id": job_id, "results": result_list, "status": "COMPLETED", "success": True, "time_taken": (end - start), "header": qobj.header.to_dict(), } return Result.from_dict(result)
[docs] @abstractmethod def run_experiment(self, experiment: QasmQobjExperiment) -> Dict[str, Any]: raise NotImplementedError("This must be implemented in the derived class")
def _validate(self, qobj): """Semantic validations of the qobj which cannot be done via schemas.""" n_qubits = qobj.config.n_qubits max_qubits = self.configuration().n_qubits if n_qubits > max_qubits: raise C3QiskitError( "Number of qubits {} ".format(n_qubits) + "is greater than maximum ({}) ".format(max_qubits) + 'for "{}".'.format( ) def _validate_initial_statevector(self): """Check initial statevector has correct dimensions and is normalised. Then transform it to a tf Tensor of correct shape Raises ------ C3QiskitError If statevector is not correctly initialised """ if self._initial_statevector is not None: psi_init = self._initial_statevector # check dimensions ref_state = np.squeeze( get_init_ground_state( self._number_of_qubits, self._number_of_levels ).numpy() ) if psi_init.shape != ref_state.shape: raise C3QiskitError( f"Initial statevector has dimensions {psi_init.shape}, expected {ref_state.shape}" ) # check normalisation norm = np.linalg.norm(psi_init) if round(norm, 12) != 1: raise C3QiskitError( f"Initial statevector is not normalized: norm {norm} != 1" ) # convert numpy array to tf Tensor # reshape to make vector: (8,) -> (8, 1) self._initial_statevector = tf.reshape( tf.constant(psi_init, dtype=tf.complex128), [-1, 1] ) def _set_options(self, qobj_config=None, backend_options=None): """Set the backend options for all experiments in a qobj""" # set runtime options for field in backend_options: if not hasattr(self._options, field): raise AttributeError( "Options field %s is not valid for this backend" % field ) self._options.update_options(**backend_options) self._initial_statevector = self.options.get("initial_statevector") # Check for custom initial statevector in backend_options first, # then config second if "initial_statevector" in backend_options: self._initial_statevector = np.array( backend_options["initial_statevector"], dtype=complex ) elif hasattr(qobj_config, "initial_statevector"): self._initial_statevector = np.array( qobj_config.initial_statevector, dtype=complex )
[docs]class C3QasmPerfectSimulator(C3QasmSimulator): """A C3-based perfect gates simulator for Qiskit Parameters ---------- C3QasmSimulator : c3.qiskit.c3_backend.C3QasmSimulator Inherits the C3QasmSimulator and implements a perfect gate simulator """ # TODO Implement gate level simulator def __init__(self, configuration=None, provider=None, **fields): raise NotImplementedError
[docs]class C3QasmPhysicsSimulator(C3QasmSimulator): """A C3-based perfect gates simulator for Qiskit Parameters ---------- C3QasmSimulator : c3.qiskit.c3_backend.C3QasmSimulator Inherits the C3QasmSimulator and implements a physics based simulator """ MAX_QUBITS_MEMORY = 10 _configuration = { "backend_name": "c3_qasm_physics_simulator", "backend_version": "0.1", "n_qubits": MAX_QUBITS_MEMORY, "url": "", "simulator": True, "local": True, "conditional": False, "open_pulse": False, "memory": False, "max_shots": 65536, "coupling_map": None, # TODO Coupling map from config file "description": "A physics based c3 simulator for qasm experiments", "basis_gates": [ "cx", "rx", ], # TODO Basis gates from config file "gates": [], } def __init__(self, configuration=None, provider=None, **fields): super().__init__( configuration=( configuration or QasmBackendConfiguration.from_dict(self._configuration) ), provider=provider, **fields, ) # Define attributes in __init__. self._local_random = np.random.RandomState() self._classical_memory = 0 self._classical_register = 0 self._statevector = 0 self._number_of_cmembits = 0 self._number_of_qubits = 0 self._shots = 0 self._memory = False self._initial_statevector = self.options.get("initial_statevector") self._qobj_config = None self._sample_measure = False self.UNSUPPORTED_OPERATIONS = ["measure", "barrier"] self.c3_exp = Experiment() @classmethod def _default_options(cls) -> C3Options: DEFAULT_OPTIONS = { "params": None, "opt_map": None, "shots": 1024, "memory": False, "initial_statevector": None, } return C3Options(**DEFAULT_OPTIONS)
[docs] def run_experiment(self, experiment: QasmQobjExperiment) -> Dict[str, Any]: """Run an experiment (circuit) and return a single experiment result Parameters ---------- experiment : QasmQobjExperiment experiment from qobj experiments list Returns ------- Dict[str, Any] A result dictionary which looks something like:: { "name": name of this experiment (obtained from qobj.experiment header) "seed": random seed used for simulation "shots": number of shots used in the simulation "data": { "counts": {'0x9: 5, ...}, "memory": ['0x9', '0xF', '0x1D', ..., '0x9'] }, "status": status string for the simulation "success": boolean "time_taken": simulation time of this single experiment } Raises ------ C3QiskitError If an error occured """ start = time.time() exp = self.c3_exp # check qubit count is adequate self._number_of_qubits = len(exp.pmap.model.subsystems) if self._number_of_qubits < experiment.config.n_qubits: raise C3QiskitError("Not enough qubits on device to run circuit") sanitized_instructions, instructions_list = self.sanitize_instructions( experiment.instructions ) # process gate that updates parameters, this should only be at the end of the circuit if sanitized_instructions[-1]["name"] == "param_update": gate = sanitized_instructions.pop(-1) param_values = gate["params"][0] param_qtys = [Qty(**param) for param in param_values] opt_map = gate["params"][1] exp.pmap.set_parameters(param_qtys, opt_map) exp.compute_propagators() try: pops = exp.evaluate([sanitized_instructions], self._initial_statevector) pop1s, _ = exp.process(pops) except KeyError as err: print(f"KeyError: {err}") raise C3QiskitError( "Possibly an unsupported gate or a SetParamsGate not at the end." ) # C3 stores labels in exp.pmap.model.state_labels meas_index = self.locate_measurements(instructions_list) self.pops_array = pop1s[0].numpy() if meas_index: counts_data = self.generate_shot_readout() else: counts_data = self.pops_array.tolist() counts = dict(zip(self.get_labels(format="qiskit"), counts_data)) state_pops = dict(zip(self.get_labels(format="c3"), self.pops_array.tolist())) # flipping state labels to match qiskit style qubit indexing convention # default is to flip labels to qiskit style, use disable_flip_labels() # if self._flip_labels: # counts = flip_labels(counts) end = time.time() exp_result = { "name":, "header": experiment.header.to_dict(), "shots": self._shots, "seed": self.seed_simulator, "status": "DONE", "success": True, "data": { "counts": counts, "state_pops": state_pops, }, "time_taken": (end - start), } return exp_result