Help needed using CreateDerivedPartitionAsync and interop with python #162

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);

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();
 }
}