<|start_filename|>common/src/main/java/io/pravega/example/tensorflow/FaceRecognizer.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.tensorflow;


import io.pravega.example.common.Embedding;
import io.pravega.example.common.VideoFrame;
import org.apache.commons.io.IOUtils;
import org.bytedeco.opencv.opencv_core.*;
import org.bytedeco.opencv.opencv_objdetect.CascadeClassifier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tensorflow.*;
import org.tensorflow.framework.ConfigProto;
import org.tensorflow.framework.GPUOptions;

import java.io.*;
import java.net.URL;
import java.nio.FloatBuffer;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.*;

import static org.bytedeco.opencv.global.opencv_core.*;
import static org.bytedeco.opencv.global.opencv_imgcodecs.*;
import static org.bytedeco.opencv.global.opencv_imgproc.*;
import static org.bytedeco.opencv.global.opencv_objdetect.CASCADE_SCALE_IMAGE;

/**
 * Class to detect and recognize faces using pre-trained models with TensorFlow Java API.
 */
public class FaceRecognizer implements Serializable, Closeable {
    // Params used for image processing
    private static final int IMAGE_DIMENSION = 160;
    private static final float SCALE = 255f;
    private static final String JPEG_BYTES_PLACEHOLDER_NAME = "image";

    private static final Logger log = LoggerFactory.getLogger(FaceRecognizer.class);

    private final Session session;
    private final Output<Float> imagePreprocessingOutput;
    private final ImageUtil imageUtil;

    private final File file;

    public FaceRecognizer() throws IOException {
        log.info("FaceRecognizer: initializing TensorFlow");
        final long t0 = System.currentTimeMillis();
        InputStream graphFile = FaceRecognizer.class.getResourceAsStream("/facenet.pb");       // Pre-trained model

        System.setProperty("org.bytedeco.javacpp.maxphysicalbytes", "0");
        System.setProperty("org.bytedeco.javacpp.maxbytes", "0");

        final ConfigProto config = ConfigProto.newBuilder()
                .setGpuOptions(GPUOptions.newBuilder()
                        .setAllowGrowth(true)
                        .setPerProcessGpuMemoryFraction(1.0/16.0)
                        .build()
                ).build();

        byte[] GRAPH_DEF = IOUtil.readAllBytesOrExit(graphFile);
        Graph graph = new Graph();
        graph.importGraphDef(GRAPH_DEF);
        session = new Session(graph, config.toByteArray());
        GraphBuilder graphBuilder = new GraphBuilder(graph);

        // Pipeline for JPEG decoding
        imagePreprocessingOutput =
                graphBuilder.div( // Divide each pixels with the MEAN
                        graphBuilder.resizeBilinear( // Resize using bilinear interpolation
                                graphBuilder.expandDims( // Increase the output tensors dimension
                                        graphBuilder.decodeJpeg(
                                                graph.opBuilder("Placeholder", JPEG_BYTES_PLACEHOLDER_NAME)
                                                        .setAttr("dtype", DataType.STRING)
                                                        .build().output(0), 3),
                                        graphBuilder.constant("make_batch", 0)),
                                graphBuilder.constant("size", new int[]{IMAGE_DIMENSION, IMAGE_DIMENSION})),
                        graphBuilder.constant("scale", SCALE));
        imageUtil = new ImageUtil();
        log.info("FaceRecognizer: done initializing TensorFlow; duration = {} ms", System.currentTimeMillis() - t0);


        // import face detection model
        String resource = "/haarcascade_frontalface_alt.xml";
        URL classifier = getClass().getResource(resource);

        // Accesses the jar file to get resources
        if (classifier.getProtocol().equals("jar")) {
            file = File.createTempFile("haarcascase_frontalface_alt", ".tmp");
            try(InputStream input = FaceRecognizer.class.getResourceAsStream(resource)
            ){
                try (OutputStream out = new FileOutputStream(file)) {

                    int read;
                    byte[] bytes = new byte[1024];

                    while ((read = input.read(bytes)) != -1) {
                        out.write(bytes, 0, read);
                    }
                }
                file.deleteOnExit();
            }
        } else {
            //this will probably work in your IDE, but not from a JAR
            file = new File(classifier.getFile());
            file.deleteOnExit();
        }

        if (!file.exists()) {
            throw new RuntimeException("Error: File " + file + " not found!");
        }
    }

    public void warmup() throws Exception {
        log.info("warmup: BEGIN");
        final long t0 = System.currentTimeMillis();
        InputStream image = FaceRecognizer.class.getResourceAsStream("/placeholder-face.jpeg");
        byte[] data = IOUtils.toByteArray(image);
        locateFaces(data);
        log.info("warmup: END; duration={} ms", System.currentTimeMillis() - t0);
    }

    @Override
    public void close() {
        session.close();
    }

    /**
     *
     * @param badgeId identifier for the face
     * @param faceLocation location of the face
     * @return a label representing location and identifier for face
     */
    public Recognition getLabel(String badgeId, BoundingBox faceLocation) {
        return new Recognition(1, badgeId, (float) 1,
                new BoxPosition((float) (faceLocation.getX()),
                        (float) (faceLocation.getY()),
                        (float) (faceLocation.getWidth()),
                        (float) (faceLocation.getHeight())));
    }

    /**
     * Extract the embeddings for the current face
     *
     * @param face in JPG format used to extract the embeddings
     * @return embeddings in a float array
     */
    public float[] embeddFace(byte[] face) {
        return executeGraph(face);
    }

    /**
     * Runs pre-trained facenet model to extract the embeddings from the image
     *
     * @param jpegBytes byte array to run the facenet model on
     * @return embeddings extracted by running the pre-trained model
     */
    private float[] executeGraph(byte[] jpegBytes) {
        // Preprocess image (decode JPEG and resize)
        try (final Tensor<?> jpegTensor = Tensor.create(jpegBytes)) {
            final List<Tensor<?>> imagePreprocessorOutputs = session
                    .runner()
                    .feed(JPEG_BYTES_PLACEHOLDER_NAME, jpegTensor)
                    .fetch(imagePreprocessingOutput.op().name())
                    .run();
            assert imagePreprocessorOutputs.size() == 1;

            try (final Tensor<Float> preprocessedInputTensor = imagePreprocessorOutputs.get(0).expect(Float.class)) {
                try (final Tensor<?> preprocessedPhaseTrainTensor = Tensor.create(false)
                     ) {
                    final List<Tensor<?>> detectorOutputs = session
                            .runner()
                            .feed("input", preprocessedInputTensor)
                            .feed("phase_train", preprocessedPhaseTrainTensor)
                            .fetch("embeddings")
                            .run();
                    assert detectorOutputs.size() == 1;
                    try (final Tensor<Float> resultTensor = detectorOutputs.get(0).expect(Float.class)) {
                        final float[] outputTensor = new float[(int) resultTensor.shape()[1]];
                        final FloatBuffer floatBuffer = FloatBuffer.wrap(outputTensor);
                        resultTensor.writeTo(floatBuffer);
                        return outputTensor;
                    }
                }
            }
        }
    }

    /**
     * @param cropBox  The location in the image to crop
     * @param faceData The data of the image in uncompressed bitmap format
     * @return jpeg format data of the image isolated to the cropped area
     */
    public byte[] cropFace(BoundingBox cropBox, Mat faceData) {
        Rect cropArea = new Rect((int) cropBox.getX(), (int) cropBox.getY(), (int) cropBox.getWidth(), (int) cropBox.getHeight());
        byte[] outData = new byte[0];
        if (0 <= cropBox.getX()
                && 0 <= cropBox.getWidth()
                && cropBox.getX() + cropBox.getWidth() <= faceData.cols()
                && 0 <= cropBox.getY()
                && 0 <= cropBox.getHeight()
                && cropBox.getY() + cropBox.getHeight() <= faceData.rows()) {

            // box within the image plane
            Mat croppedImage = new Mat(faceData, cropArea);
            outData = new byte[(int) (croppedImage.total() * croppedImage.elemSize())];
            imencode(".jpg", croppedImage, outData);
        }

        return outData;
    }

    /**
     * @param frameData data that represents the jpeg image with faces
     * @return location of the faces
     * @throws Exception when face is not scanned properly
     */
    public List<BoundingBox> locateFaces(byte[] frameData) throws Exception {
        try (CascadeClassifier faceCascade = new CascadeClassifier()
        ) {

            List<BoundingBox> recognizedBoxes;
            try (RectVector faces = new RectVector()) {

                Mat imageMat;
                try (Mat rawMat = new Mat(frameData)) {
                    imageMat = imdecode(rawMat, IMREAD_UNCHANGED);
                }
                CvArr grayImage;
                try (CvArr inputImage = new IplImage(imageMat)) {

                    //converting image to grayscale
                    grayImage = cvCreateImage(cvGetSize(inputImage), 8, 1);


                    cvCvtColor(inputImage, grayImage, COLOR_BGR2GRAY); // Convert image to grayscale
                }
                cvEqualizeHist(grayImage, grayImage);

                String classifierPath = file.getAbsolutePath();
                faceCascade.load(classifierPath);
//            log.info("facial detection model load: " + modelLoaded);

                int absoluteFaceSize = 0;
                int height = grayImage.arrayHeight();
                if (Math.round(height * 0.2f) > 0) {
                    absoluteFaceSize = Math.round(height * 0.2f);
                }

                recognizedBoxes = new ArrayList<>();

                try (Size faceSize = new Size(absoluteFaceSize, absoluteFaceSize);
                     Size emptySize = new Size()) {

                    // Identify location of the faces
                    faceCascade.detectMultiScale(cvarrToMat(grayImage), faces, 1.1, 2, CASCADE_SCALE_IMAGE, faceSize, emptySize);


                    for (int i = 0; i < faces.size(); i++) {
                        try (
                                Rect currRect = faces.get(i)
                        ) {

                            double boxX = Math.max(currRect.x() - 10, 0);
                            double boxY = Math.max(currRect.y() - 10, 0);
                            double boxWidth = Math.min(currRect.width() + 20, imageMat.arrayWidth() - boxX);
                            double boxHeight = Math.min(currRect.height() + 20, imageMat.arrayHeight() - boxY);
                            double boxConfidence = -1.0;
                            double[] boxClasses = new double[1];

                            BoundingBox currentBox = new BoundingBox(boxX, boxY, boxWidth, boxHeight, boxConfidence, boxClasses);

                            recognizedBoxes.add(currentBox);
                        }
                    }

                    log.info("recognized faces are {}", recognizedBoxes);
                }
            }

            return recognizedBoxes;
        } catch (Exception e) {
            throw new Exception(e);
        }
    }
}


<|start_filename|>common/src/main/java/io/pravega/example/tensorflow/ImageUtil.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.tensorflow;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.imageio.IIOImage;
import javax.imageio.ImageIO;
import javax.imageio.ImageWriteParam;
import javax.imageio.ImageWriter;
import javax.imageio.stream.MemoryCacheImageOutputStream;
import java.awt.*;
import java.awt.image.BufferedImage;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.util.List;

import static java.awt.image.BufferedImage.TYPE_3BYTE_BGR;

/**
 * Util class for image processing.
 */
public class ImageUtil {
    private final static Logger LOGGER = LoggerFactory.getLogger(ImageUtil.class);

    /**
     * Label image with classes and predictions given by the ThensorFLow
     *
     * @param image        buffered image to label
     * @param recognitions list of recognized objects
     * @return JPEG image in a byte array
     */
    public byte[] labelImage(final byte[] image, final List<Recognition> recognitions) {
        LOGGER.info("labelImage: BEGIN");
        byte[] bytes;
        BufferedImage bufferedImage = createImageFromBytes(image);
        float scaleX = (float) bufferedImage.getWidth() / (float) 416;
        float scaleY = (float) bufferedImage.getHeight() / (float) 416;
        Graphics2D graphics = (Graphics2D) bufferedImage.getGraphics();
        graphics.setColor(Color.green);

        for (Recognition recognition : recognitions) {
            BoxPosition box = recognition.getScaledLocation(scaleX, scaleY);
            //set font
            Font myFont = new Font("Courier New", 1, 17);
            graphics.setFont(myFont);
            //draw text
            graphics.drawString(recognition.getTitle() + " " + recognition.getConfidence(), box.getLeft(), box.getTop() - 7);
            // draw bounding box
            graphics.drawRect(box.getLeftInt(), box.getTopInt(), box.getWidthInt(), box.getHeightInt());
        }

        try {
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            ImageIO.write(bufferedImage, "jpg", baos);
            bytes = baos.toByteArray();
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            graphics.dispose();
        }


        return bytes;
    }

    public byte[] labelFace(final byte[] image, final Recognition recognition) {
        byte[] bytes = null;
        BufferedImage bufferedImage = createImageFromBytes(image);
        Graphics2D graphics = (Graphics2D) bufferedImage.getGraphics();
        graphics.setColor(Color.green);

        BoxPosition box = recognition.getLocation();
        //set font
        Font myFont = new Font("Courier New", 1, 14);
        graphics.setFont(myFont);
        //draw text
        graphics.drawString(recognition.getTitle(), box.getLeft(), box.getTop() - 7);
        // draw bounding box
        graphics.drawRect(box.getLeftInt(), box.getTopInt(), box.getWidthInt(), box.getHeightInt());

        try
        {
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            ImageIO.write(bufferedImage, "jpg", baos);
            bytes = baos.toByteArray();
        }
        catch (IOException e)
        {
            throw  new RuntimeException(e);
        }


        return bytes;
    }

    public byte[] labelFace(final byte[] image, final Recognition recognition, boolean passed) {
        byte[] bytes = null;
        BufferedImage bufferedImage = createImageFromBytes(image);
        Graphics2D graphics = (Graphics2D) bufferedImage.getGraphics();
        if(passed) {
            graphics.setColor(Color.green);
        } else {
            graphics.setColor(Color.red);
        }

        BoxPosition box = recognition.getLocation();
        //set font
        Font myFont = new Font("Courier New", 1, 14);
        graphics.setFont(myFont);
        //draw text

        System.out.println(recognition.getTitle());
        graphics.drawString(recognition.getTitle(), box.getLeft(), box.getTop() - 7);
        // draw bounding box
        System.out.println("Recognition location X: " + box.getLeftInt() + ", Height:" + box.getTopInt());
        graphics.drawRect(box.getLeftInt(),box.getTopInt(), box.getWidthInt(), box.getHeightInt());

        graphics.dispose();

        try {
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            ImageIO.write(bufferedImage, "jpg", baos);
            bytes = baos.toByteArray();
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            graphics.dispose();
        }

        LOGGER.info("labelImage: END");
        return bytes;
    }

