Add three new ZMQ publishers for TP results / Retrieve TP results

[MichaelHuth]

Publishers added: testpulse results live testpulse results 1s update testpulse results 5s update testpulse results 10s update

The TP results from TP analysis are packed in a JSON in TP_TSAnalysis and published to ZMQ. The time information when the next publishing is due is stored in the TUFXOP.

• PUB_Publish was changed to be threadsafe and added an optional argument to control if the JSON should be released.

close #2157

[MichaelHuth]

@t-b Feedback request for general design improvements.

[t-b]

General approach looks good.

Two comments:

• PUB_AddTPResultEntry: We only want a value object if it has a unit. See the other publishing functions for an example. The publishing docu can be generated with FetchAndParseMessage and OUTPUT_DOCUMENTATION_JSON_DUMP.

• PUB_CheckPublishingTime this should use TSD_ReadVar/TSD_WriteVar.

[t-b]

Git review:

ef3f1d2fd (Util: Add two conversion function for DA/AD unit string return, 2024-08-22)

Good idea!

• [x] s/in clampmode/on clampmode/

• [x] ~~Can we tweak them to also accept a channel typ parameter (XopChannelConstants) so that they also return the DA units and use these functions in GetChanAmpAssignUnit?~~ Use it GetChanAmpAssignUnit. In that way we only have a single place where we define the units depending on the clamp mode.

27dac754d (TP: Add more information that is transferred to the TP analysis thread, 2024-08-22)

Nice!

6368bbb98 (PUB: Add four zeromq publishers for TP data, 2024-08-22)

• [x] s/jst/just/ in the commit message.

• [x] Can you move the ZMQ_FILTER_* constants close to AUTO_TP_FILTER? And their names need to be something like "testpulse:", this is done so that people can easily subscribe to all messages from a certain area. See ZMQ_SUBSCRIBE in https://libzmq.readthedocs.io/en/latest/zmq_setsockopt.html. Maybe that needs a comment though.

• [x] PUB_Publish changes can be their own commit.

• [x] The test TestTPPublishing fails here with:

•runwithOpts(testcase = "TestTPPublishing")
  Start of test "MIES with HardwareBasic"
  Entering test suite "UTF_TestPulseAndTPDuringDAQ.ipf"
  Entering test case "TestPulseAndTPDuringDAQ#TestTPPublishing:ITC"
  Entering reentry "TestPulseAndTPDuringDAQ#TestTPPublishing_REENTRY"
  String mismatch (case sensitive):
str1: 0:0:0> h e a r t b e a t
str2: 0:0:0> t e s t p u l s e   r
: is false. Assertion "CHECK_EQUAL_STR(filter, expectedFilter)" failed in TestPulseAndTPDuringDAQ#TestTPPublishing_REENTRY➔FetchPublishedMessage (UTF_HelperFunctions.ipf, line 1426➔228)
  Encountered "AbortOnValue" Code 10009 in test case "TestPulseAndTPDuringDAQ#TestTPPublishing" (UTF_TestPulseAndTPDuringDAQ.ipf)
  Leaving test case "TestPulseAndTPDuringDAQ#TestTPPublishing:ITC"
  Failed with 2 errors
  Leaving test suite "UTF_TestPulseAndTPDuringDAQ.ipf"
  Test finished with 2 errors
    ▶ Assertion "CHECK_EQUAL_STR(filter, expectedFilter)" failed in TestPulseAndTPDuringDAQ#TestTPPublishing_REENTRY➔FetchPublishedMessage (UTF_HelperFunctions.ipf, line 1426➔228)
    ▶ Encountered "AbortOnValue" Code 10009 in test case "TestPulseAndTPDuringDAQ#TestTPPublishing" (UTF_TestPulseAndTPDuringDAQ.ipf)
  End of test "MIES with HardwareBasic"

The problem is likely that FetchPublishedMessage only works with a single filter. So we might need to teach FetchPublishedMessage to check fo multiple filters and return then multiple messages.

