-
Notifications
You must be signed in to change notification settings - Fork 170
/
Copy pathsystem.py
629 lines (521 loc) · 25.6 KB
/
system.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
# Do not edit this file; it was automatically generated.
import collections
import ctypes
import deprecation
import numpy
from nidaqmx import utils
from nidaqmx.system.collections._device_collection import DeviceCollection
from nidaqmx.system.collections._persisted_channel_collection import (
PersistedChannelCollection)
from nidaqmx.system.collections._persisted_scale_collection import (
PersistedScaleCollection)
from nidaqmx.system.collections._persisted_task_collection import (
PersistedTaskCollection)
from nidaqmx.utils import flatten_channel_string, unflatten_channel_string
from nidaqmx.constants import (
AOPowerUpOutputBehavior, LogicFamily, PowerUpChannelType, PowerUpStates, ResistorState,
SignalModifiers, WAIT_INFINITELY)
from nidaqmx.types import (
AOPowerUpState, CDAQSyncConnection, DOPowerUpState, DOResistorPowerUpState)
from nidaqmx.system.device import _DeviceAlternateConstructor
__all__ = ['System']
class System:
"""
Represents a DAQmx system.
Contains static properties that access tasks, scales, and global channels
stored in Measurement Automation Explorer (MAX), performs immediate
operations on DAQ hardware, and creates classes from which you can get
information about the hardware.
"""
def __init__(self, grpc_options=None):
"""
Args:
grpc_options (Optional[:class:`~nidaqmx.GrpcSessionOptions`]): Specifies
the gRPC session options.
"""
self._interpreter = utils._select_interpreter(grpc_options)
@staticmethod
def local():
"""
nidaqmx.system.system.System: Represents the local DAQmx system.
"""
return System()
@staticmethod
def remote(grpc_options):
"""
nidaqmx.system.system.System: Represents the remote DAQmx system.
Args:
grpc_options (:class:`~nidaqmx.GrpcSessionOptions`): Specifies
the gRPC session options.
"""
return System(grpc_options)
@property
def devices(self):
"""
nidaqmx.system.collections.DeviceCollection: Indicates the
collection of devices for this DAQmx system.
"""
return DeviceCollection(self._interpreter)
DriverVersion = collections.namedtuple(
'DriverVersion', ['major_version', 'minor_version',
'update_version'])
@property
def driver_version(self):
"""
collections.namedtuple: Indicates the major, minor and update
portions of the installed version of NI-DAQmx.
- major_version (int): Indicates the major portion of the
installed version of NI-DAQmx, such as 7 for version 7.0.
- minor_version (int): Indicates the minor portion of the
installed version of NI-DAQmx, such as 0 for version 7.0.
- update_version (int): Indicates the update portion of the
installed version of NI-DAQmx, such as 1 for version 9.0.1.
"""
return System.DriverVersion(self._major_version, self._minor_version,
self._update_version)
@property
def global_channels(self):
"""
nidaqmx.system.collections.PersistedChannelCollection: Indicates
the collection of global channels for this DAQmx system.
"""
return PersistedChannelCollection(self._interpreter)
@property
def scales(self):
"""
nidaqmx.system.collections.PersistedScaleCollection: Indicates
the collection of custom scales for this DAQmx system.
"""
return PersistedScaleCollection(self._interpreter)
@property
def tasks(self):
"""
nidaqmx.system.collections.PersistedTaskCollection: Indicates
the collection of saved tasks for this DAQmx system.
"""
return PersistedTaskCollection(self._interpreter)
@property
def _major_version(self):
"""
int: Indicates the major portion of the installed version of NI-
DAQmx, such as 7 for version 7.0.
"""
val = self._interpreter.get_system_info_attribute_uint32(0x1272)
return val
@property
def _minor_version(self):
"""
int: Indicates the minor portion of the installed version of NI-
DAQmx, such as 0 for version 7.0.
"""
val = self._interpreter.get_system_info_attribute_uint32(0x1923)
return val
@property
def _update_version(self):
"""
int: Indicates the update portion of the installed version of
NI-DAQmx, such as 1 for version 9.0.1.
"""
val = self._interpreter.get_system_info_attribute_uint32(0x2f22)
return val
def connect_terms(
self, source_terminal, destination_terminal,
signal_modifiers=SignalModifiers.DO_NOT_INVERT_POLARITY):
"""
Creates a route between a source and destination terminal. The
route can carry a variety of digital signals, such as triggers,
clocks, and hardware events.
Args:
source_terminal (str): Specifies the originating terminal of
the route. A DAQmx terminal constant lists all terminals
available on devices installed in the system. You also
can specify a source terminal by specifying a string
that contains a terminal name.
destination_terminal (str): Specifies the receiving terminal
of the route. A DAQmx terminal constant provides a list
of all terminals available on devices installed in the
system. You also can specify a destination terminal by
specifying a string that contains a terminal name.
signal_modifiers (Optional[nidaqmx.constants.SignalModifiers]):
Specifies whether to invert the signal this function
routes from the source terminal to the destination
terminal.
"""
self._interpreter.connect_terms(
source_terminal, destination_terminal, signal_modifiers.value)
def disconnect_terms(self, source_terminal, destination_terminal):
"""
Removes signal routes you created by using the DAQmx Connect
Terminals function. The DAQmx Disconnect Terminals function
cannot remove task-based routes, such as those you create
through timing and triggering configuration.
Args:
source_terminal (str): Specifies the originating terminal of
the route. A DAQmx terminal constant lists all terminals
available on devices installed in the system. You also
can specify a source terminal by specifying a string
that contains a terminal name.
destination_terminal (str): Specifies the receiving terminal
of the route. A DAQmx terminal constant provides a list
of all terminals available on devices installed in the
system. You also can specify a destination terminal by
specifying a string that contains a terminal name.
"""
self._interpreter.disconnect_terms(
source_terminal, destination_terminal)
def tristate_output_term(self, output_terminal):
"""
Sets a terminal to high-impedance state. If you connect an
external signal to a terminal on the I/O connector, the terminal
must be in high-impedance state. Otherwise, the device could
double-drive the terminal and damage the hardware. If you use
this function on a terminal in an active route, the function
fails and returns an error.
Args:
output_terminal (str): Specifies the terminal on the I/O
connector to set to high-impedance state. A DAQmx
terminal constant lists all available terminals on
installed devices. You also can specify an output
terminal by using a string that contains a terminal
name.
"""
self._interpreter.tristate_output_term(
output_terminal)
# region Power Up States Functions
def set_digital_power_up_states(
self, device_name, power_up_states):
"""
Updates power up states for digital physical channels.
Args:
device_name (str): Specifies the name as configured in MAX
of the device to which this operation applies.
power_up_states (List[nidaqmx.types.DOPowerUpState]):
Contains the physical channels and power up states to
set. Each element of the list contains a physical channel
and the power up state to set for that physical channel.
- physical_channel (str): Specifies the digital line or
port to modify. You cannot modify dedicated digital
input lines.
- power_up_state (:class:`nidaqmx.constants.PowerUpStates`):
Specifies the power up state to set for the physical
channel specified with the **physical_channel** input.
"""
channel_names = []
states = []
for p in power_up_states:
channel_names.append(p.physical_channel)
states.append(p.power_up_state.value)
self._interpreter.set_digital_power_up_states(device_name, channel_names, states)
def get_digital_power_up_states(self, device_name):
"""
Gets the power up states for digital physical lines.
Args:
device_name (str): Specifies the name as configured in MAX
of the device to which this operation applies.
Returns:
List[nidaqmx.types.DOPowerUpState]:
Contains the physical channels and power up states set. Each
element of the list contains a physical channel and the power
up state set for that physical channel.
- physical_channel (str): Indicates the physical channel that
was modified.
- power_up_state (:class:`nidaqmx.constants.PowerUpStates`):
Indicates the power up state set for the physical channel
specified with the **physical_channel** output.
"""
device = _DeviceAlternateConstructor(device_name, self._interpreter)
channel_names = []
for do_line in device.do_lines:
channel_names.append(do_line.name)
states = self._interpreter.get_digital_power_up_states(device_name, channel_names)
power_up_states = []
for d, p in zip(device.do_lines, states):
power_up_states.append(
DOPowerUpState(physical_channel=d.name,
power_up_state=PowerUpStates(p)))
return power_up_states
def set_digital_pull_up_pull_down_states(
self, device_name, power_up_states):
"""
Sets the resistor level to pull up or pull down for lines when
they are in tristate logic.
Args:
device_name (str): Specifies the name as configured in MAX
of the device to which this operation applies.
power_up_states (List[nidaqmx.types.DOResistorPowerUpState]):
Contains the physical channels and power up states to
set. Each element of the list contains a physical channel
and the power up state to set for that physical channel.
- physical_channel (str): Specifies the digital line or
port to modify. You cannot modify dedicated digital
input lines.
- power_up_state (:class:`nidaqmx.constants.ResistorState`):
Specifies the power up state to set for the physical
channel specified with the **physical_channel** input.
"""
channel_names = []
states = []
for p in power_up_states:
channel_names.append(p.physical_channel)
states.append(p.power_up_state.value)
self._interpreter.set_digital_pull_up_pull_down_states(device_name, channel_names, states)
def get_digital_pull_up_pull_down_states(self, device_name):
"""
Gets the resistor level for lines when they are in tristate
logic.
Args:
device_name (str): Specifies the name as configured in MAX
of the device to which this operation applies.
Returns:
List[nidaqmx.types.DOResistorPowerUpState]:
Contains the physical channels and power up states set. Each
element of the list contains a physical channel and the power
up state set for that physical channel.
- physical_channel (str): Indicates the physical channel that
was modified.
- power_up_state (:class:`nidaqmx.constants.ResistorState`):
Indicates the power up state set for the physical channel
specified with the **physical_channel** output.
"""
channel_names = []
states = []
device = _DeviceAlternateConstructor(device_name, self._interpreter)
for do_line in device.do_lines:
channel_names.append(do_line.name)
states = self._interpreter.get_digital_pull_up_pull_down_states(device_name, channel_names)
power_up_states = []
for d, p in zip(device.do_lines, states):
power_up_states.append(
DOResistorPowerUpState(
physical_channel=d.name,
power_up_state=ResistorState(p)))
return power_up_states
@deprecation.deprecated(deprecated_in="0.8.0", details="Use set_analog_power_up_states_with_output_type instead.")
def set_analog_power_up_states(self, device_name, power_up_states):
"""
Updates power up states for analog physical channels.
Args:
device_name (str): Specifies the name as configured in MAX
of the device to which this operation applies.
power_up_states (List[nidaqmx.types.AOPowerUpState]):
Contains the physical channels and power up states to
set. Each element of the list contains a physical channel
and the power up state to set for that physical channel.
- physical_channel (str): Specifies the physical channel
to modify.
- power_up_state (float): Specifies the power up state to
set for the physical channel specified with the
**physical_channel** input.
- channel_type (:class:`nidaqmx.constants.AOPowerUpOutputBehavior`):
Specifies the output type for the physical channel
specified with the **physical_channel** input.
"""
channel_names = []
states = []
channel_types = []
for p in power_up_states:
channel_names.append(p.physical_channel)
states.append(p.power_up_state)
channel_types.append(p.channel_type.value)
self._interpreter.set_analog_power_up_states(device_name, channel_names, states, channel_types)
def set_analog_power_up_states_with_output_type(
self, power_up_states):
"""
Updates power up states for analog physical channels.
Args:
power_up_states (List[nidaqmx.types.AOPowerUpState]):
Contains the physical channels and power up states to
set. Each element of the list contains a physical channel
and the power up state to set for that physical channel.
- physical_channel (str): Specifies the physical channel to
modify.
- power_up_state (float): Specifies the power up state
to set for the physical channel specified with the
**physical_channel** input.
- channel_type (:class:`nidaqmx.constants.AOPowerUpOutputBehavior`):
Specifies the output type for the physical channel
specified with the **physical_channel** input.
"""
physical_channel = flatten_channel_string(
[p.physical_channel for p in power_up_states])
state = numpy.array(
[p.power_up_state for p in power_up_states], dtype=numpy.float64)
channel_type = numpy.array(
[p.channel_type.value for p in power_up_states], dtype=numpy.int32)
self._interpreter.set_analog_power_up_states_with_output_type(physical_channel, state, channel_type)
@deprecation.deprecated(deprecated_in="0.8.0", details="Use get_analog_power_up_states_with_output_type instead.")
def get_analog_power_up_states(self, device_name):
"""
Gets the power up states for analog physical channels.
Args:
device_name (str): Specifies the name as configured in MAX
of the device to which this operation applies.
Returns:
power_up_states (List[nidaqmx.types.AOPowerUpState]):
Contains the physical channels and power up states set. Each
element of the list contains a physical channel and the
power up state set for that physical channel.
- physical_channel (str): Specifies the physical channel that
was modified.
- power_up_state (float): Specifies the power up state set
for the physical channel specified with the
**physical_channel** input.
- channel_type (:class:`nidaqmx.constants.AOPowerUpOutputBehavior`):
Specifies the output type for the physical channel
specified with the **physical_channel** input.
"""
device = _DeviceAlternateConstructor(device_name, self._interpreter)
channel_names = device.ao_physical_chans.channel_names
# Here get_analog_power_up_states is not called, since it is deprecated and passes channel_type in wrong direction
return self.get_analog_power_up_states_with_output_type(channel_names)
def get_analog_power_up_states_with_output_type(self, physical_channels):
"""
Gets the power up states for analog physical channels.
Args:
physical_channels (List[str]): Indicates the physical
channels that were modified.
Returns:
power_up_states (List[nidaqmx.types.AOPowerUpState]):
Contains the physical channels and power up states set. Each
element of the list contains a physical channel and the
power up state set for that physical channel.
- physical_channel (str): Specifies the physical channel that
was modified.
- power_up_state (float): Specifies the power up state set
for the physical channel specified with the
**physical_channel** input.
- channel_type (:class:`nidaqmx.constants.AOPowerUpOutputBehavior`):
Specifies the output type for the physical channel
specified with the **physical_channel** input.
"""
states, channel_types = self._interpreter.get_analog_power_up_states_with_output_type(
flatten_channel_string(physical_channels), len(physical_channels))
assert len(states) == len(physical_channels)
assert len(channel_types) == len(physical_channels)
power_up_states = []
for p, s, c in zip(physical_channels, states, channel_types):
power_up_states.append(
AOPowerUpState(
physical_channel=p,
power_up_state=float(s),
channel_type=PowerUpChannelType(c)))
return power_up_states
def set_digital_logic_family_power_up_state(
self, device_name, logic_family):
"""
Sets the digital logic family to use when the device powers up.
Args:
device_name (str): Specifies the name as configured in MAX
of the device to which this operation applies.
logic_family (nidaqmx.constants.LogicFamily): Specifies the
logic family set to the device to when it powers up. A
logic family corresponds to voltage thresholds that are
compatible with a group of voltage standards. Refer to
device documentation for information on the logic high
and logic low voltages for these logic families.
"""
self._interpreter.set_digital_logic_family_power_up_state(device_name, logic_family.value)
def get_digital_logic_family_power_up_state(self, device_name):
"""
Gets the digital logic family for a device.
Args:
device_name (str): Specifies the name as configured in MAX
of the device to which this operation applies.
Returns:
nidaqmx.constants.LogicFamily:
Specifies the logic family to set the device to when it powers
up. A logic family corresponds to voltage thresholds that are
compatible with a group of voltage standards. Refer to device
documentation for information on the logic high and logic low
voltages for these logic families.
"""
logic_family = self._interpreter.get_digital_logic_family_power_up_state(device_name)
return LogicFamily(logic_family)
# endregion
# region cDAQ Sync Functions
def auto_configure_cdaq_sync_connections(
self, chassis_devices_ports="", timeout=WAIT_INFINITELY):
"""
Detects and configures cDAQ Sync connections between devices.
Stop all NI-DAQmx tasks running on the devices prior to running
this function because any running tasks cause auto-configuration
to fail.
Args:
chassis_devices_ports (Optional[str]): Specifies the names of the
CompactDAQ chassis, C Series modules, or cDAQ Sync ports in
comma separated form to search. If no names are specified, all
cDAQ Sync ports on connected, non-simulated devices are
scanned.
timeout (Optional[float]): Specifies the time in seconds to
wait for the device to respond before timing out. If a
timeout occurs, no configuration is changed.
Returns:
List[nidaqmx.types.CDAQSyncConnection]:
Returns the configured port-to-port connections.
"""
self._interpreter.auto_configure_cdaq_sync_connections(chassis_devices_ports, timeout)
port_list = self._interpreter.get_auto_configured_cdaq_sync_connections()
ports = unflatten_channel_string(port_list)
output_ports = ports[::2]
input_ports = ports[1::2]
connections = []
for output_port, input_port in zip(output_ports, input_ports):
connections.append(
CDAQSyncConnection(output_port=output_port,
input_port=input_port))
return connections
def are_configured_cdaq_sync_ports_disconnected(
self, chassis_devices_ports="", timeout=WAIT_INFINITELY):
"""
Verifies configured cDAQ Sync connections between devices.
Failures generally indicate a wiring issue or that a device has
been powered off or removed. Stop all NI-DAQmx tasks running on
the devices prior to running this function because any running
tasks cause the verification process to fail.
Args:
chassis_devices_ports (Optional[str]): Specifies the names
of the CompactDAQ chassis, C Series modules, or cDAQ
Sync ports in comma separated form to search. If no
names are specified, all cDAQ Sync ports on connected,
non-simulated devices are scanned.
timeout (Optional[float]): Specifies the time in seconds to
wait for the device to respond before timing out.
Returns:
List[nidaqmx.types.CDAQSyncConnection]:
Returns the port-to-port connections that failed verification.
"""
disconnected_ports_exist = self._interpreter.are_configured_cdaq_sync_ports_disconnected(
chassis_devices_ports, timeout)
port_list = self._interpreter.get_disconnected_cdaq_sync_ports()
ports = unflatten_channel_string(port_list)
output_ports = ports[::2]
input_ports = ports[1::2]
connections = []
for output_port, input_port in zip(output_ports, input_ports):
connections.append(
CDAQSyncConnection(output_port=output_port,
input_port=input_port))
return connections
def add_cdaq_sync_connection(self, ports_to_connect):
"""
Adds a cDAQ Sync connection between devices. The connection is
not verified.
Args:
ports_to_connect (nidaqmx.types.CDAQSyncConnection):
Specifies the cDAQ Sync ports to connect.
"""
port_list = flatten_channel_string(
[ports_to_connect.output_port, ports_to_connect.input_port])
self._interpreter.add_cdaq_sync_connection(port_list)
def remove_cdaq_sync_connection(self, ports_to_disconnect):
"""
Removes a cDAQ Sync connection between devices. The connection
is not verified.
Args:
ports_to_disconnect (nidaqmx.types.CDAQSyncConnection):
Specifies the cDAQ Sync ports to disconnect.
"""
port_list = flatten_channel_string(
[ports_to_disconnect.output_port, ports_to_disconnect.input_port])
self._interpreter.remove_cdaq_sync_connection(port_list)
# endregion