    /**
     * @return JPEG image in a byte array
     */
    public byte[] createBlankJpeg(int width, int height) {
        try {
            BufferedImage outImage = new BufferedImage(width, height, TYPE_3BYTE_BGR);
            ImageWriter writer = ImageIO.getImageWritersByFormatName("jpg").next();
            ImageWriteParam writeParam = writer.getDefaultWriteParam();
            ByteArrayOutputStream outStream = new ByteArrayOutputStream();
            writer.setOutput(new MemoryCacheImageOutputStream(outStream));
            writer.write(null, new IIOImage(outImage, null, null), writeParam);
            return outStream.toByteArray();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * Saves image into target file name
     *
     * @param image  to save
     * @param target file name to save image
     * @return location of the saved image
     */
    public String saveImage(final BufferedImage image, final String target) {
        try {
            ImageIO.write(image, "jpg", new File(target));
            return target;
        } catch (IOException ex) {
            LOGGER.error("Unagle to save image {}!", target);
            throw new RuntimeException(ex);
        }
    }

    private BufferedImage createImageFromBytes(final byte[] imageData) {
        ByteArrayInputStream bais = new ByteArrayInputStream(imageData);
        try {
            return ImageIO.read(bais);
        } catch (IOException ex) {
            throw new RuntimeException();
        }
    }
}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/OrderedParallelOperator.java<|end_filename|>
package io.pravega.example.videoprocessor;

import com.google.common.base.Preconditions;

/**
 * This is currently unused as time windows appear to work just as well.
 * See OrderedVideoFrameCoProcessFunction.
 * See also MultiVideoGridExperimentalJob.java.
 */
public class OrderedParallelOperator {
    /**
     * Operator 0 expects to receive streams 0 to 1.
     * Operator 1 expects to receive streams 0 to 2.
     *
     * @param currentIndex The first index for all streams is 0.
     * @param operator The number of this operator (0 to parallelism-2).
     * @param parallelism The number of streams.
     * @return The index of the next element that this operator will emit.
     */
    static long getNextIndexToEmit(final long currentIndex, final int operator, final int parallelism) {
        Preconditions.checkArgument(0 <= currentIndex);
        Preconditions.checkArgument(0 <= operator);
        Preconditions.checkArgument( 2 <= parallelism);
        Preconditions.checkArgument(operator < parallelism - 1);
        Preconditions.checkArgument(currentIndex % parallelism <= operator + 1);
        long index = currentIndex + 1;
        final long stream = index % parallelism;
        if (stream <= operator + 1) {
            return index;
        } else {
            return (index / parallelism + 1) * parallelism;
        }
    }
}

<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/ChunkedVideoFrameSerializer.java<|end_filename|>
/*
 * Copyright (c) Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.videoprocessor;

import io.pravega.client.stream.Serializer;
import io.pravega.example.common.ChunkedVideoFrame;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;

/**
 * Pravega Serializer for ChunkedVideoFrame.
 * Uses JSON.
 */
public class ChunkedVideoFrameSerializer implements Serializer<ChunkedVideoFrame>, Serializable {
    private final ObjectMapper mapper = new ObjectMapper();

    @Override
    public ChunkedVideoFrame deserialize(ByteBuffer serializedValue) {
        final ByteArrayInputStream bin = new ByteArrayInputStream(
                serializedValue.array(),
                serializedValue.position(),
                serializedValue.remaining());
        try {
            return mapper.readValue(bin, ChunkedVideoFrame.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public ByteBuffer serialize(ChunkedVideoFrame element) {
        try {
            return ByteBuffer.wrap(mapper.writeValueAsBytes(element));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}


<|start_filename|>common/src/main/java/io/pravega/example/tensorflow/TFObjectDetector.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.tensorflow;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tensorflow.DataType;
import org.tensorflow.Graph;
import org.tensorflow.Output;
import org.tensorflow.Session;
import org.tensorflow.Tensor;
import org.tensorflow.framework.ConfigProto;
import org.tensorflow.framework.GPUOptions;

import java.io.Closeable;
import java.io.InputStream;
import java.io.Serializable;
import java.nio.FloatBuffer;
import java.util.List;

/**
 * ObjectDetector class to detect objects using pre-trained models with TensorFlow Java API.
 */
public class TFObjectDetector implements Serializable, Closeable {
    private static final Logger log = LoggerFactory.getLogger(TFObjectDetector.class);

    // Params used for image processing
    private static final int IMAGE_DIMENSION = 416;
    private static final float SCALE = 255f;
    private static final String JPEG_BYTES_PLACEHOLDER_NAME = "image";

    private final Session session;
    private final Output<Float> imagePreprocessingOutput;
    private final List<String> LABEL_DEF;
    private final YOLOClassifier yoloClassifier;
    private final ImageUtil imageUtil;

    public TFObjectDetector() {
        log.info("TFObjectDetector: initializing TensorFlow");
        final long t0 = System.currentTimeMillis();
        final ConfigProto config = ConfigProto.newBuilder()
                .setGpuOptions(GPUOptions.newBuilder()
                        .setAllowGrowth(true)
                        .setPerProcessGpuMemoryFraction(1.0/16.0)
                        .build()
                ).build();

        InputStream graphFile = getClass().getResourceAsStream("/tiny-yolo-voc.pb");       // The model
        InputStream labelFile = getClass().getResourceAsStream("/yolo-voc-labels.txt");    // labels for classes used to train model

        byte[] GRAPH_DEF = IOUtil.readAllBytesOrExit(graphFile);
        LABEL_DEF = IOUtil.readAllLinesOrExit(labelFile);
        Graph graph = new Graph();
        graph.importGraphDef(GRAPH_DEF);
        session = new Session(graph, config.toByteArray());
        GraphBuilder graphBuilder = new GraphBuilder(graph);

        imagePreprocessingOutput =
                graphBuilder.div( // Divide each pixels with the MEAN
                        graphBuilder.resizeBilinear( // Resize using bilinear interpolation
                                graphBuilder.expandDims( // Increase the output tensors dimension
                                        graphBuilder.decodeJpeg(
                                                graph.opBuilder("Placeholder", JPEG_BYTES_PLACEHOLDER_NAME)
                                                        .setAttr("dtype", DataType.STRING)
                                                        .build().output(0), 3),
                                        graphBuilder.constant("make_batch", 0)),
                                graphBuilder.constant("size", new int[]{IMAGE_DIMENSION, IMAGE_DIMENSION})),
                        graphBuilder.constant("scale", SCALE));

        yoloClassifier = new YOLOClassifier();
        imageUtil = new ImageUtil();
        log.info("TFObjectDetector: done initializing TensorFlow; duration = {} ms", System.currentTimeMillis() - t0);
    }

    public void warmup() {
        log.info("warmup: BEGIN");
        final long t0 = System.currentTimeMillis();
        detect(imageUtil.createBlankJpeg(1280, 720));
        log.info("warmup: END; duration={} ms", System.currentTimeMillis() - t0);
    }

    @Override
    public void close() {
        session.close();
    }

    /**
     * Detect objects on the given image
     *
     * @param image the location of the image
     * @return output image with objects detected
     */
    public DetectionResult detect(byte[] image) {
        log.info("detect: BEGIN");
        final long t0 = System.currentTimeMillis();
        final float[] tensorFlowOutput = executeYOLOGraph(image);
        final List<Recognition> recognitions = yoloClassifier.classifyImage(tensorFlowOutput, LABEL_DEF);
        final byte[] jpegBytes = imageUtil.labelImage(image, recognitions);
        final long dt = System.currentTimeMillis() - t0;
        final double fps = 1000.0/dt;
        log.info("detect: elapsed time for single frame: {} ms ({} fps)", dt, fps);
        log.info("detect: END");
        return new DetectionResult(recognitions, jpegBytes);
    }

    /**
     * Executes graph on the given preprocessed image
     *
     * @param jpegBytes JPEG image
     * @return output tensor returned by tensorFlow
     */
    private float[] executeYOLOGraph(byte[] jpegBytes) {
        // Preprocess image (decode JPEG and resize)
        log.info("executeYOLOGraph: BEGIN");
        try (final Tensor<?> jpegTensor = Tensor.create(jpegBytes)) {
            final List<Tensor<?>> imagePreprocessorOutputs = session
                    .runner()
                    .feed(JPEG_BYTES_PLACEHOLDER_NAME, jpegTensor)
                    .fetch(imagePreprocessingOutput.op().name())
                    .run();
            log.info("executeYOLOGraph: got imagePreprocessorOutputs");
            assert imagePreprocessorOutputs.size() == 1;
            try (final Tensor<Float> preprocessedImageTensor = imagePreprocessorOutputs.get(0).expect(Float.class)) {
                // YOLO object detection
                final List<Tensor<?>> detectorOutputs = session
                        .runner()
                        .feed("input", preprocessedImageTensor)
                        .fetch("output")
                        .run();
                log.info("executeYOLOGraph: got detectorOutputs");
                assert detectorOutputs.size() == 1;
                try (final Tensor<Float> resultTensor = detectorOutputs.get(0).expect(Float.class)) {
                    final float[] outputTensor = new float[yoloClassifier.getOutputSizeByShape(resultTensor)];
                    final FloatBuffer floatBuffer = FloatBuffer.wrap(outputTensor);
                    resultTensor.writeTo(floatBuffer);
                    return outputTensor;
                }
            }
        } finally {
            log.info("executeYOLOGraph: END");
        }
    }

    static public class DetectionResult {
        private final List<Recognition> recognitions;
        private final byte[] jpegBytes;

        public DetectionResult(List<Recognition> recognitions, byte[] jpegBytes) {
            this.recognitions = recognitions;
            this.jpegBytes = jpegBytes;
        }

        public List<Recognition> getRecognitions() {
            return recognitions;
        }

        public byte[] getJpegBytes() {
            return jpegBytes;
        }
    }
}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/MultiVideoGridJob.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.videoprocessor;

import io.pravega.client.stream.StreamCut;
import io.pravega.connectors.flink.FlinkPravegaReader;
import io.pravega.connectors.flink.FlinkPravegaWriter;
import io.pravega.connectors.flink.PravegaWriterMode;
import io.pravega.example.flinkprocessor.AbstractJob;
import io.pravega.example.common.ChunkedVideoFrame;
import io.pravega.example.common.VideoFrame;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Random;
import java.util.stream.Collectors;

import static java.lang.Math.max;

/**
 * A Flink job that reads images from multiple cameras stored in a Pravega stream, combines them
 * into a NxM grid of images (like a security camera monitor), and writes the resulting
 * images to another Pravega stream.
 */
public class MultiVideoGridJob extends AbstractJob {
    private static Logger log = LoggerFactory.getLogger(MultiVideoGridJob.class);

    /**
     * The entry point for Flink applications.
     *
     * @param args Command line arguments
     */
    public static void main(String... args) {
        VideoAppConfiguration config = new VideoAppConfiguration(args);
        log.info("config: {}", config);
        MultiVideoGridJob job = new MultiVideoGridJob(config);
        job.run();
    }

    public MultiVideoGridJob(VideoAppConfiguration config) {
        super(config);
    }

    @Override
    public VideoAppConfiguration getConfig() {
        return (VideoAppConfiguration) super.getConfig();
    }

    public void run() {
        try {
            final long periodMs = (long) (1000.0 / getConfig().getFramesPerSec());
            final int numColumns = getConfig().getParams().getInt("numColumns", 2);
            final int numRows = getConfig().getParams().getInt("numRows", 2);
            final int camerasPerMonitor = getConfig().getParams().getInt("camerasPerMonitor", numColumns * numRows);
            final int imageWidth = getConfig().getImageWidth() / numColumns;
            final int imageHeight = getConfig().getImageHeight() / numRows;
            final long slidingWindowSizeMs = getConfig().getParams().getLong("slidingWindowSizeMs", 10000);
            final int ssrc = new Random().nextInt();
            final String jobName = MultiVideoGridJob.class.getName();

            final StreamExecutionEnvironment env = initializeFlinkStreaming();
            createStream(getConfig().getInputStreamConfig());
            createStream(getConfig().getOutputStreamConfig());

            final StreamCut startStreamCut = resolveStartStreamCut(getConfig().getInputStreamConfig());
            final StreamCut endStreamCut = resolveEndStreamCut(getConfig().getInputStreamConfig());
            log.info("startStreamCut={}", startStreamCut.asText());
            log.info("endStreamCut={}", endStreamCut.asText());

            // Read chunked video frames from Pravega.
            // Operator: input-source
            // Effective parallelism: min of # of segments, getReaderParallelism()
            final FlinkPravegaReader<ChunkedVideoFrame> flinkPravegaReader = FlinkPravegaReader.<ChunkedVideoFrame>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getInputStreamConfig().getStream(), startStreamCut, endStreamCut)
                    .withDeserializationSchema(new ChunkedVideoFrameDeserializationSchema())
                    .build();
            final DataStream<ChunkedVideoFrame> inChunkedVideoFrames = env
                    .addSource(flinkPravegaReader)
                    .setParallelism(getConfig().getReaderParallelism())
                    .uid("input-source")
                    .name("input-source");
            inChunkedVideoFrames.printToErr().uid("input-source-print").name("input-source-print");

            // Assign timestamps and watermarks based on timestamp in each chunk.
            // Operator: assignTimestampsAndWatermarks
            // Effective parallelism: min of # of segments, getReaderParallelism()
            final DataStream<ChunkedVideoFrame> inChunkedVideoFramesWithTimestamps = inChunkedVideoFrames
                    .assignTimestampsAndWatermarks(
                        new BoundedOutOfOrdernessTimestampExtractor<ChunkedVideoFrame>(
                                Time.milliseconds(getConfig().getMaxOutOfOrdernessMs())) {
                            @Override
                            public long extractTimestamp(ChunkedVideoFrame element) {
                                return element.timestamp.getTime();
                            }
                        })
                    .uid("assignTimestampsAndWatermarks")
                    .name("assignTimestampsAndWatermarks");

            // Unchunk (disabled).
            // Operator: ChunkedVideoFrameReassembler
            // Effective parallelism: default parallelism (implicit rebalance before operator)
            final DataStream<VideoFrame> inVideoFrames = inChunkedVideoFramesWithTimestamps
                    .map(VideoFrame::new)
                    .uid("ChunkedVideoFrameReassembler")
                    .name("ChunkedVideoFrameReassembler");

            // For each camera and window, get the most recent frame.
            // This will emit at a maximum rate of the framesPerSec parameter.
            // Operator: lastVideoFramePerCamera
            // Effective parallelism: hash of camera
            final DataStream<VideoFrame> lastVideoFramePerCamera = inVideoFrames
                    .keyBy((KeySelector<VideoFrame, Integer>) value -> value.camera)
                    .window(TumblingEventTimeWindows.of(Time.milliseconds(periodMs)))
                    .maxBy("timestamp")
                    .uid("lastVideoFramePerCamera")
                    .name("lastVideoFramePerCamera");

            // Resize all input images.
            // Operator: ImageResizer
            // Effective parallelism: default parallelism
            lastVideoFramePerCamera.printToErr();
            final DataStream<VideoFrame> resizedVideoFrames = lastVideoFramePerCamera
                    .rebalance()
                    .map(frame -> {
                        ImageResizer resizer = new ImageResizer(imageWidth, imageHeight);
                        frame.data = resizer.resize(frame.data);
                        frame.hash = null;
                        return frame;
                    })
                    .uid("ImageResizer")
                    .name("ImageResizer");
//            resizedVideoFrames.printToErr().uid("resizedVideoFrames-print").name("resizedVideoFrames-print");;

            // Aggregate resized images. This only adds images to a hash map. The actual work is done later to increase parallelism.
            // Operator: ImageAggregator
            // Effective parallelism: hash of monitor
            final DataStream<ImageAggregatorResult> aggResults = resizedVideoFrames
                    .keyBy((KeySelector<VideoFrame, Integer>) value -> getMonitorFromCamera(value.camera, camerasPerMonitor))
                    .window(TumblingEventTimeWindows.of(Time.milliseconds(periodMs)))
                    // TODO: SlidingEventTimeWindows below does not correctly. It stops emitting records after several seconds.
//                    .window(SlidingEventTimeWindows.of(Time.milliseconds(slidingWindowSizeMs), Time.milliseconds(periodMs)))
                    .process(new ProcessWindowFunction<VideoFrame, ImageAggregatorResult, Integer, TimeWindow>() {
                        private ValueState<Integer> frameNumberState;

                        @Override
                        public void open(Configuration parameters) throws Exception {
                            frameNumberState = getRuntimeContext().getState(new ValueStateDescriptor<>("frameNumber", Integer.class));
                        }

                        @Override
                        public void process(Integer monitor, Context context, Iterable<VideoFrame> elements, Collector<ImageAggregatorResult> out) throws Exception {
                            final ImageAggregatorResult result = new ImageAggregatorResult();
                            result.monitor = monitor;
                            result.ssrc = ssrc + monitor;
                            result.frameNumber = Optional.ofNullable(frameNumberState.value()).orElse(0) + 1;
                            result.imageWidth = imageWidth;
                            result.imageHeight = imageHeight;
                            result.numColumns = numColumns;
                            result.numRows = numRows;
                            elements.forEach(element -> {
                                 result.timestamp = new Timestamp(max(result.timestamp.getTime(), element.timestamp.getTime()));
                                 final int position = getPositionInMonitorFromCamera(element.camera, camerasPerMonitor);
                                 result.videoFrames.put(position, element);
                            });
                            frameNumberState.update(result.frameNumber);
                            out.collect(result);
                        }
                    })
                    .uid("ImageAggregator")
                    .name("ImageAggregator");

            // Build output images and encode as JPEG.
            // Effective parallelism: default parallelism
            final DataStream<VideoFrame> videoFrames = aggResults
                    .rebalance() // ensure that successive frames are round robin distributed to different subtasks
                    .map(aggResult -> new VideoFrame(aggResult.encodeVideoFrame()))
                    .uid("encodeVideoFrame")
                    .name("encodeVideoFrame");

            // Ensure ordering.
            // Effective parallelism: hash of monitor
            final DataStream<VideoFrame> outVideoFrames = videoFrames
                    .keyBy((KeySelector<VideoFrame, Integer>) value -> value.camera)
                    .window(TumblingEventTimeWindows.of(Time.milliseconds(periodMs)))
                    .maxBy("timestamp")
                    .uid("outVideoFrames")
                    .name("outVideoFrames");
            outVideoFrames.printToErr().uid("outVideoFrames-print").name("outVideoFrames-print");

            // Split output video frames into chunks of 8 MiB or less.
            // Effective parallelism: hash of monitor
            final DataStream<ChunkedVideoFrame> outChunkedVideoFrames = outVideoFrames
                    .flatMap(new VideoFrameChunker(getConfig().getChunkSizeBytes()))
                    .uid("VideoFrameChunker")
                    .name("VideoFrameChunker");

            // Write chunks to Pravega encoded as JSON.
            // Effective parallelism: hash of monitor
            final FlinkPravegaWriter<ChunkedVideoFrame> flinkPravegaWriter = FlinkPravegaWriter.<ChunkedVideoFrame>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getOutputStreamConfig().getStream())
                    .withSerializationSchema(new ChunkedVideoFrameSerializationSchema())
                    .withEventRouter(frame -> String.format("%d", frame.camera))
                    .withWriterMode(PravegaWriterMode.ATLEAST_ONCE)
                    .build();
            outChunkedVideoFrames
                    .addSink(flinkPravegaWriter)
                    .uid("output-sink")
                    .name("output-sink");

            log.info("Executing {} job", jobName);
            env.execute(jobName);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private static int getMonitorFromCamera(final int camera, final int camerasPerMonitor) {
        return camera / camerasPerMonitor;
    }

    private static int getPositionInMonitorFromCamera(final int camera, final int camerasPerMonitor) {
        return camera % camerasPerMonitor;
    }

    public static class ImageAggregatorResult {
        // ID for this monitor
        public int monitor;
        public int ssrc;
        // Map from position to VideoFrame.
        public Map<Integer, VideoFrame> videoFrames = new HashMap<>();
        // Maximum timestamp from cameras.
        public Timestamp timestamp = new Timestamp(0);
        public int frameNumber;
        // Width of each input image
        public int imageWidth;
        // Height of each input image
        public int imageHeight;
        // Number of columns in grid
        public int numColumns;
        // Number of rows in grid
        public int numRows;

        public ImageAggregatorResult() {
        }

        @Override
        public String toString() {
            return "ImageAggregatorResult{" +
                    "monitor=" + monitor +
                    ", ssrc=" + ssrc +
                    ", videoFrames=" + videoFrames +
                    ", timestamp=" + timestamp +
                    ", frameNumber=" + frameNumber +
                    ", imageWidth=" + imageWidth +
                    ", imageHeight=" + imageHeight +
                    ", numColumns=" + numColumns +
                    ", numRows=" + numRows +
                    '}';
        }

        public VideoFrame encodeVideoFrame() {
            log.info("encodeVideoFrame: BEGIN: frameNumber={}", frameNumber);
            final ImageGridBuilder builder = new ImageGridBuilder(imageWidth, imageHeight, numColumns, numRows);
            videoFrames.forEach((position, frame) -> builder.addImage(position, frame.data));
            final long t0 = System.currentTimeMillis();
            VideoFrame videoFrame = new VideoFrame();
            videoFrame.camera = monitor;
            videoFrame.ssrc = ssrc;
            videoFrame.timestamp = timestamp;
            videoFrame.frameNumber = frameNumber;
            videoFrame.data = builder.getOutputImageBytes("jpg");
            videoFrame.hash = videoFrame.calculateHash();
            videoFrame.tags = new HashMap<>();
            videoFrame.tags.put("cameras", videoFrames.values().stream().map(f -> f.camera).collect(Collectors.toList()).toString());
            videoFrame.tags.put("numCameras", Integer.toString(videoFrames.size()));
            log.info("encodeVideoFrame: END: frameNumber={}, TIME={}", frameNumber, System.currentTimeMillis() - t0);
            log.trace("encodeVideoFrame: videoFrame={}", videoFrame);
            return videoFrame;
        }
    }
}


<|start_filename|>camera-recorder/src/main/java/io/pravega/example/camerarecorder/AppConfiguration.java<|end_filename|>
package io.pravega.example.camerarecorder;

import io.pravega.example.common.PravegaAppConfiguration;

public class AppConfiguration extends PravegaAppConfiguration {
    private final int imageWidth;
    private final int imageHeight;
    private final double framesPerSec;
    private final int cameraDeviceNumber;
    private final int camera;

    public AppConfiguration(String[] args) {
        super(args);
        imageWidth = Integer.parseInt(getEnvVar("IMAGE_WIDTH", "416"));
        imageHeight = Integer.parseInt(getEnvVar("IMAGE_HEIGHT", "416"));
        framesPerSec = Double.parseDouble(getEnvVar("FRAMES_PER_SEC", "1.0"));
        cameraDeviceNumber = Integer.parseInt(getEnvVar("CAMERA_DEVICE_NUMBER", "0"));
        camera = Integer.parseInt(getEnvVar("CAMERA", "3"));
    }

    public int getImageWidth() {
        return imageWidth;
    }

    public int getImageHeight() {
        return imageHeight;
    }

    public double getFramesPerSec() {
        return framesPerSec;
    }

    public int getCameraDeviceNumber() {
        return cameraDeviceNumber;
    }

    public int getCamera() {
        return camera;
    }

    @Override
    public String toString() {
        return "AppConfiguration{" +
                super.toString() +
                ", imageWidth=" + imageWidth +
                ", imageHeight=" + imageHeight +
                ", framesPerSec=" + framesPerSec +
                ", cameraDeviceNumber=" + cameraDeviceNumber +
                ", camera=" + camera +
                '}';
    }
}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/BadgesCoProcessor.java<|end_filename|>
package io.pravega.example.videoprocessor;

import io.pravega.example.common.VideoFrame;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Set;

public class BadgesCoProcessor extends CoProcessFunction<VideoFrame, Set, VideoFrame> {
    private static Logger log = LoggerFactory.getLogger(BadgesCoProcessor.class);

    private ValueState<Set> lastBadges;

    @Override
    public void open(Configuration parameters) throws Exception {
        lastBadges = getRuntimeContext().getState(new ValueStateDescriptor<>("lastBadges", Set.class));
    }

    @Override
    public void processElement1(VideoFrame value, Context ctx, Collector<VideoFrame> out) throws Exception {
        log.info("processElement1 badges:{}", lastBadges.value());
        value.lastBadges = lastBadges.value();
        value.hash = value.calculateHash();

        log.info(value.toString());

        out.collect(value);
    }

    @Override
    public void processElement2(Set value, Context ctx, Collector<VideoFrame> out) throws Exception {
        lastBadges.update(value);
//        log.info("processElement2 badges:{}", lastBadges.value());
    }
}


<|start_filename|>common/src/main/java/io/pravega/example/common/Person.java<|end_filename|>
package io.pravega.example.common;

/*
 *   This holds relevant information about the people in the Embeddings database.
 * */

public class Person {
    public String id;

    public float[] embedding;

    public Person() {
        this("Unknown", new float[1]);
    }

    public Person(String id, float[] embedding) {
        this.id = id;
        this.embedding = embedding;
    }
}


<|start_filename|>camera-recorder/src/test/java/io/pravega/example/camerarecorder/PravegaTests.java<|end_filename|>
/*
 * Copyright (c) Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.camerarecorder;

import com.fasterxml.jackson.databind.ObjectMapper;
import io.pravega.client.ClientConfig;
import io.pravega.client.EventStreamClientFactory;
import io.pravega.client.admin.ReaderGroupManager;
import io.pravega.client.admin.StreamInfo;
import io.pravega.client.admin.StreamManager;
import io.pravega.client.stream.EventRead;
import io.pravega.client.stream.EventStreamReader;
import io.pravega.client.stream.ReaderConfig;
import io.pravega.client.stream.ReaderGroupConfig;
import io.pravega.client.stream.Stream;
import io.pravega.client.stream.StreamConfiguration;
import io.pravega.client.stream.StreamCut;
import io.pravega.client.stream.impl.ByteBufferSerializer;
import io.pravega.example.common.ChunkedVideoFrame;
import io.pravega.example.common.VideoFrame;
import org.bytedeco.javacpp.BytePointer;
import org.bytedeco.javacv.CanvasFrame;
import org.bytedeco.javacv.Frame;
import org.bytedeco.javacv.FrameGrabber;
import org.bytedeco.javacv.OpenCVFrameConverter;
import org.bytedeco.javacv.OpenCVFrameGrabber;
import org.bytedeco.opencv.global.opencv_imgcodecs;
import org.bytedeco.opencv.opencv_core.Mat;
import org.bytedeco.opencv.opencv_videoio.VideoCapture;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.sql.Timestamp;
import java.time.Instant;
import java.util.UUID;

public class PravegaTests {
    private static Logger log = LoggerFactory.getLogger(PravegaTests.class);

    final private ClientConfig clientConfig;
    final private String scope = "examples";
    private String streamName = "video1";
//    final private String controllerURIStr = "tcp://10.246.21.231:9090";
    final private String controllerURIStr = "tcp://nautilus-pravega-controller.vpn4-demo.nautilus-lab-uranium.com:9090";

    public PravegaTests() throws Exception {
        URI controllerURI = URI.create(controllerURIStr);
        clientConfig = ClientConfig.builder().controllerURI(controllerURI).build();
    }

    private void createStream() {
        try (StreamManager streamManager = StreamManager.create(clientConfig)) {
            streamManager.createStream(scope, streamName, StreamConfiguration.builder().build());
        }
    }

    @Test
    @Ignore
    public void Test2() throws Exception {
        Mat image = opencv_imgcodecs.imread("c:\\temp\\capture.png");
        log.info("image={}", image);
        BytePointer png = new BytePointer();
        opencv_imgcodecs.imencode(".png", image,  png);
        log.info("png={}", png);
        byte[] pngByteArray = png.getStringBytes();
        Files.write((new File("c:\\temp\\capture2.png")).toPath(), pngByteArray);
    }

    @Test
    @Ignore
    public void Test3() throws Exception {
        VideoCapture cap = new VideoCapture();
        cap.open(1);
        Mat image = new Mat();
        log.info("Reading image from camera");
        cap.read(image);
        log.info("image={}", image);
        opencv_imgcodecs.imwrite("c:\\temp\\capture3.png", image);
    }

    @Test
    @Ignore
    public void Test4() throws Exception {
        final int WEBCAM_DEVICE_INDEX = 1;
        final int captureWidth = 640;
        final int captureHeight = 480;

        // The available FrameGrabber classes include OpenCVFrameGrabber (opencv_videoio),
        // DC1394FrameGrabber, FlyCapture2FrameGrabber, OpenKinectFrameGrabber,
        // PS3EyeFrameGrabber, VideoInputFrameGrabber, and FFmpegFrameGrabber.
        final FrameGrabber grabber = new OpenCVFrameGrabber(WEBCAM_DEVICE_INDEX);
//        final FFmpegFrameGrabber grabber = new FFmpegFrameGrabber(WEBCAM_DEVICE_INDEX);
        grabber.setImageWidth(captureWidth);
        grabber.setImageHeight(captureHeight);
        grabber.setFrameRate(15.0);
//        log.info("getFrameRate={}", grabber.getFrameRate());

        grabber.start();

        log.info("actual frame rate={}", grabber.getFrameRate());

        OpenCVFrameConverter.ToMat converterToMat = new OpenCVFrameConverter.ToMat();

        final CanvasFrame cFrame = new CanvasFrame("Capture Preview", CanvasFrame.getDefaultGamma() / grabber.getGamma());
        Frame capturedFrame;
        int frameNumber = 0;
        while ((capturedFrame = grabber.grab()) != null)
        {
            long timestamp = System.currentTimeMillis();
            log.info("frameNumber={}, timestamp={}, capturedFrame={}", frameNumber, timestamp, capturedFrame);

            Mat mat = converterToMat.convert(capturedFrame);
            BytePointer pngBytePointer = new BytePointer();
            opencv_imgcodecs.imencode(".png", mat,  pngBytePointer);
            log.info("pngBytePointer={}", pngBytePointer);
            byte[] pngByteArray = pngBytePointer.getStringBytes();
            Files.write((new File(String.format("c:\\temp\\capture4-%05d.png", frameNumber))).toPath(), pngByteArray);

            if (cFrame.isVisible())
            {
                // Show our frame in the preview
                cFrame.showImage(capturedFrame);
            }

            Thread.sleep(500);
            frameNumber++;
        }
    }

    @Test
    @Ignore
    public void TestCameraToScreen6() throws Exception {
        final int WEBCAM_DEVICE_INDEX = 0;
        final int captureWidth = 320;
        final int captureHeight = 240;
//        log.info("getDeviceDescriptions={}", Arrays.toString(VideoInputFrameGrabber.getDeviceDescriptions()));
        log.info("creating grabber");
//        final FrameGrabber grabber = new VideoInputFrameGrabber(WEBCAM_DEVICE_INDEX);
        final FrameGrabber grabber = new OpenCVFrameGrabber(WEBCAM_DEVICE_INDEX);
        grabber.setImageWidth(captureWidth);
        grabber.setImageHeight(captureHeight);
        grabber.setFrameRate(15.0);
        log.info("starting grabber");
        grabber.start();
        log.info("actual frame rate={}", grabber.getFrameRate());
        log.info("canvas gamma={}, grabber gamma={}", CanvasFrame.getDefaultGamma(), grabber.getGamma());
        final CanvasFrame cFrame = new CanvasFrame("Capture Preview", CanvasFrame.getDefaultGamma() / grabber.getGamma());
        Frame capturedFrame;
        int frameNumber = 0;
        while ((capturedFrame = grabber.grab()) != null) {
            long timestamp = System.currentTimeMillis();
            log.info("frameNumber={}, timestamp={}, capturedFrame={}", frameNumber, timestamp, capturedFrame);
            if (cFrame.isVisible()) {
                // Show our frame in the preview
                cFrame.showImage(capturedFrame);
            }
            Thread.sleep(100);
            frameNumber++;
        }
    }

    @Test
    @Ignore
    public void TestPravegaToScreen7() throws Exception {
        streamName = "grid1";
        StreamInfo streamInfo;
        try (StreamManager streamManager = StreamManager.create(clientConfig)) {
            streamInfo = streamManager.getStreamInfo(scope, streamName);
        }
        final long timeoutMs = 1000;
        ObjectMapper mapper = new ObjectMapper();
        final CanvasFrame cFrame = new CanvasFrame("Playback from Pravega");
        OpenCVFrameConverter.ToMat converter = new OpenCVFrameConverter.ToMat();
        final String readerGroup = UUID.randomUUID().toString().replace("-", "");
        final ReaderGroupConfig readerGroupConfig = ReaderGroupConfig.builder()
                .stream(Stream.of(scope, streamName), streamInfo.getTailStreamCut(), StreamCut.UNBOUNDED)
                .build();
        try (ReaderGroupManager readerGroupManager = ReaderGroupManager.withScope(scope, clientConfig)) {
            readerGroupManager.createReaderGroup(readerGroup, readerGroupConfig);
        }
        try (EventStreamClientFactory clientFactory = EventStreamClientFactory.withScope(scope, clientConfig);
             EventStreamReader<ByteBuffer> reader = clientFactory.createReader("reader",
                     readerGroup,
                     new ByteBufferSerializer(),
                     ReaderConfig.builder().build())) {
            for (;;) {
                EventRead<ByteBuffer> event = reader.readNextEvent(timeoutMs);
                if (event.getEvent() != null) {
                    ChunkedVideoFrame chunkedVideoFrame = mapper.readValue(event.getEvent().array(), ChunkedVideoFrame.class);
                    log.info("chunkedVideoFrame={}", chunkedVideoFrame);
                    VideoFrame videoFrame = new VideoFrame(chunkedVideoFrame);
                    videoFrame.validateHash();
                    Mat pngMat = new Mat(new BytePointer(videoFrame.data));
                    Mat mat = opencv_imgcodecs.imdecode(pngMat, 0);
                    Frame frame = converter.convert(mat);
                    cFrame.showImage(frame);
                }
            }
        }
    }

    @Test
    public void TestSerialization1() throws Exception {
        ObjectMapper mapper = new ObjectMapper();
        VideoFrame videoFrame = new VideoFrame();
        videoFrame.camera = 0;
        videoFrame.ssrc = 1;
        videoFrame.timestamp = Timestamp.from(Instant.now());
        videoFrame.frameNumber = 2;
        videoFrame.data = new byte[]{0, 1, 2, 3, 4, 5};
        videoFrame.hash = videoFrame.calculateHash();
        ChunkedVideoFrame chunkedVideoFrame = new ChunkedVideoFrame(videoFrame);
        String jsonString = mapper.writeValueAsString(chunkedVideoFrame);
        log.info("jsonString={}", jsonString);
    }
}


<|start_filename|>GPUTensorflowImage/Dockerfile<|end_filename|>
FROM nvidia/cuda:10.0-cudnn7-runtime-ubuntu18.04


RUN set -ex; \
    apt-get update; \
    apt-get upgrade -y && \
    apt install openjdk-8-jre  wget  libsnappy1v5 gcc ca-certificates p11-kit -y && \
    apt-get clean; \
    rm -rf /var/lib/apt/lists/*;

# Grab gosu for easy step-down from root
ENV GOSU_VERSION 1.7
ENV JAVA_HOME /usr/lib/jvm/java-1.8.0-openjdk-amd64
#ENV PATH $JAVA_HOME/bin:$PATH

RUN set -ex; \
  wget -nv -O /usr/local/bin/gosu "https://github.com/tianon/gosu/releases/download/$GOSU_VERSION/gosu-$(dpkg --print-architecture)"; \
  wget -nv -O /usr/local/bin/gosu.asc "https://github.com/tianon/gosu/releases/download/$GOSU_VERSION/gosu-$(dpkg --print-architecture).asc"; \
  export GNUPGHOME="$(mktemp -d)"; \
  for server in $(shuf -e ha.pool.sks-keyservers.net \
                          hkp://p80.pool.sks-keyservers.net:80 \
                          keyserver.ubuntu.com \
                          hkp://keyserver.ubuntu.com:80 \
                          pgp.mit.edu) ; do \
      gpg --batch --keyserver "$server" --recv-keys <KEY> && break || : ; \
  done && \
  gpg --batch --verify /usr/local/bin/gosu.asc /usr/local/bin/gosu; \
  gpgconf --kill all; \
  rm -rf "$GNUPGHOME" /usr/local/bin/gosu.asc; \
  chmod +x /usr/local/bin/gosu; \
  gosu nobody true
# Configure Flink version
ENV FLINK_VERSION=1.9.1 \
    HADOOP_SCALA_VARIANT=scala_2.12
# Prepare environment
ENV FLINK_HOME=/opt/flink
ENV PATH=$FLINK_HOME/bin:$PATH
RUN groupadd --system --gid=9999 flink && \
    useradd --system --home-dir $FLINK_HOME --uid=9999 --gid=flink flink
WORKDIR $FLINK_HOME
ENV FLINK_URL_FILE_PATH=flink/flink-${FLINK_VERSION}/flink-${FLINK_VERSION}-bin-${HADOOP_SCALA_VARIANT}.tgz
# Not all mirrors have the .asc files
ENV FLINK_TGZ_URL=https://archive.apache.org/dist/${FLINK_URL_FILE_PATH} \
    FLINK_ASC_URL=https://archive.apache.org/dist/${FLINK_URL_FILE_PATH}.asc

# For GPG verification instead of relying on key servers
COPY KEYS /KEYS
# Install Flink
RUN set -ex; \
  wget -nv -O flink.tgz "$FLINK_TGZ_URL" --no-check-certificate; \
  tar -xf flink.tgz --strip-components=1; \
  rm flink.tgz; \
  \
  chown -R flink:flink .;

RUN set -ex; \
	{ \
		echo '#!/usr/bin/env bash'; \
		echo 'set -Eeuo pipefail'; \
		echo 'ls $JAVA_HOME'; \
		echo 'cp $JAVA_HOME/jre/lib/ $JAVA_HOME/lib -r'; \
		echo 'if ! [ -d "$JAVA_HOME" ]; then echo >&2 "error: missing JAVA_HOME environment variable"; exit 1; fi'; \
	# 8-jdk uses "$JAVA_HOME/jre/lib/security/cacerts" and 8-jre and 11+ uses "$JAVA_HOME/lib/security/cacerts" directly (no "jre" directory)
		echo 'cacertsFile=; for f in "$JAVA_HOME/lib/security/cacerts" "$JAVA_HOME/jre/lib/security/cacerts"; do if [ -e "$f" ]; then echo "f = $f" && cacertsFile="$f" && echo "cacertsFile=$cacertsFile"; break; fi; done'; \
		echo 'if [ -z "$cacertsFile" ] || ! [ -f "$cacertsFile" ]; then echo >&2 "error: failed to find cacerts file in $JAVA_HOME"; exit 1; fi'; \
		echo 'trust extract --overwrite --format=java-cacerts --filter=ca-anchors --purpose=server-auth "$cacertsFile"'; \
	} > /etc/ca-certificates/update.d/docker-openjdk; \
	chmod +x /etc/ca-certificates/update.d/docker-openjdk; \
	/etc/ca-certificates/update.d/docker-openjdk; \
	# https://github.com/docker-library/openjdk/issues/331#issuecomment-498834472
	find "$JAVA_HOME/lib" -name '*.so' -exec dirname '{}' ';' | sort -u > /etc/ld.so.conf.d/docker-openjdk.conf; \
	ldconfig;

COPY ./libtensorflow_framework.so.1 /usr/lib/x86_64-linux-gnu/

# Add optional libraries into Flink classpath
RUN mkdir -p ${FLINK_HOME}/plugins/s3-fs-presto \
 && mkdir -p ${FLINK_HOME}/plugins/s3-fs-hadoop \
 && ln -fs ${FLINK_HOME}/opt/flink-s3-fs-presto-${FLINK_VERSION}.jar ${FLINK_HOME}/plugins/s3-fs-presto \
 && ln -fs ${FLINK_HOME}/opt/flink-s3-fs-hadoop-${FLINK_VERSION}.jar ${FLINK_HOME}/plugins/s3-fs-hadoop

# Configure container
COPY docker-entrypoint.sh /
ENTRYPOINT ["/docker-entrypoint.sh"]
EXPOSE 6123 8081
CMD ["help"]


<|start_filename|>person-database/src/test/java/io/pravega/example/persondatabase/PersonDatabaseTests.java<|end_filename|>
package io.pravega.example.persondatabase;

import com.fasterxml.jackson.databind.ObjectMapper;
import io.pravega.client.ClientConfig;
import io.pravega.client.EventStreamClientFactory;
import io.pravega.client.admin.ReaderGroupManager;
import io.pravega.client.admin.StreamInfo;
import io.pravega.client.admin.StreamManager;
import io.pravega.client.stream.*;
import io.pravega.client.stream.impl.ByteBufferSerializer;
import io.pravega.example.common.PravegaAppConfiguration.StreamConfig;
import io.pravega.example.common.Transaction;
import io.pravega.example.videoplayer.StreamCutBuilder;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.imageio.ImageIO;
import java.awt.image.BufferedImage;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.UUID;

public class PersonDatabaseTests {
    private static Logger log = LoggerFactory.getLogger(PersonDatabaseTests.class);

    @Test
    @Ignore
    // Read data from database stream.
    public void Test1() throws Exception {
        String scope = "examples";

        String streamName = "person-database-transaction";
        URI controllerURI = new URI("tcp://localhost:9090");
        boolean startAtTail = false;
        ClientConfig clientConfig = ClientConfig.builder().controllerURI(controllerURI).build();
        StreamConfig inputStreamConfig = new StreamConfig("examples", "INPUT_");


        StreamInfo streamInfo;
        try (StreamManager streamManager = StreamManager.create(clientConfig)) {
            streamInfo = streamManager.getStreamInfo(scope, streamName);
        }

        StreamCut startStreamCut = inputStreamConfig.getStartStreamCut();
        if (startStreamCut == StreamCut.UNBOUNDED && startAtTail) {
            startStreamCut = streamInfo.getTailStreamCut();
        }
        final String readerGroup = UUID.randomUUID().toString().replace("-", "");
        final ReaderGroupConfig readerGroupConfig = ReaderGroupConfig.builder()
                .stream(
                        Stream.of(scope, streamName),
                        startStreamCut,
                        inputStreamConfig.getEndStreamCut())
                .build();
        try (ReaderGroupManager readerGroupManager = ReaderGroupManager.withScope(scope, clientConfig)) {
            readerGroupManager.createReaderGroup(readerGroup, readerGroupConfig);
        }

        final long timeoutMs = 1000;
        final ObjectMapper mapper = new ObjectMapper();

        // Create a pravega reader and access the data
        try (EventStreamClientFactory clientFactory = EventStreamClientFactory.withScope(scope, clientConfig);
             EventStreamReader<ByteBuffer> reader = clientFactory.createReader("reader",
                     readerGroup,
                     new ByteBufferSerializer(),
                     ReaderConfig.builder().build())) {
            final StreamCutBuilder streamCutBuilder = new StreamCutBuilder(Stream.of(scope, streamName), startStreamCut);
            for (; ; ) {
                EventRead<ByteBuffer> event = reader.readNextEvent(timeoutMs);
                if (event.getEvent() != null) {
                    streamCutBuilder.addEvent(event.getPosition());
                    final StreamCut streamCutForNextEvent = streamCutBuilder.getStreamCut();
                    final Transaction transaction = mapper.readValue(event.getEvent().array(), Transaction.class);
                    log.info("Transaction={}", transaction);
                    log.info("streamCutForNextEvent={}", streamCutForNextEvent);
                    log.info("streamCutForNextEvent={}", streamCutForNextEvent.asText());
                    if (transaction.getImageData() != null) {
                        BufferedImage img = ImageIO.read(new ByteArrayInputStream(transaction.getImageData()));
                        File outputfile = new File("../images/temp/" + transaction.getImageName());
                        log.info(outputfile.getAbsolutePath());
                        outputfile.createNewFile();
                        ImageIO.write(img, "jpg", outputfile);
                    }
                }
            }
        }
    }
}


<|start_filename|>flinkprocessor/src/test/java/io/pravega/example/videoprocessor/ObjectDetectorTests.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.videoprocessor;

import io.pravega.example.common.VideoFrame;
import io.pravega.example.tensorflow.FaceRecognizer;
import io.pravega.example.tensorflow.TFObjectDetector;
import org.apache.commons.io.IOUtils;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.InputStream;
import java.nio.file.Files;

import static org.bytedeco.opencv.global.opencv_imgcodecs.imdecode;

public class ObjectDetectorTests {
    private static Logger log = LoggerFactory.getLogger(ObjectDetectorTests.class);

    @Test
    @Ignore
    public void Test1() throws Exception {
        byte[] inBytes = Files.readAllBytes((new File("test_truck.jpg")).toPath());
        byte[] outBytes = new TFObjectDetector().detect(inBytes).getJpegBytes();
        Files.write((new File("detected_test_truck.jpg")).toPath(), outBytes);
    }
}

<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/FlinkFaceRecognizerJob.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.videoprocessor;

import io.pravega.client.stream.StreamCut;
import io.pravega.connectors.flink.FlinkPravegaReader;
import io.pravega.connectors.flink.FlinkPravegaWriter;
import io.pravega.connectors.flink.PravegaWriterMode;
import io.pravega.example.common.*;
import io.pravega.example.flinkprocessor.AbstractJob;
import io.pravega.example.flinkprocessor.JsonDeserializationSchema;
import io.pravega.example.tensorflow.BoundingBox;
import io.pravega.example.tensorflow.FaceRecognizer;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.bytedeco.opencv.opencv_core.Mat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

import static org.bytedeco.opencv.global.opencv_imgcodecs.IMREAD_UNCHANGED;
import static org.bytedeco.opencv.global.opencv_imgcodecs.imdecode;

/**
 * This job matches scanned faces to known recorded faces for facial recognition by labelling recognized faces
 * and locating recognized faces.
 */
public class FlinkFaceRecognizerJob extends AbstractJob {
    private static Logger log = LoggerFactory.getLogger(FlinkFaceRecognizerJob.class);

    public FlinkFaceRecognizerJob(VideoAppConfiguration config) {
        super(config);
    }

    /**
     * The entry point for Flink applications.
     *
     * @param args Command line arguments
     */
    public static void main(String... args) {
        VideoAppConfiguration config = new VideoAppConfiguration(args);
        log.info("config: {}", config);
        FlinkFaceRecognizerJob job = new FlinkFaceRecognizerJob(config);
        job.run();
    }

    @Override
    public VideoAppConfiguration getConfig() {
        return (VideoAppConfiguration) super.getConfig();
    }

    public void run() {
        try {
            final long periodMs = (long) (1000.0 / getConfig().getFramesPerSec());
            final String jobName = FlinkFaceRecognizerJob.class.getName();
            final StreamExecutionEnvironment env = initializeFlinkStreaming();

            createStream(getConfig().getInputStreamConfig());
            createStream(getConfig().getPersonDatabaseStreamConfig());
            createStream(getConfig().getOutputStreamConfig());

            //
            // Create video datastream.
            //

            final StreamCut startStreamCut = resolveStartStreamCut(getConfig().getInputStreamConfig());
            final StreamCut endStreamCut = resolveEndStreamCut(getConfig().getInputStreamConfig());

            final FlinkPravegaReader<ChunkedVideoFrame> flinkPravegaReader = FlinkPravegaReader.<ChunkedVideoFrame>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getInputStreamConfig().getStream(), startStreamCut, endStreamCut)
                    .withDeserializationSchema(new ChunkedVideoFrameDeserializationSchema())
                    .build();
            final DataStream<ChunkedVideoFrame> inChunkedVideoFrames = env
                    .addSource(flinkPravegaReader)
                    .uid("input-source")
                    .name("input-source");

            // Assign timestamps and watermarks based on timestamp in each chunk.
            final DataStream<ChunkedVideoFrame> inChunkedVideoFramesWithTimestamps = inChunkedVideoFrames
                    .assignTimestampsAndWatermarks(
                            new BoundedOutOfOrdernessTimestampExtractor<ChunkedVideoFrame>(
                                    Time.milliseconds(getConfig().getMaxOutOfOrdernessMs())) {
                                @Override
                                public long extractTimestamp(ChunkedVideoFrame element) {
                                    return element.timestamp.getTime();
                                }
                            })
                    .uid("assignTimestampsAndWatermarks")
                    .name("assignTimestampsAndWatermarks");

            // Unchunk (disabled).
            // Operator: ChunkedVideoFrameReassembler
            final DataStream<VideoFrame> videoFrames = inChunkedVideoFramesWithTimestamps
                    .map(VideoFrame::new)
                    .uid("ChunkedVideoFrameReassembler")
                    .name("ChunkedVideoFrameReassembler");

            //
            // Create person database transaction datastream.
            //
            final FlinkPravegaReader<Transaction> flinkPravegaReaderTransactions = FlinkPravegaReader.<Transaction>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getPersonDatabaseStreamConfig().getStream())
                    .withDeserializationSchema(new JsonDeserializationSchema<>(Transaction.class))
                    .build();
            final DataStream<Transaction> personDatabaseTransactions = env
                    .addSource(flinkPravegaReaderTransactions)
                    .uid("transaction")
                    .name("transaction");
            personDatabaseTransactions.printToErr().uid("personDatabaseTransactions-print").name("personDatabaseTransactions-print");

            // Assign timestamps and watermarks based on timestamp in each event.
            final DataStream<Transaction> transactionsWithTimestamps = personDatabaseTransactions
                    .assignTimestampsAndWatermarks(
                            new BoundedOutOfOrdernessTimestampExtractor<Transaction>(
                                    Time.milliseconds(getConfig().getMaxOutOfOrdernessMs())) {
                                @Override
                                public long extractTimestamp(Transaction element) {
                                    return element.timestamp.getTime();
                                }
                            })
                    .uid("assignTimestampsAndWatermarksSensor")
                    .name("assignTimestampsAndWatermarksSensor");
            transactionsWithTimestamps.printToErr().uid("transactionsWithTimestamps-print").name("transactionsWithTimestamps-print");

            // Calculate embeddings for image data in transactions
            final DataStream<EmbeddingsTransaction> transactionEmbeddings = transactionsWithTimestamps
                    .map(new FaceRecognizerMapFunctionTransaction())
                    .uid("TransactionEmbeddings")
                    .name("transactionEmbeddings");

            // For each camera and window, get the most recent frame.
            final DataStream<VideoFrame> lastVideoFramePerCamera = videoFrames
                    .keyBy((KeySelector<VideoFrame, Integer>) value -> value.camera)
                    .window(TumblingEventTimeWindows.of(Time.milliseconds(periodMs)))
                    .maxBy("timestamp")
                    .uid("lastVideoFramePerCamera")
                    .name("lastVideoFramePerCamera");

            // This is calculating the embeddings for the located faces in the videoframe.
            final DataStream<VideoFrame> videoFrameEmbeddings = lastVideoFramePerCamera
                    .map(new FaceRecognizerMapFunctionVideoFrame())
                    .uid("videoFrameEmbeddings")
                    .name("videoFrameEmbeddings");
            videoFrameEmbeddings.printToErr().uid("videoFrameEmbeddings-print").name("videoFrameEmbeddings-print");

            final KeyedStream<VideoFrame, Integer> videoFramePerCamera = videoFrameEmbeddings
                    .keyBy((KeySelector<VideoFrame, Integer>) value -> value.camera);

            // Schema of the embeddings database in state
            // mapping is person-Id to embedding
            // This is how a person-id String maps to an embedding
            MapStateDescriptor<String, Embedding> bcStateDescriptor =
                    new MapStateDescriptor<>("embeddingBroadcastState", String.class, Embedding.class);

            // broadcast the embeddings database to all sub-tasks.
            BroadcastStream<EmbeddingsTransaction> bcedTransactions = transactionEmbeddings.broadcast(bcStateDescriptor);

            // This process tries to match faces on video frame with faces represented in embeddings database, labels
            // the recognized faces, and maintains embeddings database
            DataStream<VideoFrame> facesRecognized = videoFramePerCamera
                    .connect(bcedTransactions)
                    .process(new FaceRecognizerProcessor());

            // Split output video frames into chunks of 8 MiB or less.
            final DataStream<ChunkedVideoFrame> outChunkedVideoFrames = facesRecognized
                    .flatMap(new VideoFrameChunker(getConfig().getChunkSizeBytes()))
                    .uid("VideoFrameChunker")
                    .name("VideoFrameChunker");

            // Write chunks to Pravega encoded as JSON.
            final FlinkPravegaWriter<ChunkedVideoFrame> flinkPravegaWriter = FlinkPravegaWriter.<ChunkedVideoFrame>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getOutputStreamConfig().getStream())
                    .withSerializationSchema(new ChunkedVideoFrameSerializationSchema())
                    .withEventRouter(frame -> String.format("%d", frame.camera))
                    .withWriterMode(PravegaWriterMode.ATLEAST_ONCE)
                    .build();
            outChunkedVideoFrames
                    .addSink(flinkPravegaWriter)
                    .uid("output-sink")
                    .name("output-sink");

            log.info("Executing {} job", jobName);
            env.execute(jobName);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * A map function that uses TensorFlow and pre-trained facenet model to locate the faces in video frame, and
     * store the embeddings for the faces located.
     * The TensorFlow Session cannot be serialized so it is declared transient and
     * initialized in open().
     */
    static class FaceRecognizerMapFunctionVideoFrame extends RichMapFunction<VideoFrame, VideoFrame> {
        final private static Logger log = LoggerFactory.getLogger(FlinkFaceRecognizerJob.FaceRecognizerMapFunctionVideoFrame.class);
        private transient FaceRecognizer recognizer;

        /**
         * The first execution takes 6 minutes on a V100.
         * We warmup in open() so that map() does not timeout.
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            recognizer = new FaceRecognizer();
            recognizer.warmup();
        }

        @Override
        public void close() {
            recognizer.close();
        }

        @Override
        public VideoFrame map(VideoFrame frame) throws Exception {
            log.info("map: BEGIN: camera={}, frameNumber={}", frame.camera, frame.frameNumber);
            frame.recognizedBoxes = recognizer.locateFaces(frame.data);
            for (BoundingBox faceLocation : frame.recognizedBoxes) {
                Mat imageMat = imdecode(new Mat(frame.data), IMREAD_UNCHANGED);
                byte[] currentFaceData = recognizer.cropFace(faceLocation, imageMat);
                frame.embeddingValues.add(recognizer.embeddFace(currentFaceData));
            }
            log.info("map: END: camera={}, frameNumber={}", frame.camera, frame.frameNumber);
            return frame;
        }
    }

    /**
     * A map function that uses TensorFlow and pre-trained facenet model to locate the faces in images within transactions, and
     * store the embeddings for the faces located.
     * The TensorFlow Session cannot be serialized so it is declared transient and
     * initialized in open().
     */
    static class FaceRecognizerMapFunctionTransaction extends RichMapFunction<Transaction, EmbeddingsTransaction> {
        final private static Logger log = LoggerFactory.getLogger(FlinkFaceRecognizerJob.FaceRecognizerMapFunctionTransaction.class);
        private transient FaceRecognizer recognizer;

        /**
         * The first execution takes 6 minutes on a V100.
         * We warmup in open() so that map() does not timeout.
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            recognizer = new FaceRecognizer();
            recognizer.warmup();
        }

        @Override
        public void close() {
            recognizer.close();
        }

        @Override
        public EmbeddingsTransaction map(Transaction tran) throws Exception {
            EmbeddingsTransaction embTran = new EmbeddingsTransaction(tran.personId, tran.imageName, tran.imageData, tran.transactionType, tran.timestamp, new ArrayList<float[]>());
            if (tran.transactionType.equals("add")) {
                log.info("map: BEGIN:tran={}", tran);
                List<BoundingBox> recognizedBoxes = recognizer.locateFaces(tran.imageData);
                for (BoundingBox faceLocation : recognizedBoxes) {
                    Mat imageMat = imdecode(new Mat(embTran.imageData), IMREAD_UNCHANGED);
                    byte[] currentFaceData = recognizer.cropFace(faceLocation, imageMat);
                    embTran.embeddingValues.add(recognizer.embeddFace(currentFaceData));
                    embTran.imageData = null; // data not needed anymore, since embeddings are calculated
                }
            }
            return embTran;
        }
    }
}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/KittiSensorFusionJob.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.videoprocessor;

import io.pravega.client.stream.StreamCut;
import io.pravega.connectors.flink.FlinkPravegaReader;
import io.pravega.connectors.flink.FlinkPravegaWriter;
import io.pravega.connectors.flink.PravegaWriterMode;
import io.pravega.example.common.ChunkedVideoFrame;
import io.pravega.example.common.KittiSensorReading;
import io.pravega.example.common.VideoFrame;
import io.pravega.example.flinkprocessor.AbstractJob;
import io.pravega.example.flinkprocessor.JsonDeserializationSchema;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;

/**
 * This job demonstrates how to join different data types from different Pravega streams.
 */
public class KittiSensorFusionJob extends AbstractJob {
    private static Logger log = LoggerFactory.getLogger(KittiSensorFusionJob.class);

    /**
     * The entry point for Flink applications.
     *
     * @param args Command line arguments
     */
    public static void main(String... args) {
        VideoAppConfiguration config = new VideoAppConfiguration(args);
        log.info("config: {}", config);
        KittiSensorFusionJob job = new KittiSensorFusionJob(config);
        job.run();
    }

    public KittiSensorFusionJob(VideoAppConfiguration config) {
        super(config);
    }

    @Override
    public VideoAppConfiguration getConfig() {
        return (VideoAppConfiguration) super.getConfig();
    }

    public void run() {
        try {
            final long periodMs = (long) (1000.0 / getConfig().getFramesPerSec());
            final String jobName = KittiSensorFusionJob.class.getName();
            final StreamExecutionEnvironment env = initializeFlinkStreaming();

            createStream(getConfig().getInputStreamConfig());
            createStream(getConfig().getSensorStreamConfig());
            createStream(getConfig().getOutputStreamConfig());

            //
            // Create video datastream.
            //

            final StreamCut startStreamCut = resolveStartStreamCut(getConfig().getInputStreamConfig());
            final StreamCut endStreamCut = resolveEndStreamCut(getConfig().getInputStreamConfig());

            final FlinkPravegaReader<ChunkedVideoFrame> flinkPravegaReader = FlinkPravegaReader.<ChunkedVideoFrame>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getInputStreamConfig().getStream(), startStreamCut, endStreamCut)
                    .withDeserializationSchema(new ChunkedVideoFrameDeserializationSchema())
                    .build();
            final DataStream<ChunkedVideoFrame> inChunkedVideoFrames = env
                    .addSource(flinkPravegaReader)
                    .uid("input-source")
                    .name("input-source");

            // Assign timestamps and watermarks based on timestamp in each chunk.
            final DataStream<ChunkedVideoFrame> inChunkedVideoFramesWithTimestamps = inChunkedVideoFrames
                    .assignTimestampsAndWatermarks(
                            new BoundedOutOfOrdernessTimestampExtractor<ChunkedVideoFrame>(
                                    Time.milliseconds(getConfig().getMaxOutOfOrdernessMs())) {
                                @Override
                                public long extractTimestamp(ChunkedVideoFrame element) {
                                    return element.timestamp.getTime();
                                }
                            })
                    .uid("assignTimestampsAndWatermarks")
                    .name("assignTimestampsAndWatermarks");

            // Unchunk (disabled).
            // Operator: ChunkedVideoFrameReassembler
            final DataStream<VideoFrame> videoFrames = inChunkedVideoFramesWithTimestamps
                    .map(VideoFrame::new)
                    .uid("ChunkedVideoFrameReassembler")
                    .name("ChunkedVideoFrameReassembler");

            //
            // Create non-video sensor datastream.
            //

            final StreamCut startStreamCutSensor = resolveStartStreamCut(getConfig().getSensorStreamConfig());
            final StreamCut endStreamCutSensor = resolveEndStreamCut(getConfig().getSensorStreamConfig());

            final FlinkPravegaReader<KittiSensorReading> flinkPravegaReaderSensor = FlinkPravegaReader.<KittiSensorReading>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getSensorStreamConfig().getStream(), startStreamCutSensor, endStreamCutSensor)
                    .withDeserializationSchema(new JsonDeserializationSchema<>(KittiSensorReading.class))
                    .build();
            final DataStream<KittiSensorReading> inSensorReadings = env
                    .addSource(flinkPravegaReaderSensor)
                    .uid("input-sensor")
                    .name("input-sensor");
            inSensorReadings.printToErr().uid("inSensorReadings-print").name("inSensorReadings-print");

            // Assign timestamps and watermarks based on timestamp in each chunk.
            final DataStream<KittiSensorReading> inSensorReadingsWithTimestamps = inSensorReadings
                    .assignTimestampsAndWatermarks(
                            new BoundedOutOfOrdernessTimestampExtractor<KittiSensorReading>(
                                    Time.milliseconds(getConfig().getMaxOutOfOrdernessMs())) {
                                @Override
                                public long extractTimestamp(KittiSensorReading element) {
                                    return element.timestamp.getTime();
                                }
                            })
                    .uid("assignTimestampsAndWatermarksSensor")
                    .name("assignTimestampsAndWatermarksSensor");
            inSensorReadingsWithTimestamps.printToErr().uid("inSensorReadingsWithTimestamps-print").name("inSensorReadingsWithTimestamps-print");

            // For each camera and window, get the most recent frame.
            final DataStream<VideoFrame> lastVideoFramePerCamera = videoFrames
                    .keyBy((KeySelector<VideoFrame, Integer>) value -> value.camera)
                    .window(TumblingEventTimeWindows.of(Time.milliseconds(periodMs)))
                    .maxBy("timestamp")
                    .uid("lastVideoFramePerCamera")
                    .name("lastVideoFramePerCamera");

            // For each camera and window, get the most recent sensor reading.
            final DataStream<KittiSensorReading> lastSensorReadingsPerCamera = inSensorReadingsWithTimestamps
                    .keyBy((KeySelector<KittiSensorReading, Integer>) value -> value.camera)
                    .window(TumblingEventTimeWindows.of(Time.milliseconds(periodMs)))
                    .maxBy("timestamp")
                    .uid("lastSensorReadingsPerCamera")
                    .name("lastSensorReadingsPerCamera");

            //
            // Join video and non-video sensor data.
            //

            final DataStream<VideoFrame> outVideoFrames = lastVideoFramePerCamera
                    .join(lastSensorReadingsPerCamera)
                    .where((KeySelector<VideoFrame, Integer>) value -> value.camera)
                    .equalTo((KeySelector<KittiSensorReading, Integer>) value -> value.camera)
                    .window(TumblingEventTimeWindows.of(Time.milliseconds(periodMs)))
                    .apply((JoinFunction<VideoFrame, KittiSensorReading, VideoFrame>) (first, second) -> {
                        first.kittiSensorReadings = second;
                        return first;
                    });
            outVideoFrames.printToErr();

            // Split output video frames into chunks of 8 MiB or less.
            final DataStream<ChunkedVideoFrame> outChunkedVideoFrames = outVideoFrames
                    .flatMap(new VideoFrameChunker(getConfig().getChunkSizeBytes()))
                    .uid("VideoFrameChunker")
                    .name("VideoFrameChunker");

            // Write chunks to Pravega encoded as JSON.
            final FlinkPravegaWriter<ChunkedVideoFrame> flinkPravegaWriter = FlinkPravegaWriter.<ChunkedVideoFrame>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getOutputStreamConfig().getStream())
                    .withSerializationSchema(new ChunkedVideoFrameSerializationSchema())
                    .withEventRouter(frame -> String.format("%d", frame.camera))
                    .withWriterMode(PravegaWriterMode.ATLEAST_ONCE)
                    .build();
            outChunkedVideoFrames
                    .addSink(flinkPravegaWriter)
                    .uid("output-sink")
                    .name("output-sink");

            log.info("Executing {} job", jobName);
            env.execute(jobName);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}


<|start_filename|>common/src/main/java/io/pravega/example/common/Embedding.java<|end_filename|>
package io.pravega.example.common;

import java.sql.Timestamp;

/*
 *   This class represents a single facial embedding of a person using the algorithm.
 * */

public class Embedding {
    public String personId;
    public float[] embeddingValue;
    public String imageName; // Identifier of images used to produce embedding
    public Timestamp timestamp; // Time at which this embedding was captured to be added to database

    public Embedding(String personId, float[] embeddingValue, String imageName, Timestamp timestamp) {
        this.personId = personId;
        this.embeddingValue = embeddingValue;
        this.imageName = imageName;
        this.timestamp = timestamp;
    }

    public String getPersonId() {
        return this.personId;
    }

    public float[] getEmbeddingValue() {
        return this.embeddingValue;
    }

    public String getImageName() {
        return this.imageName;
    }

    public Timestamp getTimestamp() {
        return this.timestamp;
    }

    public String toString() {
        return "personId={" + this.personId + "}" +
                ", imageName={" + this.imageName + "}" +
                ", timestamp={" + this.timestamp.toGMTString() + "}";
    }
}


<|start_filename|>person-database/src/main/java/io/pravega/example/persondatabase/PersonDatabase.java<|end_filename|>
package io.pravega.example.persondatabase;

import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.pravega.client.EventStreamClientFactory;
import io.pravega.client.admin.StreamManager;
import io.pravega.client.stream.EventStreamWriter;
import io.pravega.client.stream.EventWriterConfig;
import io.pravega.client.stream.impl.ByteBufferSerializer;
import io.pravega.example.common.PravegaAppConfiguration;
import io.pravega.example.common.PravegaUtil;
import io.pravega.example.common.Transaction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.imageio.ImageIO;
import java.awt.image.BufferedImage;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.nio.ByteBuffer;
import java.sql.Timestamp;
import java.util.concurrent.CompletableFuture;

/*
    This class adds face images to the known face database using input jpeg images with faces in them.
 */

public class PersonDatabase implements Runnable {
    private static Logger log = LoggerFactory.getLogger(PersonDatabase.class);
    private final AppConfiguration config;

    public PersonDatabase(AppConfiguration config) {
        this.config = config;
    }

    public static void main(String... args) {
        AppConfiguration config = new AppConfiguration(args);
        log.info("config: {}", config);
        Runnable app = new PersonDatabase(config);
        app.run();
    }

    public AppConfiguration getConfig() {
        return config;
    }


    public void run() {
        if (PravegaAppConfiguration.isCreateScope()) {
            try (StreamManager streamManager = StreamManager.create(getConfig().getClientConfig())) {
                streamManager.createScope(getConfig().getDefaultScope());
            }
        }

        // Create Pravega stream.
        PravegaUtil.createStream(getConfig().getClientConfig(), getConfig().getOutputStreamConfig());

        try (EventStreamClientFactory clientFactory = EventStreamClientFactory.withScope(
                getConfig().getOutputStreamConfig().getStream().getScope(),
                getConfig().getClientConfig());
             EventStreamWriter<ByteBuffer> pravegaWriter = clientFactory.createEventWriter(
                     getConfig().getOutputStreamConfig().getStream().getStreamName(),
                     new ByteBufferSerializer(),
                     EventWriterConfig.builder().build())) {

            ObjectMapper mapper = new ObjectMapper();
            mapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);

            // args
            String personId = getConfig().getpersonId();
            String transactionType = getConfig().gettransactionType();
            String imageName = "";

            Timestamp timestamp = new Timestamp(System.currentTimeMillis());

            // Decode the jpeg image.
            BufferedImage originalImage = null;
            ByteArrayOutputStream baos = new ByteArrayOutputStream();

            byte[] imageData = null;
            if (transactionType.equals("add")) {
                File imageFile = new File(getConfig().getimagePath());
                imageName = imageFile.getName();
                originalImage = ImageIO.read(imageFile);
                ImageIO.write(originalImage, "jpg", baos);
                imageData = baos.toByteArray();
                log.info("imageName={}", imageName);
            }

            Transaction transaction = new Transaction(personId, imageName, imageData, transactionType, timestamp);
            log.info("Transaction={}", transaction);

            ByteBuffer jsonBytes = ByteBuffer.wrap(mapper.writeValueAsBytes(transaction));

            // Write to pravega
            CompletableFuture<Void> future = pravegaWriter.writeEvent(personId, jsonBytes);

            future.get();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}


<|start_filename|>common/src/test/java/io/pravega/example/tensorflow/Main.java<|end_filename|>
package io.pravega.example.tensorflow;


import org.apache.commons.io.IOUtils;

import javax.imageio.ImageIO;
import java.awt.image.BufferedImage;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.List;

//import sun.nio.ch.IOUtil;

public class Main {
    private final static String IMAGE = "before_output.jpg";

    public static void main(String[] args) {
        TFObjectDetector objectDetector = new TFObjectDetector();
        byte[]    data =  objectDetector.detect(readAllBytesOrExit(IMAGE)).getJpegBytes();

        String fIMAGE = "c:/tmp/final_output.jpg";

        BufferedImage bImage = createImageFromBytes(data);
        saveImage(bImage, fIMAGE);
    }

    private static void printToConsole(final List<Recognition> recognitions) {
        for (Recognition recognition : recognitions) {
            System.out.println("TITLE :  "+recognition.getTitle());
            System.out.println("CONFIDENCE :  "+recognition.getConfidence());
        }
    }

    public static byte[] readAllBytesOrExit(final String fileName) {
        try {
            return IOUtils.toByteArray(
                    new FileInputStream(
                            new File("c:/tmp/" + fileName)
                    )
            );
        } catch (IOException | NullPointerException ex) {

            throw new RuntimeException(ex);
        }
    }

    public static String saveImage(final BufferedImage image, final String target) {
        try {
            ImageIO.write(image,"jpg", new File(target));
            return target;
        } catch (IOException ex) {

            throw new RuntimeException(ex);
        }
    }

    private  static BufferedImage createImageFromBytes(final byte[] imageData) {
        ByteArrayInputStream bais = new ByteArrayInputStream(imageData);
        try {
            return ImageIO.read(bais);
        } catch (IOException ex) {
            throw new RuntimeException();
        }
    }
}


<|start_filename|>common/src/main/java/io/pravega/example/common/KittiSensorReading.java<|end_filename|>
/*
 * Copyright (c) Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.common;

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;


import java.sql.Timestamp;

/**
 * A class for storing a single non-video sensor reading.
 */
@JsonIgnoreProperties(ignoreUnknown = true)
public class KittiSensorReading {
    // Unique ID for this sensor.
    public int car_id;
    // The ID of the camera associated with this sensor.
    public int camera;
    // Event time of this sensor reading. We use Timestamp to have nanosecond precision for high-speed cameras.
    public Timestamp timestamp;
    // latitude of the oxts-unit (deg)
    public double lat;
    // longitude of the oxts-unit (deg)
    public double lon;
    // forward velocity, i.e. parallel to earth-surface (m/s)
    public double vf;
    // leftward velocity, i.e. parallel to earth-surface (m/s)
    public double vl;
    // forward acceleration (m/s^2)
    public double af;
    // leftward acceleration (m/s^2)
    public double al;

    @Override
    public String toString() {
        return "KittiSensorReading{" +
                "car_id=" + car_id +
                ", camera=" + camera +
                ", timestamp=" + timestamp +
                ", lat=" + lat +
                ", lon=" + lon +
                ", vf=" + vf +
                ", vl=" + vl +
                ", af=" + af +
                ", al=" + al +
                '}';
    }
}


<|start_filename|>common/src/main/java/io/pravega/example/tensorflow/PersonSerializationSchema.java<|end_filename|>
package io.pravega.example.tensorflow;

import io.pravega.example.common.Person;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;

public class PersonSerializationSchema implements SerializationSchema<Person> {
    private final ObjectMapper mapper = new ObjectMapper();

    @Override
    public byte[] serialize(Person element) {
        try {
            return mapper.writeValueAsBytes(element);
        } catch (Exception e) {
            throw new RuntimeException("Failed to serialize", e);
        }
    }
}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/FaceRecognizerProcessor.java<|end_filename|>
package io.pravega.example.videoprocessor;

import io.pravega.example.common.Embedding;
import io.pravega.example.common.EmbeddingsTransaction;
import io.pravega.example.common.VideoFrame;
import io.pravega.example.tensorflow.BoundingBox;
import io.pravega.example.tensorflow.EmbeddingsComparator;
import io.pravega.example.tensorflow.ImageUtil;
import io.pravega.example.tensorflow.Recognition;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

/*
 *   This class manages the faces of people known in state with incoming processed known face embeddings. Matches
 * embeddings of scanned faces to known faces for facial recognition upon a face being processed from camera. Upon
 * matching faces, creates video frames with labelled faces and location of faces identified.
 * */
public class FaceRecognizerProcessor
        extends KeyedBroadcastProcessFunction<String, VideoFrame, EmbeddingsTransaction, VideoFrame> {

    private static final Logger log = LoggerFactory.getLogger(FaceRecognizerProcessor.class);

    // broadcast state descriptor
    // personId -> set of embeddings
    MapStateDescriptor<String, Set<Embedding>> embeddingsDbDesc;

    @Override
    public void open(Configuration conf) {
        // initialize keyed state
        embeddingsDbDesc =
                new MapStateDescriptor<>("embeddingBroadcastState", BasicTypeInfo.STRING_TYPE_INFO,
                        (TypeInformation<Set<Embedding>>) TypeInformation.of(new TypeHint<Set<Embedding>>() {
                        }));
    }

    @Override
    public void processElement(VideoFrame frame, ReadOnlyContext ctx, Collector<VideoFrame> out) throws Exception {
        Iterable<Map.Entry<String, Set<Embedding>>> embeddigsIterable = ctx
                .getBroadcastState(this.embeddingsDbDesc)
                .immutableEntries();
        Iterator<Map.Entry<String, Set<Embedding>>> embeddingsIterator = embeddigsIterable.iterator();
        ImageUtil imageUtil = new ImageUtil();

        for (int i = 0; i < frame.recognizedBoxes.size() && frame.lastBadges != null; i++) {
            BoundingBox currFaceLocation = frame.recognizedBoxes.get(i);
            float[] currEmbedding = frame.embeddingValues.get(i);
            String match = EmbeddingsComparator.matchEmbedding(currEmbedding, embeddingsIterator, frame.lastBadges);
            Recognition recognition;
            if(!match.equals("Unknown")) {
                recognition = EmbeddingsComparator.getLabel(match, currFaceLocation);
            } else {
                recognition = EmbeddingsComparator.getLabel(match, currFaceLocation);
            }
            frame.data = imageUtil.labelFace(frame.data, recognition);
        }
        frame.hash = frame.calculateHash();
        out.collect(frame);
    }

    @Override
    public void processBroadcastElement(EmbeddingsTransaction transaction, Context ctx, Collector<VideoFrame> out)
            throws Exception {
        BroadcastState<String, Set<Embedding>> bcState = ctx.getBroadcastState(embeddingsDbDesc);

        if (transaction.transactionType.equals("add")) {
            for (int i = 0; i < transaction.embeddingValues.size(); i++) {
                float[] embeddingValues = transaction.embeddingValues.get(i);
                Embedding embedding = new Embedding(transaction.personId, embeddingValues, transaction.imageName
                        , transaction.timestamp);

                Set<Embedding> embeddingSet = bcState.get(transaction.personId);
                if (embeddingSet == null) {
                    embeddingSet = new HashSet<>();
                }
                embeddingSet.add(embedding);

                // add the new embedding to database
                bcState.put(transaction.personId, embeddingSet);
            }
        } else if (transaction.transactionType.equals("delete")) {
            // remove the embeddings related to the person in database
            bcState.remove(transaction.personId);
        }
    }
}


<|start_filename|>flinkprocessor/src/test/java/io/pravega/example/videoprocessor/ImageProcessingTests.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.videoprocessor;

import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.imageio.IIOImage;
import javax.imageio.ImageIO;
import javax.imageio.ImageWriteParam;
import javax.imageio.ImageWriter;
import javax.imageio.stream.MemoryCacheImageOutputStream;
import java.awt.*;
import java.awt.image.BufferedImage;
import java.awt.image.DataBufferByte;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.nio.file.Files;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;

import static java.awt.image.BufferedImage.TYPE_3BYTE_BGR;
import static java.lang.Math.min;

public class ImageProcessingTests {
    private static Logger log = LoggerFactory.getLogger(ImageProcessingTests.class);

    @Test
    @Ignore
    public void Test1() throws Exception {

        byte[] inBytes = Files.readAllBytes((new File("/tmp/camera0-0.png")).toPath());
        ByteArrayInputStream inStream = new ByteArrayInputStream(inBytes);
        BufferedImage inImage = ImageIO.read(inStream);
        log.info("inImage={}", inImage);

        BufferedImage outImage = new BufferedImage(inImage.getWidth()*2, inImage.getHeight()*2, inImage.getType());
        outImage.getRaster().setRect(0, 0, inImage.getData());
        log.info("outImage={}", outImage);

        ByteArrayOutputStream outStream = new ByteArrayOutputStream();
        ImageIO.write(outImage, "png", outStream);

//        ImageWriter writer = ImageIO.getImageWritersByFormatName("png").next();
//        ImageWriteParam writeParam = writer.getDefaultWriteParam();
//        writeParam.setCompressionMode(ImageWriteParam.MODE_DISABLED); // Not supported exception
//        writer.setOutput(outStream);
//        writer.write(null, new IIOImage(outImage, null, null), writeParam);

        byte[] outBytes = outStream.toByteArray();
        Files.write((new File("/tmp/out2.png")).toPath(), outBytes);
    }

    @Test
    @Ignore
    public void Test2() throws Exception {
        byte[] inBytes = Files.readAllBytes((new File("/tmp/camera0-0.png")).toPath());
        Map<Integer, byte[]> images = new HashMap<>();
        images.put(0, inBytes);
        images.put(3, inBytes);
        ImageGridBuilder builder = new ImageGridBuilder(19, 19, 4, 3);
        builder.addImage(1, inBytes);
        builder.addImages(images);
        byte[] outBytes = builder.getOutputImageBytes("png");
        Files.write((new File("/tmp/out3.png")).toPath(), outBytes);
    }

    @Test
    @Ignore
    public void Test3() throws Exception {
        byte[] inBytes = Files.readAllBytes((new File("/tmp/camera0-0.png")).toPath());
        ImageResizer resizer = new ImageResizer(100, 100);
        byte[] outBytes = resizer.resize(inBytes);
        Files.write((new File("/tmp/out4.png")).toPath(), outBytes);
    }

    @Test
    @Ignore
    public void Test4() throws Exception {
        Random rnd = new Random();
        int camera = 1;
        int frameNumber = 120;
        int width = 1000;
        int height = width;
        BufferedImage outImage = new BufferedImage(width, height, TYPE_3BYTE_BGR);

        // Image background will be random bytes to prevent compression.
        byte[] imageBuffer = ((DataBufferByte) outImage.getRaster().getDataBuffer()).getData();
        rnd.nextBytes(imageBuffer);
        Graphics2D graphics = outImage.createGraphics();

        // Write camera and frame number on image.
        float fontSize = min(width, height) * 0.13f;
        Font currentFont = graphics.getFont();
        Font newFont = currentFont.deriveFont(fontSize);
        graphics.setFont(newFont);
        int lineHeight = graphics.getFontMetrics().getHeight();
        graphics.drawString("CAMERA", 5, 5 + lineHeight);
        graphics.drawString(String.format("%04d", camera), 5, 5 + 2*lineHeight);
        graphics.drawString("FRAME", 5, 5 + 3*lineHeight);
        graphics.drawString(String.format("%05d", frameNumber), 5, 5 + 4*lineHeight);

        ImageWriter writer = ImageIO.getImageWritersByFormatName("png").next();
        ImageWriteParam writeParam = writer.getDefaultWriteParam();
//        writeParam.setCompressionMode(ImageWriteParam.MODE_EXPLICIT);
//        writeParam.setCompressionQuality(1.0f);

        ByteArrayOutputStream outStream = new ByteArrayOutputStream();
        writer.setOutput(new MemoryCacheImageOutputStream(outStream));
        writer.write(null, new IIOImage(outImage, null, null), writeParam);
        Files.write((new File("/tmp/test4.png")).toPath(), outStream.toByteArray());
    }

    @Test
    @Ignore
    public void Test5() throws Exception {
        ImageGenerator generator = new ImageGenerator(2000, 1000);
        byte[] outBytes = generator.generate(20, 1000);
        Files.write((new File("/tmp/test5.png")).toPath(), outBytes);
    }
}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/ImageResizer.java<|end_filename|>
/*
 * Copyright (c) Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.videoprocessor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.imageio.ImageIO;
import java.awt.*;
import java.awt.image.BufferedImage;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;


/**
 * Resizes images to a fixed size.
 */
public class ImageResizer {
    private static Logger log = LoggerFactory.getLogger(ImageResizer.class);

    private final int outputWidth;
    private final int outputHeight;

    public ImageResizer(int outputWidth, int outputHeight) {
        this.outputWidth = outputWidth;
        this.outputHeight = outputHeight;
    }

    /**
     * Resizes images to a fixed size.
     *
     * @param image Image file bytes
     * @return Image file bytes
     */
    public byte[] resize(byte[] image) {
        ByteArrayInputStream inStream = new ByteArrayInputStream(image);
        ByteArrayOutputStream outStream = new ByteArrayOutputStream();
        resize(inStream, outStream);
        return outStream.toByteArray();
    }

    /**
     * Resize images. The output is a PNG because this can be encoded faster than JPEG and the output is expected
     * to go to a subsequent task before it is stored.
     */
    public void resize(InputStream inStream, OutputStream outStream) {
        try {
            log.info("resize: BEGIN");
            final long t0 = System.currentTimeMillis();
            BufferedImage inImage = ImageIO.read(inStream);
            Image scaledImage = inImage.getScaledInstance(outputWidth, outputHeight, Image.SCALE_SMOOTH);
            BufferedImage outImage = new BufferedImage(outputWidth, outputHeight, inImage.getType());
            Graphics2D g2d = outImage.createGraphics();
            g2d.drawImage(scaledImage, 0, 0, null);
            g2d.dispose();
            ImageIO.write(outImage, "png", outStream);
            log.info("resize: END: TIME={}", System.currentTimeMillis() - t0);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/flinkprocessor/MemoryEaterJob.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.flinkprocessor;

import io.pravega.client.stream.StreamCut;
import io.pravega.connectors.flink.FlinkPravegaReader;
import io.pravega.connectors.flink.FlinkPravegaWriter;
import io.pravega.connectors.flink.PravegaWriterMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.LinkedList;
import java.util.List;

/**
 * Uses excessive amounts of memory to force an out-of-memory error.
 */
public class MemoryEaterJob extends AbstractJob {
    private static Logger log = LoggerFactory.getLogger(MemoryEaterJob.class);

    /**
     * The entry point for Flink applications.
     *
     * @param args Command line arguments
     */
    public static void main(String... args) {
        AppConfiguration config = new AppConfiguration(args);
        log.info("config: {}", config);
        MemoryEaterJob job = new MemoryEaterJob(config);
        job.run();
    }

    public MemoryEaterJob(AppConfiguration appConfiguration) {
        super(appConfiguration);
    }

    public void run() {
        try {
            final String jobName = MemoryEaterJob.class.getName();
            StreamExecutionEnvironment env = initializeFlinkStreaming();
            createStream(getConfig().getInputStreamConfig());
            createStream(getConfig().getOutputStreamConfig());
            final StreamCut startStreamCut = resolveStartStreamCut(getConfig().getInputStreamConfig());
            final StreamCut endStreamCut = resolveEndStreamCut(getConfig().getInputStreamConfig());
            FlinkPravegaReader<InputType> flinkPravegaReader = FlinkPravegaReader.<InputType>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getInputStreamConfig().getStream(), startStreamCut, endStreamCut)
                    .withDeserializationSchema(new JsonDeserializationSchema<>(InputType.class))
                    .build();
            DataStream<InputType> ds = env.addSource(flinkPravegaReader);
            ds.printToErr();

            DataStream<InputType> ds2 = ds.map((event) -> {
                final long totalBytes = event.size;
                final double ageSec = (System.currentTimeMillis() - event.timestamp) / 1000.0;
                log.info("event={}, ageSec={}", event, ageSec);
                // Only eat memory if event is younger than 5 seconds.
                // This should allow the processing of this event to succeed upon recovery from OOM.
                if (ageSec < 5.0) {
                    final int bytesPerBuffer = 1024 * 1024;
                    final int numBuffers = (int) (totalBytes / bytesPerBuffer);
                    log.info("Allocating {} buffers of {} bytes each.", numBuffers, bytesPerBuffer);
                    final byte[][] buf = new byte[numBuffers][bytesPerBuffer];
                    log.info("Allocated {} buffers of {} bytes each.", numBuffers, bytesPerBuffer);
                }
                return event;
            });
            ds2.printToErr();

            FlinkPravegaWriter<InputType> sink = FlinkPravegaWriter.<InputType>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getOutputStreamConfig().getStream())
                    .withSerializationSchema(new JsonSerializationSchema<>())
                    .withEventRouter((s) -> "")
                    .withWriterMode(PravegaWriterMode.ATLEAST_ONCE)
                    .build();
            ds2
                    .addSink(sink)
                    .uid("output-sink")
                    .name("output-sink");

            log.info("Executing {} job", jobName);
            env.execute(jobName);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    static class InputType {
        public long size;
        public long timestamp;

        @Override
        public String toString() {
            return "InputType{" +
                    "size=" + size +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }
}


<|start_filename|>common/src/main/java/io/pravega/example/common/EmbeddingsTransaction.java<|end_filename|>
package io.pravega.example.common;

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;

import java.io.Serializable;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/*
 *   Stores and serializes JPEG image data with their calculated embeddings.
 * */

@JsonIgnoreProperties(ignoreUnknown = true)
public class EmbeddingsTransaction extends Transaction {
    public List<float[]> embeddingValues;


    public EmbeddingsTransaction() {
        super();
        embeddingValues = new ArrayList<float[]>();
    }

    public EmbeddingsTransaction(String personId, String imageName, byte[] imageData, String transactionType, Timestamp timestamp, List<float[]> embeddingValues) {
        super(personId, imageName, imageData, transactionType, timestamp);
        this.embeddingValues = embeddingValues;
    }

    public List<float[]> getEmbeddingValues() { return this.embeddingValues; }
}


<|start_filename|>common/src/main/java/io/pravega/example/common/EventReadMetadata.java<|end_filename|>
package io.pravega.example.common;

import io.pravega.client.stream.EventPointer;
import io.pravega.client.stream.EventRead;
import io.pravega.client.stream.Position;

import java.io.Serializable;

public class EventReadMetadata implements Serializable {
    final private EventPointer eventPointer;
    final private Position position;

    public EventReadMetadata(EventRead<?> eventRead) {
        this.eventPointer = eventRead.getEventPointer();
        this.position = eventRead.getPosition();
    }

    public EventPointer getEventPointer() {
        return eventPointer;
    }

    public Position getPosition() {
        return position;
    }

    @Override
    public String toString() {
        return "EventReadMetadata{" +
                "eventPointer=" + eventPointer +
                ", position=" + position +
                '}';
    }

}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/MultiVideoGridExperimentalJob.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.videoprocessor;

import io.pravega.client.stream.StreamCut;
import io.pravega.connectors.flink.FlinkPravegaReader;
import io.pravega.connectors.flink.FlinkPravegaWriter;
import io.pravega.connectors.flink.PravegaWriterMode;
import io.pravega.example.flinkprocessor.AbstractJob;
import io.pravega.example.common.ChunkedVideoFrame;
import io.pravega.example.common.VideoFrame;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static java.lang.Math.max;

/**
 * This is similar to MultiVideoGridJob but this uses an alternative approach to parallelize processing.
 * This approach does not require watermarks.
 * However, the watermark method, used in MultiVideoGridJob, appears to work better.
 */
public class MultiVideoGridExperimentalJob extends AbstractJob {
    private static Logger log = LoggerFactory.getLogger(MultiVideoGridExperimentalJob.class);

    /**
     * The entry point for Flink applications.
     *
     * @param args Command line arguments
     */
    public static void main(String... args) {
        VideoAppConfiguration config = new VideoAppConfiguration(args);
        log.info("config: {}", config);
        MultiVideoGridExperimentalJob job = new MultiVideoGridExperimentalJob(config);
        job.run();
    }

    public MultiVideoGridExperimentalJob(VideoAppConfiguration config) {
        super(config);
    }

    @Override
    public VideoAppConfiguration getConfig() {
        return (VideoAppConfiguration) super.getConfig();
    }

    public void run() {
        try {
            final String jobName = MultiVideoGridExperimentalJob.class.getName();
            final StreamExecutionEnvironment env = initializeFlinkStreaming();
            createStream(getConfig().getInputStreamConfig());
            createStream(getConfig().getOutputStreamConfig());

            final StreamCut startStreamCut = resolveStartStreamCut(getConfig().getInputStreamConfig());
            final StreamCut endStreamCut = resolveEndStreamCut(getConfig().getInputStreamConfig());

            // Read chunked video frames from Pravega.
            // Operator: input-source
            // Effective parallelism: min of # of segments, getReaderParallelism()
            final FlinkPravegaReader<ChunkedVideoFrame> flinkPravegaReader = FlinkPravegaReader.<ChunkedVideoFrame>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getInputStreamConfig().getStream(), startStreamCut, endStreamCut)
                    .withDeserializationSchema(new ChunkedVideoFrameDeserializationSchema())
                    .build();
            final DataStream<ChunkedVideoFrame> inChunkedVideoFrames = env
                    .addSource(flinkPravegaReader)
                    .setParallelism(getConfig().getReaderParallelism())
                    .uid("input-source")
                    .name("input-source");
            inChunkedVideoFrames.printToErr();

            // Assign timestamps and watermarks based on timestamp in each chunk.
            // Operator: assignTimestampsAndWatermarks
            // Effective parallelism: min of # of segments, getReaderParallelism()
            final DataStream<ChunkedVideoFrame> inChunkedVideoFramesWithTimestamps = inChunkedVideoFrames
                    .assignTimestampsAndWatermarks(
                            new BoundedOutOfOrdernessTimestampExtractor<ChunkedVideoFrame>(
                                    Time.milliseconds(getConfig().getMaxOutOfOrdernessMs())) {
                                @Override
                                public long extractTimestamp(ChunkedVideoFrame element) {
                                    return element.timestamp.getTime();
                                }
                            })
                    .uid("assignTimestampsAndWatermarks")
                    .name("assignTimestampsAndWatermarks");
//            inChunkedVideoFramesWithTimestamps.printToErr().uid("inChunkedVideoFramesWithTimestamps-print").name("inChunkedVideoFramesWithTimestamps-print");

            // Unchunk (disabled).
            // Operator: ChunkedVideoFrameReassembler
            // Effective parallelism: default parallelism (implicit rebalance before operator)
            final DataStream<VideoFrame> inVideoFrames = inChunkedVideoFramesWithTimestamps
                    .map(VideoFrame::new)
                    .uid("ChunkedVideoFrameReassembler")
                    .name("ChunkedVideoFrameReassembler");

            // For each camera and window, get the most recent frame.
            // Operator: lastVideoFramePerCamera
            // Effective parallelism: hash of camera
            final long periodMs = (long) (1000.0 / getConfig().getFramesPerSec());
            final int imageWidth = getConfig().getImageWidth();
            final int imageHeight = getConfig().getImageHeight();
            final DataStream<VideoFrame> lastVideoFramePerCamera = inVideoFrames
                    .keyBy((KeySelector<VideoFrame, Integer>) value -> value.camera)
                    // TODO: Use a sliding window.
                    .window(TumblingEventTimeWindows.of(Time.milliseconds(periodMs)))
                    .maxBy("timestamp")
                    .uid("lastVideoFramePerCamera")
                    .name("lastVideoFramePerCamera");

            // Rebalance (round-robin) to all task slots.
            // input parallelism:  # of cameras
            // output parallelism: default parallelism
//            final DataStream<VideoFrame> rebalancedVideoFrames;
//            if (getConfig().isEnableRebalance()) {
//                rebalancedVideoFrames = lastVideoFramePerCamera.rebalance();
//            } else {
//                rebalancedVideoFrames = lastVideoFramePerCamera;
//            }

            // Resize all input images.
            // Operator: ImageResizer
            // Effective parallelism: default parallelism
            lastVideoFramePerCamera.printToErr();
            final DataStream<VideoFrame> resizedVideoFrames = lastVideoFramePerCamera
                    .rebalance()
                    .map(frame -> {
                        ImageResizer resizer = new ImageResizer(imageWidth, imageHeight);
                        frame.data = resizer.resize(frame.data);
                        frame.hash = null;
                        return frame;
                    })
                    .uid("ImageResizer")
                    .name("ImageResizer");
//            resizedVideoFrames.printToErr().uid("resizedVideoFrames-print").name("resizedVideoFrames-print");;

            // Aggregate resized images. This only adds images to a hash map. The actual work is done later to increase parallelism.
            // Operator: ImageAggregator
            // Effective parallelism: hash of monitor
            final DataStream<ImageAggregatorResult> aggResults = resizedVideoFrames
                    .keyBy((KeySelector<VideoFrame, Integer>) value -> getMonitorFromCamera(value.camera))
                    .window(TumblingEventTimeWindows.of(Time.milliseconds(periodMs)))
                    .process(new ProcessWindowFunction<VideoFrame, ImageAggregatorResult, Integer, TimeWindow>() {
                        private ValueState<Integer> frameNumberState;

                        @Override
                        public void open(Configuration parameters) throws Exception {
                            frameNumberState = getRuntimeContext().getState(new ValueStateDescriptor<>("frameNumber", Integer.class));
                        }

                        @Override
                        public void process(Integer camera, Context context, Iterable<VideoFrame> elements, Collector<ImageAggregatorResult> out) throws Exception {
                            final ImageAggregatorResult result = new ImageAggregatorResult();
                            elements.forEach(element -> {
                                result.camera = camera;
                                result.timestamp = new Timestamp(max(result.timestamp.getTime(), element.timestamp.getTime()));
                                result.videoFrames.put(element.camera, element);
                            });
                            result.frameNumber = Optional.ofNullable(frameNumberState.value()).orElse(0) + 1;
                            frameNumberState.update(result.frameNumber);
                            out.collect(result);
                        }
                    })
                    .uid("ImageAggregator")
                    .name("ImageAggregator");

            final int ssrc = new Random().nextInt();

            final DataStream<VideoFrame> outVideoFrames;
            if (true) {
                // Assign sequential index. This will be used as the frame number.
                // Operator: assignSequentialIndex
                // Effective parallelism: hash of monitor
                final DataStream<OrderedImageAggregatorResult> assignSequentialIndex = aggResults
                        .keyBy((KeySelector<ImageAggregatorResult, Integer>) value -> value.camera)
                        .process(new KeyedProcessFunction<Integer, ImageAggregatorResult, OrderedImageAggregatorResult>() {
                            private ValueState<Long> indexState;

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                indexState = getRuntimeContext().getState(new ValueStateDescriptor<>("index", Long.class));
                            }

                            @Override
                            public void processElement(ImageAggregatorResult value, Context ctx, Collector<OrderedImageAggregatorResult> out) throws Exception {
                                final long index = Optional.ofNullable(indexState.value()).orElse(0L);
                                out.collect(new OrderedImageAggregatorResult(index, value));
                                indexState.update(index + 1);
                            }
                        })
                        .uid("assignSequentialIndex")
                        .name("assignSequentialIndex");

                final int parallelism = env.getParallelism();
                final List<OutputTag<OrderedImageAggregatorResult>> outputTags = IntStream.range(0, parallelism).boxed()
                        .map(i -> new OutputTag<>(i.toString(), TypeInformation.of(OrderedImageAggregatorResult.class))).collect(Collectors.toList());

                final SingleOutputStreamOperator<Integer> splitStream = assignSequentialIndex
                        .rebalance() // ensure that frames are round robin distributed to different subtasks
                        .process(new ProcessFunction<OrderedImageAggregatorResult, Integer>() {
                            @Override
                            public void processElement(OrderedImageAggregatorResult value, Context ctx, Collector<Integer> out) throws Exception {
                                ctx.output(outputTags.get((int) value.index % parallelism), value);
                            }
                        })
                        .uid("splitStream")
                        .name("splitStream");

                final List<DataStream<OrderedImageAggregatorResult>> splits = outputTags.stream()
                        .map(splitStream::getSideOutput).collect(Collectors.toList());

                final List<SingleOutputStreamOperator<OrderedVideoFrame>> orderedVideoFrames = IntStream.range(0, parallelism).boxed()
                        .map(i ->
                                splits.get(i).map(aggResult ->
                                        new OrderedVideoFrame(aggResult.index, aggResult.value.asVideoFrame(imageWidth, imageHeight, ssrc))
                                )
                                        .uid("asVideoFrame-" + i)
                                        .name("asVideoFrame-" + i)
                        )
                        .collect(Collectors.toList());

                DataStream<OrderedVideoFrame> reduced = orderedVideoFrames.get(0);
                for (int operator = 0; operator < parallelism - 1; operator++) {
                    final KeyedStream<OrderedVideoFrame, Integer> keyed1 = reduced
                            .keyBy((KeySelector<OrderedVideoFrame, Integer>) value -> value.value.camera);
                    final KeyedStream<OrderedVideoFrame, Integer> keyed2 = orderedVideoFrames.get(operator + 1)
                            .keyBy((KeySelector<OrderedVideoFrame, Integer>) value -> value.value.camera);
                    reduced = keyed1
                            .connect(keyed2)
                            .process(new OrderedVideoFrameCoProcessFunction(operator, parallelism))
                            .uid("reduce-" + operator)
                            .name("reduce-" + operator);
                }

                outVideoFrames = reduced
                        .map(x -> x.value)
                        .uid("outVideoFrames")
                        .name("outVideoFrames");
            } else {
                // Build output images and encode as JPEG.
                // Effective parallelism: default parallelism
                final DataStream<VideoFrame> videoFrames = aggResults
                        .rebalance() // ensure that successive frames are round robin distributed to different subtasks
                        .map(aggResult -> new VideoFrame(aggResult.asVideoFrame(imageWidth, imageHeight, ssrc)))
                        .uid("asVideoFrame")
                        .name("asVideoFrame");

                // Ensure ordering.
                // Effective parallelism: hash of monitor
                outVideoFrames = videoFrames
                        .keyBy((KeySelector<VideoFrame, Integer>) value -> value.camera)
                        .window(TumblingEventTimeWindows.of(Time.milliseconds(periodMs)))
                        .maxBy("timestamp")
                        .uid("outVideoFrames")
                        .name("outVideoFrames");
            }

            outVideoFrames.printToErr().uid("outVideoFrames-print").name("outVideoFrames-print");

            // Split output video frames into chunks of 8 MiB or less.
            // Effective parallelism: hash of monitor
            final DataStream<ChunkedVideoFrame> outChunkedVideoFrames = outVideoFrames
                    .flatMap(new VideoFrameChunker(getConfig().getChunkSizeBytes()))
                    .uid("VideoFrameChunker")
                    .name("VideoFrameChunker");

            // Write chunks to Pravega encoded as JSON.
            // Effective parallelism: hash of monitor
            final FlinkPravegaWriter<ChunkedVideoFrame> flinkPravegaWriter = FlinkPravegaWriter.<ChunkedVideoFrame>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getOutputStreamConfig().getStream())
                    .withSerializationSchema(new ChunkedVideoFrameSerializationSchema())
                    .withEventRouter(frame -> String.format("%d", frame.camera))
                    .withWriterMode(PravegaWriterMode.ATLEAST_ONCE)
                    .build();
            outChunkedVideoFrames
                    .addSink(flinkPravegaWriter)
                    .uid("output-sink")
                    .name("output-sink");

            log.info("Executing {} job", jobName);
            env.execute(jobName);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    // TODO: Make this configurable.
    static int camerasPerMonitor = 2;

    static int getMonitorFromCamera(final int camera) {
        return camera / camerasPerMonitor;
    }

    static int getPositionInMonitorFromCamera(final int camera) {
        return camera % camerasPerMonitor;
    }

    public static class ImageAggregatorAccum {
        public int camera;
        // Map from camera to last VideoFrame.
        public Map<Integer, VideoFrame> videoFrames = new HashMap<>();
        // Maximum timestamp from cameras.
        public Timestamp timestamp = new Timestamp(0);
        public int frameNumber;

        public ImageAggregatorAccum() {
        }

        @Override
        public String toString() {
            return "ImageAggregatorAccum{" +
                    "camera=" + camera +
                    ",timestamp=" + timestamp +
                    ",frameNumber=" + frameNumber +
                    '}';
        }
    }

    public static class ImageAggregatorResult extends ImageAggregatorAccum {
        public ImageAggregatorResult() {
        }

        public VideoFrame asVideoFrame(int imageWidth, int imageHeight, int ssrc) {
            log.info("asVideoFrame: BEGIN: frameNumber={}", frameNumber);
            final ImageGridBuilder builder = new ImageGridBuilder(imageWidth, imageHeight, 2, 2);
            videoFrames.forEach((camera, frame) -> builder.addImage(getPositionInMonitorFromCamera(camera), frame.data));
            final long t0 = System.currentTimeMillis();
            VideoFrame videoFrame = new VideoFrame();
            videoFrame.camera = camera;
            videoFrame.ssrc = ssrc + camera;
            videoFrame.timestamp = timestamp;
            videoFrame.frameNumber = frameNumber;
            videoFrame.data = builder.getOutputImageBytes("jpg");
            videoFrame.hash = videoFrame.calculateHash();
            videoFrame.tags = new HashMap<>();
            videoFrame.tags.put("numCameras", Integer.toString(videoFrames.size()));
            videoFrame.tags.put("cameras", videoFrames.keySet().toString());

//            long sleepTime = new Random().nextInt(1000);
//            // long sleepTime = (frameNumber % 2 == 0)
//            try {
//                Thread.sleep(sleepTime);
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }

            log.info("asVideoFrame: END: frameNumber={}, TIME={}", frameNumber, System.currentTimeMillis() - t0);
            log.trace("asVideoFrame: videoFrame={}", videoFrame);
            return videoFrame;
        }
    }

    public static class OrderedImageAggregatorResult {
        final public long index;
        final public ImageAggregatorResult value;

        public OrderedImageAggregatorResult(long index, ImageAggregatorResult value) {
            this.index = index;
            this.value = value;
        }

        @Override
        public String toString() {
            return "OrderedImageAggregatorResult{" +
                    "index=" + index +
                    ", value=" + value +
                    '}';
        }
    }
}

<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/OrderedVideoFrame.java<|end_filename|>
package io.pravega.example.videoprocessor;

import io.pravega.example.common.VideoFrame;

/**
 * This is currently unused as time windows appear to work just as well.
 * See MultiVideoGridExperimentalJob.java.
 */
public class OrderedVideoFrame {
    public long index;
    public VideoFrame value;

    public OrderedVideoFrame(long index, VideoFrame value) {
        this.index = index;
        this.value = value;
    }

    @Override
    public String toString() {
        return "OrderedVideoFrame{" +
                "index=" + index +
                ", value=" + value +
                '}';
    }
}

<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/flinkprocessor/AbstractJob.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.flinkprocessor;

import io.pravega.client.admin.StreamInfo;
import io.pravega.client.admin.StreamManager;
import io.pravega.client.stream.Stream;
import io.pravega.client.stream.StreamConfiguration;
import io.pravega.client.stream.StreamCut;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * An abstract job class for Flink Pravega applications.
 */
public abstract class AbstractJob implements Runnable {
    private static Logger log = LoggerFactory.getLogger(AbstractJob.class);

    private final AppConfiguration config;

    public AbstractJob(AppConfiguration config) {
        this.config = config;
    }

    public AppConfiguration getConfig() {
        return config;
    }

    /**
     * If the Pravega stream does not exist, creates a new stream with the specified stream configuration.
     * If the stream exists, it is unchanged.
     */
    public void createStream(AppConfiguration.StreamConfig streamConfig) {
        try (StreamManager streamManager = StreamManager.create(getConfig().getPravegaConfig().getClientConfig())) {
            StreamConfiguration streamConfiguration = StreamConfiguration.builder()
                    .scalingPolicy(streamConfig.getScalingPolicy())
                    .build();
            streamManager.createStream(
                    streamConfig.getStream().getScope(),
                    streamConfig.getStream().getStreamName(),
                    streamConfiguration);
        }
    }

    /**
     * Get head and tail stream cuts for a Pravega stream.
     */
    public StreamInfo getStreamInfo(Stream stream) {
        try (StreamManager streamManager = StreamManager.create(getConfig().getPravegaConfig().getClientConfig())) {
            return streamManager.getStreamInfo(stream.getScope(), stream.getStreamName());
        }
    }

    /**
     * Convert UNBOUNDED start StreamCut to a concrete StreamCut, pointing to the current head or tail of the stream
     * (depending on isStartAtTail).
     */
    public StreamCut resolveStartStreamCut(AppConfiguration.StreamConfig streamConfig) {
        if (streamConfig.isStartAtTail()) {
            return getStreamInfo(streamConfig.getStream()).getTailStreamCut();
        } else if (streamConfig.getStartStreamCut() == StreamCut.UNBOUNDED) {
            return getStreamInfo(streamConfig.getStream()).getHeadStreamCut();
        } else {
            return streamConfig.getStartStreamCut();
        }
    }

    /**
     * For bounded reads (indicated by isEndAtTail), convert UNBOUNDED end StreamCut to a concrete StreamCut,
     * pointing to the current tail of the stream.
     * For unbounded reads, returns UNBOUNDED.
     */
    public StreamCut resolveEndStreamCut(AppConfiguration.StreamConfig streamConfig) {
        if (streamConfig.isEndAtTail()) {
            return getStreamInfo(streamConfig.getStream()).getTailStreamCut();
        } else {
            return streamConfig.getEndStreamCut();
        }
    }

    public StreamExecutionEnvironment initializeFlinkStreaming() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(getConfig().getParallelism());
        if (!getConfig().isEnableOperatorChaining()) {
            env.disableOperatorChaining();
        }
        if (getConfig().isEnableCheckpoint()) {
            env.enableCheckpointing(getConfig().getCheckpointIntervalMs());
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(getConfig().getCheckpointIntervalMs() / 2);
            env.getCheckpointConfig().setCheckpointTimeout(getConfig().getCheckpointIntervalMs() * 2);
            env.getCheckpointConfig().setFailOnCheckpointingErrors(true);
        }
        log.info("Parallelism={}, MaxParallelism={}", env.getParallelism(), env.getMaxParallelism());

        // We can't use MemoryStateBackend because it can't store large state.
        if (env instanceof LocalStreamEnvironment && (env.getStateBackend() == null || env.getStateBackend() instanceof MemoryStateBackend)) {
            log.warn("Using FsStateBackend instead of MemoryStateBackend");
            env.setStateBackend(new FsStateBackend("file:///tmp/flink-state", true));
        }

        // Stop immediately on any errors.
        if (env instanceof LocalStreamEnvironment) {
            log.warn("Using noRestart restart strategy");
            env.setRestartStrategy(RestartStrategies.noRestart());
        }

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        return env;
    }

    public ExecutionEnvironment initializeFlinkBatch() {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        int parallelism = getConfig().getParallelism();
        if (parallelism > 0) {
            env.setParallelism(parallelism);
        }
        log.info("Parallelism={}", env.getParallelism());
        return env;
    }
}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/ChunkedVideoFrameSerializationSchema.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.videoprocessor;

import io.pravega.connectors.flink.serialization.PravegaSerializationSchema;
import io.pravega.example.common.ChunkedVideoFrame;

/**
 * Serializes ChunkedVideoFrame to JSON.
 */
public class ChunkedVideoFrameSerializationSchema extends PravegaSerializationSchema<ChunkedVideoFrame> {
    public ChunkedVideoFrameSerializationSchema() {
        super(new ChunkedVideoFrameSerializer());
    }
}


<|start_filename|>common/src/main/java/io/pravega/example/tensorflow/EmbeddingsComparator.java<|end_filename|>
package io.pravega.example.tensorflow;

import io.pravega.example.common.Embedding;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Map;
import java.util.Set;

/**
 * This class allows to build labels, and compare embeddings
 */
public class EmbeddingsComparator {
    private static final Logger log = LoggerFactory.getLogger(EmbeddingsComparator.class);

    private static final double THRESHOLD = 0.97;   // Threshold for matching embeddings

    /**
     * @param otherEmbedding     The current facial embedding found in the image
     * @param embeddingsDatabase The database of facial embeddings to compare to
     * @return The name of the person the embedding matches with in the embeddings database
     */
    public static String matchEmbedding(float[] otherEmbedding, Iterator<Map.Entry<String, Set<Embedding>>> embeddingsDatabase) {
        String match = "Unknown";
        double minDiff = 1.0;

        while (embeddingsDatabase.hasNext()) {
            Map.Entry<String, Set<Embedding>> embeddingEntry = embeddingsDatabase.next();
            String personId = embeddingEntry.getKey();
            Set<Embedding> embeddingSet = embeddingEntry.getValue();

            log.info("Current embedding considered is for {}", personId);

            for (Embedding embedding : embeddingSet) {
                double diff = compareEmbeddings(embedding.embeddingValue, otherEmbedding);
                // Matches if within threshold
                if (diff < THRESHOLD && diff < minDiff) {
                    match = personId;
                    minDiff = diff;
                }
            }
        }

        return match;
    }

    /**
     * @param otherEmbedding     The current facial embedding found in the image
     * @param embeddingsDatabase The database of facial embeddings to compare to
     * @param badgeList List of badges scanned to compare scanned face to
     * @return The name of the person the embedding matches with in the embeddings database and badges scanned
     */
    public static String matchEmbedding(float[] otherEmbedding, Iterator<Map.Entry<String, Set<Embedding>>> embeddingsDatabase, Set<String> badgeList) {
        String match = "Unknown";
        double minDiff = 1.0;

        while (embeddingsDatabase.hasNext()) {
            Map.Entry<String, Set<Embedding>> embeddingEntry = embeddingsDatabase.next();
            String personId = embeddingEntry.getKey();
            Set<Embedding> embeddingSet = embeddingEntry.getValue();

            // only match with badges scanned
            if(badgeList != null && !badgeList.contains(personId)) {
                log.info("badge list has something, so skip this comparison");
                continue;
            }

            log.info("Current embedding considered is for {}", personId);

            for (Embedding embedding : embeddingSet) {
                double diff = compareEmbeddings(embedding.embeddingValue, otherEmbedding);
                // Matches if within threshold
                if (diff < THRESHOLD && diff < minDiff) {
                    match = personId;
                    minDiff = diff;
                }
            }
        }

        return match;
    }


    /**
     * @param origEmbedding  The data for a facial embedding
     * @param otherEmbedding The data for another facial embedding
     * @return The difference between the facial embeddings
     */
    public static double compareEmbeddings(float[] origEmbedding, float[] otherEmbedding) {
        double sumDiffSq = 0;

        for (int i = 0; i < origEmbedding.length; i++) {
            sumDiffSq += Math.pow(origEmbedding[i] - otherEmbedding[i], 2);
        }

        return Math.sqrt(sumDiffSq);
    }

    /**
     * @param badgeId      identifier for the face
     * @param faceLocation location of the face
     * @return a label representing location and identifier for face
     */
    public static Recognition getLabel(String badgeId, BoundingBox faceLocation) {
        Recognition recognition = new Recognition(1, badgeId, (float) 1,
                new BoxPosition((float) (faceLocation.getX()),
                        (float) (faceLocation.getY()),
                        (float) (faceLocation.getWidth()),
                        (float) (faceLocation.getHeight())));

        return recognition;
    }
}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/ChunkedVideoFrameDeserializationSchema.java<|end_filename|>
/*
 * Copyright (c) Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.videoprocessor;

import io.pravega.client.stream.EventRead;
import io.pravega.connectors.flink.serialization.PravegaDeserializationSchema;
import io.pravega.example.common.ChunkedVideoFrame;
import io.pravega.example.common.EventReadMetadata;

/**
 * Deserializes ChunkedVideoFrame from JSON.
 */
public class ChunkedVideoFrameDeserializationSchema extends PravegaDeserializationSchema<ChunkedVideoFrame> {
    public ChunkedVideoFrameDeserializationSchema() {
        super(ChunkedVideoFrame.class, new ChunkedVideoFrameSerializer());
    }

    @Override
    public ChunkedVideoFrame extractEvent(EventRead<ChunkedVideoFrame> eventRead) {
        final ChunkedVideoFrame event = eventRead.getEvent();
        if (event != null) {
            event.eventReadMetadata = new EventReadMetadata(eventRead);
        }
        return event;
    }
}


<|start_filename|>person-database/src/main/java/io/pravega/example/persondatabase/AppConfiguration.java<|end_filename|>
package io.pravega.example.persondatabase;

import io.pravega.example.common.PravegaAppConfiguration;

public class AppConfiguration extends PravegaAppConfiguration {
    private final String personId;
    private final String imagePath;
    private final String transactionType;

    public AppConfiguration(String[] args) {
        super(args);
        personId = getEnvVar("personId", "");
        imagePath = getEnvVar("imagePath", "");
        transactionType = getEnvVar("transactionType", "");
    }

    public String getpersonId() {
        return personId;
    }

    public String getimagePath() {
        return imagePath;
    }

    public String gettransactionType() {
        return transactionType;
    }

    @Override
    public String toString() {
        return "AppConfiguration{" +
                super.toString() +
                ", personId=" + personId +
                ", imagePath=" + imagePath +
                ", transactionType=" + transactionType +
                '}';
    }
}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/OrderedVideoFrameCoProcessFunction.java<|end_filename|>
package io.pravega.example.videoprocessor;

import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.MessageFormat;
import java.util.Optional;

/**
 * This class can be used to merge two streams of OrderedVideoFrame and produce ordered output.
 * It buffers elements received out-of-order in the Flink state.
 * This is currently unused as time windows appear to work just as well.
 * See MultiVideoGridExperimentalJob.java.
 */
class OrderedVideoFrameCoProcessFunction extends CoProcessFunction<OrderedVideoFrame, OrderedVideoFrame, OrderedVideoFrame> {
    private static Logger log = LoggerFactory.getLogger(OrderedVideoFrameCoProcessFunction.class);

    private final int operator;
    private final int parallelism;
    private ValueState<Long> nextOutputIndexState;
    private MapState<Long, OrderedVideoFrame> outOfOrderElementsState;

    public OrderedVideoFrameCoProcessFunction(int operator, int parallelism) {
        this.operator = operator;
        this.parallelism = parallelism;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        nextOutputIndexState = getRuntimeContext().getState(new ValueStateDescriptor<>("nextFrameNumberToEmit", Long.class));
        outOfOrderElementsState = getRuntimeContext().getMapState(new MapStateDescriptor<>("outOfOrderElements", Long.class, OrderedVideoFrame.class));
    }

    @Override
    public void processElement1(OrderedVideoFrame value, Context ctx, Collector<OrderedVideoFrame> out) throws Exception {
        processElementHelper(value, ctx, out);
    }

    @Override
    public void processElement2(OrderedVideoFrame value, Context ctx, Collector<OrderedVideoFrame> out) throws Exception {
        processElementHelper(value, ctx, out);
    }

    private void processElementHelper(OrderedVideoFrame value, Context ctx, Collector<OrderedVideoFrame> out) throws Exception {
        long nextOutputIndex = Optional.ofNullable(nextOutputIndexState.value()).orElse(0L);
        if (nextOutputIndex == value.index) {
            // We can emit this element now.
            out.collect(value);
            nextOutputIndex = OrderedParallelOperator.getNextIndexToEmit(nextOutputIndex, operator, parallelism);
            // Emit buffered elements.
            for(;;) {
                final OrderedVideoFrame buffered = outOfOrderElementsState.get(nextOutputIndex);
                if (buffered == null) {
                    break;
                }
                // Emit the buffered element now.
                log.info("Emitting buffered element; nextOutputIndex={}, buffered={}", nextOutputIndex, buffered);
                out.collect(buffered);
                outOfOrderElementsState.remove(nextOutputIndex);
                nextOutputIndex = OrderedParallelOperator.getNextIndexToEmit(nextOutputIndex, operator, parallelism);
            }
            nextOutputIndexState.update(nextOutputIndex);
        } else if (nextOutputIndex < value.index) {
            // Add early element to buffer.
            log.info("Buffering early element; nextOutputIndex={}, value={}", nextOutputIndex, value);
            outOfOrderElementsState.put(value.index, value);
        } else {
            throw new IllegalStateException(MessageFormat.format("Unexpected element order; nextOutputIndex={0}, value.index={1}", nextOutputIndex, value.index));
        }
    }
}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/BadgesProcessor.java<|end_filename|>
package io.pravega.example.videoprocessor;

import io.pravega.example.common.Embedding;
import io.pravega.example.common.Transaction;
import io.pravega.example.common.VideoFrame;
import io.pravega.example.tensorflow.FaceRecognizer;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Set;

public class BadgesProcessor extends KeyedBroadcastProcessFunction<String, VideoFrame, Set, VideoFrame> {
    private final Logger log = LoggerFactory.getLogger(BadgesProcessor.class);

    MapStateDescriptor<Void, Set> badgeDesc;

    @Override
    public void open(Configuration conf) throws Exception {
        // initialize keyed state
        badgeDesc =
                new MapStateDescriptor<Void, Set>("badgesBroadcastState", Void.class, Set.class);
    }
    @Override
    public void processElement(VideoFrame value, ReadOnlyContext ctx, Collector<VideoFrame> out) throws Exception {
        Set lastBadges = ctx
                .getBroadcastState(this.badgeDesc)
                .get(null);

        log.info("processing badges:{}", lastBadges);

        value.lastBadges = lastBadges;
        value.hash = value.calculateHash();

        out.collect(value);
    }

    @Override
    public void processBroadcastElement(Set value, Context ctx, Collector<VideoFrame> out) throws Exception {
        BroadcastState<Void, Set> bcState = ctx.getBroadcastState(badgeDesc);
        bcState.clear();

        if(value.size() > 0) {
            bcState.put(null, value);
            log.info("trying to add element:{}", value);
        }
    }
}


<|start_filename|>flinkprocessor/src/test/java/io/pravega/example/videoprocessor/OrderedParallelOperatorTests.java<|end_filename|>
package io.pravega.example.videoprocessor;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.junit.Assert.*;

public class OrderedParallelOperatorTests {
    private static Logger log = LoggerFactory.getLogger(OrderedParallelOperatorTests.class);

    @Test
    public void TestGetNextIndexToEmit() throws Exception {
        // parallelism 2
        assertEquals(1, OrderedParallelOperator.getNextIndexToEmit(0, 0, 2));
        assertEquals(2, OrderedParallelOperator.getNextIndexToEmit(1, 0, 2));

        // parallelism 3
        assertEquals(1, OrderedParallelOperator.getNextIndexToEmit(0, 0, 3));
        assertEquals(3, OrderedParallelOperator.getNextIndexToEmit(1, 0, 3));
        assertEquals(4, OrderedParallelOperator.getNextIndexToEmit(3, 0, 3));
        assertEquals(6, OrderedParallelOperator.getNextIndexToEmit(4, 0, 3));
        assertEquals(1, OrderedParallelOperator.getNextIndexToEmit(0, 1, 3));
        assertEquals(2, OrderedParallelOperator.getNextIndexToEmit(1, 1, 3));
        assertEquals(3, OrderedParallelOperator.getNextIndexToEmit(2, 1, 3));

        // parallelism 4
        assertEquals(1, OrderedParallelOperator.getNextIndexToEmit(0, 0, 4));
        assertEquals(4, OrderedParallelOperator.getNextIndexToEmit(1, 0, 4));
        assertEquals(5, OrderedParallelOperator.getNextIndexToEmit(4, 0, 4));
        assertEquals(1, OrderedParallelOperator.getNextIndexToEmit(0, 1, 4));
        assertEquals(2, OrderedParallelOperator.getNextIndexToEmit(1, 1, 4));
        assertEquals(4, OrderedParallelOperator.getNextIndexToEmit(2, 1, 4));
        assertEquals(1, OrderedParallelOperator.getNextIndexToEmit(0, 2, 4));
        assertEquals(2, OrderedParallelOperator.getNextIndexToEmit(1, 2, 4));
        assertEquals(3, OrderedParallelOperator.getNextIndexToEmit(2, 2, 4));
        assertEquals(4, OrderedParallelOperator.getNextIndexToEmit(3, 2, 4));
        assertEquals(5, OrderedParallelOperator.getNextIndexToEmit(4, 2, 4));
    }
}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/ImageGridBuilder.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.videoprocessor;

import javax.imageio.IIOImage;
import javax.imageio.ImageIO;
import javax.imageio.ImageWriteParam;
import javax.imageio.ImageWriter;
import javax.imageio.stream.MemoryCacheImageOutputStream;
import java.awt.image.BufferedImage;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Map;


/**
 * Combines multiple images into a square grid of images.
 * Images must be the correct size.
 * See example output in /images/grid-sample.png.
 */
public class ImageGridBuilder {
    private final int imageWidth;
    private final int imageHeight;
    private final int numColumns;
    private final int numRows;
    private final int margin;
    private final int statusWidth;
    private final BufferedImage outImage;

    /**
     *
     * @param imageWidth    Width of each input image.
     * @param imageHeight   Height of each input image.
     * @param numColumns    Number of columns.
     * @param numRows       Number of rows.
     */
    public ImageGridBuilder(int imageWidth, int imageHeight, int numColumns, int numRows) {
        this.imageWidth = imageWidth;
        this.imageHeight = imageHeight;
        this.numColumns = numColumns;
        this.numRows = numRows;
        margin = 1;
        statusWidth = 0;
        int outputWidth = (imageWidth + margin) * this.numColumns - margin + statusWidth;
        int outputHeight = (imageHeight + margin) * this.numRows - margin;
        outImage = new BufferedImage(outputWidth, outputHeight, BufferedImage.TYPE_INT_RGB);
    }

    /**
     *
     * @param position 0-based position.
     * @param image    Image file bytes.
     */
    public void addImage(int position, byte[] image) {
        try {
            ByteArrayInputStream inStream = new ByteArrayInputStream(image);
            BufferedImage inImage = ImageIO.read(inStream);
            int x = (position % numColumns) * (imageWidth + margin);
            int y = (position / numColumns) * (imageHeight + margin);
            outImage.getRaster().setRect(x, y, inImage.getData());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     *
     * @param images Map from position to image file bytes.
     */
    public void addImages(Map<Integer, byte[]> images) {
        images.forEach(this::addImage);
    }

    /**
     *
     * @param format "png" for PNG output.
     * @return Image file bytes.
     */
    public byte[] getOutputImageBytes(String format) {
        try {
            ByteArrayOutputStream outStream = new ByteArrayOutputStream();
            ImageWriter writer = ImageIO.getImageWritersByFormatName(format).next();
            ImageWriteParam writeParam = writer.getDefaultWriteParam();
            if (format.equals("jpg")) {
                writeParam.setCompressionMode(ImageWriteParam.MODE_EXPLICIT);
                writeParam.setCompressionQuality(0.5f);
            }
            writer.setOutput(new MemoryCacheImageOutputStream(outStream));
            writer.write(null, new IIOImage(outImage, null, null), writeParam);
            return outStream.toByteArray();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}


<|start_filename|>common/src/main/java/io/pravega/example/common/Transaction.java<|end_filename|>
package io.pravega.example.common;

import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;

import java.io.Serializable;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/*
 *   Stores and serializes JPEG images in stream.
 * */

@JsonIgnoreProperties(ignoreUnknown = true)
public class Transaction implements Serializable {
    public String personId;
    public String imageName;
    public byte[] imageData;    // JPEG image
    public String transactionType;
    public Timestamp timestamp;


    public Transaction() {

    }

    public Transaction(String personId, String imageName, byte[] imageData, String transactionType, Timestamp timestamp) {
        this.personId = personId;
        this.imageName = imageName;
        this.imageData = imageData;
        this.transactionType = transactionType;
        this.timestamp = timestamp;
    }

    public String getPersonId() {
        return this.personId;
    }

    public String getImageName() {
        return this.imageName;
    }

    public byte[] getImageData() {
        return this.imageData;
    }

    public String getTransactionType() {
        return this.transactionType;
    }

    public Timestamp getTimestamp() {
        return this.timestamp;
    }

    public String toString() {
        String dataStr = "null";
        int dataLength = 0;

        // crop the data
        if (this.imageData != null) {
            dataLength = this.imageData.length;
            int sizeToPrint = dataLength;
            int maxSizeToPrint = 10;
            if (sizeToPrint > maxSizeToPrint) {
                sizeToPrint = maxSizeToPrint;
            }
            dataStr = Arrays.toString(Arrays.copyOf(this.imageData, sizeToPrint));
        }
        return "personId={" + personId + "}, " +
                "imageName={" + imageName + "}, " +
                "transactionType={" + transactionType + "}, " +
                "data(" + dataLength + ")={" + dataStr + "}, " +
                "timestamp={" + timestamp.toGMTString() + "}";
    }
}


<|start_filename|>common/src/main/java/io/pravega/example/tensorflow/QRCode.java<|end_filename|>
package io.pravega.example.tensorflow;

import java.awt.image.BufferedImage;
import java.io.*;
import java.util.HashMap;
import java.util.Map;

import javax.imageio.ImageIO;

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.zxing.BarcodeFormat;
import com.google.zxing.BinaryBitmap;
import com.google.zxing.EncodeHintType;
import com.google.zxing.MultiFormatReader;
import com.google.zxing.MultiFormatWriter;
import com.google.zxing.NotFoundException;
import com.google.zxing.Result;
import com.google.zxing.WriterException;
import com.google.zxing.client.j2se.BufferedImageLuminanceSource;
import com.google.zxing.client.j2se.MatrixToImageWriter;
import com.google.zxing.common.BitMatrix;
import com.google.zxing.common.HybridBinarizer;
import com.google.zxing.qrcode.decoder.ErrorCorrectionLevel;
import org.apache.commons.io.IOUtils;

import static org.bytedeco.opencv.global.opencv_imgcodecs.imencode;

public class QRCode {

    public static void main(String[] args) throws WriterException, IOException,
            NotFoundException {

        // Example QRCode production
        JsonObject identification = new JsonObject();
        identification.addProperty("Id", "Ashish.Bitwara");

        String qrCodeData = identification.toString();
        String filePath = "./common/src/main/resources/QRCode.png";
        InputStream imgFile = new FileInputStream(filePath);
        byte[] imageBytes = IOUtils.toByteArray(imgFile);

        String charset = "UTF-8"; // or "ISO-8859-1"
        Map hintMap = new HashMap();
        hintMap.put(EncodeHintType.ERROR_CORRECTION, ErrorCorrectionLevel.L);

        createQRCode(qrCodeData, filePath, charset, 200, 200);
        System.out.println("QR Code image created successfully!");

        String readValue = readQRCode(imageBytes);
        System.out.println("Data read from QR Code: "
                + readValue);

        Gson g = new Gson();
        JsonObject json = g.fromJson(readValue, JsonObject.class);

        String id = json.get("Id").getAsString();

        System.out.println(id);
    }

    /**
     *
     * @param qrCodeData data that is needed to be converted into a QR code
     * @param filePath location where the QR code needs to be stored
     * @param charset which type of charset needs to be used for input data
     * @param qrCodeheight height of the QR code
     * @param qrCodewidth width of the QR code
     * @throws WriterException
     * @throws IOException
     */
    public static void createQRCode(String qrCodeData, String filePath,
                                    String charset, int qrCodeheight, int qrCodewidth)
            throws WriterException, IOException {
        BitMatrix matrix = new MultiFormatWriter().encode(
                new String(qrCodeData.getBytes(charset), charset),
                BarcodeFormat.QR_CODE, qrCodewidth, qrCodeheight);
        MatrixToImageWriter.writeToFile(matrix, filePath.substring(filePath
                .lastIndexOf('.') + 1), new File(filePath));
    }

    public static String readQRCode(byte[] imageBytes)
            throws IOException, NotFoundException {
        BinaryBitmap binaryBitmap = new BinaryBitmap(new HybridBinarizer(
                new BufferedImageLuminanceSource(
                        ImageIO.read(new ByteArrayInputStream(imageBytes)))));
        Result qrCodeResult = new MultiFormatReader().decode(binaryBitmap);
        return qrCodeResult.getText();
    }
}


<|start_filename|>common/src/main/java/io/pravega/client/stream/impl/EventPointerExposed.java<|end_filename|>
package io.pravega.client.stream.impl;

import io.pravega.client.segment.impl.Segment;

public class EventPointerExposed extends EventPointerImpl {
    public EventPointerExposed(EventPointerInternal eventPointer) {
        super(eventPointer.getSegment(), eventPointer.getEventStartOffset(), eventPointer.getEventLength());
    }

    @Override
    public Segment getSegment() {
        return super.getSegment();
    }

    @Override
    public long getEventStartOffset() {
        return super.getEventStartOffset();
    }

    @Override
    public int getEventLength() {
        return super.getEventLength();
    }
}


<|start_filename|>camera-recorder/src/main/java/io/pravega/example/camerarecorder/CameraRecorderTwoStreams.java<|end_filename|>
/*
 * Copyright (c) 2019 Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.camerarecorder;

import com.fasterxml.jackson.databind.ObjectMapper;
import io.pravega.client.EventStreamClientFactory;
import io.pravega.client.admin.StreamManager;
import io.pravega.client.stream.EventStreamWriter;
import io.pravega.client.stream.EventWriterConfig;
import io.pravega.client.stream.impl.ByteBufferSerializer;
import io.pravega.example.common.ChunkedVideoFrame;
import io.pravega.example.common.PravegaAppConfiguration;
import io.pravega.example.common.PravegaUtil;
import io.pravega.example.common.VideoFrame;
import org.bytedeco.javacpp.BytePointer;
import org.bytedeco.javacv.CanvasFrame;
import org.bytedeco.javacv.Frame;
import org.bytedeco.javacv.OpenCVFrameConverter;
import org.bytedeco.opencv.global.opencv_imgcodecs;
import org.bytedeco.opencv.global.opencv_videoio;
import org.bytedeco.opencv.opencv_core.Mat;
import org.bytedeco.opencv.opencv_videoio.VideoCapture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.rmi.ConnectIOException;
import java.sql.Timestamp;
import java.util.Random;
import java.util.concurrent.CompletableFuture;

/**
 * Reads video images from a web cam and writes them to two Pravega streams.
 */
public class CameraRecorderTwoStreams implements Runnable {
    private static Logger log = LoggerFactory.getLogger(CameraRecorderTwoStreams.class);

    private final AppConfiguration config;

    public static void main(String... args) {
        AppConfiguration config = new AppConfiguration(args);
        log.info("config: {}", config);
        Runnable app = new CameraRecorderTwoStreams(config);
        app.run();
    }

    public CameraRecorderTwoStreams(AppConfiguration appConfiguration) {
        config = appConfiguration;
    }

    public AppConfiguration getConfig() {
        return config;
    }

    public void run() {
        try {
            if (getConfig().isCreateScope()) {
                try (StreamManager streamManager = StreamManager.create(getConfig().getClientConfig())) {
                    streamManager.createScope(getConfig().getDefaultScope());
                }
            }

            // Initialize camera.
            final int captureWidth = getConfig().getImageWidth();
            final int captureHeight = getConfig().getImageHeight();
            log.info("creating grabber");
            final VideoCapture cap = new VideoCapture(getConfig().getCameraDeviceNumber());

            if(!cap.open(getConfig().getCameraDeviceNumber())) {
                throw new ConnectIOException("Cannot open the camera");
            }

            log.info("starting video capture");
            cap.set(opencv_videoio.CAP_PROP_FPS, getConfig().getFramesPerSec());
            cap.set(opencv_videoio.CAP_PROP_FRAME_WIDTH, captureWidth);
            cap.set(opencv_videoio.CAP_PROP_FRAME_HEIGHT, captureHeight);

            final double actualFramesPerSec = cap.get(opencv_videoio.CAP_PROP_FPS);
            log.info("actual frame rate={}", actualFramesPerSec);
            final boolean dropFrames = actualFramesPerSec > getConfig().getFramesPerSec();
            long lastTimestamp = 0;

            // Initialize capture preview window.
            final CanvasFrame cFrame = new CanvasFrame("Capture Preview", CanvasFrame.getDefaultGamma() / 2.2);

            // Create Pravega stream.
            PravegaUtil.createStream(getConfig().getClientConfig(), getConfig().getOutputStreamConfig());
            PravegaUtil.createStream(getConfig().getClientConfig(), new PravegaAppConfiguration.StreamConfig("examples","OUTPUT_"));

            try (EventStreamClientFactory clientFactory = EventStreamClientFactory.withScope(
                    getConfig().getOutputStreamConfig().getStream().getScope(),
                    getConfig().getClientConfig());
                 EventStreamWriter<ByteBuffer> pravegaWriter = clientFactory.createEventWriter(
                         getConfig().getOutputStreamConfig().getStream().getStreamName(),
                         new ByteBufferSerializer(),
                         EventWriterConfig.builder().build());
                 EventStreamWriter<ByteBuffer> pravegaWriterCopy = clientFactory.createEventWriter(
                         getConfig().getOutputStreamConfig().getStream().getStreamName() + "-copy",
                         new ByteBufferSerializer(),
                         EventWriterConfig.builder().build())
            ) {

                ObjectMapper mapper = new ObjectMapper();
                OpenCVFrameConverter.ToMat converterToMat = new OpenCVFrameConverter.ToMat();

                int frameNumber = 0;
                int ssrc = new Random().nextInt();
                Frame capturedFrame;

                Mat mat = new Mat();

                while (cap.read(mat)) {
                    capturedFrame = converterToMat.convert(mat);
                    long timestamp = System.currentTimeMillis();
                    // drop frames to adjust speed of camera recorder
                    if (dropFrames && timestamp - lastTimestamp < 1000 / getConfig().getFramesPerSec()) {
                        log.debug("Dropping captured frame to maintain desired frames per second");
                        continue;
                    }

                    log.info("frameNumber={}, timestamp={}, capturedFrame={}", frameNumber, timestamp, capturedFrame);

                    // Convert captured frame to JPEG.
                    BytePointer jpgBytePointer = new BytePointer();
                    opencv_imgcodecs.imencode(".jpg", mat,  jpgBytePointer);
                    log.info("jpgBytePointer={}", jpgBytePointer);
                    byte[] jpgByteArray = jpgBytePointer.getStringBytes();
                    if (false) {
                        Files.write((new File(String.format("capture-%05d.jpg", frameNumber))).toPath(), jpgByteArray);
                    }

                    // Create VideoFrame. We assume that it fits in a single chunk (< 1 MB).
                    VideoFrame videoFrame = new VideoFrame();
                    videoFrame.camera = getConfig().getCamera();
                    videoFrame.ssrc = ssrc;
                    videoFrame.timestamp = new Timestamp(timestamp);
                    videoFrame.frameNumber = frameNumber;
                    videoFrame.data = jpgByteArray;
                    videoFrame.hash = videoFrame.calculateHash();
                    ChunkedVideoFrame chunkedVideoFrame = new ChunkedVideoFrame(videoFrame);
                    ByteBuffer jsonBytes = ByteBuffer.wrap(mapper.writeValueAsBytes(chunkedVideoFrame));

                    // Write to Pravega.
                    CompletableFuture<Void> future = pravegaWriter.writeEvent(Integer.toString(videoFrame.camera), jsonBytes);
                    future = pravegaWriterCopy.writeEvent(Integer.toString(videoFrame.camera), jsonBytes);

                    // Show our frame in the preview window..
                    if (cFrame.isVisible()) {
                        cFrame.showImage(capturedFrame);
                    }

                    frameNumber++;
                    lastTimestamp = timestamp;
                }
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

<|start_filename|>common/src/main/java/io/pravega/example/common/ExtendedEventPointer.java<|end_filename|>
package io.pravega.example.common;

import io.pravega.client.segment.impl.Segment;
import io.pravega.client.stream.EventPointer;
import io.pravega.client.stream.StreamCut;
import io.pravega.client.stream.impl.EventPointerExposed;
import io.pravega.client.stream.impl.StreamCutImpl;
import io.pravega.client.stream.impl.StreamCutInternal;

import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

/**
 * Identifies an event in a Pravega stream.
 * This can be used for fetching this single event or creating a reader group that starts at this event
 * or ends immediately after this event.
 */
public class ExtendedEventPointer implements Serializable {
    final private byte[] eventPointerBytes;
    final private String startStreamCutText;
    final private String endStreamCutText;
    final private String description;

    public ExtendedEventPointer() {
        eventPointerBytes = new byte[0];
        startStreamCutText = "";
        endStreamCutText = "";
        description = "";
    }

    /**
     * TODO: The generated StreamCuts are only valid if the current event is in the same segment as initialStreamCut.
     * This assumption breaks if scaling or (possibly) when the segment owner changes due to a failure.
     */
    public ExtendedEventPointer(EventReadMetadata eventReadMetadata, StreamCut initialStreamCut) {
        final EventPointer eventPointer = eventReadMetadata.getEventPointer();
        ByteBuffer buf = eventReadMetadata.getEventPointer().toBytes();
        eventPointerBytes = new byte[buf.remaining()];
        buf.get(eventPointerBytes);
        final EventPointerExposed ep = new EventPointerExposed(eventReadMetadata.getEventPointer().asImpl());
        final StreamCutInternal sc = initialStreamCut.asImpl();
        final Map<Segment, Long> positions = new HashMap<>(sc.getPositions());
        positions.put(ep.getSegment(), ep.getEventStartOffset());
        final StreamCut startStreamCut = new StreamCutImpl(sc.getStream(), positions);
        startStreamCutText = startStreamCut.asText();
        positions.put(ep.getSegment(), ep.getEventStartOffset() + ep.getEventLength());
        final StreamCut endStreamCut = new StreamCutImpl(sc.getStream(), positions);
        endStreamCutText = endStreamCut.asText();
        description = "ExtendedEventPointer{" +
                "eventPointer=" + eventPointer +
                ", startStreamCut=" + startStreamCut +
                ", endStreamCut=" + endStreamCut +
                '}';
    }

    public byte[] getEventPointerBytes() {
        return eventPointerBytes;
    }

    public String getStartStreamCutText() {
        return startStreamCutText;
    }

    public String getEndStreamCutText() {
        return endStreamCutText;
    }

    public String getDescription() {
        return description;
    }

    @Override
    public String toString() {
        return description;
    }
}


<|start_filename|>flinkprocessor/src/main/java/io/pravega/example/videoprocessor/FlinkObjectDetectorJob.java<|end_filename|>
/*
 * Copyright (c) Dell Inc., or its subsidiaries. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 */
package io.pravega.example.videoprocessor;

import io.pravega.client.stream.StreamCut;
import io.pravega.connectors.flink.FlinkPravegaReader;
import io.pravega.connectors.flink.FlinkPravegaWriter;
import io.pravega.connectors.flink.PravegaWriterMode;
import io.pravega.example.common.ChunkedVideoFrame;
import io.pravega.example.common.ExtendedEventPointer;
import io.pravega.example.common.VideoFrame;
import io.pravega.example.flinkprocessor.AbstractJob;
import io.pravega.example.tensorflow.TFObjectDetector;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.text.MessageFormat;


/**
 * This job reads and writes a video stream from Pravega and writes frame metadata to the console.
 */
public class FlinkObjectDetectorJob extends AbstractJob {
    // Logger initialization
    private static Logger log = LoggerFactory.getLogger(FlinkObjectDetectorJob.class);

    /**
     * The entry point for Flink applications.
     *
     * @param args Command line arguments
     */
    public static void main(String[] args) throws IOException, InterruptedException {
        VideoAppConfiguration config = new VideoAppConfiguration(args);
        log.info("config: {}", config);
        FlinkObjectDetectorJob job = new FlinkObjectDetectorJob(config);
        job.run();
    }

    public FlinkObjectDetectorJob(VideoAppConfiguration config) {
        super(config);
    }

    @Override
    public VideoAppConfiguration getConfig() {
        return (VideoAppConfiguration) super.getConfig();
    }

    public void run() {
        try {
            final long periodMs = (long) (1000.0 / getConfig().getFramesPerSec());
            final String jobName = FlinkObjectDetectorJob.class.getName();
            final StreamExecutionEnvironment env = initializeFlinkStreaming();
            final int mode = getConfig().getParams().getInt("mode", 2);
            log.info("mode={}", mode);
            createStream(getConfig().getInputStreamConfig());
            createStream(getConfig().getOutputStreamConfig());
            final StreamCut startStreamCut = resolveStartStreamCut(getConfig().getInputStreamConfig());
            final StreamCut endStreamCut = resolveEndStreamCut(getConfig().getInputStreamConfig());
            log.info("startStreamCut={}", startStreamCut.asText());
            log.info("endStreamCut={}", endStreamCut.asText());

            // Read chunked video frames from Pravega.
            // Operator: input-source
            // Effective parallelism: min of # of segments, getReaderParallelism()
            final FlinkPravegaReader<ChunkedVideoFrame> flinkPravegaReader = FlinkPravegaReader.<ChunkedVideoFrame>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getInputStreamConfig().getStream(), startStreamCut, endStreamCut)
                    .withDeserializationSchema(new ChunkedVideoFrameDeserializationSchema())
                    .build();
            final DataStream<ChunkedVideoFrame> inChunkedVideoFrames = env
                    .addSource(flinkPravegaReader)
                    .setParallelism(getConfig().getReaderParallelism())
                    .uid("input-source")
                    .name("input-source");
            inChunkedVideoFrames
                    .printToErr()
                    .setParallelism(getConfig().getReaderParallelism())
                    .uid("input-source-print")
                    .name("input-source-print");

            // Calculate source event pointer.
            final DataStream<ChunkedVideoFrame> inChunkedVideoFramedWithSource = inChunkedVideoFrames.
                    map(videoFrame -> {
                            videoFrame.sourceEventPointer = new ExtendedEventPointer(videoFrame.eventReadMetadata, startStreamCut);
                            return videoFrame;
                    })
                    .uid("inChunkedVideoFramedWithSource")
                    .name("inChunkedVideoFramedWithSource");
            inChunkedVideoFramedWithSource.printToErr().uid("inChunkedVideoFramedWithSource-print").name("inChunkedVideoFramedWithSource-print");

            final DataStream<VideoFrame> outVideoFrames;

            if (mode == 0) {
                // BUG: THIS RESULTS IN OUT-OF-ORDER WRITES!

                // Unchunk (disabled).
                // Operator: ChunkedVideoFrameReassembler
                // Effective parallelism: default parallelism (implicit rebalance before operator) ???
                final DataStream<VideoFrame> inVideoFrames = inChunkedVideoFramedWithSource
                        .map(VideoFrame::new)
                        .uid("ChunkedVideoFrameReassembler")
                        .name("ChunkedVideoFrameReassembler");

                // Identify objects with YOLOv3.
                // Effective parallelism: default parallelism
                final DataStream<VideoFrame> objectDetectedFrames = inVideoFrames
                        .map(new TFObjectDetectorMapFunction())
                        .uid("objectDetectedFrames")
                        .name("objectDetectedFrames");
                objectDetectedFrames.printToErr().uid("objectDetectedFrames-print").name("objectDetectedFrames-print");
                outVideoFrames = objectDetectedFrames;

            } else if (mode == 1) {
                // BUG: THIS RESULTS IN OUT-OF-ORDER WRITES!

                // Unchunk (disabled).
                // Operator: ChunkedVideoFrameReassembler
                // Effective parallelism: default parallelism (implicit rebalance before operator) ???
                final DataStream<VideoFrame> inVideoFrames = inChunkedVideoFramedWithSource
                        .map(VideoFrame::new)
                        .uid("ChunkedVideoFrameReassembler")
                        .name("ChunkedVideoFrameReassembler");

                // Identify objects with YOLOv3.
                // Effective parallelism: default parallelism
                final DataStream<VideoFrame> objectDetectedFrames = inVideoFrames
                        .rebalance()
                        .map(new TFObjectDetectorMapFunction())
                        .uid("objectDetectedFrames")
                        .name("objectDetectedFrames");
                objectDetectedFrames.printToErr().uid("objectDetectedFrames-print").name("objectDetectedFrames-print");
                outVideoFrames = objectDetectedFrames;

            } else if (mode == 2) {

                // Assign timestamps and watermarks based on timestamp in each chunk.
                // Operator: assignTimestampsAndWatermarks
                // Effective parallelism: min of # of segments, getReaderParallelism()
                final DataStream<ChunkedVideoFrame> inChunkedVideoFramesWithTimestamps = inChunkedVideoFramedWithSource
                        .assignTimestampsAndWatermarks(
                                new BoundedOutOfOrdernessTimestampExtractor<ChunkedVideoFrame>(
                                        Time.milliseconds(getConfig().getMaxOutOfOrdernessMs())) {
                                    @Override
                                    public long extractTimestamp(ChunkedVideoFrame element) {
                                        return element.timestamp.getTime();
                                    }
                                })
                        .uid("assignTimestampsAndWatermarks")
                        .name("assignTimestampsAndWatermarks");

                // Unchunk (disabled).
                // Operator: ChunkedVideoFrameReassembler
                // Effective parallelism: default parallelism (implicit rebalance before operator) ???
                final DataStream<VideoFrame> inVideoFrames = inChunkedVideoFramesWithTimestamps
                        .map(VideoFrame::new)
                        .uid("ChunkedVideoFrameReassembler")
                        .name("ChunkedVideoFrameReassembler");

                // For each camera and window, get the most recent frame.
                // This will emit at a maximum rate of the framesPerSec parameter.
                // Operator: lastVideoFramePerCamera
                // Effective parallelism: hash of camera
                final DataStream<VideoFrame> lastVideoFramePerCamera = inVideoFrames
                        .keyBy((KeySelector<VideoFrame, Integer>) value -> value.camera)
                        .window(TumblingEventTimeWindows.of(Time.milliseconds(periodMs)))
                        .maxBy("timestamp")
                        .uid("lastVideoFramePerCamera")
                        .name("lastVideoFramePerCamera");

                // Identify objects with YOLOv3.
                // Effective parallelism: default parallelism
                final DataStream<VideoFrame> objectDetectedFrames = lastVideoFramePerCamera
                        .rebalance()
                        .map(new TFObjectDetectorMapFunction())
                        .uid("objectDetectedFrames")
                        .name("objectDetectedFrames");
                objectDetectedFrames.printToErr().uid("objectDetectedFrames-print").name("objectDetectedFrames-print");

                // Ensure ordering.
                // Effective parallelism: hash of camera
                outVideoFrames = objectDetectedFrames
                        .keyBy((KeySelector<VideoFrame, Integer>) value -> value.camera)
                        .window(TumblingEventTimeWindows.of(Time.milliseconds(periodMs)))
                        .maxBy("timestamp")
                        .uid("outVideoFrames")
                        .name("outVideoFrames");
                outVideoFrames.printToErr().uid("outVideoFrames-print").name("outVideoFrames-print");

                // Validate strictly increasing frame number. Throws an exception if events are out of order.
                if (false) {
                    final DataStream<VideoFrame> verifyOrderedVideoFrames = outVideoFrames
                            .keyBy((KeySelector<VideoFrame, Integer>) value -> value.camera)
                            .process(new KeyedProcessFunction<Integer, VideoFrame, VideoFrame>() {
                                private ValueState<Long> lastFrameNumberState;

                                @Override
                                public void open(Configuration parameters) {
                                    lastFrameNumberState = getRuntimeContext().getState(new ValueStateDescriptor<>("lastFrameNumber", Long.class));
                                }

                                @Override
                                public void processElement(VideoFrame value, Context ctx, Collector<VideoFrame> out) throws Exception {
                                    final Long lastFrameNumber = lastFrameNumberState.value();
                                    if (lastFrameNumber != null) {
                                        if (value.frameNumber <= lastFrameNumber) {
                                            log.error(MessageFormat.format(
                                                    "Unexpected frame number; current={0}, last={1}, camera={2}",
                                                    value.frameNumber, lastFrameNumber, value.camera));
                                        }
                                    }
                                    out.collect(value);
                                    lastFrameNumberState.update((long) value.frameNumber);
                                }
                            })
                            .uid("verifyOrderedVideoFrames")
                            .name("verifyOrderedVideoFrames");
                }
            } else {
                throw new IllegalArgumentException(MessageFormat.format("Unknown mode {0}", mode));
            }

            // Split output video frames into chunks of 8 MiB or less.
            // Effective parallelism: hash of camera
            final DataStream<ChunkedVideoFrame> chunkedVideoFrames = outVideoFrames
                    .flatMap(new VideoFrameChunker(getConfig().getChunkSizeBytes()))
                    .uid("VideoFrameChunker")
                    .name("VideoFrameChunker");

            // Write chunks to Pravega encoded as JSON.
            // Effective parallelism: hash of camera
            final FlinkPravegaWriter<ChunkedVideoFrame> writer = FlinkPravegaWriter.<ChunkedVideoFrame>builder()
                    .withPravegaConfig(getConfig().getPravegaConfig())
                    .forStream(getConfig().getOutputStreamConfig().getStream())
                    .withSerializationSchema(new ChunkedVideoFrameSerializationSchema())
                    .withEventRouter(frame -> String.format("%d", frame.camera))
                    .withWriterMode(PravegaWriterMode.ATLEAST_ONCE)
                    .build();
            chunkedVideoFrames
                    .addSink(writer)
                    .uid("output-sink")
                    .name("output-sink");

            log.info("Executing {} job", jobName);
            env.execute(jobName);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * A map function that uses TensorFlow.
     * The TensorFlow Session cannot be serialized so it is declared transient and
     * initialized in open().
     */
    static class TFObjectDetectorMapFunction extends RichMapFunction<VideoFrame, VideoFrame> {
        final private static Logger log = LoggerFactory.getLogger(TFObjectDetectorMapFunction.class);
        private transient TFObjectDetector tfObjectDetector;

        /**
         * The first execution takes 6 minutes on a V100.
         * We warmup in open() so that map() does not timeout.
         */
        @Override
        public void open(Configuration parameters) {
            tfObjectDetector = new TFObjectDetector();
            tfObjectDetector.warmup();
        }

        @Override
        public void close() {
            tfObjectDetector.close();
        }

        @Override
        public VideoFrame map(VideoFrame frame) {
            log.info("map: BEGIN: camera={}, frameNumber={}", frame.camera, frame.frameNumber);
            final TFObjectDetector.DetectionResult result = tfObjectDetector.detect(frame.data);
            frame.data = result.getJpegBytes();
            frame.recognitions = result.getRecognitions();
            frame.hash = frame.calculateHash();
            log.info("map: END: camera={}, frameNumber={}", frame.camera, frame.frameNumber);
            return frame;
        }
    }
}
