use channel names as signal names, instead of inventing new name

src/qibolab/instruments/zhinst.py

@@ -50,6 +50,16 @@
 """Maximum number of subsequences in a single sequence."""
 
 
+def acquire_signal_name(qubit: Qubit) -> str:
+    """Construct and return a name for qubit's acquire signal line.
+
+    For other types of signals (drive, flux, etc.) we use the channel name as signal name. Currently we do not have a concept
+    of an acquire channel, so there is no source for acquire signal names, hence this function exists.
+    FIXME: this function will become redundant and shall be removed once all the channel refactoring is done.
+    """
+    return f"acquire{qubit.name}"
+
+
 def select_pulse(pulse, pulse_type):
     """Pulse translation."""
 
@@ -218,7 +228,7 @@ def __init__(self, sweeper, qubit=None, sequence=None, pulse=None):
                     channel=qubit.flux.name,
                     qubit=qubit.name,
                 )
-                self.signal = f"flux{qubit.name}"
+                self.signal = qubit.flux.name
             if isinstance(qubit, Coupler):
                 pulse = CouplerFluxPulse(
                     start=0,
@@ -228,7 +238,7 @@ def __init__(self, sweeper, qubit=None, sequence=None, pulse=None):
                     channel=qubit.flux.name,
                     qubit=qubit.name,
                 )
-                self.signal = f"couplerflux{qubit.name}"
+                self.signal = qubit.flux.name
 
             self.pulse = pulse
 
@@ -241,7 +251,7 @@ def __init__(self, sweeper, qubit=None, sequence=None, pulse=None):
         elif sweeper.parameter is Parameter.start:
             if pulse:
                 self.pulse = pulse
-                self.signal = f"flux{qubit}"
+                self.signal = qubit.flux.name
 
                 self.zhpulse = ZhPulse(pulse).zhpulse
 
@@ -335,13 +345,13 @@ def calibration_step(self, qubits, couplers, options):
         for qubit in qubits.values():
             if qubit.flux is not None:
                 self.register_flux_line(qubit)
-            if len(self.sequence[f"drive{qubit.name}"]) != 0:
+            if len(self.sequence[qubit.drive.name]) != 0:
                 self.register_drive_line(
                     qubit=qubit,
                     intermediate_frequency=qubit.drive_frequency
                     - qubit.drive.local_oscillator.frequency,
                 )
-            if len(self.sequence[f"readout{qubit.name}"]) != 0:
+            if len(self.sequence[qubit.readout.name]) != 0:
                 self.register_readout_line(
                     qubit=qubit,
                     intermediate_frequency=qubit.readout_frequency
@@ -371,7 +381,7 @@ def register_readout_line(self, qubit, intermediate_frequency, options):
         """
 
         q = qubit.name  # pylint: disable=C0103
-        self.signal_map[f"measure{q}"] = self.device_setup.logical_signal_groups[
+        self.signal_map[qubit.readout.name] = self.device_setup.logical_signal_groups[
             f"q{q}"
         ].logical_signals["measure_line"]
         self.calibration[f"/logical_signal_groups/q{q}/measure_line"] = (
@@ -390,9 +400,11 @@ def register_readout_line(self, qubit, intermediate_frequency, options):
             )
         )
 
-        self.signal_map[f"acquire{q}"] = self.device_setup.logical_signal_groups[
-            f"q{q}"
-        ].logical_signals["acquire_line"]
+        self.signal_map[acquire_signal_name(qubit)] = (
+            self.device_setup.logical_signal_groups[f"q{q}"].logical_signals[
+                "acquire_line"
+            ]
+        )
 
         oscillator = lo.Oscillator(
             frequency=intermediate_frequency,
@@ -420,7 +432,7 @@ def register_readout_line(self, qubit, intermediate_frequency, options):
     def register_drive_line(self, qubit, intermediate_frequency):
         """Registers qubit drive line to calibration and signal map."""
         q = qubit.name  # pylint: disable=C0103
-        self.signal_map[f"drive{q}"] = self.device_setup.logical_signal_groups[
+        self.signal_map[qubit.drive.name] = self.device_setup.logical_signal_groups[
             f"q{q}"
         ].logical_signals["drive_line"]
         self.calibration[f"/logical_signal_groups/q{q}/drive_line"] = (
@@ -442,7 +454,7 @@ def register_drive_line(self, qubit, intermediate_frequency):
     def register_flux_line(self, qubit):
         """Registers qubit flux line to calibration and signal map."""
         q = qubit.name  # pylint: disable=C0103
-        self.signal_map[f"flux{q}"] = self.device_setup.logical_signal_groups[
+        self.signal_map[qubit.flux.name] = self.device_setup.logical_signal_groups[
             f"q{q}"
         ].logical_signals["flux_line"]
         self.calibration[f"/logical_signal_groups/q{q}/flux_line"] = (
@@ -457,7 +469,7 @@ def register_flux_line(self, qubit):
     def register_couplerflux_line(self, coupler):
         """Registers qubit flux line to calibration and signal map."""
         c = coupler.name  # pylint: disable=C0103
-        self.signal_map[f"couplerflux{c}"] = self.device_setup.logical_signal_groups[
+        self.signal_map[coupler.flux.name] = self.device_setup.logical_signal_groups[
             f"qc{c}"
         ].logical_signals["flux_line"]
         self.calibration[f"/logical_signal_groups/qc{c}/flux_line"] = (
@@ -505,8 +517,13 @@ def create_sub_sequence(
         """
         for quantum_element in quantum_elements.values():
             q = quantum_element.name  # pylint: disable=C0103
-            measurements = self.sequence[f"readout{q}"]
-            pulses = self.sequence[f"{line_name}{q}"]
+            measurements = (
+                self.sequence[quantum_element.readout.name]
+                if isinstance(quantum_element, Qubit)
+                else []
+            )
+            channel_name = getattr(quantum_element, line_name).name
+            pulses = self.sequence[channel_name]
             pulse_sequences = [[] for _ in measurements]
             pulse_sequences.append([])
             measurement_index = 0
@@ -515,7 +532,7 @@ def create_sub_sequence(
                     if pulse.pulse.finish > measurements[measurement_index].pulse.start:
                         measurement_index += 1
                 pulse_sequences[measurement_index].append(pulse)
-            self.sub_sequences[f"{line_name}{q}"] = pulse_sequences
+            self.sub_sequences[channel_name] = pulse_sequences
 
     def create_sub_sequences(
         self, qubits: Dict[str, Qubit], couplers: Dict[str, Coupler]
@@ -529,7 +546,7 @@ def create_sub_sequences(
         self.sub_sequences = {}
         self.create_sub_sequence("drive", qubits)
         self.create_sub_sequence("flux", qubits)
-        self.create_sub_sequence("couplerflux", couplers)
+        self.create_sub_sequence("flux", couplers)
 
     def experiment_flow(
         self,
@@ -568,8 +585,8 @@ def play(self, qubits, couplers, sequence, options):
         results = {}
         for qubit in qubits.values():
             q = qubit.name  # pylint: disable=C0103
-            if len(self.sequence[f"readout{q}"]) != 0:
-                for i, ropulse in enumerate(self.sequence[f"readout{q}"]):
+            if len(self.sequence[qubit.readout.name]) != 0:
+                for i, ropulse in enumerate(self.sequence[qubit.readout.name]):
                     data = np.array(self.results.get_data(f"sequence{q}_{i}"))
                     if options.acquisition_type is AcquisitionType.DISCRIMINATION:
                         data = (
@@ -589,7 +606,7 @@ def sequence_zh(self, sequence, qubits, couplers):
 
         # Fill the sequences with pulses according to their lines in temporal order
         for pulse in sequence:
-            zhsequence[f"{pulse.type.name.lower()}{pulse.qubit}"].append(ZhPulse(pulse))
+            zhsequence[pulse.channel].append(ZhPulse(pulse))
 
         def nt_loop(sweeper):
             if not self.nt_sweeps:
@@ -601,7 +618,7 @@ def nt_loop(sweeper):
         for sweeper in self.sweepers.copy():
             if sweeper.parameter.name in SWEEPER_SET:
                 for pulse in sweeper.pulses:
-                    aux_list = zhsequence[f"{pulse.type.name.lower()}{pulse.qubit}"]
+                    aux_list = zhsequence[pulse.channel]
                     if (
                         sweeper.parameter is Parameter.frequency
                         and pulse.type is PulseType.READOUT
@@ -642,7 +659,7 @@ def nt_loop(sweeper):
             # This may not place the Zhsweeper when the start occurs among different sections or lines
             if sweeper.parameter.name in SWEEPER_START:
                 pulse = sweeper.pulses[0]
-                aux_list = zhsequence[f"{pulse.type.name.lower()}{pulse.qubit}"]
+                aux_list = zhsequence[pulse.channel]
                 for element in aux_list:
                     if pulse == element.pulse:
                         if isinstance(aux_list[aux_list.index(element)], ZhPulse):
@@ -838,14 +855,14 @@ def flux(self, exp: lo.Experiment, qubits: Dict[str, Qubit]):
             q = qubit.name  # pylint: disable=C0103
             time = 0
             previous_section = None
-            for i, sequence in enumerate(self.sub_sequences[f"flux{q}"]):
+            for i, sequence in enumerate(self.sub_sequences[qubit.flux.name]):
                 section_uid = f"sequence_flux{q}_{i}"
                 with exp.section(uid=section_uid, play_after=previous_section):
                     for j, pulse in enumerate(sequence):
                         if not isinstance(pulse, ZhSweeperLine):
                             pulse.zhpulse.uid += f"{i}_{j}"
                             exp.delay(
-                                signal=f"flux{q}",
+                                signal=qubit.flux.name,
                                 time=round(pulse.pulse.start * NANO_TO_SECONDS, 9)
                                 - time,
                             )
@@ -857,7 +874,7 @@ def flux(self, exp: lo.Experiment, qubits: Dict[str, Qubit]):
                         elif isinstance(pulse, ZhSweeper):
                             self.play_sweep(exp, qubit, pulse, section="flux")
                         elif isinstance(pulse, ZhPulse):
-                            exp.play(signal=f"flux{q}", pulse=pulse.zhpulse)
+                            exp.play(signal=qubit.flux.name, pulse=pulse.zhpulse)
                 previous_section = section_uid
 
     def drive(self, exp: lo.Experiment, qubits: Dict[str, Qubit]):
@@ -871,13 +888,13 @@ def drive(self, exp: lo.Experiment, qubits: Dict[str, Qubit]):
             q = qubit.name  # pylint: disable=C0103
             time = 0
             previous_section = None
-            for i, sequence in enumerate(self.sub_sequences[f"drive{q}"]):
+            for i, sequence in enumerate(self.sub_sequences[qubit.drive.name]):
                 section_uid = f"sequence_drive{q}_{i}"
                 with exp.section(uid=section_uid, play_after=previous_section):
                     for j, pulse in enumerate(sequence):
                         if not isinstance(pulse, ZhSweeperLine):
                             exp.delay(
-                                signal=f"drive{q}",
+                                signal=qubit.drive.name,
                                 time=round(pulse.pulse.start * NANO_TO_SECONDS, 9)
                                 - time,
                             )
@@ -889,22 +906,22 @@ def drive(self, exp: lo.Experiment, qubits: Dict[str, Qubit]):
                                 self.play_sweep(exp, qubit, pulse, section="drive")
                             elif isinstance(pulse, ZhPulse):
                                 exp.play(
-                                    signal=f"drive{q}",
+                                    signal=qubit.drive.name,
                                     pulse=pulse.zhpulse,
                                     phase=pulse.pulse.relative_phase,
                                 )
                         elif isinstance(pulse, ZhSweeperLine):
-                            exp.delay(signal=f"drive{q}", time=pulse.zhsweeper)
+                            exp.delay(signal=qubit.drive.name, time=pulse.zhsweeper)
 
-                    if len(self.sequence[f"readout{q}"]) > 0 and isinstance(
-                        self.sequence[f"readout{q}"][0], ZhSweeperLine
+                    if len(self.sequence[qubit.readout.name]) > 0 and isinstance(
+                        self.sequence[qubit.readout.name][0], ZhSweeperLine
                     ):
                         exp.delay(
-                            signal=f"drive{q}",
-                            time=self.sequence[f"readout{q}"][0].zhsweeper,
+                            signal=qubit.drive.name,
+                            time=self.sequence[qubit.readout.name][0].zhsweeper,
                         )
-                        self.sequence[f"readout{q}"].remove(
-                            self.sequence[f"readout{q}"][0]
+                        self.sequence[qubit.readout.name].remove(
+                            self.sequence[qubit.readout.name][0]
                         )
 
                 previous_section = section_uid
@@ -913,16 +930,15 @@ def find_subsequence_finish(
         self,
         measurement_number: int,
         line: str,
-        quantum_elements: Union[Dict[str, Qubit], Dict[str, Coupler]],
+        quantum_elements: Union[Qubit, Coupler],
     ) -> Tuple[int, str]:
         """Find the finishing time and qubit for a given sequence.
 
         Args:
             measurement_number (int): number of the measure pulse.
             line (str): line from which measure the finishing time.
                 e.g.: "drive", "flux", "couplerflux"
-            quantum_elements (dict[str, Qubit]|dict[str, Coupler]): qubits or couplers from
-                which measure the finishing time.
+            quantum_elements: qubits or couplers from which to measure the finishing time.
 
         Returns:
             time_finish (int): Finish time of the last pulse of the subsequence
@@ -933,14 +949,10 @@ def find_subsequence_finish(
         time_finish = 0
         sequence_finish = "None"
         for quantum_element in quantum_elements:
-            if (
-                len(self.sub_sequences[f"{line}{quantum_element}"])
-                <= measurement_number
-            ):
+            channel_name = getattr(quantum_element, line).name
+            if len(self.sub_sequences[channel_name]) <= measurement_number:
                 continue
-            for pulse in self.sub_sequences[f"{line}{quantum_element}"][
-                measurement_number
-            ]:
+            for pulse in self.sub_sequences[channel_name][measurement_number]:
                 if pulse.pulse.finish > time_finish:
                     time_finish = pulse.pulse.finish
                     sequence_finish = f"{line}{quantum_element}"
@@ -955,12 +967,11 @@ def measure_relax(self, exp, qubits, couplers, relaxation_time, acquisition_type
         qubit_readout_schedule = defaultdict(list)
         iq_angle_readout_schedule = defaultdict(list)
         for qubit in qubits.values():
-            q = qubit.name  # pylint: disable=C0103
             iq_angle = qubit.iq_angle
-            if len(self.sequence[f"readout{q}"]) != 0:
-                for i, pulse in enumerate(self.sequence[f"readout{q}"]):
+            if len(self.sequence[qubit.readout.name]) != 0:
+                for i, pulse in enumerate(self.sequence[qubit.readout.name]):
                     readout_schedule[i].append(pulse)
-                    qubit_readout_schedule[i].append(q)
+                    qubit_readout_schedule[i].append(qubit)
                     iq_angle_readout_schedule[i].append(iq_angle)
 
         weights = {}
@@ -973,7 +984,7 @@ def measure_relax(self, exp, qubits, couplers, relaxation_time, acquisition_type
         ):
             qd_finish = self.find_subsequence_finish(i, "drive", qubits_readout)
             qf_finish = self.find_subsequence_finish(i, "flux", qubits_readout)
-            cf_finish = self.find_subsequence_finish(i, "couplerflux", couplers)
+            cf_finish = self.find_subsequence_finish(i, "flux", couplers.values())
             finish_times = np.array(
                 [
                     qd_finish,
@@ -989,11 +1000,12 @@ def measure_relax(self, exp, qubits, couplers, relaxation_time, acquisition_type
                 play_after = f"sequence_{latest_sequence['line']}_{i}"
             # Section on the outside loop allows for multiplex
             with exp.section(uid=f"sequence_measure_{i}", play_after=play_after):
-                for pulse, q, iq_angle in zip(pulses, qubits_readout, iq_angles):
+                for pulse, qubit, iq_angle in zip(pulses, qubits_readout, iq_angles):
+                    q = qubit.name
                     pulse.zhpulse.uid += str(i)
 
                     exp.delay(
-                        signal=f"acquire{q}",
+                        signal=acquire_signal_name(qubit),
                         time=self.smearing * NANO_TO_SECONDS,
                     )
 
@@ -1040,18 +1052,18 @@ def measure_relax(self, exp, qubits, couplers, relaxation_time, acquisition_type
 
                     measure_pulse_parameters = {"phase": 0}
 
-                    if i == len(self.sequence[f"readout{q}"]) - 1:
+                    if i == len(self.sequence[qubit.readout.name]) - 1:
                         reset_delay = relaxation_time * NANO_TO_SECONDS
                     else:
                         reset_delay = 0
 
                     exp.measure(
-                        acquire_signal=f"acquire{q}",
+                        acquire_signal=acquire_signal_name(qubit),
                         handle=f"sequence{q}_{i}",
                         integration_kernel=weight,
                         integration_kernel_parameters=None,
                         integration_length=None,
-                        measure_signal=f"measure{q}",
+                        measure_signal=qubit.readout.name,
                         measure_pulse=pulse.zhpulse,
                         measure_pulse_length=round(
                             pulse.pulse.duration * NANO_TO_SECONDS, 9
@@ -1116,8 +1128,8 @@ def sweep(self, qubits, couplers, sequence: PulseSequence, options, *sweepers):
         results = {}
         for qubit in qubits.values():
             q = qubit.name  # pylint: disable=C0103
-            if len(self.sequence[f"readout{q}"]) != 0:
-                for i, ropulse in enumerate(self.sequence[f"readout{q}"]):
+            if len(self.sequence[qubit.readout.name]) != 0:
+                for i, ropulse in enumerate(self.sequence[qubit.readout.name]):
                     exp_res = self.results.get_data(f"sequence{q}_{i}")
                     # if using singleshot, the first axis contains shots,
                     # i.e.: (nshots, sweeper_1, sweeper_2)
@@ -1151,12 +1163,13 @@ def sweep_recursion(self, qubits, couplers, exp, exp_calib, exp_options):
 
         if sweeper.parameter is Parameter.frequency:
             for pulse in sweeper.pulses:
-                line = "drive" if pulse.type is PulseType.DRIVE else "measure"
+                line = "drive" if pulse.type is PulseType.DRIVE else "readout"
                 zhsweeper = ZhSweeper(
                     pulse, sweeper, qubits[sweeper.pulses[0].qubit]
                 ).zhsweeper
                 zhsweeper.uid = "frequency"  # Changing the name from "frequency" breaks it f"frequency_{i}
-                exp_calib[f"{line}{pulse.qubit}"] = lo.SignalCalibration(
+                channel_name = getattr(qubits[pulse.qubit], line).name
+                exp_calib[channel_name] = lo.SignalCalibration(
                     oscillator=lo.Oscillator(
                         frequency=zhsweeper,
                         modulation_type=lo.ModulationType.HARDWARE,