dev: reformat code

leeq/compiler/lbnl_qubic/circuit_list_compiler.py

@@ -48,7 +48,9 @@ def compare_dicts(dict1, dict2, rtol=1e-05, atol=1e-08):
         value2 = dict2[key]
 
         # If values are both numbers, use np.allclose for comparison
-        if isinstance(value1, (int, float)) and isinstance(value2, (int, float)):
+        if isinstance(
+                value1, (int, float)) and isinstance(
+                value2, (int, float)):
             if not np.allclose(value1, value2, rtol=rtol, atol=atol):
                 return False
 
@@ -96,7 +98,8 @@ def __init__(self, leeq_channel_to_qubic_channel: dict[str, str]):
         # lpb has changed, the waveform is dirty. Otherwise, it kept clean.
         self._envelope_dirty = False
         # Indicates if the waveform shape has been changed. If not we can skip uploading
-        # the frequency to board to save time. It is done by checking the parameters for each lpbs.
+        # the frequency to board to save time. It is done by checking the
+        # parameters for each lpbs.
         self._frequency_dirty = False
         # Indicates if the command (lpb or phase etc) has been changed.
         # If not we can omit uploading the waveform to board to save time.
@@ -152,7 +155,8 @@ def func(dt, **kwargs):
                 Tuple[np.ndarray, np.ndarray]: The time list and the pulse shape envelope.
             """
 
-            # amp will be modified by the qubic system, so we always pass amp = 1
+            # amp will be modified by the qubic system, so we always pass amp =
+            # 1
             kwargs = kwargs.copy()
             kwargs['amp'] = 1
 
@@ -203,7 +207,8 @@ def compile_lpb(self, context: ExperimentContext,
         self._circuit_list = []
         self._phase_shift = {}
 
-        # Keep track of the compiled UUID, and remove untouched UUID from the dirty tracking data structures
+        # Keep track of the compiled UUID, and remove untouched UUID from the
+        # dirty tracking data structures
         self._compiled_lpb_uuid = []
         self._command_dirty = False
         self._frequency_dirty = False
@@ -220,7 +225,8 @@ def compile_lpb(self, context: ExperimentContext,
             }
         }
 
-        # Now look at the compiled UUID, and remove unseen UUID from the dirty tracking data structures
+        # Now look at the compiled UUID, and remove unseen UUID from the dirty
+        # tracking data structures
         for lpb_id in list(self._sweep_lpb_uuid_to_last_selection.keys()):
             if lpb_id not in self._compiled_lpb_uuid:
                 del self._sweep_lpb_uuid_to_last_selection[lpb_id]
@@ -250,11 +256,12 @@ def _check_parameter_if_dirty(self, lpb: LogicalPrimitive):
             lpb (LogicalPrimitive): a lpb to be checked.
         """
 
-        # Track the UUID of the lpbs, if they are not seen we remove them from the tracking data structure
+        # Track the UUID of the lpbs, if they are not seen we remove them from
+        # the tracking data structure
         self._compiled_lpb_uuid.append(lpb.uuid)
 
-        _parameter_diff = lambda x: x in parameters and \
-                                    parameters[x] != self._lpb_uuid_to_parameter_last_compiled[lpb.uuid][x]
+        def _parameter_diff(x): return x in parameters and \
+            parameters[x] != self._lpb_uuid_to_parameter_last_compiled[lpb.uuid][x]
 
         parameters = lpb.get_parameters()
         lpb_id = lpb.uuid
@@ -269,7 +276,8 @@ def _check_parameter_if_dirty(self, lpb: LogicalPrimitive):
                 # Passed check
                 return
 
-            last_compiled_value = self._lpb_uuid_to_parameter_last_compiled[lpb_id].copy()
+            last_compiled_value = self._lpb_uuid_to_parameter_last_compiled[lpb_id].copy(
+            )
 
             if _parameter_diff('freq'):
                 self._frequency_dirty = True
@@ -292,11 +300,13 @@ def _check_parameter_if_dirty(self, lpb: LogicalPrimitive):
                 del last_compiled_value['phase']
 
             if parameters != last_compiled_value:
-                # Something else has changed as well, not sure if its shape or command, then mark them both dirty
+                # Something else has changed as well, not sure if its shape or
+                # command, then mark them both dirty
                 self._command_dirty = True
                 self._envelope_dirty = True
 
