47class SpectrumDigitiserAnalogChannel(AbstractSpectrumAnalogChannel, SpectrumDigitiserAnalogChannelInterface):
 48    """Class for controlling an individual channel of a spectrum digitiser. Channels are constructed automatically when
 49    a `SpectrumDigitiserCard` or `SpectrumDigitiserStarHub` is instantiated, and can then be accessed via the
 50    `.channels` property."""
 51
 52    def __init__(self, channel_number: int, parent_device: SpectrumDigitiserInterface) -> None:
 53
 54        if parent_device.type != CardType.SPCM_TYPE_AI:
 55            raise SpectrumCardIsNotADigitiser(parent_device.type)
 56
 57        # pass unused args up the inheritance hierarchy
 58        super().__init__(channel_number=channel_number, parent_device=parent_device)
 59
 60        self._full_scale_value = self._parent_device.read_spectrum_device_register(SPC_MIINST_MAXADCVALUE)
 61        # used frequently so store locally instead of reading from device each time:
 62        self._vertical_range_mv = self.vertical_range_in_mv
 63        self._vertical_offset_in_percent = self.vertical_offset_in_percent
 64
 65    def _get_settings_as_dict(self) -> dict:
 66        return {
 67            SpectrumDigitiserAnalogChannel.input_path.__name__: self.input_path,
 68            SpectrumDigitiserAnalogChannel.input_coupling.__name__: self.input_coupling,
 69            SpectrumDigitiserAnalogChannel.input_impedance.__name__: self.input_impedance,
 70            SpectrumDigitiserAnalogChannel.vertical_range_in_mv.__name__: self.vertical_range_in_mv,
 71            SpectrumDigitiserAnalogChannel.vertical_offset_in_percent.__name__: self.vertical_offset_in_percent,
 72        }
 73
 74    def _set_settings_from_dict(self, settings: dict) -> None:
 75        self.set_input_path(settings[SpectrumDigitiserAnalogChannel.input_path.__name__])
 76        self.set_input_coupling(settings[SpectrumDigitiserAnalogChannel.input_coupling.__name__])
 77        self.set_input_impedance(settings[SpectrumDigitiserAnalogChannel.input_impedance.__name__])
 78        self.set_vertical_range_in_mv(settings[SpectrumDigitiserAnalogChannel.vertical_range_in_mv.__name__])
 79        self.set_vertical_offset_in_percent(
 80            settings[SpectrumDigitiserAnalogChannel.vertical_offset_in_percent.__name__]
 81        )
 82
 83    def convert_raw_waveform_to_voltage_waveform(self, raw_waveform: ndarray) -> ndarray:
 84        vertical_offset_mv = 0.01 * float(self._vertical_range_mv * self._vertical_offset_in_percent)
 85        return 1e-3 * (
 86            float(self._vertical_range_mv) * raw_waveform / float(self._full_scale_value) + vertical_offset_mv
 87        )
 88
 89    @property
 90    def vertical_range_in_mv(self) -> int:
 91        """The currently set input range of the channel in mV.
 92
 93        Returns:
 94            vertical_range (int): The currently set vertical range in mV.
 95        """
 96        self._vertical_range_mv = self._parent_device.read_spectrum_device_register(
 97            VERTICAL_RANGE_COMMANDS[self._number]
 98        )
 99        return self._vertical_range_mv
