Fixed device signatures, updated tests.

Author: nwittler

c3/generator/devices.py

@@ -2,7 +2,7 @@
 import tempfile
 import warnings
 import hjson
-from typing import Callable, Dict, Any
+from typing import Callable, Dict, Any, List
 import tensorflow as tf
 import numpy as np
 from c3.signal.pulse import Envelope, Carrier
@@ -26,7 +26,7 @@ class Device(C3obj):
 
     Parameters
     ----------
-    resolution: np.float64
+    resolution: float
         Number of samples per second this device operates at.
     """
 
@@ -69,32 +69,32 @@ def asdict(self) -> Dict[str, Any]:
     def __str__(self) -> str:
         return hjson.dumps(self.asdict(), default=hjson_encode)
 
-    def calc_slice_num(self, t_start: np.float64, t_end: np.float64) -> None:
+    def calc_slice_num(self, t_start: float = 0.0, t_end: float = 0.0) -> None:
         """
         Effective number of time slices given start, end and resolution.
 
         Parameters
         ----------
-        t_start: np.float64
+        t_start: float
             Starting time for this device.
-        t_end: np.float64
+        t_end: float
             End time for this device.
         """
         res = self.resolution
         self.slice_num = int(np.abs(t_start - t_end) * res)
         # return self.slice_num
 
     def create_ts(
-        self, t_start: np.float64, t_end: np.float64, centered: bool = True
+        self, t_start: float = 0, t_end: float = 0, centered: bool = True
     ) -> tf.constant:
         """
         Compute time samples.
 
         Parameters
         ----------
-        t_start: np.float64
+        t_start: float
             Starting time for this device.
-        t_end: np.float64
+        t_end: float
             End time for this device.
         centered: boolean
             Sample in the middle of an interval, otherwise at the beginning.
@@ -121,6 +121,31 @@ def create_ts(
         ts = tf.linspace(t_start, t_end, num)
         return ts
 
+    def process(
+        self, instr: Instruction, chan: str, signals: List[Dict[str, Any]]
+    ) -> Dict[str, Any]:
+        """To be implemented by inheriting class.
+
+        Parameters
+        ----------
+        instr : Instruction
+            Information about the instruction or gate to be excecuted.
+        chan : str
+            Identifier of the drive line
+        signals : List[Dict[str, Any]]
+            List of potentially multiple input signals to this device.
+
+        Returns
+        -------
+        Dict[str, Any]
+            Output signal of this device.
+
+        Raises
+        ------
+        NotImplementedError
+        """
+        raise NotImplementedError()
+
 
 @dev_reg_deco
 class Readout(Device):
@@ -176,7 +201,7 @@ def __init__(self, **props):
         self.outputs = props.pop("outputs", 1)
 
     def process(
-        self, instr: Instruction, chan: str, mixed_signal: Dict[str, Any]
+        self, instr: Instruction, chan: str, mixed_signal: List[Dict[str, Any]]
     ) -> Dict[str, Any]:
         """Transform signal from value of V to Hz.
 
@@ -191,8 +216,8 @@ def process(
             Waveform as control amplitudes
         """
         v2hz = self.params["V_to_Hz"].get_value()
-        self.signal["values"] = mixed_signal["values"] * v2hz
-        self.signal["ts"] = mixed_signal["ts"]
+        self.signal["values"] = mixed_signal[0]["values"] * v2hz
+        self.signal["ts"] = mixed_signal[0]["ts"]
         return self.signal
 
 
@@ -202,7 +227,7 @@ class Crosstalk(Device):
     Device to phenomenologically include crosstalk in the model by explicitly mixing
     drive lines.
 
-    Parameters
+    Parameters^
     ----------
 
     crosstalk_matrix: tf.constant
@@ -234,7 +259,9 @@ def __init__(self, **props):
         self.outputs = props.pop("outputs", 1)
         self.params["crosstalk_matrix"] = props.pop("crosstalk_matrix", None)
 
-    def process(self, signal: Dict[str, Any]) -> Dict[str, Any]:
+    def process(
+        self, instr: Instruction, chan: str, signals: List[Dict[str, Any]]
+    ) -> Dict[str, Any]:
         """
         Mix channels in the input signal according to a crosstalk matrix.
 
@@ -258,11 +285,11 @@ def process(self, signal: Dict[str, Any]) -> Dict[str, Any]:
 
         """
         xtalk = self.params["crosstalk_matrix"]
-        signals = [signal[ch]["values"] for ch in self.crossed_channels]
-        crossed_signals = xtalk.get_value() @ signals
+        signal = [signals[0][ch]["values"] for ch in self.crossed_channels]
+        crossed_signals = xtalk.get_value() @ signal
         for indx, ch in enumerate(self.crossed_channels):
-            signal[ch]["values"] = crossed_signals[indx]
-        return signal
+            signals[0][ch]["values"] = crossed_signals[indx]
+        return signals[0]
 
 
 @dev_reg_deco
@@ -277,7 +304,7 @@ def __init__(self, **props):
         self.sampling_method = props.pop("sampling_method", "nearest")
 
     def process(
-        self, instr: Instruction, chan: str, awg_signal: Dict[str, Any]
+        self, instr: Instruction, chan: str, awg_signal: List[Dict[str, Any]]
     ) -> Dict[str, Any]:
         """Resample the awg values to higher resolution.
 
@@ -297,12 +324,12 @@ def process(
             Inphase and Quadrature compontent of the upsampled signal.
         """
         ts = self.create_ts(instr.t_start, instr.t_end, centered=True)
-        old_dim = awg_signal["inphase"].shape[0]
+        old_dim = awg_signal[0]["inphase"].shape[0]
         new_dim = ts.shape[0]
         # TODO add following zeros
         inphase = tf.reshape(
             tf.image.resize(
-                tf.reshape(awg_signal["inphase"], shape=[1, old_dim, 1]),
+                tf.reshape(awg_signal[0]["inphase"], shape=[1, old_dim, 1]),
                 size=[1, new_dim],
                 method=self.sampling_method,
             ),
@@ -311,7 +338,7 @@ def process(
         inphase = tf.cast(inphase, tf.float64)
         quadrature = tf.reshape(
             tf.image.resize(
-                tf.reshape(awg_signal["quadrature"], shape=[1, old_dim, 1]),
+                tf.reshape(awg_signal[0]["quadrature"], shape=[1, old_dim, 1]),
                 size=[1, new_dim],
                 method=self.sampling_method,
             ),
@@ -335,7 +362,7 @@ def __init__(self, **props):
         # super().__init__(**props)
 
     def process(
-        self, instr: Instruction, chan: str, Hz_signal: Dict[str, Any]
+        self, instr: Instruction, chan: str, Hz_signal: List[Dict[str, Any]]
     ) -> Dict[str, Any]:
         """Apply a filter function to the signal."""
         raise Exception("C3:ERROR Not yet implemented.")
@@ -560,9 +587,11 @@ def process(self, instr, chan, iq_signal):
         offset = tf.exp(-((-1 - cen) ** 2) / (2 * sigma * sigma))
         # TODO make sure ratio of risetime and resolution is an integer
         risefun = gauss - offset
-        inphase = self.convolve(iq_signal["inphase"], risefun / tf.reduce_sum(risefun))
+        inphase = self.convolve(
+            iq_signal[0]["inphase"], risefun / tf.reduce_sum(risefun)
+        )
         quadrature = self.convolve(
-            iq_signal["quadrature"], risefun / tf.reduce_sum(risefun)
+            iq_signal[0]["quadrature"], risefun / tf.reduce_sum(risefun)
         )
         self.signal = {
             "inphase": inphase,
@@ -602,7 +631,7 @@ def process(self, instr, chan, iq_signal):
             Bandwidth limited IQ signal.
 
         """
-        res_diff = (iq_signal["ts"][1] - iq_signal["ts"][0]) / self.resolution - 1
+        res_diff = (iq_signal[0]["ts"][1] - iq_signal[0]["ts"][0]) / self.resolution - 1
         if res_diff > 1e-8:
             raise Exception(
                 "C3:Error:Actual time resolution differs from desired by {res_diff:1.3g}."
@@ -621,17 +650,17 @@ def process(self, instr, chan, iq_signal):
         offset = tf.exp(-((-1 - cen) ** 2) / (2 * sigma * sigma))
 
         risefun = gauss - offset
-        inphase = tf_convolve(iq_signal["inphase"], risefun / tf.reduce_sum(risefun))
+        inphase = tf_convolve(iq_signal[0]["inphase"], risefun / tf.reduce_sum(risefun))
         quadrature = tf_convolve(
-            iq_signal["quadrature"], risefun / tf.reduce_sum(risefun)
+            iq_signal[0]["quadrature"], risefun / tf.reduce_sum(risefun)
         )
 
         inphase = tf.math.real(inphase)
         quadrature = tf.math.real(quadrature)
         self.signal = {
             "inphase": inphase,
             "quadrature": quadrature,
-            "ts": iq_signal["ts"],
+            "ts": iq_signal[0]["ts"],
         }
         return self.signal
 
@@ -845,7 +874,7 @@ def __init__(self, **props):
         self.inputs = props.pop("inputs", 2)
         self.outputs = props.pop("outputs", 1)
 
-    def process(self, instr: Instruction, chan: str, in1: dict, in2: dict):
+    def process(self, instr: Instruction, chan: str, inputs: List[Dict[str, Any]]):
         """Combine signal from AWG and LO.
 
         Parameters
@@ -860,6 +889,7 @@ def process(self, instr: Instruction, chan: str, in1: dict, in2: dict):
         dict
             Mixed signal.
         """
+        in1, in2 = inputs
         i1 = in1["inphase"]
         q1 = in1["quadrature"]
         i2 = in2["inphase"]
@@ -1004,7 +1034,9 @@ def __init__(self, **props):
         self.freq_noise = props.pop("freq_noise", 0)
         self.amp_noise = props.pop("amp_noise", 0)
 
-    def process(self, instr: Instruction, chan: str) -> dict:
+    def process(
+        self, instr: Instruction, chan: str, signal: List[Dict[str, Any]]
+    ) -> dict:
         # TODO check somewhere that there is only 1 carrier per instruction
         ts = self.create_ts(instr.t_start, instr.t_end, centered=True)
         dt = ts[1] - ts[0]
@@ -1099,7 +1131,7 @@ def asdict(self) -> dict:
     # TODO create DC function
 
     # TODO make AWG take offset from the previous point
-    def create_IQ(self, instr: Instruction, chan: str) -> dict:
+    def create_IQ(self, instr: Instruction, chan: str, inputs) -> dict:
         """
         Construct the in-phase (I) and quadrature (Q) components of the signal.
         These are universal to either experiment or simulation.
@@ -1137,7 +1169,7 @@ def create_IQ(self, instr: Instruction, chan: str) -> dict:
         }
         return self.signal[chan]
 
-    def create_IQ_pwc(self, instr: Instruction, chan: str) -> dict:
+    def create_IQ_pwc(self, instr: Instruction, chan: str, inputs) -> dict:
         """
         Construct the in-phase (I) and quadrature (Q) components of the signal.
         These are universal to either experiment or simulation.

c3/generator/generator.py

@@ -11,11 +11,11 @@
 
 from typing import List, Callable, Dict
 import hjson
-import numpy as np
 import tensorflow as tf
 from c3.c3objs import hjson_decode, hjson_encode
 from c3.signal.gates import Instruction
 from c3.generator.devices import devices as dev_lib
+from c3.generator.devices import Device
 
 
 class Generator:
@@ -37,10 +37,10 @@ def __init__(
         self,
         devices: dict = None,
         chains: dict = None,
-        resolution: np.float64 = 0.0,
+        resolution: float = 0.0,
         callback: Callable = None,
     ):
-        self.devices = {}
+        self.devices: Dict[str, Device] = {}
         if devices:
             self.devices = devices
         self.chains = {}
@@ -199,7 +199,7 @@ def generate_signals(self, instr: Instruction) -> dict:
 
                 # calculate the output and store it in the stack
                 dev = self.devices[dev_id]
-                output = dev.process(instr, chan, *inputs)
+                output = dev.process(instr, chan, inputs)
                 signal_stack[chan][dev_id] = output
 
                 # remove inputs if they are not needed anymore
@@ -219,5 +219,7 @@ def generate_signals(self, instr: Instruction) -> dict:
         # Hack to use crosstalk. Will be generalized to a post-processing module.
         # TODO: Rework of the signal generation for larger chips, similar to qiskit
         if "crosstalk" in self.devices:
-            gen_signal = self.devices["crosstalk"].process(signal=gen_signal)
+            gen_signal = self.devices["crosstalk"].process(
+                None, None, signals=[gen_signal]
+            )
         return gen_signal

test/test_crosstalk.py

@@ -6,15 +6,15 @@
 @pytest.mark.unit
 def test_crosstalk(get_xtalk_pmap, get_test_signal) -> None:
     xtalk = get_xtalk_pmap.generator.devices["crosstalk"]
-    new_sig = xtalk.process(signal=get_test_signal)
+    new_sig = xtalk.process(None, None, signals=[get_test_signal])
     assert new_sig == get_test_signal
 
 
 @pytest.mark.unit
 def test_crosstalk_flip(get_xtalk_pmap, get_test_signal) -> None:
     xtalk = get_xtalk_pmap.generator.devices["crosstalk"]
     xtalk.params["crosstalk_matrix"].set_value([[0, 1], [1, 0]])
-    new_sig = xtalk.process(signal=get_test_signal)
+    new_sig = xtalk.process(None, None, signals=[get_test_signal])
     assert (new_sig["TC2"]["values"].numpy() == np.linspace(0, 100, 101)).all()
     assert (new_sig["TC1"]["values"].numpy() == np.linspace(100, 200, 101)).all()
 
@@ -23,7 +23,7 @@ def test_crosstalk_flip(get_xtalk_pmap, get_test_signal) -> None:
 def test_crosstalk_mix(get_xtalk_pmap, get_test_signal) -> None:
     xtalk = get_xtalk_pmap.generator.devices["crosstalk"]
     xtalk.params["crosstalk_matrix"].set_value([[0.5, 0.5], [0.5, 0.5]])
-    new_sig = xtalk.process(signal=get_test_signal)
+    new_sig = xtalk.process(None, None, signals=[get_test_signal])
     assert (new_sig["TC2"]["values"].numpy() == new_sig["TC1"]["values"].numpy()).all()
 
 

test/test_envelope.py

@@ -47,3 +47,6 @@ def test_envelope_netzero() -> None:
         shape.numpy()
         == np.array([1.0, 1.0, 1.0, 1.0, 1.0, 0.0, -1.0, -1.0, -1.0, -1.0, -1.0, -0.0])
     )
+
+
+test_envelope_netzero()