test_stream_counter_readers_writers.py 文件源码

def test_one_sample_uint32(self, x_series_device, seed):
        # Reset the pseudorandom number generator with seed.
        random.seed(seed)

        number_of_pulses = random.randint(2, 50)
        frequency = random.uniform(1000, 10000)

        # Select random counters from the device.
        counters = random.sample(self._get_device_counters(x_series_device), 2)

        with nidaqmx.Task() as write_task, nidaqmx.Task() as read_task:
            write_task.co_channels.add_co_pulse_chan_freq(
                counters[0], freq=frequency)
            write_task.timing.cfg_implicit_timing(
                samps_per_chan=number_of_pulses)

            read_task.ci_channels.add_ci_count_edges_chan(counters[1])
            read_task.ci_channels.all.ci_count_edges_term = (
                '/{0}InternalOutput'.format(counters[0]))

            reader = CounterReader(read_task.in_stream)

            read_task.start()
            write_task.start()

            write_task.wait_until_done(timeout=2)

            value_read = reader.read_one_sample_uint32()
            assert value_read == number_of_pulses