Add methods to retrieve info from channels config in backends (Qiskit…

…#4097)

* Add get_channel_qubits method to get info from channels config

* Add tests

* Update docstrings

* Remove unwanted imports

* Review suggestions

* Fix get_qubit_channels

* Deprecation warning to use control(qubits: List) instead of control(channel: int)

* typos

* update the control based on the new backend schema

* typos

* More logic fixes

* lint

* lint

* add older code

* More logic fixes

* lint

* Update docstrings and typehints

* Unused imports

* Release notes

* Review suggestions

* Bug fix for get_qubit_channels(0)

Updated tests too

* Review Suggestions

* lint

* Change error message of _get_channel_prefix_index

* Added raise error for backends without 'channel' information

* Update docstrings for raise errors

* Fix error messages and docstrings

* make self.channels private

* Remove collections.Counter

* Revert "Remove collections.Counter"

This reverts commit d602738.

* change list() -> set()

qiskit/providers/models/backendconfiguration.py

@@ -13,13 +13,17 @@
 # that they have been altered from the originals.
 
 """Backend Configuration Classes."""
+import re
 import copy
+import warnings
 from types import SimpleNamespace
-from typing import Dict, List
+from typing import Dict, List, Any, Iterable, Union
+from collections import defaultdict
 
 from qiskit.exceptions import QiskitError
 from qiskit.providers.exceptions import BackendConfigurationError
-from qiskit.pulse.channels import DriveChannel, MeasureChannel, ControlChannel, AcquireChannel
+from qiskit.pulse.channels import (Channel, DriveChannel, MeasureChannel,
+                                   ControlChannel, AcquireChannel)
 
 
 class GateConfig:
@@ -384,6 +388,7 @@ def __init__(self,
                  display_name=None,
                  description=None,
                  tags=None,
+                 channels: Dict[str, Any] = None,
                  **kwargs):
         """
         Initialize a backend configuration that contains all the extra configuration that is made
@@ -437,6 +442,8 @@ def __init__(self,
             display_name (str): Alternate name field for the backend
             description (str): A description for the backend
             tags (list): A list of string tags to describe the backend
+            channels: An optional dictionary containing information of each channel -- their
+                purpose, type, and qubits operated on.
             **kwargs: Optional fields.
         """
         self.n_uchannels = n_uchannels
@@ -454,6 +461,13 @@ def __init__(self,
         self.dt = dt * 1e-9  # pylint: disable=invalid-name
         self.dtm = dtm * 1e-9
 
+        if channels is not None:
+            self._channels = channels
+
+            (self._qubit_channel_map,
+             self._channel_qubit_map,
+             self._control_channels) = self._parse_channels(channels=channels)
+
         if channel_bandwidth is not None:
             self.channel_bandwidth = [[min_range * 1e9, max_range * 1e9] for
                                       (min_range, max_range) in channel_bandwidth]
@@ -570,16 +584,90 @@ def acquire(self, qubit: int) -> AcquireChannel:
             raise BackendConfigurationError("Invalid index for {}-qubit systems.".format(qubit))
         return AcquireChannel(qubit)
 
-    def control(self, channel: int) -> ControlChannel:
+    def control(self, qubits: Iterable[int] = None,
+                channel: int = None) -> List[ControlChannel]:
         """
         Return the secondary drive channel for the given qubit -- typically utilized for
         controlling multiqubit interactions. This channel is derived from other channels.
 
+        Args:
+            qubits: Tuple or list of qubits of the form `(control_qubit, target_qubit)`.
+            channel: Deprecated.
+
+        Raises:
+            BackendConfigurationError: If the ``qubits`` is not a part of the system or if
+                the backend does not provide `channels` information in its configuration.
+
+        Returns:
+            List of control channels.
+        """
+        if channel is not None:
+            warnings.warn('The channel argument has been deprecated in favor of qubits. '
+                          'This method will now return accurate ControlChannels determined '
+                          'by qubit indices.',
+                          DeprecationWarning)
+            qubits = [channel]
+        try:
+            if isinstance(qubits, list):
+                qubits = tuple(qubits)
+            return self._control_channels[qubits]
+        except KeyError:
+            raise BackendConfigurationError("Couldn't find the ControlChannel operating on qubits "
+                                            "{} on {}-qubit system. The ControlChannel information"
+                                            " is retrieved from the "
+                                            " backend.".format(qubits, self.n_qubits))
+        except AttributeError:
+            raise BackendConfigurationError("This backend - '{}' does not provide channel "
+                                            "information.".format(self.backend_name))
+
+    def get_channel_qubits(self, channel: Channel) -> List[int]:
+        """
+        Return a list of indices for qubits which are operated on directly by the given ``channel``.
+
+        Raises:
+            BackendConfigurationError: If ``channel`` is not a found or if
+                the backend does not provide `channels` information in its configuration.
+
         Returns:
-            Qubit control channel.
+            List of qubits operated on my the given ``channel``.
         """
-        # TODO: Determine this from the hamiltonian.
-        return ControlChannel(channel)
+        try:
+            return self._channel_qubit_map[channel]
+        except KeyError:
+            raise BackendConfigurationError("Couldn't find the Channel - {}".format(channel))
+        except AttributeError:
+            raise BackendConfigurationError("This backend - '{}' does not provide channel "
+                                            "information.".format(self.backend_name))
+
+    def get_qubit_channels(self, qubit: Union[int, Iterable[int]]) -> List[Channel]:
+        r"""Return a list of channels which operate on the given ``qubit``.
+
+        Raises:
+            BackendConfigurationError: If ``qubit`` is not a found or if
+                the backend does not provide `channels` information in its configuration.
+
+        Returns:
+            List of ``Channel``\s operated on my the given ``qubit``.
+        """
+        channels = set()
+        try:
+            if isinstance(qubit, int):
+                for key in self._qubit_channel_map.keys():
+                    if qubit in key:
+                        channels.update(self._qubit_channel_map[key])
+                if len(channels) == 0:
+                    raise KeyError
+            elif isinstance(qubit, list):
+                qubit = tuple(qubit)
+                channels.update(self._qubit_channel_map[qubit])
+            elif isinstance(qubit, tuple):
+                channels.update(self._qubit_channel_map[qubit])
+            return list(channels)
+        except KeyError:
+            raise BackendConfigurationError("Couldn't find the qubit - {}".format(qubit))
+        except AttributeError:
+            raise BackendConfigurationError("This backend - '{}' does not provide channel "
+                                            "information.".format(self.backend_name))
 
     def describe(self, channel: ControlChannel) -> Dict[DriveChannel, complex]:
         """
@@ -610,3 +698,56 @@ def describe(self, channel: ControlChannel) -> Dict[DriveChannel, complex]:
         for u_chan_lo in self.u_channel_lo[channel.index]:
             result[DriveChannel(u_chan_lo.q)] = u_chan_lo.scale
         return result
+
+    def _parse_channels(self, channels: Dict[set, Any]) -> Dict[Any, Any]:
+        r"""
+        Generates a dictionaries of ``Channel``\s, and tuple of qubit(s) they operate on.
+
+        Args:
+            channels: An optional dictionary containing information of each channel -- their
+                purpose, type, and qubits operated on.
+
+        Returns:
+            qubit_channel_map: Dictionary mapping tuple of qubit(s) to list of ``Channel``\s.
+            channel_qubit_map: Dictionary mapping ``Channel`` to list of qubit(s).
+            control_channels: Dictionary mapping tuple of qubit(s), to list of
+                ``ControlChannel``\s.
+        """
+        qubit_channel_map = defaultdict(list)
+        channel_qubit_map = defaultdict(list)
+        control_channels = defaultdict(list)
+        channels_dict = {
+            DriveChannel.prefix: DriveChannel,
+            ControlChannel.prefix: ControlChannel,
+            MeasureChannel.prefix: MeasureChannel,
+            'acquire': AcquireChannel
+        }
+        for channel, config in channels.items():
+            channel_prefix, index = self._get_channel_prefix_index(channel)
+            channel_type = channels_dict[channel_prefix]
+            qubits = tuple(config['operates']['qubits'])
+            if channel_prefix in channels_dict:
+                qubit_channel_map[qubits].append(channel_type(index))
+                channel_qubit_map[(channel_type(index))].extend(list(qubits))
+                if channel_prefix == ControlChannel.prefix:
+                    control_channels[qubits].append(channel_type(index))
+        return dict(qubit_channel_map), dict(channel_qubit_map), dict(control_channels)
+
+    def _get_channel_prefix_index(self, channel: str) -> str:
+        """Return channel prefix and index from the given ``channel``.
+
+        Args:
+            channel: Name of channel.
+
+        Raises:
+            BackendConfigurationError: If invalid channel name is found.
+
+        Return:
+            Channel name and index. For example, if ``channel=acquire0``, this method
+            returns ``acquire`` and ``0``.
+        """
+        channel_prefix = re.match(r"(?P<channel>[a-z]+)(?P<index>[0-9]+)", channel)
+        try:
+            return channel_prefix.group('channel'), int(channel_prefix.group('index'))
+        except AttributeError:
+            raise BackendConfigurationError("Invalid channel name - '{}' found.".format(channel))

releasenotes/notes/fix-control-channel-logic-cdb45b7adae394d8.yaml

@@ -0,0 +1,19 @@
+---
+features:
+  - |
+    Added methods --
+    :mod:`qiskit.providers.models.PulseBackendConfiguration.get_channel_qubits(channel)`
+    to get a list of all qubits operated by the given channel and
+    :mod:`qiskit.providers.models.PulseBackendConfiguration.get_qubit_channel(qubits)`
+    to get a list of channels operating on the given qubit.
+
+deprecations:
+  - |
+    Previously, ``control(channel)`` returned ``ControlChannel(channel)``. This logic
+    was not accurate (Issue `#3311 <https://github.com/Qiskit/qiskit-terra/issues/3311>`__).
+    The parameter ``channel`` has been deprecated. ControlChannel(index) is now generated from
+    the backend configuration ``channels`` which has the information of all channels and the
+    qubits they operate on. Now, the method
+    :mod:`qiskit.providers.models.PulseBackendConfiguration.control` takes the parameter --
+    ``qubits`` of the form **(control_qubit, target_qubit)**  and type **list or tuple**,
+    and returns a list of control channel(s).

test/python/providers/test_backendconfiguration.py

@@ -14,6 +14,7 @@
 """
 Test that the PulseBackendConfiguration methods work as expected with a mocked Pulse backend.
 """
+import collections
 from qiskit.test import QiskitTestCase
 from qiskit.test.mock import FakeProvider
 
@@ -54,7 +55,42 @@ def test_get_channels(self):
         with self.assertRaises(BackendConfigurationError):
             # Check that an error is raised if the system doesn't have that many qubits
             self.assertEqual(self.config.acquire(10), AcquireChannel(10))
-        self.assertEqual(self.config.control(0), ControlChannel(0))
+        self.assertEqual(self.config.control(qubits=[0, 1]), [ControlChannel(0)])
+        with self.assertRaises(BackendConfigurationError):
+            # Check that an error is raised if key not found in self._qubit_channel_map
+            self.config.control(qubits=(10, 1))
+
+    def test_get_channel_qubits(self):
+        """Test to get all qubits operated on a given channel."""
+        self.assertEqual(self.config.get_channel_qubits(channel=DriveChannel(0)), [0])
+        self.assertEqual(self.config.get_channel_qubits(channel=ControlChannel(0)), [0, 1])
+        backend_3q = self.provider.get_backend('fake_openpulse_3q')
+        self.assertEqual(backend_3q.configuration().get_channel_qubits(ControlChannel(2)), [2, 1])
+        self.assertEqual(backend_3q.configuration().get_channel_qubits(ControlChannel(1)), [1, 0])
+        with self.assertRaises(BackendConfigurationError):
+            # Check that an error is raised if key not found in self._channel_qubit_map
+            self.config.get_channel_qubits(MeasureChannel(10))
+
+    def test_get_qubit_channels(self):
+        """Test to get all channels operated on a given qubit."""
+        self.assertTrue(self._test_lists_equal(
+            actual=self.config.get_qubit_channels(qubit=(1,)),
+            expected=[DriveChannel(1), MeasureChannel(1), AcquireChannel(1)]
+        ))
+        self.assertTrue(self._test_lists_equal(
+            actual=self.config.get_qubit_channels(qubit=1),
+            expected=[ControlChannel(0), ControlChannel(1), AcquireChannel(1),
+                      DriveChannel(1), MeasureChannel(1)]
+        ))
+        backend_3q = self.provider.get_backend('fake_openpulse_3q')
+        self.assertTrue(self._test_lists_equal(
+            actual=backend_3q.configuration().get_qubit_channels(1),
+            expected=[MeasureChannel(1), ControlChannel(0), ControlChannel(2),
+                      AcquireChannel(1), DriveChannel(1), ControlChannel(1)]
+        ))
+        with self.assertRaises(BackendConfigurationError):
+            # Check that an error is raised if key not found in self._channel_qubit_map
+            self.config.get_qubit_channels(10)
 
     def test_get_rep_times(self):
         """Test whether rep time property is the right size"""
@@ -65,3 +101,13 @@ def test_get_rep_times(self):
             self.assertAlmostEqual(self.config.rep_times[i], time)
         for i, time in enumerate(_rep_times_us):
             self.assertEqual(round(self.config.rep_times[i]*1e6), time)
+
+    def test_get_channel_prefix_index(self):
+        """Test private method to get channel and index."""
+        self.assertEqual(self.config._get_channel_prefix_index('acquire0'), ('acquire', 0))
+        with self.assertRaises(BackendConfigurationError):
+            self.config._get_channel_prefix_index("acquire")
+
+    def _test_lists_equal(self, actual, expected):
+        """Test if 2 lists are equal. It returns ``True`` is lists are equal."""
+        return collections.Counter(actual) == collections.Counter(expected)

test/python/pulse/test_schedule.py

@@ -112,10 +112,11 @@ def test_can_create_valid_schedule(self):
         sched = Schedule()
         sched = sched.append(Play(gp0, self.config.drive(0)))
         with self.assertWarns(DeprecationWarning):
-            sched = sched.insert(0, PersistentValue(value=0.2 + 0.4j)(self.config.control(0)))
+            sched = sched.insert(0, PersistentValue(value=0.2 + 0.4j)(self.config.control(
+                [0, 1])[0]))
         sched = sched.insert(60, ShiftPhase(-1.57, self.config.drive(0)))
         sched = sched.insert(30, Play(gp1, self.config.drive(0)))
-        sched = sched.insert(60, Play(gp0, self.config.control(0)))
+        sched = sched.insert(60, Play(gp0, self.config.control([0, 1])[0]))
         sched = sched.insert(80, Snapshot("label", "snap_type"))
         sched = sched.insert(90, ShiftPhase(1.57, self.config.drive(0)))
         sched = sched.insert(90, Acquire(10,
@@ -145,10 +146,10 @@ def test_can_create_valid_schedule_with_syntax_sugar(self):
         sched = Schedule()
         sched += Play(gp0, self.config.drive(0))
         with self.assertWarns(DeprecationWarning):
-            sched |= PersistentValue(value=0.2 + 0.4j)(self.config.control(0))
+            sched |= PersistentValue(value=0.2 + 0.4j)(self.config.control([0, 1])[0])
         sched |= ShiftPhase(-1.57, self.config.drive(0)) << 60
         sched |= Play(gp1, self.config.drive(0)) << 30
-        sched |= Play(gp0, self.config.control(0)) << 60
+        sched |= Play(gp0, self.config.control(qubits=[0, 1])[0]) << 60
         sched |= Snapshot("label", "snap_type") << 60
         sched |= ShiftPhase(1.57, self.config.drive(0)) << 90
         sched |= Acquire(10,
@@ -457,7 +458,7 @@ def test_delay_measure_channel(self):
     def test_delay_control_channel(self):
         """Test Delay on ControlChannel"""
 
-        control_ch = self.config.control(0)
+        control_ch = self.config.control([0, 1])[0]
         pulse = SamplePulse(np.full(10, 0.1))
         # should pass as is an append
         sched = Delay(self.delay_time, control_ch) + Play(pulse, control_ch)