Help needed using CreateDerivedPartitionAsync and interop with python

[Fastjur]

Hi,

I am a bit stuck trying to get something to work. I would greatly appreciate some help!

The scenario is the following:

• I have recorded webcam (and other data) to stores.
• I want to add some derived partitions by sending this data over to python, using NetMQWriter, processing it in python, and sending it back, using NetMQSource.

However:

• The python processing is slower than the reading of the stores (and sending it to python).

This is causing the pipeline created in await CreateDerivedPartitionAsync to close well before all the messages are sent back from python to PSI. This is causing only about half of the data to be processed, and even though python is still happily sending messages over ZeroMQ, the pipeline has already stopped listening to it. How can I ensure that the pipeline stays alive until I have processed all the messages? Is there someway I can keep it running until the NetMQSource has received back a message with OriginatingTime equals the last OriginatingTime sent to the NetMQWriter?

Here are some snippets of the code which I think are important:

The code that opens the store, reads the webcam stream and pipes it to a the EmotionsDetector component

await dataset.CreateDerivedPartitionAsync((pipeline, importer, exporter) =>
{
    IProducer<Shared<Image>> webcamStream = importer.OpenStream<Shared<Image>>(ParticipantPipeline.WEBCAM_STREAM_NAME);

    EmotionsDetector emotionsDetector = new EmotionsDetector(pipeline);
    webcamStream.Do(frame => this.DrawFrame(frame)).PipeTo(emotionsDetector);

    IProducer<DetectedFace> largestFace = emotionsDetector.Out.Select(e => e.OrderByDescending(e2 => e2.Box.Width * e2.Box.Height).ElementAt(0));
    largestFace.Select(f => f.Box).Write(StoreNamesJurriaan.FACE_RECTANGLES_STREAM_NAME, exporter);

    IProducer<Emotions> emotions = largestFace.Select(f => f.Emotions);
    emotions.Select(e => e.Angry).Write(StoreNamesJurriaan.ANGRY_STREAM_NAME, exporter);
    emotions.Select(e => e.Disgust).Write(StoreNamesJurriaan.DISGUST_STREAM_NAME, exporter);
    emotions.Select(e => e.Fear).Write(StoreNamesJurriaan.FEAR_STREAM_NAME, exporter);
    emotions.Select(e => e.Happy).Write(StoreNamesJurriaan.HAPPY_STREAM_NAME, exporter);
    emotions.Select(e => e.Neutral).Write(StoreNamesJurriaan.NEUTRAL_STREAM_NAME, exporter);
    emotions.Select(e => e.Sad).Write(StoreNamesJurriaan.SAD_STREAM_NAME, exporter);
    emotions.Select(e => e.Surprise).Write(StoreNamesJurriaan.SURPRISE_STREAM_NAME, exporter);

    //webcamStream.Do(frame => this.DrawFrame(frame));
},
StoreNamesJurriaan.DETECTED_FACES_STORE_NAME,
false,
StoreNamesJurriaan.DETECTED_FACES_STORE_NAME,
detectedFacesStoreWithVersionPath,
this.asFastAsPossible == true ? ReplayDescriptor.ReplayAll : ReplayDescriptor.ReplayAllRealTime,
null,
false,
this.replayProgress);

The EmotionsDetector component:
I hoped to fix this using the Start() and Stop() functions, but they are never called. I feel like I should be fixing it somewhere in those methods, but I just cannot seem to get it working. I have tried many many versions of this component (e.g. a ConsumerProducer, only a Consumer). But it feels like I am not constructing the component correctly.

public class EmotionsDetector : IConsumer<Shared<Image>>, IProducer<List<DetectedFace>>, ISourceComponent, IDisposable
{
    NetMQWriter<byte[]> frameWriter;
    NetMQSource<dynamic> faceDetectionSource;
    Connector<Shared<Image>> inFrames;