-        self._lpb_uuid_to_parameter_last_compiled[lpb_id] = lpb.get_parameters()
+        self._lpb_uuid_to_parameter_last_compiled[lpb_id] = lpb.get_parameters(
+        )
 
     @_compile_lpb.register
     def _(self, lpb: LogicalPrimitive):
@@ -333,8 +343,7 @@ def _(self, lpb: LogicalPrimitive):
             parameters = lpb.get_parameters()
         else:
             parameters = exp_manager.status().get_modified_lpb_parameters_from_channel_callback(
-                channel=lpb.channel, parameters=lpb.get_parameters()
-            )
+                channel=lpb.channel, parameters=lpb.get_parameters())
 
         phase_shift = 0
 
@@ -347,11 +356,14 @@ def _(self, lpb: LogicalPrimitive):
             phase_shift += self._phase_shift[lpb.channel][lpb.transition_name]
 
         # Here we can't put all the parameters through, as they will be used as hashes to determine if the
-        # waveform has changed. We only put the parameters that are used in the waveform generation.
+        # waveform has changed. We only put the parameters that are used in the
+        # waveform generation.
         env_func = self._get_envelope_function(parameters['shape'])
         required_parameters = inspect.signature(env_func).parameters
-        parameters_keeping = {key: parameters[key] for key in required_parameters if (key in parameters) and \
-                              key not in ['amp', 'freq', 'phase']}
+        parameters_keeping = {
+            key: parameters[key] for key in required_parameters if (
+                key in parameters) and key not in [
+                'amp', 'freq', 'phase']}
 
         env = {"env_func": env_func, "paradict": parameters_keeping}
 
@@ -431,16 +443,17 @@ def _(self, lpb: MeasurementPrimitive):
             modified_parameters = lpb.get_parameters()
         else:
             modified_parameters = exp_manager.status().get_modified_lpb_parameters_from_channel_callback(
-                channel=lpb.channel, parameters=lpb.get_parameters()
-            )
+                channel=lpb.channel, parameters=lpb.get_parameters())
 
         # Here we can't put all the parameters through, as they will be used as hashes to determine if the
-        # waveform has changed. We only put the parameters that are used in the waveform generation.
+        # waveform has changed. We only put the parameters that are used in the
+        # waveform generation.
         env_func = self._get_envelope_function(modified_parameters['shape'])
         required_parameters = inspect.signature(env_func).parameters
-        parameters_keeping = {key: modified_parameters[key] for key in required_parameters
-                              if (key in modified_parameters) and \
-                              key not in ['amp', 'freq', 'phase']}
+        parameters_keeping = {
+            key: modified_parameters[key] for key in required_parameters if (
+                key in modified_parameters) and key not in [
+                'amp', 'freq', 'phase']}
 
         env = {"env_func": env_func, "paradict": parameters_keeping}
 
@@ -456,7 +469,8 @@ def _(self, lpb: MeasurementPrimitive):
 
         # A delay is introduced between the start of the drive and start of the measurement.
         # First it takes roughly 100ns for the signal to arrive the resonator in the fridge and
-        # comback. Also the delay prevents the on-board crosstalk of the qubit drive signal being picked up
+        # comback. Also the delay prevents the on-board crosstalk of the qubit
+        # drive signal being picked up
         delay_between_drive_and_measure = {
             'name': 'delay',
             't': 100e-9,
@@ -485,14 +499,15 @@ def _(self, lpb: MeasurementPrimitive):
         }
 
         if primitive_scope in self._qubic_channel_to_lpb_uuid and self._qubic_channel_to_lpb_uuid[
-            primitive_scope] != lpb.uuid:
+                primitive_scope] != lpb.uuid:
             msg = "Two measurement primitives exists for a same channel, which is not supported."
             logger.error(msg)
             raise NotImplementedError(msg)
 
         self._qubic_channel_to_lpb_uuid[primitive_scope] = lpb.uuid
 
-        return [drive_pulse, delay_between_drive_and_measure, demodulate_pulse], {primitive_scope + ".rdrv"}
+        return [drive_pulse, delay_between_drive_and_measure,
+                demodulate_pulse], {primitive_scope + ".rdrv"}
 
     @_compile_lpb.register
     def _(self, lpb: LogicalPrimitiveBlockSerial):
@@ -530,7 +545,8 @@ def _(self, lpb: LogicalPrimitiveBlockSerial):
         # the delay primitive to have the entire scope of the block.
         if len(
                 child_circuits[0]) == 1 and child_circuits[0][0]["name"] == "delay":
