Merge pull request #63 from KCL-BMEIS/62-raw-waveforms-in-conveninece…

…-methods

#62 raw waveforms in convenience methods

.github/workflows/spectrumdevice-docs-pages.yml

@@ -16,7 +16,7 @@ jobs:
     - name: Set up Python 3.10
       uses: actions/setup-python@v3
       with:
-        python-version: "3.10"
+        python-version: "3.12"
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip

.github/workflows/spectrumdevice-integration-tests.yml

@@ -13,7 +13,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.10", "3.11", "3.12"]
 
     steps:
     - uses: actions/checkout@v3

.github/workflows/spectrumdevice-unit-tests.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.10", "3.11", "3.12"]
 
     steps:
     - uses: actions/checkout@v3

setup.cfg

@@ -16,7 +16,6 @@ classifiers =
  Operating System :: POSIX :: Linux
  Operating System :: Microsoft :: Windows :: Windows 10
  Operating System :: MacOS :: MacOS X
- Programming Language :: Python :: 3.9
  Programming Language :: Python :: 3.10
  Programming Language :: Python :: 3.11
  Programming Language :: Python :: 3.12
@@ -34,8 +33,8 @@ package_dir =
     = src
 include_package_data = True
 install_requires =
- numpy>=1.21.4
-python_requires = >=3.8
+ numpy>=1.26.2
+python_requires = >=3.10
 
 [options.packages.find]
 where = src

src/example_scripts/pulse_generator_example.py

@@ -28,7 +28,7 @@ def pulse_generator_example(mock_mode: bool) -> None:
 
     # Set the card's sample rate. This affects the precision with which pulse timings can be chosen, and the min and max
     # allowed pulse periods
-    card.set_sample_rate_in_hz(100000)
+    card.set_sample_rate_in_hz(8000000)
     # Enable a single channel of the card. Although not used in this example, the number of enabled channels affects
     # the precision with which pulse timings can be chosen and the min and max allowed pulse periods
     card.set_enabled_analog_channels([0])
@@ -62,7 +62,7 @@ def pulse_generator_example(mock_mode: bool) -> None:
     # The period is the length of the whole pulse (high-voltage length + 0V length)
     # The duty cycle is the high-voltage length divided by the period
     pulse_output_settings = PulseGeneratorOutputSettings(
-        period_in_seconds=1e-3, duty_cycle=0.5, num_pulses=2, delay_in_seconds=0.0, output_inversion=False
+        period_in_seconds=1e-3, duty_cycle=0.01, num_pulses=1000, delay_in_seconds=0.0, output_inversion=False
     )
     pulse_gen.configure_output(pulse_output_settings, coerce=False)
 

src/spectrumdevice/devices/awg/synthesis.py

@@ -1,10 +1,10 @@
-from numpy import float_, iinfo, issubdtype, signedinteger, pi, sin, linspace, int_, ones, int16
+from numpy import float64, iinfo, issubdtype, signedinteger, pi, sin, linspace, int_, ones, int16
 from numpy.typing import NDArray
 
 
 def make_full_scale_sine_waveform(
     frequency_in_hz: float, sample_rate_in_hz: int, num_cycles: float, dtype: type = int16
-) -> tuple[NDArray[float_], NDArray[int_]]:
+) -> tuple[NDArray[float64], NDArray[int_]]:
     """Create a sine waveform covering the full range of the given data type. The resulting waveform is intended to
     be transferred to the AWG's on-board memory for generation.
 
@@ -25,7 +25,7 @@ def make_full_scale_sine_waveform(
 
 def make_full_scale_rect_waveform(
     sample_rate_in_hz: int, duration_in_seconds: float, dtype: type = int16
-) -> tuple[NDArray[float_], NDArray[int_]]:
+) -> tuple[NDArray[float64], NDArray[int_]]:
     """Create a rectangular waveform covering the full range of the given data type. The resulting waveform is intended
     to be transferred to the AWG's on-board memory for generation.
 

src/spectrumdevice/devices/digitiser/abstract_spectrum_digitiser.py

@@ -7,7 +7,7 @@
 from abc import ABC
 from typing import List
 
