Merge pull request #41 from KCL-BMEIS/26_awg_super

Arbitrary waveform generator support

README.md

@@ -9,7 +9,7 @@ for controlling devices:
 |----------------------------|-----------------------------------------------------------------------|
 | `SpectrumDigitiserCard`    | Controlling individual digitiser cards                                |
 | `SpectrumDigitiserStarHub` | Controlling digitiser cards aggregated with a StarHub                 |
-| `SpectrumAWGCard`          | Controlling individual AWG cards (Not yet implemented)                |
+| `SpectrumAWGCard`          | Controlling individual AWG cards                                      |
 | `SpectrumAWGStarHub`       | Controlling AWG cards aggregated with a StarHub (Not yet implemented) |
 
 `spectrumdevice` also includes mock classes for testing software without drivers installed or hardware connected:
@@ -157,11 +157,12 @@ independently configuring each channel.
 For example, to change the vertical range of channel 2 of a digitiser card to 1V:
 
 ```python
-card.channels[2].set_vertical_range_in_mv(1000)
+card.analog_channels[2].set_vertical_range_in_mv(1000)
 ```
 and then print the vertical offset:
+
 ```python
-print(card.channels[2].vertical_offset_in_percent)
+print(card.analog_channels[2].vertical_offset_in_percent)
 ```
 
 ### Configuring everything at once
@@ -230,6 +231,7 @@ buffer = transfer_buffer_factory(
   buffer_type=BufferType.SPCM_BUF_DATA,  # must be SPCM_BUF_DATA to transfer samples from digitiser
   direction=BufferDirection.SPCM_DIR_CARDTOPC,  # must be SPCM_DIR_CARDTOPC to transfer samples from digitiser
   size_in_samples=size_in_samples,
+  bytes_per_sampe=card.bytes_per_sample,
   board_memory_offset_bytes=board_memory_offset_bytes,
   notify_size_in_pages=notify_size_in_pages
 )

src/example_scripts/awg_example.py

@@ -0,0 +1,55 @@
+from time import sleep
+
+from matplotlib.pyplot import plot, show
+from numpy import int16, iinfo, linspace, sin, pi
+
+from spectrumdevice.devices.awg.awg_card import SpectrumAWGCard
+from spectrumdevice.settings import TriggerSettings, TriggerSource, ExternalTriggerMode
+from spectrumdevice.settings.channel import OutputChannelStopLevelMode
+from spectrumdevice.settings.device_modes import GenerationMode
+
+PULSE_RATE_HZ = 5000
+NUM_PULSES = 5
+NUM_CYCLES = 2
+FREQUENCY = 20e3
+SAMPLE_RATE = 125000000
+
+
+if __name__ == "__main__":
+
+    card = SpectrumAWGCard(device_number=0)
+    print(card)
+
+    trigger_settings = TriggerSettings(
+        trigger_sources=[TriggerSource.SPC_TMASK_EXT0],
+        external_trigger_mode=ExternalTriggerMode.SPC_TM_POS,
+        external_trigger_level_in_mv=200,
+    )
+    card.configure_trigger(trigger_settings)
+
+    full_scale_min_value = iinfo(int16).min
+    full_scale_max_value = iinfo(int16).max
+
+    duration = NUM_CYCLES / FREQUENCY
+    t = linspace(0, duration, int(duration * SAMPLE_RATE + 1))
+    analog_wfm = (sin(2 * pi * FREQUENCY * t) * full_scale_max_value).astype(int16)
+    card.set_sample_rate_in_hz(SAMPLE_RATE)
+    card.set_generation_mode(GenerationMode.SPC_REP_STD_SINGLERESTART)
+    card.set_num_loops(NUM_PULSES)
+    card.transfer_waveform(analog_wfm)
+    card.analog_channels[0].set_stop_level_mode(OutputChannelStopLevelMode.SPCM_STOPLVL_ZERO)
+    card.analog_channels[0].set_is_switched_on(True)
+    card.analog_channels[0].set_signal_amplitude_in_mv(1000)
+
+    card.start()
+
+    for _ in range(NUM_PULSES):
+        card.force_trigger_event()
+        sleep(1 / PULSE_RATE_HZ)
+        print("generated pulse")
+
+    card.stop()
+    card.disconnect()
+
+    plot(t * 1e6, analog_wfm)
+    show()

src/example_scripts/continuous_averaging_fifo_mode.py