• [x] Can we check the exact numeric values of the published data (with some tolerance) instead of using IsInteger or != NaN? The data is in the labnotebook or TPStorage (and really only these two).

• [x] Test in UTF_ZeroMQPublishing.ipf missing.

• [x] This test is also required to fill in the documentation for PUB_TPResult like for the other PUB_XXX functions.

• [x] Can we name the parameters in TP_PrepareAnalysisDF, so that the code in TP_TSAnalysis is easier to grasp? See NWB_ASYNC_SerializeStruct/NWB_ASYNC_DeserializeStruct.

• [x] GetTPResultAsyncBuffer: numCols should only be calculated if required. The normal case it that the wave exists with the correct size.

• [x] Can we NaN the wave in the upgrade case? Just to be sure that we don't have bogus data lying around?

• [x] The wave upgrade cleanup in GetTPStorage is welcome, but should be its own commit.

• [x] GetTPStorage: The number of layers is different in upgrade compared to creation.

• [x] GetTPStorage: Are the label names in layer 31 to 39 from TP_ANALYSIS_DATA_LABELS? If yes we need a comment to say so.

9c33299d7 (TP: Added two functions that allow to retrieve info about TPs, 2024-08-22)

commit message:

• [x] s/both function/both functions/

• [x] s/meta information/metadata/

• [x] Should be its own commit:

the former function TP_GetStoredTPs was renamed to TP_GetConsecutiveTPsUptoMarker to prevent misleading naming.

• [x] The code to recreate the DAC data should be its own function and not duplicated.

• [x] TP_GetStoredTPsFromCycle/TP_GetStoredTP: I would expect the returned metadata (tpresult, or the waves in tpsResult) to be 1D.

• [x] Why do both functions not require a headstage parameter? If I understand the code in SCOPE_UpdateOscilloscopeData correctly a tpmarker is constant for all headstages of a given TP. Maybe make the test use two headstages with different clamp modes to verify that.

[MichaelHuth]

• [x] Change TP return functions to have an headstage argument and slice also HS. Then return tpADC, tpDAC, tpStorage slice as 1d wave.

• [x] Add test that DimLabels of layers are correctly transferred to ROWs (if Redimension does that itself)

• [x] Use outData wave directly for PUB_TP* argument, scrap tpzmq struct

• [x] add wave getter for outData, use MoveWave in TP_TSAnalysis with named free wave

[t-b]

@MichaelHuth Please rebase

[t-b]

CI is failing.

Review: 92ad402b3 (Util: Add two conversion function for DA/AD unit string return, 2024-08-22) 557177403 (TP: Add more information that is transferred to the TP analysis thread, 2024-08-22)

Looks all good.

31d177113 (Getters: Cleanup of GetTPStorage wave upgrade, 2024-08-25)

Now looking at this change in isolation, I realize that this is not doing the right thing.

Consider a wave upgrade for version 13. The old code set layers 24 to 30. But the new code sets layers 17 and 20 to 30. Overwriting valuable data!

28023219d (PUB: Preparation to add four zeromq publishers for TP data, 2024-08-25) 00941de29 (PUB: Add four publishers to publish TP data, 2024-08-23)

TestTPPublishing_REENTRY:

Instead of

Make/FREE/T filters = {ZMQ_FILTER_TPRESULT_NOW, ZMQ_FILTER_TPRESULT_1S, ZMQ_FILTER_TPRESULT_5S, ZMQ_FILTER_TPRESULT_10S}

you can just write

WAVE filters = DataGenerators#PUB_TPFilters()

00578959c (TP: Rename TP_GetStoredTPs to TP_GetConsecutiveTPsUptoMarker, 2024-08-25)

We nowadays have IsValidHeadstage.

e86dda871 (TP: Added two functions that allow to retrieve info about TPs, 2024-08-22)

Looks good!

[MichaelHuth]

