refactor: delete deprecated attributess, move connect() in parent class

src/qibolab/instruments/qblox/cluster_qcm_bb.py

@@ -110,10 +110,6 @@ def __init__(self, name: str, address: str):
         """
         super().__init__(name, address)
 
-        self._debug_folder: str = ""
-        self._sequencers: dict[Sequencer] = {}
-        self._device_num_output_ports = 2
-
     def _set_default_values(self):
         # disable all sequencer connections
         self.device.disconnect_outputs()
@@ -135,35 +131,12 @@ def _set_default_values(self):
         for port_num, value in self.OUT_PORT_PATH.items():
             self.device.sequencers[port_num].set(f"connect_out{port_num}", value)
 
-    def connect(self):
-        """Connects to the instrument using the instrument settings in the
-        runcard.
-
-        Once connected, it creates port classes with properties mapped
-        to various instrument parameters, and initialises the the
-        underlying device parameters. It uploads to the module the port
-        settings loaded from the runcard.
-        """
-        if self.is_connected:
-            return
-        # test connection with module. self.device is initialized in QbloxController connect()
-        if not self.device.present():
-            raise ConnectionError(f"Module {self.device.name} not present")
-        # once connected, initialise the parameters of the device to the default values
-        self._device_num_sequencers = len(self.device.sequencers)
-        self._set_default_values()
-        # then set the value loaded from the runcard
-        try:
-            for port in self._ports:
-                self._sequencers[port] = []
-                self._ports[port].hardware_mod_en = True
-                self._ports[port].nco_freq = 0
-                self._ports[port].nco_phase_offs = 0
-        except Exception as error:
-            raise RuntimeError(
-                f"Unable to initialize port parameters on module {self.name}: {error}"
-            )
-        self.is_connected = True
+    def _setup_ports(self):
+        for port in self._ports:
+            self._sequencers[port] = []
+            self._ports[port].hardware_mod_en = True
+            self._ports[port].nco_freq = 0
+            self._ports[port].nco_phase_offs = 0
 
     def setup(self, **settings):
         """Cache the settings of the runcard and instantiate the ports of the
@@ -194,7 +167,7 @@ def _get_next_sequencer(self, port, frequency, qubits: dict):
         # select the qubit with flux line, if present, connected to the specific port
         qubit = None
         for _qubit in qubits.values():
-            if _qubit.flux is not None and _qubit.flux.port == self.ports(port):
+            if _qubit.flux is not None and _qubit.flux.port == self._ports[port]:
                 qubit = _qubit
 
         # select a new sequencer and configure it as required

src/qibolab/instruments/qblox/cluster_qcm_rf.py