100
101    def set_vertical_range_in_mv(self, vertical_range: int) -> None:
102        """Set the input range of the channel in mV. See Spectrum documentation for valid values.
103
104        Args:
105            vertical_range (int): The desired vertical range in mV.
106        """
107        self._parent_device.write_to_spectrum_device_register(VERTICAL_RANGE_COMMANDS[self._number], vertical_range)
108        self._vertical_range_mv = vertical_range
109
110    @property
111    def vertical_offset_in_percent(self) -> int:
112        """The currently set input offset of the channel in percent of the vertical range.
113
114        Returns:
115            offset (int): The currently set vertical offset in percent.
116        """
117        self._vertical_offset_in_percent = self._parent_device.read_spectrum_device_register(
118            VERTICAL_OFFSET_COMMANDS[self._number]
119        )
120        return self._vertical_offset_in_percent
121
122    def set_vertical_offset_in_percent(self, offset: int) -> None:
123        """Set the input offset of the channel in percent of the vertical range. See spectrum documentation for valid
124        values.
125
126        Args:
127            offset (int): The desired vertical offset in percent.
128        """
129        self._parent_device.write_to_spectrum_device_register(VERTICAL_OFFSET_COMMANDS[self._number], offset)
130        self._vertical_offset_in_percent = offset
131
132    @property
133    def input_impedance(self) -> InputImpedance:
134        """The current input impedance setting of the channel (50 Ohm or 1 MOhm)"""
135        impedance_binary_value = self._parent_device.read_spectrum_device_register(
136            INPUT_IMPEDANCE_COMMANDS[self._number]
137        )
138        return InputImpedance(impedance_binary_value)
139
140    def set_input_impedance(self, input_impedance: InputImpedance) -> None:
141        self._parent_device.write_to_spectrum_device_register(
142            INPUT_IMPEDANCE_COMMANDS[self._number], input_impedance.value
143        )
144
145    @property
146    def input_coupling(self) -> InputCoupling:
147        """The coupling (AC or DC) setting of the channel. Only available on some hardware."""
148        coupling_binary_value = self._parent_device.read_spectrum_device_register(INPUT_COUPLING_COMMANDS[self._number])
149        return InputCoupling(coupling_binary_value)
150
151    def set_input_coupling(self, input_coupling: InputCoupling) -> None:
152        self._parent_device.write_to_spectrum_device_register(
153            INPUT_COUPLING_COMMANDS[self._number], input_coupling.value
154        )
155
156    @property
157    def input_path(self) -> InputPath:
158        """The input path setting of the channel. Only available on some hardware."""
159        path_binary_value = self._parent_device.read_spectrum_device_register(INPUT_PATH_COMMANDS[self._number])
160        return InputPath(path_binary_value)
161
162    def set_input_path(self, input_path: InputPath) -> None:
163        self._parent_device.write_to_spectrum_device_register(INPUT_PATH_COMMANDS[self._number], input_path.value)

