Read samples from all channels.
(self)
| 238 | print(f"DAQmx started: {self.num_channels} channels @ {SAMPLE_RATE} Hz") |
| 239 | |
| 240 | def read(self) -> np.ndarray: |
| 241 | """Read samples from all channels.""" |
| 242 | samples = self.reader.read_many_sample( |
| 243 | self._buffer, number_of_samples_per_channel=SAMPLES_PER_READ |
| 244 | ) |
| 245 | self.samples_read += samples |
| 246 | return self._buffer[:, :samples] |
| 247 | |
| 248 | def close(self) -> None: |
| 249 | if self.task: |
no outgoing calls
no test coverage detected