How can you implement DAQmx streaming & callback functions using betterproto stubs?

[sjdemartini]

Per the suggestion in the Wiki here https://github.com/ni/grpc-device/wiki/Creating-a-gRPC-Client, I am using the betterproto compiler to generate python NI clients. (This has proved critical in order to support both typing and asyncio behavior.)

I read the Wiki section here on NI DAQmx callback functions, and have looked closely at the analog-input-every-n-samples.py example (which uses the vanilla Google proto client).

I have attempted to replicate this behavior using betterproto clients; however, I haven't figured out a straightforward way to "finish" the read task once the "DoneEvent" occurs. It appears that betterproto does not provide access to the gRPC stream while using the async generator from the the "register_" methods (see source code for the _unary_stream method which is called by register_every_n_samples_event, register_done_event, and register_signal_event).

As such, when receiving a done message in register_done_event, there doesn't appear to be a way to tell the "read samples" stream to complete/cancel.

Here's some example code I have put together which contains the meat of the relevant logic from the example above (also configured to collect 1000 samples per channel at 10000 Hz):

response = await raise_if_error_async(
    daq_client.get_task_attribute_u_int32(
        task=task,
        attribute_raw=nidaqmx_grpc.TaskUInt32Attributes.TASK_ATTRIBUTE_NUM_CHANS,
    )
)
number_of_channels = response.value

async def read_data():
    async for every_n_samples_response in daq_client.register_every_n_samples_event(
        task=task,
        n_samples=100,
        every_n_samples_event_type_raw=nidaqmx_grpc.EveryNSamplesEventType.EVERY_N_SAMPLES_EVENT_TYPE_ACQUIRED_INTO_BUFFER,
    ):
        await _raise_if_error_sync(every_n_samples_response)
        read_response = await raise_if_error_async(
            daq_client.read_analog_f64(
                task=task,
                num_samps_per_chan=100,
                fill_mode_raw=nidaqmx_grpc.GroupBy.GROUP_BY_GROUP_BY_CHANNEL,
                array_size_in_samps=number_of_channels * 100,
            )
        )

        print(
            f"Acquired {len(read_response.read_array)} samples",
            f"({read_response.samps_per_chan_read} samples per channel)",
        )
        print("Read Data (first 10 samples):", read_response.read_array[:10])

    print("read_data Finished")

async def wait_for_done():
    async for done_response in daq_client.register_done_event(task=task):
        await _raise_if_error_sync(done_response)
        break

    print("wait_for_done Finished")

start_task = raise_if_error_async(daq_client.start_task(task=task))

print("Starting")
await asyncio.gather(read_data(), wait_for_done(), start_task)
print("Done")

This will end up hanging indefinitely. The print statements for Acquired 100 samples will show up all 10 times (for the total 1000 finite samples), and wait_for_done Finished will print out as well somewhere along the way. But read_data Finished and Done will never print and the code will never finish executing, since it appears to be hanging waiting for further messages on the register_every_n_samples_event stream that will never come.