35class SpectrumDigitiserIOLine(AbstractSpectrumIOLine, SpectrumDigitiserIOLineInterface):
36    """Class for controlling multipurpose IO lines of a digitiser, e.g. X0, X1, X2 and X3."""
37
38    def __init__(self, parent_device: AbstractSpectrumCard, **kwargs: Any) -> None:
39        if parent_device.type != CardType.SPCM_TYPE_AI:
40            raise SpectrumCardIsNotADigitiser(parent_device.type)
41        super().__init__(parent_device=parent_device, **kwargs)  # pass unused args up the inheritance hierarchy
42
43    def _get_io_line_mode_settings_mask(self, mode: IOLineMode) -> int:
44        return 0  # no settings required for DigOut

 57class SpectrumDigitiserCard(
 58    AbstractSpectrumCard[SpectrumDigitiserAnalogChannelInterface, SpectrumDigitiserIOLineInterface],
 59    AbstractSpectrumDigitiser,
 60):
 61    """Class for controlling individual Spectrum digitiser cards."""
 62
 63    def __init__(self, device_number: int, ip_address: Optional[str] = None) -> None:
 64        """
 65        Args:
 66            device_number (int): Index of the card to control. If only one card is present, set to 0.
 67            ip_address (Optional[str]): If connecting to a networked card, provide the IP address here as a string.
 68
 69        """
 70        # pass unused args up the inheritance hierarchy
 71        super().__init__(device_number=device_number, ip_address=ip_address)
 72
 73        if self.type != CardType.SPCM_TYPE_AI:
 74            raise SpectrumCardIsNotADigitiser(self.type)
 75        self._acquisition_mode = self.acquisition_mode
 76        self._timestamper: Optional[Timestamper] = None
 77        self._batch_size = 1
 78
 79    def _init_analog_channels(self) -> Sequence[SpectrumDigitiserAnalogChannelInterface]:
 80        num_modules = self.read_spectrum_device_register(SPC_MIINST_MODULES)
 81        num_channels_per_module = self.read_spectrum_device_register(SPC_MIINST_CHPERMODULE)
 82        total_channels = num_modules * num_channels_per_module
 83        return tuple(
 84            [SpectrumDigitiserAnalogChannel(channel_number=n, parent_device=self) for n in range(total_channels)]
 85        )
 86
 87    def _init_io_lines(self) -> Sequence[SpectrumDigitiserIOLineInterface]:
 88        if (self.model_number.value & TYP_SERIESMASK) == TYP_M2PEXPSERIES:
 89            return tuple([SpectrumDigitiserIOLine(channel_number=n, parent_device=self) for n in range(4)])
 90        else:
 91            raise NotImplementedError("Don't know how many IO lines other types of card have. Only M2P series.")
 92
 93    def enable_timestamping(self) -> None:
 94        self._timestamper = Timestamper(self, self._handle)
 95
 96    def wait_for_acquisition_to_complete(self) -> None:
 97        """Blocks until the current acquisition has finished, or the timeout is reached.
 98
 99        In Standard Single mode (SPC_REC_STD_SINGLE), this should be called after `start()`. Once the call
100            to `wait_for_acquisition_to_complete()` returns, the newly acquired samples are in the on_device buffer and
101            ready for transfer to the `TransferBuffer` using `start_transfer()`.
102
103        In FIFO mode (SPC_REC_FIFO_MULTI), the card will continue to acquire samples until
104            `stop()` is called, so `wait_for_acquisition_to_complete()` should not be used.
105
106        """
107        self.write_to_spectrum_device_register(SPC_M2CMD, M2CMD_CARD_WAITREADY)
108
109    def get_raw_waveforms(self) -> List[List[NDArray[int16]]]:
110        """Get a list of the most recently transferred waveforms, in channel order, as 16-bit integers.
111
112        This method copies and reshapes the samples in the `TransferBuffer` into a list of lists of 1D NumPy arrays
113        (waveforms) and returns the list.
114
115        In Standard Single mode (SPC_REC_STD_SINGLE), `get_waveforms()` should be called after
116        `wait_for_transfer_to_complete()` has returned.
117
118        In FIFO mode (SPC_REC_FIFO_MULTI), while the card is continuously acquiring samples and transferring them to the
119        `TransferBuffer`, this method should be called in a loop . The method will block until each new transfer is
120        received, so the loop will run at the same rate as the acquisition (in SPC_REC_FIFO_MULTI mode, for example,
121        this would the rate at which your trigger source was running).
122
123        Returns:
124             waveforms (List[List[NDArray[int16]]]): A list of lists of 1D numpy arrays, one inner list per acquisition
125             and one array per enabled channel, in channel order. To average the acquisitions:
126                `np.array(waveforms).mean(axis=0)`
127
128        """
129        if self._transfer_buffer is None:
130            raise SpectrumNoTransferBufferDefined("Cannot find a samples transfer buffer")
131
132        num_read_bytes = 0
133        num_samples_per_frame = self.acquisition_length_in_samples * len(self.enabled_analog_channel_nums)
134        num_expected_bytes_per_frame = num_samples_per_frame * self._transfer_buffer.data_array.itemsize
135        raw_samples = zeros(num_samples_per_frame * self._batch_size, dtype=self._transfer_buffer.data_array.dtype)
136
137        if self.acquisition_mode in (AcquisitionMode.SPC_REC_STD_SINGLE, AcquisitionMode.SPC_REC_STD_AVERAGE):
138            raw_samples = self._transfer_buffer.copy_contents()
139
140        elif self.acquisition_mode in (AcquisitionMode.SPC_REC_FIFO_MULTI, AcquisitionMode.SPC_REC_FIFO_AVERAGE):
141            self.wait_for_transfer_chunk_to_complete()
142
143            while num_read_bytes < (num_expected_bytes_per_frame * self._batch_size):
144                num_available_bytes = self.read_spectrum_device_register(SPC_DATA_AVAIL_USER_LEN)
145                position_of_available_bytes = self.read_spectrum_device_register(SPC_DATA_AVAIL_USER_POS)
146
147                # Don't allow reading over the end of the transfer buffer
148                if (
149                    position_of_available_bytes + num_available_bytes
150                ) > self._transfer_buffer.data_array_length_in_bytes:
151                    num_available_bytes = self._transfer_buffer.data_array_length_in_bytes - position_of_available_bytes
152
153                # Don't allow reading over the end of the current acquisition:
154                if (num_read_bytes + num_available_bytes) > (num_expected_bytes_per_frame * self._batch_size):
155                    num_available_bytes = (num_expected_bytes_per_frame * self._batch_size) - num_read_bytes
156
157                num_available_samples = num_available_bytes // self._transfer_buffer.data_array.itemsize
158                num_read_samples = num_read_bytes // self._transfer_buffer.data_array.itemsize
159
160                raw_samples[
161                    num_read_samples : num_read_samples + num_available_samples
162                ] = self._transfer_buffer.read_chunk(position_of_available_bytes, num_available_bytes)
163                self.write_to_spectrum_device_register(SPC_DATA_AVAIL_CARD_LEN, num_available_bytes)
164
165                num_read_bytes += num_available_bytes
166
167        waveforms_in_columns = raw_samples.reshape(
168            (self._batch_size, self.acquisition_length_in_samples, len(self.enabled_analog_channel_nums))
169        )
170
171        repeat_acquisitions = []
172        for n in range(self._batch_size):
173            repeat_acquisitions.append([waveform for waveform in waveforms_in_columns[n, :, :].T])
174
175        return repeat_acquisitions
176
177    def get_waveforms(self) -> List[List[NDArray[float64]]]:
178        """Get a list of the most recently transferred waveforms, in channel order, in Volts as floats.
179
180        See get_raw_waveforms() for details.
181
182        Returns:
183             waveforms (List[List[NDArray[float64]]]): A list of lists of 1D numpy arrays, one inner list per acquisition
184             and one array per enabled channel, in channel order. To average the acquisitions:
185                `np.array(waveforms).mean(axis=0)`
186
187        """
188        raw_repeat_acquisitions = self.get_raw_waveforms()
189        repeat_acquisitions = []
190        for n in range(self._batch_size):
191            repeat_acquisitions.append(
192                [
193                    cast(
194                        SpectrumDigitiserAnalogChannel, self.analog_channels[ch_num]
195                    ).convert_raw_waveform_to_voltage_waveform(squeeze(waveform))
196                    for ch_num, waveform in zip(self.enabled_analog_channel_nums, raw_repeat_acquisitions[n])
197                ]
198            )
199        return repeat_acquisitions
200
201    def get_timestamp(self) -> Optional[datetime.datetime]:
202        """Get timestamp for the last acquisition"""
203        if self._timestamper is not None:
204            return self._timestamper.get_timestamp()
205        else:
206            return None
207
208    @property
209    def acquisition_length_in_samples(self) -> int:
210        """The current recording length (per channel) in samples.
211
212        Returns:
213            length_in_samples (int): The current recording length ('acquisition length') in samples."""
214        return self.read_spectrum_device_register(SPC_MEMSIZE)
215
216    def set_acquisition_length_in_samples(self, length_in_samples: int) -> None:
217        """Change the recording length (per channel). In FIFO mode, it will be quantised according to the step size
218          allowed by the connected card type.
219
220        Args:
221            length_in_samples (int): The desired recording length ('acquisition length'), in samples.
222        """
223        length_in_samples = self._coerce_num_samples_if_fifo(length_in_samples)
224        self.write_to_spectrum_device_register(SPC_SEGMENTSIZE, length_in_samples)
225        self.write_to_spectrum_device_register(SPC_MEMSIZE, length_in_samples)
226
227    @property
228    def post_trigger_length_in_samples(self) -> int:
229        """The number of samples of the recording that will contain data received after the trigger event.
230
231        Returns:
232            length_in_samples (int): The currently set post trigger length in samples.
233        """
234        return self.read_spectrum_device_register(SPC_POSTTRIGGER)
235
236    def set_post_trigger_length_in_samples(self, length_in_samples: int) -> None:
237        """Change the number of samples of the recording that will contain data received after the trigger event.
238        In FIFO mode, this will be quantised according to the minimum step size allowed by the connected card.
239
240        Args:
241            length_in_samples (int): The desired post trigger length in samples."""
242        length_in_samples = self._coerce_num_samples_if_fifo(length_in_samples)
243        if self.acquisition_mode == AcquisitionMode.SPC_REC_FIFO_MULTI:
244            if (self.acquisition_length_in_samples - length_in_samples) < get_memsize_step_size(self._model_number):
245                logger.warning(
246                    "FIFO mode: coercing post trigger length to maximum allowed value (step-size samples less than "
247                    "the acquisition length)."
248                )
249                length_in_samples = self.acquisition_length_in_samples - get_memsize_step_size(self._model_number)
250        self.write_to_spectrum_device_register(SPC_POSTTRIGGER, length_in_samples)
251
252    def _coerce_num_samples_if_fifo(self, value: int) -> int:
253        if self.acquisition_mode == AcquisitionMode.SPC_REC_FIFO_MULTI:
254            if mod(value, get_memsize_step_size(self._model_number)) != 0:
255                logger.warning(
256                    f"FIFO mode: coercing length to nearest {get_memsize_step_size(self._model_number)}" f" samples"
257                )
258                value = int(value - mod(value, get_memsize_step_size(self._model_number)))
259        return value
260
261    @property
262    def number_of_averages(self) -> int:
263        return self.read_spectrum_device_register(SPC_AVERAGES)
264
265    def set_number_of_averages(self, num_averages: int) -> None:
266        if num_averages > 0:
267            self.write_to_spectrum_device_register(SPC_AVERAGES, num_averages)
268        else:
269            raise ValueError("Number of averages must be greater than 0.")
270
271    @property
272    def acquisition_mode(self) -> AcquisitionMode:
273        """The currently enabled card mode. Will raise an exception if the current mode is not supported by
274        `spectrumdevice`.
275
276        Returns:
277            mode (`AcquisitionMode`): The currently enabled card acquisition mode."""
278        return AcquisitionMode(self.read_spectrum_device_register(SPC_CARDMODE))
279
280    def set_acquisition_mode(self, mode: AcquisitionMode) -> None:
281        """Change the currently enabled card mode. See `AcquisitionMode` and the Spectrum documentation
282        for the available modes.
283
284        Args:
285            mode (`AcquisitionMode`): The desired acquisition mode."""
286        self.write_to_spectrum_device_register(SPC_CARDMODE, mode.value)
287
288    @property
289    def batch_size(self) -> int:
290        return self._batch_size
291
292    def set_batch_size(self, batch_size: int) -> None:
293        self._batch_size = batch_size
294
295    def define_transfer_buffer(self, buffer: Optional[Sequence[TransferBuffer]] = None) -> None:
296        """Create or provide a `TransferBuffer` object for receiving acquired samples from the device.
297
298        If no buffer is provided, and no buffer has previously been defined, then one will be created: in FIFO mode,
299         with a notify size of 10 pages or the size of the acquisition, whichever is smaller; in Standard Single mode,
300         one with the correct length and no notify size. A separate buffer for transferring Timestamps will also be
301         created using the Timestamper class.
302
303        Args:
304            buffer (Optional[List[`TransferBuffer`]]): A length-1 list containing a pre-constructed
305                `TransferBuffer` set up for card-to-PC transfer of samples ("data"). The size of the buffer should be
306                chosen according to the current number of active channels, the acquisition length and the number
307                of acquisitions which you intend to download at a time using get_waveforms().
308        """
309        self._set_or_update_transfer_buffer_attribute(buffer)
310        if self._transfer_buffer is not None:
311            set_transfer_buffer(self._handle, self._transfer_buffer)
312
313    def _set_or_update_transfer_buffer_attribute(self, buffer: Optional[Sequence[TransferBuffer]]) -> None:
314        if buffer:
315            self._transfer_buffer = buffer[0]
316            if self._transfer_buffer.direction != BufferDirection.SPCM_DIR_CARDTOPC:
317                raise ValueError("Digitisers need a transfer buffer with direction BufferDirection.SPCM_DIR_CARDTOPC")
318            if self._transfer_buffer.type != BufferType.SPCM_BUF_DATA:
319                raise ValueError("Digitisers need a transfer buffer with type BufferDirection.SPCM_BUF_DATA")
320        elif self._transfer_buffer is None:
321            if self.acquisition_mode in (AcquisitionMode.SPC_REC_FIFO_MULTI, AcquisitionMode.SPC_REC_FIFO_AVERAGE):
322                samples_per_batch = (
323                    self.acquisition_length_in_samples * len(self.enabled_analog_channel_nums) * self._batch_size
324                )
325                pages_per_batch = samples_per_batch * self.bytes_per_sample / PAGE_SIZE_IN_BYTES
326
327                if pages_per_batch < DEFAULT_NOTIFY_SIZE_IN_PAGES:
328                    notify_size = pages_per_batch
329                else:
330                    notify_size = DEFAULT_NOTIFY_SIZE_IN_PAGES
331
332                # Make transfer buffer big enough to hold all samples in the batch
333                self._transfer_buffer = create_samples_acquisition_transfer_buffer(
334                    size_in_samples=samples_per_batch,
335                    notify_size_in_pages=notify_size,
336                    bytes_per_sample=self.bytes_per_sample,
337                )
338            elif self.acquisition_mode in (AcquisitionMode.SPC_REC_STD_SINGLE, AcquisitionMode.SPC_REC_STD_AVERAGE):
339                self._transfer_buffer = create_samples_acquisition_transfer_buffer(
340                    size_in_samples=self.acquisition_length_in_samples * len(self.enabled_analog_channel_nums),
341                    notify_size_in_pages=0,
342                    bytes_per_sample=self.bytes_per_sample,
343                )
344            else:
345                raise ValueError("AcquisitionMode not recognised")
346
347    def __str__(self) -> str:
348        return f"Card {self._visa_string}"

 31class SpectrumDigitiserStarHub(
 32    AbstractSpectrumStarHub[
 33        SpectrumDigitiserCard, SpectrumDigitiserAnalogChannelInterface, SpectrumDigitiserIOLineInterface
 34    ],
 35    AbstractSpectrumDigitiser,
 36):
 37    """Composite class of `SpectrumDigitiserCard` for controlling a StarHub digitiser device, for example the Spectrum
 38    NetBox. StarHub digitiser devices are composites of more than one Spectrum digitiser card. Acquisition from the
 39    child cards of a StarHub is synchronised, aggregating the channels of all child cards. This class enables the
 40    control of a StarHub device as if it were a single Spectrum card."""
 41
 42    def __init__(self, device_number: int, child_cards: tuple[SpectrumDigitiserCard, ...], master_card_index: int):
 43        """
 44        Args:
 45            device_number (int): The index of the StarHub to connect to. If only one StarHub is present, set to 0.
 46            child_cards (Sequence[`SpectrumDigitiserCard`]): A list of `SpectrumCard` objects defining the child cards
 47                located within the StarHub, correctly constructed with their IP addresses and/or device numbers.
 48            master_card_index (int): The position within child_cards where the master card (the card which controls the
 49                clock) is located.
 50        """
 51        super().__init__(device_number=device_number, child_cards=child_cards, master_card_index=master_card_index)
 52        self._acquisition_mode = self.acquisition_mode
 53
 54    def define_transfer_buffer(self, buffer: Optional[Sequence[TransferBuffer]] = None) -> None:
 55        """Create or provide `CardToPCDataTransferBuffer` objects for receiving acquired samples from the child cards.
 56        If no buffers are provided, they will be created with the correct size and a board_memory_offset_bytes of 0. See
 57        `SpectrumDigitiserCard.define_transfer_buffer()` for more information
 58
 59        Args:
 60            buffer (Optional[`CardToPCDataTransferBuffer`]): A list containing pre-constructed
 61            `CardToPCDataTransferBuffer` objects, one for each child card. The size of the buffers should be chosen
 62            according to the current number of active channels in each card and the acquisition length.
 63        """
 64        if buffer:
 65            for card, buff in zip(self._child_cards, buffer):
 66                card.define_transfer_buffer([buff])
 67        else:
 68            for card in self._child_cards:
 69                card.define_transfer_buffer()
 70
 71    def wait_for_acquisition_to_complete(self) -> None:
 72        """Wait for each card to finish its acquisition. See `SpectrumDigitiserCard.wait_for_acquisition_to_complete()`
 73        for more information."""
 74        for card in self._child_cards:
 75            card.wait_for_acquisition_to_complete()
 76
 77    def get_waveforms(self) -> List[List[NDArray[float64]]]:
 78        """Get a list of the most recently transferred waveforms, as floating point voltages.
 79
 80        This method gets the waveforms from each child card and joins them into a new list, ordered by channel number.
 81        See `SpectrumDigitiserCard.get_waveforms()` for more information.
 82
 83        Returns:
 84            waveforms (List[List[NDArray[float64]]]): A list lists of 1D numpy arrays, one inner list per acquisition,
 85              and one array per enabled channel, in channel order.
 86        """
 87        return self._get_waveforms_in_threads(SpectrumDigitiserCard.get_waveforms)
 88
 89    def get_raw_waveforms(self) -> List[List[NDArray[int16]]]:
 90        """Get a list of the most recently transferred waveforms, as integers.
 91
 92        This method gets the waveforms from each child card and joins them into a new list, ordered by channel number.
 93        See `SpectrumDigitiserCard.get_waveforms()` for more information.
 94
 95        Returns:
 96            waveforms (List[List[NDArray[int16]]]): A list lists of 1D numpy arrays, one inner list per acquisition,
 97              and one array per enabled channel, in channel order.
 98        """
 99        return self._get_waveforms_in_threads(SpectrumDigitiserCard.get_raw_waveforms)