@t-b I diagnosed an issue with the JSON construction for publishing in this PR that requires an update of the JSON XOP to the current version where we fixed the handling of adding values that are large integers. The current effect is that for certain calculated values from the TP analysis the JSON XOP returns an error. That can happen randomly (here about 1 in 100), depending on the noise.

[t-b]

@t-b I diagnosed an issue with the JSON construction for publishing in this PR that requires an update of the JSON XOP to the current version where we fixed the handling of adding values that are large integers.

Let's update the JSON XOP then, 7a3992638 (Update JSON-XOP, 2023-09-25) shows how this is done. We probably would like to update the MacOSX XOP as well.

[t-b]

We return a wave reference wave from the get tp function and I was curious if we are handling recursive wave references correctly, yes we do.

Function Dostuff()

	make/o/WAVE/N=1 data
	make/o/WAVE/N=1 elem1
	make/o/WAVE/N=1 elem2

	data[0] = elem1
	elem1[0] = elem2
	
	print data, elem1, elem2

	print zeromq_test_serializewave(data)
End

gives

•dostuff()
  data[0]= {1.11244e+09}
elem1[0]= {1.11244e+09}
elem2[0]= {0}
  {
    "data": {
        "raw": [
            {
                "data": {
                    "raw": [
                        {
                            "data": {
                                "raw": [
                                    null
                                ]
                            },
                            "date": {
                                "modification": 1725915394
                            },
                            "dimension": {
                                "size": [
                                    1
                                ]
                            },
                            "type": "WAVE_TYPE"
                        }
                    ]
                },
                "date": {
                    "modification": 1725915394
                },
                "dimension": {
                    "size": [
                        1
                    ]
                },
                "type": "WAVE_TYPE"
            }
        ]
    },
    "date": {
        "modification": 1725915394
    },
    "dimension": {
        "size": [
            1
        ]
    },
    "type": "WAVE_TYPE"
}

[t-b]

Fixed conflicts.

[timjarsky]

@campagnola, can you confirm that this PR has the necessary functionality to pass TP data from MIES to ACQ4?

[campagnola]

I will work with Ben to get this tested. Neither of us is very familiar with Igor, though; is the protocol documented somewhere?

[t-b]

To test this I would do:

• Install the user installer on a rig
• Start Igor Pro
• Lock a device
• Start the TP

In a python console:

• Call the function FFI_GetAvailableMessageFilters via ZeroMQ, the JSON message should look like

{
  "version" : 1,
   "CallFunction" : {
     "name" : "FFI_GetAvailableMessageFilters"
   }
}

python example code is at https://github.com/AllenInstitute/ZeroMQ-XOP/tree/main/examples#python-client.

• Create a zeromq sub socket and subscribe to one or more of the topics returned from FFI_GetAvailableMessageFilters. The new ones are:

StrConstant ZMQ_FILTER_TPRESULT_NOW       = "testpulse:results live"
StrConstant ZMQ_FILTER_TPRESULT_1S        = "testpulse:results 1s update"
StrConstant ZMQ_FILTER_TPRESULT_5S        = "testpulse:results 5s update"
StrConstant ZMQ_FILTER_TPRESULT_10S       = "testpulse:results 10s update"

• Look at the sent messages, their layout is defined at the function comment for PUB_TPResult.

We already do have tests that things work like we think it should, so this is more a request to check that it works like you think it should. The idea of the PR is that programs interfacing with MIES (aka ACQ4) don't need to poll the TP values with FFI_ReturnTPValues but get a message with all the data in the requested interval.

Another thing we added is the possibility to query the stored TPs via TP_GetStoredTP.

[t-b]

rebased against latest main.

[t-b]

Fixed conflicts.

[t-b]

Fixed conflicts.

[t-b]

@campagnola Anything I can do to help getting this tested?

[MoxenWolf]