    public EmotionsDetector(Pipeline pipeline)
    {
        this.frameWriter = new NetMQWriter<byte[]>(
            pipeline,
            "webcamFrames",
            "tcp://127.0.0.1:12345",
            MessagePackFormat.Instance
        );

        this.inFrames = pipeline.CreateConnector<Shared<Image>>(nameof(this.inFrames));
        this.In = this.inFrames.In;

        this.inFrames.Out
            .EncodeJpeg(90)
            .Select(frame => frame.Resource.GetBuffer())
            .PipeTo(frameWriter);

        this.faceDetectionSource = new NetMQSource<dynamic>(
            pipeline,
            "faces",
            "tcp://127.0.0.1:12346",
            MessagePackFormat.Instance
        );

        int i = 0;

        this.Out = faceDetectionSource
            .Select((faces, e) => ((IEnumerable<dynamic>)faces)
                .Select(faceMsg =>
                {
                    DetectedFace detectedFace = new DetectedFace
                    {
                        Box = new Rectangle(faceMsg["box"][0], faceMsg["box"][1], faceMsg["box"][2], faceMsg["box"][3]),
                        //Box = new Rectangle(0, 1, 20, 30),

                        Emotions = new Emotions
                        {
                            Angry = faceMsg["emotions"]["angry"],
                            Disgust = faceMsg["emotions"]["disgust"],
                            Fear = faceMsg["emotions"]["fear"],
                            Happy = faceMsg["emotions"]["happy"],
                            Neutral = faceMsg["emotions"]["neutral"],
                            Sad = faceMsg["emotions"]["sad"],
                            Surprise = faceMsg["emotions"]["surprise"]
                        }
                    };
                    Console.WriteLine(i++ + ": EmotionsDetector: " + detectedFace);
                    return detectedFace;
                }).ToList()
            ).Out;
    }

    public Emitter<List<DetectedFace>> Out { get; private set; }

    public Receiver<Shared<Image>> In { get; }

    public void Dispose()
    {
        Console.WriteLine("Disposing EmotionsDetector");
        this.frameWriter?.Dispose();
        this.faceDetectionSource?.Dispose();
    }

    public void Start(Action<DateTime> notifyCompletionTime)
    {
        notifyCompletionTime(DateTime.MaxValue);
    }

    public void Stop(DateTime finalOriginatingTime, Action notifyCompleted)
    {
        faceDetectionSource.Stop(finalOriginatingTime, notifyCompleted);
        this.Dispose();
    }
}

Any help is greatly appreciated!

[AshleyF]

I think that what you’re running into is a common scenario and is pointing out that we need to improve the infrastructure. You’re correct that that pipeline is shutting down while results are still pending from the Python side. What we really need is a mechanism by which to signal the Python side when a \psi stream closes (data store exhausted) and similarly a mechanism for Python to signal when a ZeroMQ stream is complete and make NetMQSource finite. We will open a work item on our end to look into this.

Here is a workaround in the meantime: Firstly, the pipeline is shutting down when the store is complete because the Importer component internally proposes a replay time of the extents of the store. We’ll need to propose an unbounded replay time of our own to allow the pipeline to remain running beyond this:

pipeline.ProposeReplayTime(TimeInterval.LeftBounded(DateTime.UtcNow));

Next, we need to create a finite source that will “hold open” the pipeline. For example, to force the pipeline to continue running for 60 seconds:

Generators.Repeat(pipeline, true, 60, TimeSpan.FromSeconds(1));

This assumes that we know that processing will require less than 60 seconds. Another solution may be to wait for a period of wall-clock time (say, 10 seconds) after the last message received from the Python side:

var lastReceivedWallClockTime = DateTime.MaxValue;
this.faceDetectionSource.Do((_, e) => lastReceivedWallClockTime = DateTime.UtcNow);

IEnumerable<bool> WaitForQuiescence()
{
    while (DateTime.UtcNow - lastReceivedWallClockTime < TimeSpan.FromSeconds(10))
    {
        yield return true;
    }
}

