Delay datastream node

[zoeith]

I'm trying to make a node to delay data moving on a stream. The node should receive data, wait a waittime (just using time.sleep() which blocks the thread/process), then send it to the outputs['data'] stream. The issue is that in general we don't know the type of data coming in so the stream needs to be setup when the node is created (or configured).

If I don't mind the delay time blocking everything, this works perfectly (not using nodegroup, etc.). But the delay shouldn't block other nodes doing whatever they are doing, so the Timer node should be created in a separate process using a manager and nodegroup. Or so I thought. Passing from the main thread a dictionary to create the new datastream has landed me in all sorts of problems. Here's an example how it could be instantiated:

    delay_timer = ng.create_node('Timer', datastream=dict(streamtype='analogsignal',dtype=np.uint8,
                                        shape=(-1,),protocol='tcp', transfermode='plaindata',
                                        compression='')

And on my fork, I have my current implementation of the Timer node.

Trying to create a node this way has ended up giving me the following kind of errors:
TypeError: Cannot make proxy to <class 'numpy.uint8'> without proxy server.

Even if I hard code this in the Timer node definition. Later on, when I try to connect the nodes up like so
delay_timer.inputs['data'].connect(data_generating_node.outputs['command'])
then the problem is similar but now it can't encode a stream.

      File "C:\Users\Zoey\Anaconda3\envs\pyacq\lib\site-packages\pyacq-0.2.0.dev0-py3.7.egg\pyacq\core\rpc\serializer.py", line 98, in encode
    raise TypeError("Cannot make proxy to %r without proxy server." % obj)
    TypeError: Cannot make proxy to <pyacq.core.stream.stream.OutputStream object at 0x00000227D0377E08> without proxy server.

In the example, the connection is the other way around -- a remote node outputs to a local viewer. Is this a bug or is there some way around this I'm missing? Thanks for any help.

p.s. I don't mind that this kind of software delay is imprecise.

p.p.s. I made a node to communicate with an Arduino running a simplified gcode. And another node type to generate the gcode. Ultimately, I want the instructions to the arduino to be delayed sometimes, which is why I need this delay node. Those nodes also need to run in a separate process/thread so the serial communications don't block the rest of the system (data acquisition stuff with nidaqmx).

[samuelgarcia]

The issue is that in general we don't know the type of data coming in so the stream needs to be setup when the node is created > (or configured).
This is why a Node have 3 steps:
init() create the node
configure() make some configuration
connect input/ouput
initialize() finalize the configuration after connection

time.sleep()
I am not sure if the time sleep is the good approach because doing this you acculate the input buffer.
And so if te next buffer arrive while the previous is sleeping you will get an overflow.
The main idea is to deal with one buffer as soon as possible.

What I would do is to to create a queue and put the coming buffer in the queue with a timestamp and check if a buffer in the queue that corresponde to delay came be pop and send to output.

For the error
you should put "uint8" instead of np.uint8 when using remote because the dict serialized.

I hope it helps.