-from spectrumdevice.measurement import Measurement
+from spectrumdevice.measurement import Measurement, RawWaveformType, VoltageWaveformType
 from spectrumdevice.devices.abstract_device import AbstractSpectrumDevice
 from spectrumdevice.devices.digitiser.digitiser_interface import (
     SpectrumDigitiserAnalogChannelInterface,
@@ -76,14 +76,18 @@ def configure_acquisition(self, settings: AcquisitionSettings) -> None:
         if settings.timestamping_enabled:
             self.enable_timestamping()
 
-    def execute_standard_single_acquisition(self) -> Measurement:
+    def execute_standard_single_acquisition(self, raw: bool = False) -> Measurement:
         """Carry out a single measurement in standard single mode and return the acquired waveforms.
 
         This method automatically carries out a standard single mode acquisition, including handling the creation
         of a `TransferBuffer` and the retrieval of the acquired waveforms. After being called, it will wait until a
         trigger event is received before carrying out the acquisition and then transferring and returning the acquired
         waveforms. The device must be configured in SPC_REC_STD_SINGLE acquisition mode.
 
+        Args:
+            raw (bool, optional): Set to true to obtain raw (i.e. 16-bit integer) waveforms, instead of floating point
+            voltage waveforms.
+
         Returns:
             measurement (Measurement): A Measurement object. The `.waveforms` attribute of `measurement` will be a list
                 of 1D NumPy arrays, each array containing the waveform data received on one channel, in channel order.
@@ -101,11 +105,13 @@ def execute_standard_single_acquisition(self) -> Measurement:
         self.define_transfer_buffer()
         self.start_transfer()
         self.wait_for_transfer_chunk_to_complete()
-        waveforms = self.get_waveforms()[0]
+        waveforms: list[RawWaveformType] | list[VoltageWaveformType] = (
+            self.get_raw_waveforms()[0] if raw else self.get_waveforms()[0]
+        )
         self.stop()  # Only strictly required for Mock devices. Should not affect hardware.
         return Measurement(waveforms=waveforms, timestamp=self.get_timestamp())
 
-    def execute_finite_fifo_acquisition(self, num_measurements: int) -> List[Measurement]:
+    def execute_finite_fifo_acquisition(self, num_measurements: int, raw: bool = False) -> List[Measurement]:
         """Carry out a finite number of FIFO mode measurements and then stop the acquisitions.
 
         This method automatically carries out a defined number of measurement in Multi FIFO mode, including handling the
@@ -118,6 +124,8 @@ def execute_finite_fifo_acquisition(self, num_measurements: int) -> List[Measure
 
         Args:
             num_measurements (int): The number of measurements to carry out.
+            raw (bool, optional): Set to true to obtain raw (i.e. 16-bit integer) waveforms, instead of floating point
+                voltage waveforms.
         Returns:
             measurements (List[Measurement]): A list of Measurement objects with length `num_measurements`. Each
                 Measurement object has a `waveforms` attribute containing a list of 1D NumPy arrays. Each array is a
@@ -133,9 +141,10 @@ def execute_finite_fifo_acquisition(self, num_measurements: int) -> List[Measure
         self.execute_continuous_fifo_acquisition()
         measurements = []
         for _ in range(num_measurements // self.batch_size):
-            measurements += [
-                Measurement(waveforms=frame, timestamp=self.get_timestamp()) for frame in self.get_waveforms()
-            ]
+            waveforms: list[list[RawWaveformType]] | list[list[VoltageWaveformType]] = (
+                self.get_raw_waveforms() if raw else self.get_waveforms()
+            )
+            measurements += [Measurement(waveforms=frame, timestamp=self.get_timestamp()) for frame in waveforms]
         self.stop()
         return measurements
 

src/spectrumdevice/devices/digitiser/digitiser_card.py

@@ -7,7 +7,7 @@
 import logging
 from typing import List, Optional, Sequence, cast
 
-from numpy import float_, int16, mod, squeeze, zeros
+from numpy import float64, int16, mod, squeeze, zeros
 from numpy.typing import NDArray
 
 from spectrum_gmbh.py_header.regs import (
@@ -173,13 +173,13 @@ def get_raw_waveforms(self) -> List[List[NDArray[int16]]]:
 
         return repeat_acquisitions
 
-    def get_waveforms(self) -> List[List[NDArray[float_]]]:
+    def get_waveforms(self) -> List[List[NDArray[float64]]]:
         """Get a list of the most recently transferred waveforms, in channel order, in Volts as floats.
 
         See get_raw_waveforms() for details.
 
         Returns:
-             waveforms (List[List[NDArray[float_]]]): A list of lists of 1D numpy arrays, one inner list per acquisition
+             waveforms (List[List[NDArray[float64]]]): A list of lists of 1D numpy arrays, one inner list per acquisition
              and one array per enabled channel, in channel order. To average the acquisitions:
                 `np.array(waveforms).mean(axis=0)`
 

src/spectrumdevice/devices/digitiser/digitiser_interface.py

@@ -8,14 +8,14 @@
 from datetime import datetime
 from typing import List, Optional
 
-from numpy import float_, int16, ndarray
-from numpy.typing import NDArray
+from numpy import ndarray
 
 from spectrumdevice.devices.abstract_device.device_interface import SpectrumDeviceInterface
 from spectrumdevice.devices.abstract_device.channel_interfaces import (
     SpectrumAnalogChannelInterface,
     SpectrumIOLineInterface,
 )
+from spectrumdevice.measurement import VoltageWaveformType, RawWaveformType
 from spectrumdevice.settings import AcquisitionMode, AcquisitionSettings
 from spectrumdevice import Measurement
 from spectrumdevice.settings.channel import InputImpedance, InputCoupling, InputPath
@@ -105,11 +105,11 @@ def execute_continuous_fifo_acquisition(self) -> None:
         raise NotImplementedError()
 
     @abstractmethod
-    def get_raw_waveforms(self) -> List[List[NDArray[int16]]]:
+    def get_raw_waveforms(self) -> List[List[RawWaveformType]]:
         raise NotImplementedError()
 
     @abstractmethod
-    def get_waveforms(self) -> List[List[NDArray[float_]]]:
+    def get_waveforms(self) -> List[List[VoltageWaveformType]]:
         raise NotImplementedError()
 
     @abstractmethod

src/spectrumdevice/devices/digitiser/digitiser_star_hub.py

@@ -7,7 +7,7 @@
 from threading import Thread
 from typing import Callable, Dict, List, Optional, Sequence, TypeVar
 
-from numpy import float_, int16
+from numpy import float64, int16
 from numpy.typing import NDArray
 
 from spectrumdevice.devices.abstract_device import (
@@ -23,7 +23,7 @@
 from spectrumdevice.settings.device_modes import AcquisitionMode
 
 
-WAVEFORM_TYPE_VAR = TypeVar("WAVEFORM_TYPE_VAR", NDArray[float_], NDArray[int16])
+WAVEFORM_TYPE_VAR = TypeVar("WAVEFORM_TYPE_VAR", NDArray[float64], NDArray[int16])
 
 
 # noinspection PyTypeChecker
@@ -73,14 +73,14 @@ def wait_for_acquisition_to_complete(self) -> None:
         for card in self._child_cards:
             card.wait_for_acquisition_to_complete()
 
-    def get_waveforms(self) -> List[List[NDArray[float_]]]:
+    def get_waveforms(self) -> List[List[NDArray[float64]]]:
         """Get a list of the most recently transferred waveforms, as floating point voltages.
 
         This method gets the waveforms from each child card and joins them into a new list, ordered by channel number.
         See `SpectrumDigitiserCard.get_waveforms()` for more information.
 
         Returns:
-            waveforms (List[List[NDArray[float_]]]): A list lists of 1D numpy arrays, one inner list per acquisition,
+            waveforms (List[List[NDArray[float64]]]): A list lists of 1D numpy arrays, one inner list per acquisition,
               and one array per enabled channel, in channel order.
         """
         return self._get_waveforms_in_threads(SpectrumDigitiserCard.get_waveforms)

src/spectrumdevice/measurement.py

@@ -1,16 +1,19 @@
 from datetime import datetime
 from dataclasses import dataclass
-from typing import List, Optional
 
-from numpy import float_
+from numpy import int16, float64
 from numpy.typing import NDArray
 
 
+VoltageWaveformType = NDArray[float64]
+RawWaveformType = NDArray[int16]
+
+
 @dataclass
 class Measurement:
     """Measurement is a dataclass for storing a set of waveforms generated by a single acquisition, with a timestamp."""
 
-    waveforms: List[NDArray[float_]]
-    """Contains the acquired waveforms as a list of 1D NumPy arrays"""
-    timestamp: Optional[datetime]
+    waveforms: list[VoltageWaveformType] | list[RawWaveformType]
+    """Contains the acquired waveforms as a list of 1D NumPy arrays or either floats or ints"""
+    timestamp: datetime | None
     """The time at which the acquisition was triggered, as a datetime.datetime object"""