Is there some way around this, or a suggested approach here? (I'm guessing it could be done by re-implementing the logic of register_every_n_samples_event so as to not use the built-in _unary_stream but instead one where you can control the stream yourself and cancel it.) An official betterproto version of the example would be greatly appreciated.

---

_{Side question: is the await asyncio.gather(read_data(), wait_for_done(), start_task) actually guaranteed to work, or is there a possibility that the callbacks may not actually get registered before starting the task due to a race condition (since we can't call stream.initial_metadata() as in the vanilla example)? This relates to the same problem above of not being able to use the stream directly.}

[astarche]

Tough question, thanks!

You can get it to work using aclose the same way the other example uses cancel in wait_for_done. But, I can't figure out a way to get past your "Side Question" without adding a sleep. That's frustrating. Maybe that could be added as a feature to betterproto (not sure how that would work). That project hasn't been very active lately, unfortunately.

The snippet below worked for me. I think you're also correct that you could mix in lower-level calls as an alternative. betterproto wraps grpclib internally.

You could also consider using mypy-protobuf to add type hints to the default protoc gen code. And/or use blocking reads instead of events if that can work for you.

This is an unfortunate combination of betterprotos minimalist streaming API and our challenging "callback wrapper" streaming implementation. Eventually, I'd like to add true "data streaming" APIs. This is good feedback on interest in that type of API and specific challenges with what we have now.

       every_n_samples_stream = daq_service.register_every_n_samples_event(
            task=task,
            n_samples=100,
           every_n_samples_event_type_raw=nidaqmx_grpc.EveryNSamplesEventType.EVERY_N_SAMPLES_EVENT_TYPE_ACQUIRED_INTO_BUFFER,
        )

        done_stream = daq_service.register_done_event(task=task)

        async def read_data():
            async for every_n_samples_response in every_n_samples_stream:
                await raise_if_error(every_n_samples_response)
                read_response = await raise_if_error(
                    await daq_service.read_analog_f64(
                        task=task,
                        num_samps_per_chan=100,
                        fill_mode_raw=nidaqmx_grpc.GroupBy.GROUP_BY_GROUP_BY_CHANNEL,
                        array_size_in_samps=100,
                    )
                )

                print(
                    f"Acquired {len(read_response.read_array)} samples",
                    f"({read_response.samps_per_chan_read} samples per channel)",
                )
                print("Read Data (first 10 samples):", read_response.read_array[:10])

            print("read_data Finished")

        async def wait_for_done():
            async for done_response in done_stream:
                await every_n_samples_stream.aclose()
                await done_stream.aclose()
                await raise_if_error(done_response)

            print("wait_for_done Finished")

        acquisition = asyncio.gather(read_data(), wait_for_done())

        # How to wait for initial metadata in betterproto?
        await sleep(2)

        await raise_if_error(await daq_service.start_task(task=task))

        print("Starting")
        await acquisition
        print("Done")

[sjdemartini]

Thanks so much for the detailed reply!

That all makes sense to me, and I agree that it seems betterproto isn't well suited to the finer-grained control unfortunately. I attempted to use mypy-protobuf and the standard grpc library originally, but unfortunately typing isn't yet supported for the asyncio version of the grpc code (shabbyrobe/grpc-stubs#22, nipunn1313/mypy-protobuf#216). That made it significantly harder to manage and validate that the code was being used properly (and made things clunkier in my editor), hence the choice to use betterproto.

I've tested out the code you pasted above, and I think it mostly works. However, it seems that you can actually prematurely close the every_n_samples_stream and not record all of the data you intended to. You can reproduce that more easily by increasing the sample clock rate (to say 10000) and keeping the rest of the values the same. You'll see that it usually ends up printing "Acquired 100 samples" less than 10 times (i.e., the expected amount with 1000 samples divided by the 100 samples per event). Perhaps this bug is actually present in the original every_n_samples_stream.cancel() logic in the analog-input-every-n-samples.py file? If so, is the "right" way to ensure you collect all samples to sum up read_response.samps_per_chan_read across iterations, and then using a break in the every_n_samples_stream once you reach the samps_per_chan total? That could presumably get around my original problem of needing to close the stream, too, though it's slightly cumbersome/unintuitive to have to check when you're done.

Good to know about the start race-condition, and your suggestion to use a sleep makes sense and is probably good enough. Again, appreciate your time and help!

[astarche]

You'll see that it usually ends up printing "Acquired 100 samples" less than 10 times (i.e., the expected amount with 1000 samples divided by the 100 samples per event).

Ouch.

Perhaps this bug is actually present in the original every_n_samples_stream.cancel() logic in the analog-input-every-n-samples.py file?

Yep, you're right. I was hoping that gRPC's queues would maintain ordering through all of that, but looks like it breaks down somewhere.

#402

so, is the "right" way to ensure you collect all samples to sum up read_response.samps_per_chan_read across iterations, and then using a break in the every_n_samples_stream once you reach the samps_per_chan total?

Well, definitely if you can avoid relying on the done event, you remove that complexity. And I'm not sure of how to fix the ordering issue. Counting samples or polling IsTaskDone could be good alternatives if it works for your setup.

[sjdemartini]

Thanks for opening that issue and for investigating further! 👍

Polling IsTaskDone (hadn't thought of this) or counting both seem like reasonable enough workarounds for our setup, so we'll go with one of those approaches when we need a finite number of samples.