test_stream_counter_readers_writers.py 文件源码

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

        high_ticks = random.randint(100, 1000)
        low_ticks = random.randint(100, 1000)
        starting_edge = random.choice([Edge.RISING, Edge.FALLING])

        # 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_ticks(
                counters[0],
                '/{0}/100kHzTimebase'.format(x_series_device.name),
                high_ticks=high_ticks, low_ticks=low_ticks)
            write_task.timing.cfg_implicit_timing(
                sample_mode=AcquisitionType.CONTINUOUS)

            read_task.ci_channels.add_ci_pulse_chan_ticks(
                counters[1], source_terminal='/{0}/100kHzTimebase'.format(
                    x_series_device.name),
                min_val=100, max_val=1000)
            read_task.ci_channels.all.ci_pulse_ticks_term = (
                '/{0}InternalOutput'.format(counters[0]))
            read_task.ci_channels.all.ci_pulse_ticks_starting_edge = (
                starting_edge)

            read_task.start()
            write_task.start()

            reader = CounterReader(read_task.in_stream)
            value_read = reader.read_one_sample_pulse_ticks()
            write_task.stop()

            assert numpy.isclose(
                value_read.high_tick, high_ticks, rtol=0.05, atol=1)
            assert numpy.isclose(
                value_read.low_tick, low_ticks, rtol=0.05, atol=1)