diff --git a/qiskit_ibm_provider/ibm_backend.py b/qiskit_ibm_provider/ibm_backend.py index 3dba0a133..ca4dbdd07 100644 --- a/qiskit_ibm_provider/ibm_backend.py +++ b/qiskit_ibm_provider/ibm_backend.py @@ -58,11 +58,12 @@ ) from .utils import validate_job_tags from .utils.options import QASM2Options, QASM3Options -from .utils.backend_converter import ( - convert_to_target, -) from .utils.converters import local_to_utc -from .utils.json_decoder import defaults_from_server_data, properties_from_server_data +from .utils.json_decoder import ( + defaults_from_server_data, + properties_from_server_data, + target_from_server_data, +) from .api.exceptions import RequestsApiError @@ -237,11 +238,6 @@ def __getattr__(self, name: str) -> Any: self.__class__.__name__, name ) ) - # Lazy load properties and pulse defaults and construct the target object. - self._get_properties() - self._get_defaults() - self._convert_to_target() - # Check if the attribute now is available on IBMBackend class due to above steps try: return super().__getattribute__(name) except AttributeError: @@ -257,35 +253,32 @@ def __getattr__(self, name: str) -> Any: ) ) - def _get_properties(self, datetime: Optional[python_datetime] = None) -> None: - """Gets backend properties and decodes it""" - if not self._properties: - if datetime: - datetime = local_to_utc(datetime) - api_properties = self.provider._runtime_client.backend_properties( - self.name, datetime=datetime - ) - if api_properties: - backend_properties = properties_from_server_data(api_properties) - self._properties = backend_properties - - def _get_defaults(self) -> None: - """Gets defaults if pulse backend and decodes it""" - if not self._defaults: - api_defaults = self.provider._runtime_client.backend_pulse_defaults( - self.name - ) - if api_defaults: - self._defaults = defaults_from_server_data(api_defaults) + def _get_target( + self, + *, + datetime: Optional[python_datetime] = None, + refresh: bool = False, + ) -> Target: + """Gets target from configuration, properties and pulse defaults.""" + if datetime: + if not isinstance(datetime, python_datetime): + raise TypeError("'{}' is not of type 'datetime'.") + datetime = local_to_utc(datetime) - def _convert_to_target(self) -> None: - """Converts backend configuration, properties and defaults to Target object""" - if not self._target: - self._target = convert_to_target( + if datetime or refresh or self._target is None: + client = getattr(self.provider, "_runtime_client") + api_properties = client.backend_properties(self.name, datetime=datetime) + api_pulse_defaults = client.backend_pulse_defaults(self.name) + target = target_from_server_data( configuration=self._configuration, - properties=self._properties, - defaults=self._defaults, + pulse_defaults=api_pulse_defaults, + properties=api_properties, ) + if datetime: + # Don't cache result. + return target + self._target = target + return self._target @classmethod def _default_options(cls) -> Options: @@ -324,20 +317,14 @@ def target(self) -> Target: Returns: Target """ - self._get_properties() - self._get_defaults() - self._convert_to_target() - return self._target + return self._get_target() def target_history(self, datetime: Optional[python_datetime] = None) -> Target: """A :class:`qiskit.transpiler.Target` object for the backend. Returns: Target with properties found on `datetime` """ - self._get_properties(datetime=datetime) - self._get_defaults() - self._convert_to_target() - return self._target + return self._get_target(datetime=datetime) def run( self, diff --git a/qiskit_ibm_provider/utils/backend_converter.py b/qiskit_ibm_provider/utils/backend_converter.py deleted file mode 100644 index fe1052787..000000000 --- a/qiskit_ibm_provider/utils/backend_converter.py +++ /dev/null @@ -1,198 +0,0 @@ -# This code is part of Qiskit. -# -# (C) Copyright IBM 2022. -# -# 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. - -"""Converters for migration from IBM Quantum BackendV1 to BackendV2.""" - -from typing import Any, Dict, List - -from qiskit.transpiler.target import Target, InstructionProperties -from qiskit.utils.units import apply_prefix -from qiskit.circuit.library.standard_gates import ( - IGate, - SXGate, - XGate, - CXGate, - RZGate, - ECRGate, - CZGate, -) -from qiskit.circuit import IfElseOp, WhileLoopOp, ForLoopOp -from qiskit.circuit.parameter import Parameter -from qiskit.circuit.gate import Gate -from qiskit.circuit.delay import Delay -from qiskit.circuit.measure import Measure -from qiskit.circuit.reset import Reset -from qiskit.providers.models import ( - BackendConfiguration, - BackendProperties, - PulseDefaults, -) - -from ..ibm_qubit_properties import IBMQubitProperties - - -def convert_to_target( - configuration: BackendConfiguration, - properties: BackendProperties = None, - defaults: PulseDefaults = None, -) -> Target: - """Uses configuration, properties and pulse defaults - to construct and return Target class. - """ - name_mapping = { - "id": IGate(), - "sx": SXGate(), - "x": XGate(), - "cx": CXGate(), - "rz": RZGate(Parameter("λ")), - "reset": Reset(), - "ecr": ECRGate(), - "cz": CZGate(), - } - control_flow_map = { - "if_else": IfElseOp, - "while_loop": WhileLoopOp, - "for_loop": ForLoopOp, - } - custom_gates = {} - target = None - # Parse from properties if it exsits - if properties is not None: - qubit_properties = qubit_props_list_from_props(properties=properties) - target = Target( - num_qubits=configuration.n_qubits, qubit_properties=qubit_properties - ) - # Parse instructions - gates: Dict[str, Any] = {} - for gate in properties.gates: - name = gate.gate - if name in name_mapping: - if name not in gates: - gates[name] = {} - elif name not in custom_gates: - custom_gate = Gate(name, len(gate.qubits), []) - custom_gates[name] = custom_gate - gates[name] = {} - - qubits = tuple(gate.qubits) - gate_props = {} - for param in gate.parameters: - if param.name == "gate_error": - gate_props["error"] = param.value - if param.name == "gate_length": - gate_props["duration"] = apply_prefix(param.value, param.unit) - gates[name][qubits] = InstructionProperties(**gate_props) - for gate, props in gates.items(): - if gate in name_mapping: - inst = name_mapping.get(gate) - else: - inst = custom_gates[gate] - target.add_instruction(inst, props) - # Create measurement instructions: - measure_props = {} - for qubit, _ in enumerate(properties.qubits): - measure_props[(qubit,)] = InstructionProperties( - duration=properties.readout_length(qubit), - error=properties.readout_error(qubit), - ) - target.add_instruction(Measure(), measure_props) - # Parse from configuration because properties doesn't exist - else: - target = Target(num_qubits=configuration.n_qubits) - for gate in configuration.gates: - name = gate.name - gate_props = ( - {tuple(x): None for x in gate.coupling_map} # type: ignore[misc] - if hasattr(gate, "coupling_map") - else {None: None} - ) - gate_len = len(gate.coupling_map[0]) if hasattr(gate, "coupling_map") else 0 - if name in name_mapping: - target.add_instruction(name_mapping[name], gate_props) - else: - custom_gate = Gate(name, gate_len, []) - target.add_instruction(custom_gate, gate_props) - target.add_instruction(Measure()) - # parse global configuration properties - if hasattr(configuration, "dt"): - target.dt = configuration.dt - if hasattr(configuration, "timing_constraints"): - target.granularity = configuration.timing_constraints.get("granularity") - target.min_length = configuration.timing_constraints.get("min_length") - target.pulse_alignment = configuration.timing_constraints.get("pulse_alignment") - target.aquire_alignment = configuration.timing_constraints.get( - "acquire_alignment" - ) - # If a pulse defaults exists use that as the source of truth - if defaults is not None: - inst_map = defaults.instruction_schedule_map - for inst in inst_map.instructions: - for qarg in inst_map.qubits_with_instruction(inst): - sched = inst_map.get(inst, qarg) - if inst in target: - try: - qarg = tuple(qarg) - except TypeError: - qarg = (qarg,) - if inst == "measure": - for qubit in qarg: - target[inst][(qubit,)].calibration = sched - else: - target[inst][qarg].calibration = sched - if "delay" not in target: - target.add_instruction( - Delay(Parameter("t")), {(bit,): None for bit in range(target.num_qubits)} - ) - # Handle control flow opeartions as globally support in the target - for control_flow_op_name, op_class in control_flow_map.items(): - if ( - control_flow_op_name in getattr(configuration, "supported_instructions", {}) - or control_flow_op_name in configuration.basis_gates - ): - target.add_instruction(op_class, name=control_flow_op_name) - - return target - - -def qubit_props_list_from_props( - properties: BackendProperties, -) -> List[IBMQubitProperties]: - """Uses BackendProperties to construct - and return a list of IBMQubitProperties. - """ - qubit_props: List[IBMQubitProperties] = [] - for qubit, _ in enumerate(properties.qubits): - try: - t_1 = properties.t1(qubit) - except Exception: # pylint: disable=broad-except - t_1 = None - try: - t_2 = properties.t2(qubit) - except Exception: # pylint: disable=broad-except - t_2 = None - try: - frequency = properties.frequency(qubit) - except Exception: # pylint: disable=broad-except - t_2 = None - try: - anharmonicity = properties.qubit_property(qubit, "anharmonicity")[0] - except Exception: # pylint: disable=broad-except - anharmonicity = None - qubit_props.append( - IBMQubitProperties( # type: ignore[no-untyped-call] - t1=t_1, - t2=t_2, - frequency=frequency, - anharmonicity=anharmonicity, - ) - ) - return qubit_props diff --git a/qiskit_ibm_provider/utils/json_decoder.py b/qiskit_ibm_provider/utils/json_decoder.py index 78b3ede2b..6d7e8646e 100644 --- a/qiskit_ibm_provider/utils/json_decoder.py +++ b/qiskit_ibm_provider/utils/json_decoder.py @@ -11,17 +11,33 @@ # that they have been altered from the originals. """Custom JSON decoder.""" - -from typing import Dict, Union, List, Any +from typing import Dict, Tuple, Union, List, Any, Optional import json +import logging import dateutil.parser from qiskit.providers.models import ( + QasmBackendConfiguration, + PulseBackendConfiguration, PulseDefaults, BackendProperties, + Command, ) +from qiskit.providers.models.backendproperties import Gate as GateSchema +from qiskit.circuit.controlflow import IfElseOp, WhileLoopOp, ForLoopOp +from qiskit.circuit.gate import Gate, Instruction +from qiskit.circuit.parameter import Parameter +from qiskit.circuit.library.standard_gates import get_standard_gate_name_mapping +from qiskit.pulse.calibration_entries import PulseQobjDef +from qiskit.transpiler.target import Target, InstructionProperties +from qiskit.qobj.pulse_qobj import PulseLibraryItem +from qiskit.qobj.converters.pulse_instruction import QobjToInstructionConverter +from qiskit.utils import apply_prefix from .converters import utc_to_local, utc_to_local_all +from ..ibm_qubit_properties import IBMQubitProperties + +logger = logging.getLogger(__name__) def defaults_from_server_data(defaults: Dict) -> PulseDefaults: @@ -69,6 +85,165 @@ def properties_from_server_data(properties: Dict) -> BackendProperties: return BackendProperties.from_dict(properties) +def target_from_server_data( + configuration: Union[QasmBackendConfiguration, PulseBackendConfiguration], + pulse_defaults: Optional[Dict] = None, + properties: Optional[Dict] = None, +) -> Target: + """Decode transpiler target from backend data set. + + This function directly generate ``Target`` instance without generate + intermediate legacy objects such as ``BackendProperties`` and ``PulseDefaults``. + + Args: + configuration: Backend configuration. + pulse_defaults: Backend pulse defaults dictionary. + properties: Backend property dictionary. + + Returns: + A ``Target`` instance. + """ + required = ["measure", "delay"] + + # Load Qiskit object representation + qiskit_inst_mapping = get_standard_gate_name_mapping() + qiskit_control_flow_mapping = { + "if_else": IfElseOp, + "while_loop": WhileLoopOp, + "for_loop": ForLoopOp, + } + + in_data = {"num_qubits": configuration.n_qubits} + + # Parse global configuration properties + if hasattr(configuration, "dt"): + in_data["dt"] = configuration.dt + if hasattr(configuration, "timing_constraints"): + in_data.update(configuration.timing_constraints) + + # Create instruction property placeholder from backend configuration + supported_instructions = set(getattr(configuration, "supported_instructions", [])) + basis_gates = set(getattr(configuration, "basis_gates", [])) + gate_configs = {gate.name: gate for gate in configuration.gates} + inst_name_map = {} # type: Dict[str, Instruction] + prop_name_map = {} # type: Dict[str, Dict[Tuple[int, ...], InstructionProperties]] + all_instructions = set.union(supported_instructions, basis_gates, set(required)) + + # Create name to Qiskit instruction object repr mapping + for name in all_instructions: + if name in qiskit_control_flow_mapping: + continue + if name in qiskit_inst_mapping: + inst_name_map[name] = qiskit_inst_mapping[name] + elif name in gate_configs: + this_config = gate_configs[name] + params = list(map(Parameter, getattr(this_config, "parameters", []))) + coupling_map = getattr(this_config, "coupling_map", []) + inst_name_map[name] = Gate( + name=name, + num_qubits=len(coupling_map[0]) if coupling_map else 0, + params=params, + ) + else: + logger.warning( + "Definition of instruction %s is not found in the Qiskit namespace and " + "GateConfig is not provided by the BackendConfiguration payload. " + "Qiskit Gate model cannot be instantiated for this instruction and " + "this instruction is silently excluded from the Target. " + "Please add new gate class to Qiskit or provide GateConfig for this name.", + name, + ) + all_instructions.remove(name) + continue + + # Create placeholder for the instruction properties + for name, spec in gate_configs.items(): + if hasattr(spec, "coupling_map"): + coupling_map = spec.coupling_map + prop_name_map[name] = dict.fromkeys(map(tuple, coupling_map)) + else: + prop_name_map[name] = None + if "delay" not in prop_name_map: + # Case for real IBM backend. They don't have delay in gate configuration. + prop_name_map["delay"] = {(q,): None for q in range(configuration.n_qubits)} + + # Populate instruction properties + if properties: + in_data["qubit_properties"] = list( + map(_decode_qubit_property, properties["qubits"]) + ) + for gate_spec in map(GateSchema.from_dict, properties["gates"]): + name = gate_spec.gate + qubits = tuple(gate_spec.qubits) + if name not in all_instructions: + logger.info( + "Gate property for instruction %s on qubits %s is found " + "in the BackendProperties payload. However, this gate is not included in the " + "basis_gates or supported_instructions, or maybe the gate model " + "is not defined in the Qiskit namespace. This gate is ignored.", + name, + qubits, + ) + continue + inst_prop = _decode_instruction_property(gate_spec) + if prop_name_map[name] is None: + prop_name_map[name] = {} + prop_name_map[name][qubits] = inst_prop + # Measure instruction property is stored in qubit property in IBM + measure_props = list(map(_decode_measure_property, properties["qubits"])) + prop_name_map["measure"] = {} + for qubit, measure_prop in enumerate(measure_props): + qubits = (qubit,) + prop_name_map["measure"][qubits] = measure_prop + + # Define pulse qobj converter and command sequence for lazy conversion + if pulse_defaults: + pulse_lib = list( + map(PulseLibraryItem.from_dict, pulse_defaults["pulse_library"]) + ) + converter = QobjToInstructionConverter(pulse_lib) + for cmd in map(Command.from_dict, pulse_defaults["cmd_def"]): + name = cmd.name + qubits = tuple(cmd.qubits) + if name not in all_instructions or qubits not in prop_name_map[name]: + logger.info( + "Gate calibration for instruction %s on qubits %s is found " + "in the PulseDefaults payload. However, this entry is not defined in " + "the gate mapping of Target. This calibration is ignored.", + name, + qubits, + ) + continue + entry = PulseQobjDef(converter=converter, name=cmd.name) + entry.define(cmd.sequence) + try: + prop_name_map[name][qubits].calibration = entry + except AttributeError: + logger.info( + "The PulseDefaults payload received contains an instruction %s on " + "qubits %s which is not present in the configuration or properties payload.", + name, + qubits, + ) + + # Add parsed properties to target + target = Target(**in_data) + for inst_name in all_instructions: + if inst_name in qiskit_control_flow_mapping: + # Control flow operator doesn't have gate property. + target.add_instruction( + instruction=qiskit_control_flow_mapping[inst_name], + name=inst_name, + ) + else: + target.add_instruction( + instruction=inst_name_map[inst_name], + properties=prop_name_map.get(inst_name, None), + ) + + return target + + def decode_pulse_qobj(pulse_qobj: Dict) -> None: """Decode a pulse Qobj. @@ -156,3 +331,61 @@ def _decode_pulse_qobj_instr(pulse_qobj_instr: Dict) -> None: pulse_qobj_instr["parameters"]["amp"] = _to_complex( pulse_qobj_instr["parameters"]["amp"] ) + + +def _decode_qubit_property(qubit_specs: List[Dict]) -> IBMQubitProperties: + """Decode qubit property data to generate IBMQubitProperty instance. + + Args: + qubit_specs: List of qubit property dictionary. + + Returns: + An ``IBMQubitProperty`` instance. + """ + in_data = {} + for spec in qubit_specs: + name = spec["name"] + if name in IBMQubitProperties.__slots__: + in_data[name] = apply_prefix( + value=spec["value"], unit=spec.get("unit", None) + ) + return IBMQubitProperties(**in_data) # type: ignore[no-untyped-call] + + +def _decode_instruction_property(gate_spec: GateSchema) -> InstructionProperties: + """Decode gate property data to generate InstructionProperties instance. + + Args: + gate_spec: List of gate property dictionary. + + Returns: + An ``InstructionProperties`` instance. + """ + in_data = {} + for param in gate_spec.parameters: + if param.name == "gate_error": + in_data["error"] = param.value + if param.name == "gate_length": + in_data["duration"] = apply_prefix(value=param.value, unit=param.unit) + return InstructionProperties(**in_data) + + +def _decode_measure_property(qubit_specs: List[Dict]) -> InstructionProperties: + """Decode qubit property data to generate InstructionProperties instance. + + Args: + qubit_specs: List of qubit property dictionary. + + Returns: + An ``InstructionProperties`` instance. + """ + in_data = {} + for spec in qubit_specs: + name = spec["name"] + if name == "readout_error": + in_data["error"] = spec["value"] + if name == "readout_length": + in_data["duration"] = apply_prefix( + value=spec["value"], unit=spec.get("unit", None) + ) + return InstructionProperties(**in_data)