test_stream_counter_readers_writers.py 文件源码

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

        number_of_samples = 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), 3)

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

            read_task.ci_channels.add_ci_freq_chan(
                counters[2], min_val=1000, max_val=10000, edge=Edge.RISING)
            read_task.ci_channels.all.ci_freq_term = (
                '/{0}InternalOutput'.format(counters[1]))
            read_task.timing.cfg_implicit_timing(
                samps_per_chan=number_of_samples)

            read_task.start()
            write_task.start()
            write_task.wait_until_done(timeout=2)

            reader = CounterReader(read_task.in_stream)

            values_read = numpy.zeros(number_of_samples, dtype=numpy.float64)
            reader.read_many_sample_double(
                values_read, number_of_samples_per_channel=number_of_samples,
                timeout=2)

            expected_values = [frequency for _ in range(number_of_samples)]

            numpy.testing.assert_allclose(
                values_read, expected_values, rtol=0.05)