Hey Thomas! Here's an update. We (Luke and I) tested the new messages being published for Test Pulses. We would like for the raw data (binary rather than JSON if possible), including command and recording, to be included in the published test pulse message. Ideally, this would be sent as a additional part of message like this... [id, json_dump, raw_data_new]

Does this sound possible?

[t-b]

We (Luke and I) tested the new messages being published for Test Pulses.

So the new publishing methods worked and TP fetching methods worked?

We would like for the raw data (binary rather than JSON if possible), including command and recording, to be included in the published test pulse message. Ideally, this would be sent as a additional part of message like this... [id, json_dump, raw_data_new]

Doable, for sure.

I need to look into how we could transfer binary data. Would something like BSON work or anything else from the list at 1 be good on your end? But even if we send the data in binary form this would still amount to some data e.g. with ITC hardware and default TP settings you have 2 arrays with 6666 points single precision floats which makes it ~50kB (6666 * 4 * 2) per 33ms aka 1.5MB/s. Just to mention that, as you also have to then manage that data ;)

The calculation assumes that you use the method to get an update for each TP.

[campagnola]

So the new publishing methods worked and TP fetching methods worked?

Almost -- it looked like the TP fetching method returned less data than was expected (hundreds of samples rather than thousands), but I didn't look into this carefully.

I need to look into how we could transfer binary data.

I have used both JSONB and MsgPack in the past (and they work). However, the easiest method I have found is just to send multipart messages in zeromq, in which one part contains regular JSON metadata, and subsequent part(s) contain raw binary data. This is much more efficient because we don't have to parse / copy the raw data at all when we receive it -- we can actually just create an array in python that references the memory zmq has already allocated for the message part.

[ghost]

Hey @t-b . Is there anything else you need from @campagnola or myself in order to complete this request?

[timjarsky]

@campagnola, Would analyzing the TP in MIES and passing the results be simpler? Can you provide an overview of the TP analysis that will be applied to the binary data? This approach assumes a degree of stability in the TP analysis. I'm also motivated to improve the TP analysis in MIES.

[t-b]

Hey @t-b . Is there anything else you need from @campagnola or myself in order to complete this request?

I think I'm good from you guys. I'll be chatting with Tim to see how/if we implement that.

[t-b]

Just to mention: There is already a way to query the TP data from MIES TP_GetStoredTPs.

[campagnola]

Just to mention: There is already a way to query the TP data from MIES TP_GetStoredTPs.

As I mentioned above, when we tried to access TP data in this way, it appeared we would receive far fewer samples of data than expected given the duration of the test pulse * sample rate.

That said, i order to keep TP data communication efficient, I think it would be best to just send the complete TP data with the original notification message coming from MIES. Happy to discuss in a call at some point if you'd like.

[t-b]

Happy to discuss in a call at some point if you'd like.

Yes let's chat about that.

Just to mention: There is already a way to query the TP data from MIES TP_GetStoredTPs.

As I mentioned above, when we tried to access TP data in this way, it appeared we would receive far fewer samples of data than expected given the duration of the test pulse * sample rate.

Looks like I overlooked that. Sorry. I'll look into that, because we will send the same data out anyway.

[t-b]

Conclusion from today's meeting:

• Add new ZeroMQ pub message with the raw AD data attached as another multipart message block
• Only every TP version
• Keep JSON with metadata
• Don't require to store every TP in MIES with that

[t-b]

@MoxenWolf Ready to test. The new filter name is testpulse:results live with data and has three frames. Filter, json message and then the raw AD data (single precision float). As you are only using one HS the raw AD data wave is 1D.

[t-b]

@MoxenWolf I've chatted with Tim about the zeromq log files getting too large. And we decided that it is the easiest fix for now to disable logging. This is now done in this branch if you get c9fdc24c2 (GetZeroMQXOPFlags: Turn off logging, 2025-02-25) which is the latest version of this branch.

[t-b]

@campagnola @ben-at-allen Ready for next round of testing. The JSON now holds all amplifier settings the MIES GUI holds.