Stream iq#582
Conversation
* Can serve antennas_iq and/or bfiq and/or rawacf and/or fitacf (fitacf as DMAP) * Receiver responsible for unpickling SliceData and storing as HDF5 or DMAP * realtime simulator needs SliceData object for testing now, not DMAP data * Messages sent with topic of "file_type/slice_id", for filtering with receivers zmq SUB socket.
* Use one socket for data_write -> realtime comms, added header to message to differentiate between file types * Updated realtime_sim.py to accurately test realtime.py functionality (realtime_server function). * Moved flags for streaming to CLI args for realtime.py, instead of data_write.py (more logical grouping, realtime will always receive data from data_write and will be in charge of streaming it or not).
|
TODO: test which messages are all sent over-the-wire if, for example, a subscriber only wants FITACF data. I suspect that all messages are sent, as zmq specifies that messages are filtered on the subscriber end. If so, we may be able to fix this by splitting the single publisher in the realtime module to one publisher per file type (possibly and per slice ID), then adding an intermediary. The intermediary would be the endpoint for external connections, and the subscription filtering might be able to happen at the intermediary (which is running on the radar computer) as opposed to on the external subscriber. |
* Sends pickled dictionaries over the wire, then depickles and turns into SliceData object subscriber-side. * Created script to subscribe, filter by topic, and write to file.
|
It seems that the Publisher also filters messages for certain transports, including TCP (which is used in this implementation), from zeromq/libzmq#3611. So, no further work to do. |
|
Testing idea: Run |
|
Tested by running |
tjk584
left a comment
tjk584
left a comment
There was a problem hiding this comment.
Looks good, just a few clarifying questions
Also, documentation should be added on this before a new release
| socket = ctx.socket(zmq.SUB) | ||
| socket.setsockopt(zmq.SUBSCRIBE, topic.encode("utf-8")) | ||
| socket.setsockopt(zmq.RCVTIMEO, 100) # time out after 100 ms | ||
| socket.connect(f"tcp://127.0.0.1:{port}") |
There was a problem hiding this comment.
Is this set to localhost because our radar sites are configured with autossh?
There was a problem hiding this comment.
Yup, this could be folded into a command-line option but for now this will work for our purposes.
| parser.add_argument( | ||
| "in_ports", nargs="+", type=int, help="Ports to accept Borealis data from" | ||
| ) | ||
| parser.add_argument("--topic", default="bfiq", help="Topic to subscribe to") |
There was a problem hiding this comment.
What are the possible topics?
There was a problem hiding this comment.
The topics are formatted as [file_type]/[slice_id], e.g. antennas_iq/0, bfiq/2, etc. The subscription filter matches prefixes, so specifying a topic of bf will match bf* in the message topic. In the future, you could extend this script to handle multiple topic subscriptions (multiple calls of socket.setsockopt(zmq.SUBSCRIBE, topic)), and you will receive multiple types of message, or receive all types from just a single slice ID or something.
|
|
||
| # Check if message was intended for realtime, and drop the message if so | ||
| if sender.decode("utf-8") == options.dw_to_rt_identity: | ||
| if receiver.decode("utf-8") == options.rt_to_dw_identity: |
There was a problem hiding this comment.
Not a bug, I don't remember why I changed this, I think it felt more clear at the time to check the recipient
| def _write_correlations( | ||
| def _package_write_stream( |
There was a problem hiding this comment.
I like the consistent function names
2 participants
Allows Borealis to stream any type of data (antennas_iq, bfiq, rawacf, or fitacf) over the wire.
file_type/slice_id(e.g.antennas_iq/0,fitacf/2).SUBsocket to connect, and they have the option to filter messages by type or slice ID. They also need Borealis in order to unpickle theSliceDatapackets being sent over the socket, if accepting antennas_iq/bfiq/rawacf messages.