-            child_circuits[0][0]["scope"] = list(set(x.split('.')[0] for x in block_scope))
+            child_circuits[0][0]["scope"] = list(
+                set(x.split('.')[0] for x in block_scope))
             child_scopes[0] = block_scope
 
         compiled_circuit.extend(child_circuits[0])
@@ -542,20 +558,21 @@ def _(self, lpb: LogicalPrimitiveBlockSerial):
             # For this operation we do not need a barrier.
             if len(
                     child_circuits[i]) == 1 and child_circuits[i][0]["name"] == "delay":
-                child_circuits[i][0]["scope"] = list(set(x.split('.')[0] for x in block_scope))
+                child_circuits[i][0]["scope"] = list(
+                    set(x.split('.')[0] for x in block_scope))
                 if len(block_scope) > 1:
-                    # If the block contains more than 1 channel, we need a barrier
-                    compiled_circuit.append(
-                        {"name": "barrier", "scope": list(set(x.split('.')[0] for x in block_scope))}
-                    )
+                    # If the block contains more than 1 channel, we need a
+                    # barrier
+                    compiled_circuit.append({"name": "barrier", "scope": list(
+                        set(x.split('.')[0] for x in block_scope))})
             else:
                 # If we are only dealing with one channel, and the previous and current child are on the same channel,
-                # we do not need a barrier. Otherwise add a barrier at block scope.
+                # we do not need a barrier. Otherwise add a barrier at block
+                # scope.
                 union_scope = child_scopes[i - 1].union(child_scopes[i])
                 if len(block_scope) > 1 and len(union_scope) > 1:
-                    compiled_circuit.append(
-                        {"name": "barrier", "scope": list(set(x.split('.')[0] for x in block_scope))}
-                    )
+                    compiled_circuit.append({"name": "barrier", "scope": list(
+                        set(x.split('.')[0] for x in block_scope))})
 
             compiled_circuit.extend(child_circuits[i])
 
@@ -608,8 +625,8 @@ def _(self, lpb: LogicalPrimitiveBlockParallel):
         compiled_circuit = sum(child_circuits, [])
         block_scope = set(block_scope)
         if len(block_scope) > 1:
-            compiled_circuit.append(
-                {"name": "barrier", "scope": list(set(x.split('.')[0] for x in block_scope))})
+            compiled_circuit.append({"name": "barrier", "scope": list(
+                set(x.split('.')[0] for x in block_scope))})
 
         return compiled_circuit, block_scope
 
@@ -641,7 +658,7 @@ def _(self, lpb: DelayPrimitive):
         self._check_parameter_if_dirty(lpb)
 
         return [{"name": "delay", "t": lpb.get_delay_time() /
-                                       1e6}], set()  # In seconds
+                 1e6}], set()  # In seconds
 
     @_compile_lpb.register
     def _(self, lpb: PhaseShift):
@@ -654,18 +671,17 @@ def _(self, lpb: PhaseShift):
             parameters = lpb.get_parameters()
         else:
             parameters = exp_manager.status().get_modified_lpb_parameters_from_channel_callback(
-                channel=lpb.channel, parameters=lpb.get_parameters()
-            )
+                channel=lpb.channel, parameters=lpb.get_parameters())
 
         if lpb.channel not in self._phase_shift:
             self._phase_shift[lpb.channel] = {}
 
         for k, m in parameters["transition_multiplier"].items():
             if k not in self._phase_shift[lpb.channel]:
                 self._phase_shift[lpb.channel][k] = m * \
-                                                    parameters["phase_shift"]
+                    parameters["phase_shift"]
             else:
                 self._phase_shift[lpb.channel][k] += m * \
-                                                     parameters["phase_shift"]
+                    parameters["phase_shift"]
 
         return [], set()

leeq/compiler/utils/pulse_shape_utils.py

@@ -109,7 +109,11 @@ def _load_built_in_pulse_shapes(self):
                 self.register_pulse_shape(
                     pulse_shape_name, pulse_shape_function)
 
-    def calculate_integrated_area(self, pulse_shape_name: str, sampling_rate: float = 1e3, **kwargs):
+    def calculate_integrated_area(
+            self,
+            pulse_shape_name: str,
+            sampling_rate: float = 1e3,
+            **kwargs):
         """
         Calculate the integrated area of the pulse shape with the given parameters.
 
@@ -122,7 +126,8 @@ def calculate_integrated_area(self, pulse_shape_name: str, sampling_rate: float
             float: The integrated area of the pulse shape.
         """
 
