[QHC-760] Implement OPX1000 with QOP320

requirements.txt

@@ -9,7 +9,7 @@ urllib3==1.26.12
 qpysequence==0.10.4
 h5py==3.11.0
 papermill==2.4.0
-qm-qua==1.2.0
+qm-qua==1.2.1a2
 qualang-tools==0.17.6
 networkx==3.2.1
 ruamel.yaml==0.18.6

src/qililab/instruments/quantum_machines/quantum_machines_cluster.py

@@ -19,11 +19,18 @@
 from typing import Any, cast
 
 import numpy as np
-from qm import DictQuaConfig, QmJob, QuantumMachine, QuantumMachinesManager, SimulationConfig
+from qm import (
+    DictQuaConfig,
+    OPX1000ControllerConfigType,
+    Program,
+    QmJob,
+    QuantumMachine,
+    QuantumMachinesManager,
+    SimulationConfig,
+)
 from qm.api.v2.job_api import JobApi
 from qm.jobs.running_qm_job import RunningQmJob
 from qm.octave import QmOctaveConfig
-from qm.program import Program
 
 from qililab.instruments.instrument import Instrument, ParameterNotFound
 from qililab.instruments.utils import InstrumentFactory
@@ -440,7 +447,7 @@ def initial_setup(self):
     def turn_on(self):
         """Turns on the instrument."""
         if not self._is_connected_to_qm:
-            self._qm = self._qmm.open_qm(config=self._config, close_other_machines=False)
+            self._qm = self._qmm.open_qm(config=self._config, close_other_machines=True)
             self._compiled_program_cache = {}
             self._is_connected_to_qm = True
 
@@ -475,7 +482,7 @@ def append_configuration(self, configuration: dict):
             self._config = cast(DictQuaConfig, merged_configuration)
             # If we are already connected, reopen the connection with the new configuration
             if self._is_connected_to_qm:
-                self._qm = self._qmm.open_qm(config=self._config, close_other_machines=False)  # type: ignore[assignment]
+                self._qm = self._qmm.open_qm(config=self._config, close_other_machines=True)  # type: ignore[assignment]
                 self._compiled_program_cache = {}
 
     def run_octave_calibration(self):
@@ -484,7 +491,7 @@ def run_octave_calibration(self):
         for element in elements:
             self._qm.calibrate_element(element)
 
-    def get_controller_type_from_bus(self, bus: str) -> str | None:
+    def get_controller_type_from_bus(self, bus: str) -> str:
         """Gets the OPX controller name of the bus used
 
         Args:
@@ -497,20 +504,23 @@ def get_controller_type_from_bus(self, bus: str) -> str | None:
             str | None: Alias of the controller, either opx1 or opx1000.
         """
 
-        if "RF_inputs" in self._config["elements"][bus]:
-            octave = self._config["elements"][bus]["RF_inputs"]["port"][0]
-            controller_name = self._config["octaves"][octave]["connectivity"]
-        elif "mixInputs" in self._config["elements"][bus]:
-            controller_name = self._config["elements"][bus]["mixInputs"]["I"][0]
-        elif "singleInput" in self._config["elements"][bus]:
-            controller_name = self._config["elements"][bus]["singleInput"]["port"][0]
+        if "RF_inputs" in self.config["elements"][bus]:
+            octave = self._config["elements"][bus]["RF_inputs"]["port"]
+            if "connectivity" in self._config["octaves"][octave[0]]:
+                controller_name = self._config["octaves"][octave[0]]["connectivity"]
+            else:
+                controller_name = self._config["octaves"][octave[0]]["RF_outputs"][octave[1]]["I_connection"][0]
+        elif "mixInputs" in self.config["elements"][bus]:
+            controller_name = self.config["elements"][bus]["mixInputs"]["I"][0]
+        elif "singleInput" in self.config["elements"][bus]:
+            controller_name = self.config["elements"][bus]["singleInput"]["port"][0]
 
         for controller in self.settings.controllers:
             if controller["name"] == controller_name:
                 return controller["type"] if "type" in controller else "opx1"
         raise AttributeError(f"Controller with bus {bus} does not exist")
 
