Implementation proposition for qiskit hardware API simplification

hardware_architecture.py

import logging
import pickle
import typing as ty
from pathlib import Path

import networkx as nx
from qiskit.circuit.quantumregister import Qubit
from qiskit.dagcircuit.dagcircuit import DAGNode

logger = logging.getLogger("IBMQHardwareArchitecture")


class HardwareArchitecture(nx.DiGraph):
    def __init__(self, incoming_graph_data=None, **kwargs):
        """Base class for hardware architecture.

        :param incoming_graph_data: forwarded to
            :py:method:`networkx.Digraph.__init__`.
        :param kwargs: forwarded to
            :py:method:`networkx.Digraph.__init__`.
        """
        super().__init__(incoming_graph_data, **kwargs)
        self._qubit_number = 0

    def _is_valid_qubit(self, qubit: int) -> bool:
        """Check if the given qubit index is in the hardware or not."""
        return qubit < self._qubit_number

    def add_qubit(self, **qubit_data) -> int:
        """Add a qubit to the hardware architecture.

        :param qubit_data: Any data related to the added qubit. For example,
            qubit decoherence times, single qubit operations error rate or
            execution time, ...
        :return: the index of the added qubit.
        """
        self.add_node(self._qubit_number, **qubit_data)
        self._qubit_number += 1
        return self._qubit_number

    def add_link(self, control: int, target: int, **link_data):
        """Add a directed link to the hardware architecture.

        The control and target indices should:
        1. not be equal.
        2. already be present in the hardware description.
        If one of control or target does not exist, calling this function
        will raise an exception.

        :param control: index of a qubit in the hardware.
        :param target: index of a qubit in the hardware.
        :return: the index of the added qubit.
        :raise RuntimeError: when control or target is the index of a qubit
            that is not in the hardware yet (i.e. no call to
            :py:method:`add_qubit` returned this index) or when control and
            target represent the same qubit.
        """
        if control == target:
            raise RuntimeError("Cannot add self-links to hardware architecture.")
        if not self._is_valid_qubit(control):
            raise RuntimeError(
                "The given control qubit is not registered in the hardware yet."
            )
        if not self._is_valid_qubit(target):
            raise RuntimeError(
                "The given target qubit is not registered in the hardware yet."
            )
        self.add_edge(control, target, **link_data)

    @property
    def qubit_number(self):
        return self._qubit_number