Generators.Sequence(pipeline, WaitForQuiescence(), TimeSpan.FromSeconds(1));

Or we could do what you suggested and wait for matching originating times coming back from Python. This assumes that no messages will be dropped and that Python will return a result for every frame:

var lastSentOriginatingTime = DateTime.MaxValue;
this.inFrames.Do((_, e) => lastSentOriginatingTime = e.OriginatingTime);

var lastReceivedOriginatingTime = DateTime.MinValue;
this.faceDetectionSource.Do((_, e) => lastReceivedOriginatingTime = e.OriginatingTime);

IEnumerable<bool> WaitForPending()
{
    while (lastReceivedOriginatingTime < lastSentOriginatingTime)
    {
        yield return true;
    }
}

Generators.Sequence(pipeline, WaitForPending(), TimeSpan.FromSeconds(1));

Additionally, to explain why the Start(…) and Stop(…) in your EmotionsDetector class are not being called by the \psi runtime:
It looks like your intention is for this class to be a component, and specifically an ISourceComponent, but because it creates no Receivers or Emitters with itself (this) as the owner, it actually will not show up as a component to the \psi runtime. It does appear to have In and Out properties, but these are being set to the inFrames.In (Connector) and faceDetectionSource…Out. So it’s really just a class that happens to be instantiating and wiring together various components, but isn’t itself a component, and thus the runtime does not consider it as such. To make it a proper component, it would need to instantiate at least one Emitter and/or Receiver that it owns.

Hopefully this helps to unblock you!

[Fastjur]

Hi Ashley,

Thank you very much for your detailed answer. I think your proposed solution of keeping the pipeline alive until I have received back the last frame is very well suited for my case!

Regarding the EmotionsDetector class. I had a feeling that I was doing something wrong there indeed. I followed the guide "bridging to python", specifically, the part where a sample component is created: https://github.com/microsoft/psi/wiki/Python-Interop-Walkthrough#bundling-as-a-component, but then altered it to attempt to prevent the pipeline from shutting down. I think the fact that the guide there is for a 'live' scenario is messing things up here, as that way the pipeline will not shut down until told to do so.

Regarding the 'EmotionsDetector' component. I think the proper term for this would be a 'reactive component' if I am not mistaken, as for every input Shared<Image>, I expect to output a List<DetectedFace>. I attempted to make this an ISourceComponent such that I could call notifyCompletionTime with the originating time of the last received input frame. I hoped that this would keep the pipeline alive, but that was not the case (as it was not called). I do not need this to be an ISourceComponent, and it will only be used in the scenario of replaying data. In your experience, does it make sense to create this as a component? If so, what would be the proper interfaces for it? I had a hard time defining how to create the In and Out properties, hence I used the connector as proposed in the python interop guide to pipe the webcam frames to it.

Also, if time permits for you, two follow up questions:

1. I have temporarily worked around this issue right now by simply creating 2 pipelines, one that sends and one that receives, starting and stopping them separately and manually. (I will however construct it back to the createDerivedPartitionAsync method, as that feels much cleaner). What I noticed is that I seem to be leaking memory / replaying the data too quickly. Our data recordings are about an hour-long, halfway through the NetMQWriter simply closes (even though taskmanager is not reporting high memory usage). I attempted to put a DeliveryPolicy in the EncodeJpeg method to throttle, but that does not seem to do much. Where in the original scenario (as described in the first message) could I prevent the data from being read too quickly? Is it possible to signal from the EmotionsDetector class to slow down when it has, let's say, more than 100 frames still pending from python?

2. Very short question, I could not find a way to count the amount of Shared<Image> frames before they're all read and sent out. I would like to know this so I can show progress. Is there any way to count the amount of messages in a stream before running through all of them? I could not find much regarding the Count and LongCount methods. Maybe I have missed some documentation about this?

Thanks again very much for your help. I am at least already able to keep the pipeline alive right now :)

[Fastjur]

Hi Ashley,

I have just answered my own two questions :)

Thanks again for your previous answer!