@@ -125,11 +125,7 @@ def __init__(self, name: str, address: str):
         >>> qcm_module = QcmRf(name="qcm_rf", address="192.168.1.100:2", cluster=cluster_instance)
         """
         super().__init__(name, address)
-        self.settings = {}
-
-        self._debug_folder: str = ""
-        self._sequencers: dict[Sequencer] = {}
-        self._device_num_output_ports = 2
+        self.settings: dict = {}
 
     def _set_default_values(self):
         # disable all sequencer connections
@@ -154,39 +150,16 @@ def _set_default_values(self):
         self.device.sequencers[self.DEFAULT_SEQUENCERS["o2"]].set("connect_out1", "IQ")
         self.device.sequencers[self.DEFAULT_SEQUENCERS["o2"]].set("connect_out0", "off")
 
-    def connect(self):
-        """Connects to the instrument using the instrument settings in the
-        runcard.
-
-        Once connected, it creates port classes with properties mapped
-        to various instrument parameters, and initialises the the
-        underlying device parameters. It uploads to the module the port
-        settings loaded from the runcard.
-        """
-        if self.is_connected:
-            return
-        # test connection with module. self.device is initialized in QbloxController connect()
-        if not self.device.present():
-            raise ConnectionError(f"Module {self.device.name} not present")
-        # once connected, initialise the parameters of the device to the default values
-        self._device_num_sequencers = len(self.device.sequencers)
-        self._set_default_values()
-        # then set the value loaded from the runcard
-        try:
-            for port in self.settings:
-                self._sequencers[port] = []
-                if self.settings[port]["lo_frequency"]:
-                    self._ports[port].lo_enabled = True
-                    self._ports[port].lo_frequency = self.settings[port]["lo_frequency"]
-                self._ports[port].attenuation = self.settings[port]["attenuation"]
-                self._ports[port].hardware_mod_en = True
-                self._ports[port].nco_freq = 0
-                self._ports[port].nco_phase_offs = 0
-        except Exception as error:
-            raise RuntimeError(
-                f"Unable to initialize port parameters on module {self.name}: {error}"
-            )
-        self.is_connected = True
+    def _setup_ports(self):
+        for port in self.settings:
+            self._sequencers[port] = []
+            if self.settings[port]["lo_frequency"]:
+                self._ports[port].lo_enabled = True
+                self._ports[port].lo_frequency = self.settings[port]["lo_frequency"]
+            self._ports[port].attenuation = self.settings[port]["attenuation"]
+            self._ports[port].hardware_mod_en = True
+            self._ports[port].nco_freq = 0
+            self._ports[port].nco_phase_offs = 0
 
     def setup(self, **settings):
         """Cache the settings of the runcard and instantiate the ports of the

src/qibolab/instruments/qblox/cluster_qrm_rf.py

@@ -135,14 +135,11 @@ def __init__(self, name: str, address: str):
         """
 
         super().__init__(name, address)
-        self.classification_parameters: dict = {}
-        self.settings: dict = {}
+        self.classification_parameters = {}
+        self.settings = {}
 
-        self._debug_folder: str = ""
         self._input_ports_keys = ["i1"]
         self._output_ports_keys = ["o1"]
-        self._sequencers: dict[Sequencer] = {"o1": []}
-        self._device_num_output_ports = 1
         self._execution_time: float = 0
 
     def _set_default_values(self):
@@ -172,47 +169,24 @@ def _set_default_values(self):
         target.set("connect_out0", "IQ")
         target.set("connect_acq", "in0")
 
-    def connect(self):
-        """Connects to the instrument using the instrument settings in the
-        runcard.
-
-        Once connected, it creates port classes with properties mapped
-        to various instrument parameters, and initialises the the
-        underlying device parameters. It uploads to the module the port
-        settings loaded from the runcard.
-        """
-        if self.is_connected:
-            return
-        # test connection with module. self.device is initialized in QbloxController connect()
-        if not self.device.present():
-            raise ConnectionError(f"Module {self.device.name} not present")
-        # once connected, initialise the parameters of the device to the default values
-        self._device_num_sequencers = len(self.device.sequencers)
-        self._set_default_values()
-        # then set the value loaded from the runcard
-        try:
-            if "o1" in self.settings:
-                self._ports["o1"].attenuation = self.settings["o1"]["attenuation"]
-                if self.settings["o1"]["lo_frequency"]:
-                    self._ports["o1"].lo_enabled = True
-                    self._ports["o1"].lo_frequency = self.settings["o1"]["lo_frequency"]
-                self._ports["o1"].hardware_mod_en = True
-                self._ports["o1"].nco_freq = 0
-                self._ports["o1"].nco_phase_offs = 0
-
-            if "i1" in self.settings:
-                self._ports["i1"].hardware_demod_en = True
-                self._ports["i1"].acquisition_hold_off = self.settings["i1"][
-                    "acquisition_hold_off"
-                ]
-                self._ports["i1"].acquisition_duration = self.settings["i1"][
-                    "acquisition_duration"
-                ]
-        except Exception as error:
-            raise RuntimeError(
-                f"Unable to initialize port parameters on module {self.name}: {error}"
-            )
-        self.is_connected = True
+    def _setup_ports(self):
+        if "o1" in self.settings:
+            self._ports["o1"].attenuation = self.settings["o1"]["attenuation"]
+            if self.settings["o1"]["lo_frequency"]:
+                self._ports["o1"].lo_enabled = True
+                self._ports["o1"].lo_frequency = self.settings["o1"]["lo_frequency"]
+            self._ports["o1"].hardware_mod_en = True
+            self._ports["o1"].nco_freq = 0
+            self._ports["o1"].nco_phase_offs = 0
+
+        if "i1" in self.settings:
+            self._ports["i1"].hardware_demod_en = True
+            self._ports["i1"].acquisition_hold_off = self.settings["i1"][
+                "acquisition_hold_off"
+            ]
+            self._ports["i1"].acquisition_duration = self.settings["i1"][
+                "acquisition_duration"
+            ]
 
     def setup(self, **settings):
         """Cache the settings of the runcard and instantiate the ports of the