@@ -73,6 +73,7 @@ def continuous_averaging_multi_fifo_example(
         # Retrieve streamed waveform data until desired time has elapsed
         measurements_list = []
         while (monotonic() - start_time) < acquisition_duration_in_seconds:
+            print(f"Asking for waveforms at {monotonic() - start_time}")
             measurements_list += [
                 Measurement(waveforms=frame, timestamp=card.get_timestamp()) for frame in card.get_waveforms()
             ]

src/example_scripts/standard_single_mode.py

@@ -30,6 +30,8 @@ def standard_single_mode_example(
         card = SpectrumDigitiserCard(device_number=device_number, ip_address=ip_address)
     else:
         # Set up a mock device
+        for item in MockSpectrumDigitiserCard.__mro__:
+            print(item)
         card = MockSpectrumDigitiserCard(
             device_number=device_number,
             model=ModelNumber.TYP_M2P5966_X4,
@@ -75,7 +77,7 @@ def standard_single_mode_example(
     from matplotlib.pyplot import plot, show, xlabel, tight_layout, ylabel
 
     meas = standard_single_mode_example(
-        mock_mode=False, trigger_source=TriggerSource.SPC_TMASK_EXT0, device_number=1, ip_address="169.254.13.35"
+        mock_mode=True, trigger_source=TriggerSource.SPC_TMASK_EXT0, device_number=1, ip_address="169.254.13.35"
     )
 
     # Plot waveforms

src/example_scripts/star_hub_example.py

@@ -22,7 +22,9 @@ def connect_to_star_hub_example(
             # Connect to each card in the hub.
             child_cards.append(SpectrumDigitiserCard(device_number=n, ip_address=ip_address))
         # Connect to the hub itself
-        return SpectrumDigitiserStarHub(device_number=0, child_cards=child_cards, master_card_index=master_card_index)
+        return SpectrumDigitiserStarHub(
+            device_number=0, child_cards=tuple(child_cards), master_card_index=master_card_index
+        )
     else:
         mock_child_cards = []
         for n in range(num_cards):
@@ -49,8 +51,8 @@ def connect_to_star_hub_example(
     num_measurements = 5
     hub = connect_to_star_hub_example(mock_mode=False, num_cards=2, master_card_index=1, ip_address="169.254.13.35")
 
-    print(f"{hub} contains {len(hub.channels)} channels in total:")
-    for channel in hub.channels:
+    print(f"{hub} contains {len(hub.analog_channels)} channels in total:")
+    for channel in hub.analog_channels:
         print(channel)
 
     # Trigger settings

src/spectrumdevice/__init__.py

@@ -58,7 +58,7 @@
 
 from spectrumdevice.measurement import Measurement
 from .devices.digitiser.digitiser_card import SpectrumDigitiserCard
-from .devices.digitiser.digitiser_channel import SpectrumDigitiserChannel
+from .devices.digitiser.digitiser_channel import SpectrumDigitiserAnalogChannel
 from .devices.digitiser.digitiser_star_hub import SpectrumDigitiserStarHub
 from .devices.mocks import MockSpectrumDigitiserCard, MockSpectrumDigitiserStarHub
 from .devices.abstract_device import (
@@ -70,7 +70,7 @@
 from .devices.digitiser.abstract_spectrum_digitiser import AbstractSpectrumDigitiser
 
 __all__ = [
-    "SpectrumDigitiserChannel",
+    "SpectrumDigitiserAnalogChannel",
     "SpectrumDigitiserCard",
     "SpectrumDigitiserStarHub",
     "MockSpectrumDigitiserCard",

src/spectrumdevice/devices/abstract_device/__init__.py

@@ -8,7 +8,7 @@
 from spectrumdevice.devices.abstract_device.abstract_spectrum_channel import AbstractSpectrumChannel
 from spectrumdevice.devices.abstract_device.abstract_spectrum_device import AbstractSpectrumDevice
 from spectrumdevice.devices.abstract_device.abstract_spectrum_hub import AbstractSpectrumStarHub
-from spectrumdevice.devices.abstract_device.device_interface import SpectrumChannelInterface, SpectrumDeviceInterface
+from spectrumdevice.devices.abstract_device.interfaces import SpectrumChannelInterface, SpectrumDeviceInterface
 
 
 __all__ = [

src/spectrumdevice/devices/abstract_device/abstract_spectrum_card.py

@@ -9,7 +9,7 @@
 from abc import ABC, abstractmethod
 from functools import reduce
 from operator import or_
-from typing import List, Optional, Sequence, Tuple
+from typing import Any, List, Optional, Sequence, Tuple, TypeVar, Generic
 
 from spectrum_gmbh.regs import (
     M2CMD_DATA_STARTDMA,
@@ -31,9 +31,11 @@
     SPC_TIMEOUT,
     SPC_TRIG_ANDMASK,
     SPC_TRIG_ORMASK,
+    M2CMD_CARD_FORCETRIGGER,
+    SPC_MIINST_BYTESPERSAMPLE,
 )
 from spectrumdevice.devices.abstract_device.abstract_spectrum_device import AbstractSpectrumDevice
-from spectrumdevice.devices.abstract_device.device_interface import SpectrumChannelInterface
+from spectrumdevice.devices.abstract_device.interfaces import SpectrumAnalogChannelInterface, SpectrumIOLineInterface
 from spectrumdevice.exceptions import (
     SpectrumExternalTriggerNotEnabled,
     SpectrumInvalidNumberOfEnabledChannels,
@@ -67,25 +69,33 @@
 logger = logging.getLogger(__name__)
 
 
-class AbstractSpectrumCard(AbstractSpectrumDevice, ABC):
+# Use a Generic and Type Variables to allow subclasses of AbstractSpectrumCard to define whether they own AWG analog
+# channels or Digitiser analog channels and IO lines
+AnalogChannelInterfaceType = TypeVar("AnalogChannelInterfaceType", bound=SpectrumAnalogChannelInterface)
+IOLineInterfaceType = TypeVar("IOLineInterfaceType", bound=SpectrumIOLineInterface)
+
+
+class AbstractSpectrumCard(AbstractSpectrumDevice, Generic[AnalogChannelInterfaceType, IOLineInterfaceType], ABC):
     """Abstract superclass implementing methods common to all individual "card" devices (as opposed to "hub" devices)."""
 
-    def __init__(self, device_number: int = 0, ip_address: Optional[str] = None):
+    def __init__(self, device_number: int, ip_address: Optional[str] = None, **kwargs: Any):
         """
         Args:
             device_number (int): Index of the card to control. If only one card is present, set to 0.
             ip_address (Optional[str]): If connecting to a networked card, provide the IP address here as a string.
 
         """
+        super().__init__()  # required for proper MRO resolution
         if ip_address is not None:
             self._visa_string = _create_visa_string_from_ip(ip_address, device_number)
         else:
             self._visa_string = f"/dev/spcm{device_number}"
         self._connect(self._visa_string)
         self._model_number = ModelNumber(self.read_spectrum_device_register(SPC_PCITYP))
         self._trigger_sources: List[TriggerSource] = []
-        self._channels = self._init_channels()
-        self._enabled_channels: List[int] = [0]
+        self._analog_channels = self._init_analog_channels()
+        self._io_lines = self._init_io_lines()
+        self._enabled_analog_channels: List[int] = [0]
         self._transfer_buffer: Optional[TransferBuffer] = None
         self.apply_channel_enabling()
 
@@ -118,6 +128,8 @@ def start_transfer(self) -> None:
         For digitisers in FIFO mode (SPC_REC_FIFO_MULTI), `start_transfer()` should be called immediately after
         `start()` has been called, so that the waveform data can be continuously streamed into the transfer buffer as it
         is acquired.
+
+        # todo: docstring for AWG transfers
         """
         self.write_to_spectrum_device_register(SPC_M2CMD, M2CMD_DATA_STARTDMA)
 
@@ -134,6 +146,8 @@ def stop_transfer(self) -> None:
         For digitisers in FIFO mode (SPC_REC_FIFO_MULTI), samples are transferred continuously during acquisition,
         and transfer will automatically stop when `stop()` is called as there will be no more
         samples to transfer, so `stop_transfer()` should not be used.
+
+        # todo: docstring for AWG
         """
         self.write_to_spectrum_device_register(SPC_M2CMD, M2CMD_DATA_STOPDMA)
 
@@ -146,6 +160,9 @@ def wait_for_transfer_chunk_to_complete(self) -> None:
         be read using the `get_waveforms()` method.
 
         For digitisers in FIFO mode (SPC_REC_FIFO_MULTI) this method is internally used by get_waveforms().
+
+        # todo: update the above docstring to take into account cases where notify size < data lemgth
+        # todo: docstring for AWG
         """
         self.write_to_spectrum_device_register(SPC_M2CMD, M2CMD_DATA_WAITDMA)
 
@@ -182,7 +199,7 @@ def __eq__(self, other: object) -> bool:
             raise NotImplementedError(f"Cannot compare {self.__class__} with {other.__class__}")
 
     @property
-    def channels(self) -> Sequence[SpectrumChannelInterface]:
+    def analog_channels(self) -> Sequence[AnalogChannelInterfaceType]:
         """A tuple containing the channels that belong to the card.
 
         Properties of the individual channels can be set by calling the methods of the
@@ -192,24 +209,37 @@ def channels(self) -> Sequence[SpectrumChannelInterface]:
             channels (Sequence[`SpectrumChannelInterface`]): A tuple of objects conforming to the
             `SpectrumChannelInterface` interface.
         """
-        return self._channels
+        return self._analog_channels
+
+    @property
+    def io_lines(self) -> Sequence[IOLineInterfaceType]:
+        """A tuple containing the Multipurpose IO Lines that belong to the card.
+
+        Properties of the individual channels can be set by calling the methods of the
+            returned objects directly.
+
+        Returns:
+            channels (Sequence[`SpectrumIOLineInterface`]): A tuple of objects conforming to the
+            `SpectrumIOLineInterface` interface.
+        """
+        return self._io_lines
 
     @property
-    def enabled_channels(self) -> List[int]:
+    def enabled_analog_channels(self) -> List[int]:
         """The indices of the currently enabled channels.
         Returns:
             enabled_channels (List[int]): The indices of the currently enabled channels.
         """
-        return self._enabled_channels
+        return self._enabled_analog_channels
 
-    def set_enabled_channels(self, channels_nums: List[int]) -> None:
+    def set_enabled_analog_channels(self, channels_nums: List[int]) -> None:
         """Change which channels are enabled.
 
         Args:
             channels_nums (List[int]): The integer channel indices to enable.
         """
         if len(channels_nums) in [1, 2, 4, 8]:
-            self._enabled_channels = channels_nums
+            self._enabled_analog_channels = channels_nums
             self.apply_channel_enabling()
         else:
             raise SpectrumInvalidNumberOfEnabledChannels(f"{len(channels_nums)} cannot be enabled at once.")
@@ -353,7 +383,7 @@ def set_external_trigger_pulse_width_in_samples(self, width: int) -> None:
     def apply_channel_enabling(self) -> None:
         """Apply the enabled channels chosen using set_enable_channels(). This happens automatically and does not
         usually need to be called."""
-        enabled_channel_spectrum_values = [self.channels[i].name.value for i in self._enabled_channels]
+        enabled_channel_spectrum_values = [self.analog_channels[i].name.value for i in self._enabled_analog_channels]
         if len(enabled_channel_spectrum_values) in [1, 2, 4, 8]:
             bitwise_or_of_enabled_channels = reduce(or_, enabled_channel_spectrum_values)
             self.write_to_spectrum_device_register(SPC_CHENABLE, bitwise_or_of_enabled_channels)
@@ -363,7 +393,11 @@ def apply_channel_enabling(self) -> None:
             )
 
     @abstractmethod
-    def _init_channels(self) -> Sequence[SpectrumChannelInterface]:
+    def _init_analog_channels(self) -> Sequence[AnalogChannelInterfaceType]:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def _init_io_lines(self) -> Sequence[IOLineInterfaceType]:
         raise NotImplementedError()
 
     @property
@@ -452,6 +486,14 @@ def __str__(self) -> str:
     def type(self) -> CardType:
         return CardType(self.read_spectrum_device_register(SPC_FNCTYPE))
 
+    def force_trigger_event(self) -> None:
+        """Force a trigger event to occur"""
+        self.write_to_spectrum_device_register(SPC_M2CMD, M2CMD_CARD_FORCETRIGGER)
+
+    @property
+    def bytes_per_sample(self) -> int:
+        return self.read_spectrum_device_register(SPC_MIINST_BYTESPERSAMPLE)
+
 
 def _create_visa_string_from_ip(ip_address: str, instrument_number: int) -> str:
     return f"TCPIP[0]::{ip_address}::inst{instrument_number}::INSTR"