-        pulse_shape_envelope = self.compile_pulse_shape(pulse_shape_name, sampling_rate=sampling_rate, **kwargs)
+        pulse_shape_envelope = self.compile_pulse_shape(
+            pulse_shape_name, sampling_rate=sampling_rate, **kwargs)
         time_step = 1 / sampling_rate
         return np.real(pulse_shape_envelope.sum() * time_step)
 

leeq/compiler/utils/pulse_shapes/basic_shapes.py

@@ -14,8 +14,11 @@
 
 
 def gaussian(
-        sampling_rate: int, amp: float, phase: float, width: float, trunc: float
-) -> np.array:
+        sampling_rate: int,
+        amp: float,
+        phase: float,
+        width: float,
+        trunc: float) -> np.array:
     """
     Generates a complex Gaussian wave packet.
 
@@ -39,9 +42,9 @@ def gaussian(
     # Generate and return the Gaussian wave packet.
     # You may extend the function to incorporate it in the future.
     return (
-            amp
-            * np.exp(1.0j * phase)
-            * np.exp(-(((t - gauss_width) / gauss_width) ** 2)).astype("complex64")
+        amp
+        * np.exp(1.0j * phase)
+        * np.exp(-(((t - gauss_width) / gauss_width) ** 2)).astype("complex64")
     )
 
 
@@ -89,7 +92,8 @@ def gaussian_drag(
     shape = np.exp(-(((t - gauss_width) / gauss_width) ** 2), dtype="cfloat")
 
     # Compute the imaginary part using the derivative of the Gaussian
-    shape.imag = shape.real * 2.0 * t / (2.0 * np.pi * alpha2) / gauss_width ** 2
+    shape.imag = shape.real * 2.0 * t / \
+        (2.0 * np.pi * alpha2) / gauss_width ** 2
 
     # Return the pulse shape, with applied amplitude and phase modulation
     return amp * np.exp(1.0j * phase) * shape
@@ -136,9 +140,9 @@ def blackman(
 
     # Update the real part of midshape based on the Blackman window formula
     midshape += 1 - (
-            a0
-            - a1 * np.cos((2.0 * np.pi * (t + width / 2.0)) / width, dtype="cfloat")
-            + a2 * np.cos(4.0 * np.pi * (t + width / 2.0) / width, dtype="cfloat")
+        a0
+        - a1 * np.cos((2.0 * np.pi * (t + width / 2.0)) / width, dtype="cfloat")
+        + a2 * np.cos(4.0 * np.pi * (t + width / 2.0) / width, dtype="cfloat")
     )
 
     # Return the scaled and phased shape
@@ -205,16 +209,16 @@ def blackman_drag(
 
     # Apply the Blackman function
     midshape += 1 - (
-            a0
-            - a1 * np.cos((2.0 * np.pi * (t + width / 2.0)) / width, dtype="cfloat")
-            + a2 * np.cos(4.0 * np.pi * (t + width / 2.0) / width, dtype="cfloat")
+        a0
+        - a1 * np.cos((2.0 * np.pi * (t + width / 2.0)) / width, dtype="cfloat")
+        + a2 * np.cos(4.0 * np.pi * (t + width / 2.0) / width, dtype="cfloat")
     )
 
     # Handle imaginary part with a DRAG correction term
     alpha2 = alpha * 2.0
     midshape.imag = -(
-            a1 * 2.0 * np.pi / width * np.sin((2.0 * np.pi * (t + width / 2.0)) / width)
-            - a2 * 4.0 * np.pi / width * np.sin(4.0 * np.pi * (t + width / 2.0) / width)
+        a1 * 2.0 * np.pi / width * np.sin((2.0 * np.pi * (t + width / 2.0)) / width)
+        - a2 * 4.0 * np.pi / width * np.sin(4.0 * np.pi * (t + width / 2.0) / width)
     ) / (2.0 * np.pi * alpha2)
 
     # Return the pulse shaped with phase and amplitude
@@ -255,10 +259,10 @@ def soft_square(
 
     t -= 0.5 * delay
     y = (
-            amp
-            * np.exp(1.0j * (phase + phase_shift))
-            * 0.5
-            * (np.tanh((t + 0.5 * width) / rise) - np.tanh((t - 0.5 * width) / rise))
+        amp
+        * np.exp(1.0j * (phase + phase_shift))
+        * 0.5
+        * (np.tanh((t + 0.5 * width) / rise) - np.tanh((t - 0.5 * width) / rise))
     )
 
     if ex_delay > 0: