Merge pull request #1325 from qiboteam/emulator-custom-sampling-rate

Author: alecandido

src/qibolab/_core/instruments/emulator/emulator.py

@@ -39,21 +39,27 @@
 class EmulatorController(Controller):
     """Emulator controller."""
 
+    sampling_rate_: float = 1
+    """Sampling rate used during simulation."""
     engine: SimulationEngine = QutipEngine()
     """SimulationEngine. Default is QutipEngine."""
     bounds: str = "emulator/bounds"
+    """Bounds for emulator."""
+
+    @property
+    def sampling_rate(self) -> float:
+        return self.sampling_rate_
+
+    @sampling_rate.setter
+    def sampling_rate(self, value: float):
+        self.sampling_rate_ = value
 
     def connect(self):
         """Dummy connect method."""
 
     def disconnect(self):
         """Dummy disconnect method."""
 
-    @property
-    def sampling_rate(self):
-        """Sampling rate of emulator."""
-        return 1
-
     def play(
         self,
         configs: dict[str, Config],
@@ -127,7 +133,7 @@ def _play_sequence(
         the various measurements included in the sequence.
         """
         sequence_ = update_sequence(sequence, updates)
-        tlist_ = tlist(sequence_, self.sampling_rate)
+        tlist_ = tlist(sequence_, self.sampling_rate, per_sample=2)
         configs_ = update_configs(configs, updates)
         config = cast(HamiltonianConfig, configs_["hamiltonian"])
         hamiltonian = config.hamiltonian(config=configs_, engine=self.engine)
@@ -151,8 +157,13 @@ def _pulse_hamiltonian(
         """Construct Hamiltonian time dependent term for qutip simulation."""
 
         channels = [
-            [operator, channel_time(waveforms)]
-            for operator, waveforms in hamiltonians(sequence, configs, self.engine)
+            [
+                operator,
+                channel_time(waveforms, sampling_rate=self.sampling_rate),
+            ]
+            for operator, waveforms in hamiltonians(
+                sequence, configs, self.engine, self.sampling_rate
+            )
         ]
         return OperatorEvolution(channels) if len(channels) > 0 else None
 
@@ -206,21 +217,25 @@ def hamiltonian(
     hamiltonian: HamiltonianConfig,
     hilbert_space_index: int,
     engine: SimulationEngine,
+    sampling_rate: float,
 ) -> tuple[Operator, list[Modulated]]:
     n = hamiltonian.transmon_levels
     op = engine.expand(
         config.operator(n=n, engine=engine), hamiltonian.dims, hilbert_space_index
     )
     waveforms = (
-        waveform(pulse, config, hamiltonian.qubits[hilbert_space_index])
+        waveform(pulse, config, hamiltonian.qubits[hilbert_space_index], sampling_rate)
         for pulse in pulses
         if isinstance(pulse, (Pulse, Delay, VirtualZ))
     )
     return (op, [w for w in waveforms if w is not None])
 
 
 def hamiltonians(
-    sequence: PulseSequence, configs: dict[str, Config], engine: SimulationEngine
+    sequence: PulseSequence,
+    configs: dict[str, Config],
+    engine: SimulationEngine,
+    sampling_rate: float,
 ) -> Iterable[tuple[Operator, list[Modulated]]]:
     hconfig = cast(HamiltonianConfig, configs["hamiltonian"])
     return (
@@ -230,14 +245,18 @@ def hamiltonians(
             hconfig,
             index(ch, hconfig),
             engine,
+            sampling_rate,
         )
         for ch in sequence.channels
         # TODO: drop the following, and treat acquisitions just as empty channels
         if not isinstance(configs[ch], AcquisitionConfig)
     )
 
 
-def channel_time(waveforms: Iterable[Modulated]) -> Callable[[float], float]:
+def channel_time(
+    waveforms: Iterable[Modulated],
+    sampling_rate: int,
+) -> Callable[[float], float]:
     """Wrap time function for specific channel.
 
     Used to avoid late binding issues.
@@ -247,13 +266,12 @@ def time(t: float) -> float:
         cumulative_time = 0
         cumulative_phase = 0
         for pulse in waveforms:
-            pulse_duration = pulse.duration  # TODO: pass sampling rate
             pulse_phase = pulse.phase
-            if cumulative_time <= t < cumulative_time + pulse_duration:
+            if cumulative_time <= t < cumulative_time + pulse.duration:
                 relative_time = t - cumulative_time
-                index = int(relative_time)  # TODO: pass sampling rate
+                index = int(np.floor(relative_time * sampling_rate))
                 return pulse(t, index, cumulative_phase)
-            cumulative_time += pulse_duration
+            cumulative_time += pulse.duration
             cumulative_phase += pulse_phase
         return 0
 

src/qibolab/_core/instruments/emulator/hamiltonians.py

@@ -361,19 +361,23 @@ def waveform(
     pulse: PulseLike,
     config: Config,
     qubit: Qubit,
+    sampling_rate: float,
 ) -> Optional[ControlLine]:
     """Convert pulse to hamiltonian."""
     if not isinstance(config, (DriveEmulatorConfig, FluxEmulatorConfig)):
         return None
 
     if isinstance(pulse, Pulse):
         if isinstance(config, DriveEmulatorConfig):
-            return ModulatedDrive(pulse=pulse, config=config)
+            return ModulatedDrive(
+                pulse=pulse, config=config, sampling_rate=sampling_rate
+            )
         if isinstance(config, FluxEmulatorConfig):
             return FluxPulse(
                 pulse=pulse,
                 config=config,
                 qubit=qubit,
+                sampling_rate=sampling_rate,
             )
     if isinstance(pulse, Delay):
         return ModulatedDelay(duration=pulse.duration)

src/qibolab/_core/instruments/emulator/results.py

@@ -75,6 +75,7 @@ def acquisitions(sequence: PulseSequence) -> dict[PulseId, float]:
             if isinstance(ev, (Acquisition, Readout)):
                 acq[ev.id] = time
             time += ev.duration
+
     return acq
 
 
@@ -121,7 +122,6 @@ def results(
         hamiltonian.nqubits,
         hamiltonian.transmon_levels,
     )
-
     assert options.nshots is not None
     sampled = shots(np.moveaxis(probabilities, -2, 0), options.nshots)
     # move measurements dimension to the front, getting ready for extraction

tests/instruments/emulator/platforms/fixed-frequency-qutrits/platform.py

@@ -28,7 +28,9 @@ def create() -> Hardware:
     }
     # register the instruments
     instruments = {
-        "emulator": EmulatorController(address="0.0.0.0", channels=channels),
+        "emulator": EmulatorController(
+            address="0.0.0.0", channels=channels, sampling_rate_=2
+        ),
     }
 
     return Hardware(

tests/instruments/emulator/platforms/qubit/platform.py

@@ -27,7 +27,9 @@ def create() -> Platform:
 
     # register the instruments
     instruments = {
-        "dummy": EmulatorController(address="0.0.0.0", channels=channels),
+        "dummy": EmulatorController(
+            address="0.0.0.0", channels=channels, sampling_rate_=1
+        ),
     }
 
     return Platform.load(

tests/instruments/emulator/test_hamiltonians.py

@@ -20,8 +20,14 @@ def test_dummy_waveform():
     acq_config = AcquisitionConfig(delay=0, smearing=0)
     dc_config = DcConfig(offset=0)
     qubit = Qubit()
-    assert waveform(pulse=_PulseLike(), config=acq_config, qubit=qubit) is None
-    assert waveform(pulse=_PulseLike(), config=dc_config, qubit=qubit) is None
+    assert (
+        waveform(pulse=_PulseLike(), config=acq_config, qubit=qubit, sampling_rate=1)
+        is None
+    )
+    assert (
+        waveform(pulse=_PulseLike(), config=dc_config, qubit=qubit, sampling_rate=1)
+        is None
+    )
 
 
 @pytest.mark.parametrize(
@@ -36,7 +42,7 @@ def test_dummy_waveform():
 def test_iq_waveform(pulse, level):
     iq_config = DriveEmulatorConfig(frequency=5e9)
     qubit = Qubit(frequency=5e9)
-    modulated = waveform(pulse=pulse, config=iq_config, qubit=qubit)
+    modulated = waveform(pulse=pulse, config=iq_config, qubit=qubit, sampling_rate=1)
     if isinstance(pulse, Pulse):
         assert isinstance(modulated, ModulatedDrive)
         assert pytest.approx(modulated.omega) == 2 * np.pi * 5