100
101    def _get_waveforms_in_threads(
102        self, get_waveforms_method: Callable[[SpectrumDigitiserCard], List[List[WAVEFORM_TYPE_VAR]]]
103    ) -> List[List[WAVEFORM_TYPE_VAR]]:
104        """Gets waveforms from child cards in separate threads, using the SpectrumDigitiserCard method provided."""
105
106        card_ids_and_waveform_sets: Dict[str, list[list[WAVEFORM_TYPE_VAR]]] = {}
107
108        def _get_waveforms(digitiser_card: SpectrumDigitiserCard) -> None:
109            this_cards_waveforms = get_waveforms_method(digitiser_card)
110            card_ids_and_waveform_sets[str(digitiser_card)] = this_cards_waveforms
111
112        threads = [Thread(target=_get_waveforms, args=(card,)) for card in self._child_cards]
113
114        for thread in threads:
115            thread.start()
116        for thread in threads:
117            thread.join()
118
119        waveform_sets_all_cards_ordered = []
120        for n in range(self.batch_size):
121            waveforms_in_this_batch = []
122            for card in self._child_cards:
123                waveforms_in_this_batch += card_ids_and_waveform_sets[str(card)][n]
124            waveform_sets_all_cards_ordered.append(waveforms_in_this_batch)
125
126        return waveform_sets_all_cards_ordered
127
128    def get_timestamp(self) -> Optional[datetime.datetime]:
129        """Get timestamp for the last acquisition"""
130        return self._triggering_card.get_timestamp()
131
132    def enable_timestamping(self) -> None:
133        self._triggering_card.enable_timestamping()
134
135    @property
136    def acquisition_length_in_samples(self) -> int:
137        """The currently set recording length, which should be the same for all child cards. If different recording
138        lengths are set, an exception is raised. See `SpectrumDigitiserCard.acquisition_length_in_samples` for more
139        information.
140
141        Returns:
142            length_in_samples: The currently set acquisition length in samples."""
143        lengths = []
144        for d in self._child_cards:
145            lengths.append(d.acquisition_length_in_samples)
146        return check_settings_constant_across_devices(lengths, __name__)
147
148    def set_acquisition_length_in_samples(self, length_in_samples: int) -> None:
149        """Set a new recording length for all child cards. See `SpectrumDigitiserCard.set_acquisition_length_in_samples()`
150        for more information.
151
152        Args:
153            length_in_samples (int): The desired acquisition length in samples."""
154        for d in self._child_cards:
155            d.set_acquisition_length_in_samples(length_in_samples)
156
157    @property
158    def post_trigger_length_in_samples(self) -> int:
159        """The number of samples recorded after a trigger is received. This should be consistent across all child
160        cards. If different values are found across the child cards, an exception is raised. See
161        `SpectrumDigitiserCard.post_trigger_length_in_samples` for more information.
162
163        Returns:
164            length_in_samples (int): The current post trigger length in samples.
165        """
166        lengths = []
167        for d in self._child_cards:
168            lengths.append(d.post_trigger_length_in_samples)
169        return check_settings_constant_across_devices(lengths, __name__)
170
171    def set_post_trigger_length_in_samples(self, length_in_samples: int) -> None:
172        """Set a new post trigger length for all child cards. See `SpectrumDigitiserCard.set_post_trigger_length_in_samples()`
173        for more information.
174
175        Args:
176            length_in_samples (int): The desired post trigger length in samples.
177        """
178        for d in self._child_cards:
179            d.set_post_trigger_length_in_samples(length_in_samples)
180
181    @property
182    def acquisition_mode(self) -> AcquisitionMode:
183        """The acquisition mode, which should be the same for all child cards. If it's not, an exception is raised.
184        See `SpectrumDigitiserCard.acquisition_mode` for more information.
185
186        Returns:
187            mode (`AcquisitionMode`): The currently enabled acquisition mode.
188        """
189        modes = []
190        for d in self._child_cards:
191            modes.append(d.acquisition_mode)
192        return AcquisitionMode(check_settings_constant_across_devices([m.value for m in modes], __name__))
193
194    def set_acquisition_mode(self, mode: AcquisitionMode) -> None:
195        """Change the acquisition mode for all child cards. See `SpectrumDigitiserCard.set_acquisition_mode()` for more
196        information.
197
198        Args:
199            mode (`AcquisitionMode`): The desired acquisition mode."""
200        for d in self._child_cards:
201            d.set_acquisition_mode(mode)
202
203    @property
204    def batch_size(self) -> int:
205        batch_sizes = []
206        for d in self._child_cards:
207            batch_sizes.append(d.batch_size)
208        return check_settings_constant_across_devices(batch_sizes, __name__)
209
210    def set_batch_size(self, batch_size: int) -> None:
211        for d in self._child_cards:
212            d.set_batch_size(batch_size)
213
214    def force_trigger(self) -> None:
215        for d in self._child_cards:
216            d.force_trigger()
217
218    @property
219    def type(self) -> CardType:
220        return self._child_cards[0].type
221
222    @property
223    def model_number(self) -> ModelNumber:
224        return self._child_cards[0].model_number
225
226    @property
227    def analog_channels(self) -> Sequence[SpectrumDigitiserAnalogChannelInterface]:
228        """A tuple containing of all the channels of the child cards of the hub. See `AbstractSpectrumCard.channels` for
229        more information.
230
231        Returns:
232            channels (Sequence[`SpectrumDigitiserAnalogChannelInterface`]):
233            A tuple of `SpectrumDigitiserAnalogChannelInterface` objects.
234        """
235        return super().analog_channels