Initialize DAQmx task and channels.
(self)
| 202 | self.samples_read = 0 |
| 203 | |
| 204 | def setup(self) -> None: |
| 205 | """Initialize DAQmx task and channels.""" |
| 206 | self.task = nidaqmx.Task() |
| 207 | |
| 208 | for ch_cfg in self.channel_config: |
| 209 | physical_channel = f"{DEVICE_NAME}/{ch_cfg['channel']}" |
| 210 | v_min, v_max = ch_cfg["voltage_range"] |
| 211 | term_config = TERMINAL_MAP.get( |
| 212 | ch_cfg["terminal"].upper(), TerminalConfiguration.RSE |
| 213 | ) |
| 214 | |
| 215 | self.task.ai_channels.add_ai_voltage_chan( |
| 216 | physical_channel, |
| 217 | name_to_assign_to_channel=ch_cfg.get("name", ch_cfg["channel"]), |
| 218 | min_val=v_min, |
| 219 | max_val=v_max, |
| 220 | terminal_config=term_config, |
| 221 | ) |
| 222 | ch_type = ch_cfg.get("type", "voltage") |
| 223 | if ch_type == "current": |
| 224 | print( |
| 225 | f" {ch_cfg['channel']}: {ch_type} (shunt={ch_cfg.get('shunt_resistor', 1.0)} ohm)" |
| 226 | ) |
| 227 | else: |
| 228 | print(f" {ch_cfg['channel']}: {ch_type} ({v_min}V to {v_max}V)") |
| 229 | |
| 230 | self.task.timing.cfg_samp_clk_timing( |
| 231 | rate=SAMPLE_RATE, |
| 232 | sample_mode=AcquisitionType.CONTINUOUS, |
| 233 | samps_per_chan=max(SAMPLES_PER_READ * 10, 10), |
| 234 | ) |
| 235 | |
| 236 | self.reader = AnalogMultiChannelReader(self.task.in_stream) |
| 237 | self.task.start() |
| 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.""" |