@@ -352,7 +326,7 @@ def process_pulse_sequence(
         self._free_sequencers_numbers = [*range(self._device_num_sequencers)]
 
         # split the collection of instruments pulses by ports
-        port_pulses = self.filter_port_pulse(sequence, qubits, self.ports[port])
+        port_pulses = self.filter_port_pulse(sequence, qubits, self._ports[port])
 
         # initialise the list of sequencers required by the port
         self._sequencers[port] = []

src/qibolab/instruments/qblox/module.py

@@ -1,13 +1,17 @@
 """Qblox Cluster QCM driver."""
 
+from abc import abstractmethod
+
 from qblox_instruments.qcodes_drivers.qcm_qrm import QcmQrm as QbloxQrmQcm
 from qibo.config import log
 
 from qibolab.instruments.abstract import Instrument
-from qibolab.instruments.qblox.port import QbloxInputPort, QbloxOutputPort
 from qibolab.pulses import Pulse, PulseSequence
 from qibolab.qubits import Qubit
 
+from .port import QbloxInputPort, QbloxOutputPort
+from .sequencer import Sequencer
+
 
 class ClusterModule(Instrument):
     """This class defines common features shared by all Qblox modules (QCM-BB,
@@ -45,6 +49,8 @@ def __init__(self, name: str, address: str):
         self._free_sequencers_numbers: list[int] = []
         self._used_sequencers_numbers: list[int] = []
         self._unused_sequencers_numbers: list[int] = []
+        self._debug_folder: str = ""
+        self._sequencers: dict[str, list[Sequencer]] = {}
 
     def ports(self, name: str, out: bool = True):
         """Adds an entry to the dictionary `self._ports` with key 'name' and
@@ -73,6 +79,36 @@ def count(cls):
         self._ports[name] = port_cls(self, port_number=count(port_cls), port_name=name)
         return self._ports[name]
 
+    @abstractmethod
+    def _setup_ports(self):
+        pass
+
+    def connect(self):
+        """Connects to the instrument using the instrument settings in the
+        runcard.
+
+        Once connected, it creates port classes with properties mapped
+        to various instrument parameters, and initialises the the
+        underlying device parameters. It uploads to the module the port
+        settings loaded from the runcard.
+        """
+        if self.is_connected:
+            return
+        # test connection with module. self.device is initialized in QbloxController connect()
+        if not self.device.present():
+            raise ConnectionError(f"Module {self.device.name} not present")
+        # once connected, initialise the parameters of the device to the default values
+        self._device_num_sequencers = len(self.device.sequencers)
+        self._set_default_values()
+        # then set the value loaded from the runcard
+        try:
+            self._setup_ports()
+        except Exception as error:
+            raise RuntimeError(
+                f"Unable to initialize port parameters on module {self.name}: {error}"
+            )
+        self.is_connected = True
+
     def clone_sequencer_params(self, first_sequencer: int, next_sequencer: int):
         """Clone the values of all writable parameters from the first_sequencer
         into the next_sequencer."""