-    def get_controller_from_element(self, element: dict, key: str | None) -> tuple[str, int, int | None]:
+    def get_controller_from_element(self, element: dict, key: str | None = None) -> tuple[str, int, int | None]:
         """Get controller name, port and FEM (if applicable) from element
 
         Args:
@@ -819,6 +829,17 @@ def get_parameter_of_bus(self, bus: str, parameter: Parameter) -> float | str |
                 return settings_config_dict["controllers"][con_name]["analog_inputs"][out_value]["offset"]  # type: ignore[typeddict-item]
             return settings_config_dict["controllers"][con_name]["fems"][con_fem]["analog_inputs"][out_value]["offset"]  # type: ignore[typeddict-item]
 
+        if parameter == Parameter.IS_AMPLIFIED:
+            opx_type = self.get_controller_type_from_bus(bus=bus)
+            if opx_type != "opx1000":
+                return False
+            key = "I" if ("rf_inputs" in element or "mix_inputs" in element) else None
+            controller, port, fem = self.get_controller_from_element(element=element, key=key)
+            opx1000_controller_config = cast(
+                OPX1000ControllerConfigType, settings_config_dict["controllers"][controller]
+            )
+            return opx1000_controller_config["fems"][fem]["analog_outputs"][con_port]["output_mode"] == "amplified"
+
         raise ParameterNotFound(f"Could not find parameter {parameter} in instrument {self.name}")
 
     def compile(self, program: Program) -> str:

src/qililab/platform/platform.py

@@ -823,13 +823,17 @@ def compile_qprogram(
                 for bus in buses
                 if isinstance(bus.system_control, ReadoutSystemControl)
             }
+            is_amplified: dict[str, bool] = {
+                bus.alias: bool(bus.get_parameter(parameter=Parameter.IS_AMPLIFIED) or False) for bus in buses
+            }
 
             compiler = QuantumMachinesCompiler()
             return compiler.compile(
                 qprogram=qprogram,
                 bus_mapping=bus_mapping,
                 thresholds=thresholds,
                 threshold_rotations=threshold_rotations,
+                is_amplified=is_amplified,
                 calibration=calibration,
             )
         raise NotImplementedError("Compiling QProgram for a mixture of instruments is not supported.")

src/qililab/qprogram/quantum_machines_compiler.py

@@ -116,7 +116,8 @@ class QuantumMachinesCompiler:
 
     FREQUENCY_COEFF = 1
     PHASE_COEFF = 2 * np.pi
-    VOLTAGE_COEFF = 2
+    VOLTAGE_COEFF = 2.0
+    VOLTAGE_AMPLIFICATION_COEFF = 5.0
     WAIT_COEFF = 4
     MINIMUM_TIME = 4
 
@@ -152,6 +153,7 @@ def compile(
         bus_mapping: dict[str, str] | None = None,
         thresholds: dict[str, float] | None = None,
         threshold_rotations: dict[str, float] | None = None,
+        is_amplified: dict[str, bool] | None = None,
         calibration: Calibration | None = None,
     ) -> QuantumMachinesCompilationOutput:
         """Compile QProgram to QUA's Program.
@@ -591,10 +593,12 @@ def __add_weights_to_configuration(self, weights: IQPair, rotation: float):
         # Return weights names
         return A, B, C, D
 
-    def __add_waveform_to_configuration(self, waveform: Waveform):
+    def __add_waveform_to_configuration(self, waveform: Waveform, is_amplified: bool = False):
         waveform_name = QuantumMachinesCompiler.__hash_waveform(waveform)
         if waveform_name not in self._configuration["waveforms"]:
-            self._configuration["waveforms"][waveform_name] = QuantumMachinesCompiler.__waveform_to_config(waveform)
+            self._configuration["waveforms"][waveform_name] = QuantumMachinesCompiler.__waveform_to_config(
+                waveform, is_amplified=is_amplified
+            )
         return waveform_name
 
     @staticmethod
@@ -609,12 +613,19 @@ def __hash_waveform(waveform: Waveform):
         return hash_result.hexdigest()[:8]
 
     @staticmethod
-    def __waveform_to_config(waveform: Waveform):
+    def __waveform_to_config(waveform: Waveform, is_amplified: bool = False):
         if isinstance(waveform, Square):
             amplitude = waveform.amplitude / QuantumMachinesCompiler.VOLTAGE_COEFF
+            if is_amplified:
+                amplitude *= QuantumMachinesCompiler.VOLTAGE_AMPLIFICATION_COEFF
+                if amplitude == 2.5:
+                    amplitude = 2.499
             return {"type": "constant", "sample": amplitude}
 
         envelope = waveform.envelope() / QuantumMachinesCompiler.VOLTAGE_COEFF
+        if is_amplified:
+            envelope *= QuantumMachinesCompiler.VOLTAGE_AMPLIFICATION_COEFF
+            envelope = np.where(envelope == 2.5, 2.499, envelope)
         return {"type": "arbitrary", "samples": envelope.tolist()}
 
     @staticmethod

src/qililab/typings/enums.py

@@ -344,6 +344,7 @@ class Parameter(str, Enum):
     B = "b"
     T_PHI = "t_phi"
     GATE_OPTIONS = "options"
+    IS_AMPLIFIED = "is_amplified"
 
     @classmethod
     def to_yaml(cls, representer, node):

tests/instruments/quantum_machines/test_quantum_machines_cluster.py

@@ -638,6 +638,7 @@ def test_set_parameter_of_bus_method_raises_exception_when_parameter_not_found(
             ("drive_q0", Parameter.LO_FREQUENCY, "qmm"),
             ("drive_q0_rf", Parameter.LO_FREQUENCY, "qmm_with_octave"),
             ("drive_q0", Parameter.IF, "qmm"),
+            ("drive_q0_rf", Parameter.IF, "qmm_with_octave_custom_connectivity"),
             ("readout_q0", Parameter.GAIN, "qmm"),
             ("drive_q0_rf", Parameter.GAIN, "qmm_with_octave"),
             ("readout_q0", Parameter.TIME_OF_FLIGHT, "qmm"),

tests/platform/test_platform.py

@@ -775,6 +775,7 @@ def test_execute_qprogram_with_quantum_machines(self, platform_quantum_machines:
             patch.object(QuantumMachinesCluster, "compile") as compile_program,
             patch.object(QuantumMachinesCluster, "run_compiled_program") as run_compiled_program,
             patch.object(QuantumMachinesCluster, "get_acquisitions") as get_acquisitions,
+            patch.object(QuantumMachinesCluster, "get_controller_type_from_bus"),
         ):
             cluster = platform_quantum_machines.get_element("qmm")
             config.return_value = cluster.settings.to_qua_config()