class IBMQHardwareArchitecture(HardwareArchitecture):
    @staticmethod
    def _get_value(value: float, unit: str):
        """Internal method to harmonise units.

        Default units are:
        - nanoseconds for time.
        - Hz for frequencies.
        """
        # We want time in nanoseconds
        if unit == "us":
            return value * 10 ** 3
        elif unit == "ns":
            return value

        # We want everything in Hertz
        elif unit == "GHz":
            return value * 10 ** 9

        elif unit == "":
            return value

        else:
            raise RuntimeError(f"Unsupported unit: '{unit}'")

    @staticmethod
    def _get_link_properties_dict(
        backend_properties, source: int, sink: int
    ) -> ty.Dict[str, float]:
        """Extract properties dictionnary from the backend properties object.

        The properties dictionnary returned by this method is composed of:
        - property names (gate_error, gate_length) as keys.
        - property values as values.

        :param backend_properties: result of calling the `properties()` method on
            one of the IBMQ backends.
        :param source: source of the link we are interested in.
        :param sink: sink of the link we are interested in.

        :raise RuntimeError: if the link (source, sink) does not exist.
        """
        for gate in backend_properties.gates:
            if gate.gate != "cx":
                continue
            if len(gate.qubits) != 2:
                continue
            if gate.qubits[0] == source and gate.qubits[1] == sink:
                # Return the dictionary
                properties = {
                    param.name: IBMQHardwareArchitecture._get_value(
                        param.value, param.unit
                    )
                    for param in gate.parameters
                }
                return properties
        # If we finish here this means that the link was not added to the hardware.
        raise RuntimeError(
            f'Trying to recover properties of link "{source} -> {sink}" that '
            "is not present in the hardware architecture."
        )

    @staticmethod
    def _get_qubit_properties_dict(
        backend_properties, qubit_index,
    ) -> ty.Dict[str, ty.Dict[str, float]]:
        """Extract properties dictionnary from the backend properties object.

        The properties dictionnary returned by this method is composed of:
        - single-qubit gate name (id, u1, u2, u3) as keys.
        - a dictionnary as keys with:
          - property names (gate_error, gate_length) as keys.
          - property values as values.

        :param backend_properties: result of calling the `properties()` method on
            one of the IBMQ backends.
        :param qubit_index: index of the qubit we are interested in.

        :raise KeyError: if the given qubit_index is not present in the hardware.
        """
        qubit_properties = backend_properties.qubits
        gates_properties = backend_properties.gates
        properties = dict()
        for prop in qubit_properties[qubit_index]:
            properties[prop.name] = IBMQHardwareArchitecture._get_value(
                prop.value, prop.unit
            )
        for gate in gates_properties:
            # Filter the gates that are not interesting
            if len(gate.qubits) > 1:
                continue
            if gate.qubits[0] != qubit_index:
                continue
            properties[gate.gate] = {
                param.name: IBMQHardwareArchitecture._get_value(param.value, param.unit)
                for param in gate.parameters
            }
        return properties

    @staticmethod
    def _get_backend(
        backend_name: str,
        hub: str = "ibm-q",
        group: str = "open",
        project: str = "main",
    ):
        """Returns the backend identified by the given name."""
        from qiskit import IBMQ
        from qiskit.providers.ibmq.exceptions import (
            IBMQAccountCredentialsNotFound,
            IBMQAccountMultipleCredentialsFound,
            IBMQAccountCredentialsInvalidUrl,
        )

        logger.info("Loading IBMQ account.")
        try:
            IBMQ.load_account()
            provider = IBMQ.get_provider(hub=hub, group=group, project=project)
        except (
            IBMQAccountMultipleCredentialsFound,
            IBMQAccountCredentialsNotFound,
            IBMQAccountCredentialsInvalidUrl,
        ):
            logger.error(
                "WARNING: No valid IBMQ credentials found on disk.\n"
                "You must store your credentials using "
                "IBMQ.save_account(token, url).\n"
            )
            raise
        logger.info("Connected to IBMQ account!")
        logger.info(f"Getting backend '{backend_name}'.")
        matching_backends = provider.backends(backend_name)
        if not matching_backends:
            raise RuntimeError(f"No backend matching the search '{backend_name}'.")
        backend = matching_backends[0]
        return backend

    def __init__(
        self, backend_name: str, **kwargs,
    ):
        """The architecture of any IBMQ hardware.
        :param backend_name: name of the IBMQ backend that will be used to
            build the architecture.
        :param kwargs: forwarded to :py:method:`networkx.Digraph.__init__`.
        """
        super().__init__(**kwargs)
        self._ignored_gates = {"barrier"}

        backend = IBMQHardwareArchitecture._get_backend(backend_name)

        # Get the configuration data
        backend_configuration = backend.configuration()
        qubit_number = backend_configuration.n_qubits
        coupling_map = backend_configuration.coupling_map
        # Get the properties of the backend
        backend_properties = backend.properties()
        qubit_indices = list()
        # Add the qubits with their properties (error rates?)
        for qubit_index in range(qubit_number):
            added_qubit_index = self.add_qubit(
                **IBMQHardwareArchitecture._get_qubit_properties_dict(
                    backend_properties, qubit_index,
                )
            )
            qubit_indices.append(added_qubit_index)
        # Add the links between qubits
        for source, sink in coupling_map:
            self.add_link(
                source,
                sink,
                **IBMQHardwareArchitecture._get_link_properties_dict(
                    backend_properties, source, sink
                ),
            )

    @property
    def qubits(self) -> ty.Iterable[int]:
        yield from range(self.qubit_number)

    @property
    def links(self) -> ty.Iterable[ty.Tuple[int, int]]:
        yield from self.edges()

    def draw(
        self,
        edge_label_property: str = None,
        pos: ty.Dict[ty.Tuple[int, int], ty.Tuple[float, float]] = None,
    ):
        """Draw the hardware topology.
        :param edge_label_property: Key of the link property that should be plotted
            alongside qubits and links. If None (default), no property is drawn on 
            the resulting plot.
        :param pos: dictionary with nodes (tuples of ints) as keys and the
            associated position (tuple of floats) as values. Default behaviour (None) is
            to use networkx.spring_layout to compute the positions.
        """
        # Draw the circuit on the specified positions, with the node labels.
        if pos is None:
            pos = nx.spring_layout(self)
        qubit_labels = {i: str(i) for i in self.nodes}
        nx.draw(self, pos, labels=qubit_labels)
        # Also include edge weights if asked
        if edge_label_property is not None:
            edge_labels = nx.get_edge_attributes(self, edge_label_property)
            for key, val in edge_labels.items():
                edge_labels[key] = str(round(val, 2))
            nx.draw_networkx_edge_labels(self, pos, edge_labels=edge_labels)

    def get_link_execution_time(self, source: int, sink: int) -> float:
        """Returns the execution time of the CNOT gate between source and sink in \
        nano-seconds."""
        return self.edges[source, sink]["gate_length"]

    def get_link_error_rate(self, source: int, sink: int) -> float:
        """Returns the error rate of the CNOT gate between source and sink."""
        return self.edges[source, sink]["gate_error"]

    def get_qubit_execution_time(self, qubit_index: int, operation: str) -> float:
        """Returns the execution time of operation on the given qubit.
        :param qubit_index: index of the qubit we are interested in.
        :param operation: name of the quantum operation. For IBMQ hardware, it can be
            either "id", "u1", "u2" or "u3".
        :return: the execution time of operation on the given qubit.
        """
        return self.nodes[qubit_index][operation]["gate_length"]

    def get_qubit_error_rate(self, qubit_index: int, operation: str) -> float:
        """Returns the error rate of operation on the given qubit.
        :param qubit_index: index of the qubit we are interested in.
        :param operation: name of the quantum operation. For IBMQ hardware, it can be
            either "id", "u1", "u2" or "u3".
        :return: the error rate of operation on the given qubit.
        """
        return self.nodes[qubit_index][operation]["gate_error"]

    def get_qubit_T1(self, qubit_index: int) -> float:
        """Returns the T1 characteristic time for the given qubit index."""
        return self.nodes[qubit_index]["T1"]

    def get_qubit_T2(self, qubit_index: int) -> float:
        """Returns the T2 characteristic time for the given qubit index."""
        return self.nodes[qubit_index]["T2"]

    def get_qubit_frequency(self, qubit_index: int) -> float:
        """Returns the qubit frequency of the qubit pointed by qubit_index."""
        return self.nodes[qubit_index]["frequency"]

    def get_qubit_readout_error(self, qubit_index: int) -> float:
        """Returns the qubit readout error of the qubit pointed by qubit_index."""
        return self.nodes[qubit_index]["readout_error"]

    def is_ignored_operation(self, op: DAGNode) -> bool:
        return op.name in self._ignored_gates

    def can_natively_execute_operation(
        self, op: DAGNode, mapping: ty.Dict[Qubit, int],
    ) -> bool:
        if self.is_ignored_operation(op):
            return True

        if len(op.qargs) == 1:
            # If this is a 1-qubit operation, then the hardware can always execute it
            # natively.
            return True
        elif len(op.qargs) == 2:
            source = mapping[op.qargs[0]]
            sink = mapping[op.qargs[1]]
            return (source, sink) in self.edges
        else:
            raise RuntimeError(
                f"Found invalid operation acting on {len(op.qargs)} qubits. "
                f"Ignoring the operation {op.name} and exiting."
            )

    def save(self, path: Path):
        with open(str(path), "wb") as f:
            logger.info(f"Saving IBMQHardwareArchitecture instance in '{path}'.")
            pickle.dump(self, f)

    @staticmethod
    def load(path: Path) -> "IBMQHardwareArchitecture":
        with open(str(path), "rb") as f:
            logger.info(f"Loading IBMQHardwareArchitecture instance from '{path}'.")
            return pickle.load(f)