class ImageRotationStage extends PronghornStage {

    @Override
    public void run() {

        // Read in a next frame if we do not have one.
        if (nextInputFrameTimestamp == NONE && PipeReader.tryReadFragment(imagePipeIn)) {
            assert PipeReader.getMsgIdx(imagePipeIn) == ImageSchema.MSG_FRAMESTART_1;
            nextInputFrameWidth = PipeReader.readInt(imagePipeIn, ImageSchema.MSG_FRAMESTART_1_FIELD_WIDTH_101);
            nextInputFrameHeight = PipeReader.readInt(imagePipeIn, ImageSchema.MSG_FRAMESTART_1_FIELD_HEIGHT_201);
            nextInputFrameTimestamp = PipeReader.readLong(imagePipeIn, ImageSchema.MSG_FRAMESTART_1_FIELD_TIMESTAMP_301);
            nextInputFrameSize = PipeReader.readInt(imagePipeIn, ImageSchema.MSG_FRAMESTART_1_FIELD_FRAMEBYTES_401);
            assert inputFrameHead == NONE || inputFrameHead == FRAME_BUFFERED;
            inputFrameHead = 0;

            PipeReader.releaseReadLock(imagePipeIn);
        }

        // Read in a cropping request if we do not have one.
        if (rotationRequestTimestamp == NONE && PipeReader.tryReadFragment(imageRotationPipeIn)) {
            assert PipeReader.getMsgIdx(imageRotationPipeIn) == ImageRotationSchema.MSG_ROTATIONREQUEST_1;

            // Read in rotation data.
            long rotationRequestNumerator = PipeReader.readLong(imageRotationPipeIn, ImageRotationSchema.MSG_ROTATIONREQUEST_1_FIELD_NUMERATOR_101);
            long rotationRequestDenominator = PipeReader.readLong(imageRotationPipeIn, ImageRotationSchema.MSG_ROTATIONREQUEST_1_FIELD_DENOMINATOR_201);
            rotationRequestTimestamp = PipeReader.readLong(imageRotationPipeIn, ImageRotationSchema.MSG_ROTATIONREQUEST_1_FIELD_TIMESTAMP_301);

            // Calculate theta degrees from rotation data.
            double theta = rotationRequestNumerator / rotationRequestDenominator;

            // Calculate alpha, beta, and gamma for rotation data.
            rotationAlpha = -1 * Math.tan(theta / 2);
            rotationBeta = Math.sin(theta);

            PipeReader.releaseReadLock(imageRotationPipeIn);
        }

        // Read in the next frame if the next frame's timestamp is older than the rotation request.
        if (nextInputFrameTimestamp != NONE && nextInputFrameTimestamp < rotationRequestTimestamp) {

            // Ensure image frame buffer is correctly sized.
            if (inputFrame == null || inputFrame.length != nextInputFrameSize) {
                inputFrame = new byte[nextInputFrameSize];
            }

            // Fill frame arrays.
            while (PipeReader.tryReadFragment(imagePipeIn)) {
                assert PipeReader.getMsgIdx(imagePipeIn) == ImageSchema.MSG_FRAMECHUNK_2;

                // Read data.
                inputFrameHead += PipeReader.readBytes(imagePipeIn, ImageSchema.MSG_FRAMECHUNK_2_FIELD_ROWBYTES_102, inputFrame, inputFrameHead);

                // If frame is full, we're done!
                if (inputFrameHead >= inputFrame.length) {
                    inputFrameTimestamp = nextInputFrameTimestamp;
                    inputFrameWidth = nextInputFrameWidth;
                    inputFrameHeight = nextInputFrameHeight;
                    inputFrameHead = FRAME_BUFFERED;
                    nextInputFrameTimestamp = NONE;
                    break;
                }

                PipeReader.releaseReadLock(imagePipeIn);
            }
        }

        // If we have an input frame buffered, have an active rotation request,
        // and the input frame predates the rotation request, write out a rotation.
        if (outputFrameHead != NONE ||
            (inputFrameHead == FRAME_BUFFERED && rotationRequestTimestamp != NONE && inputFrameTimestamp < rotationRequestTimestamp)) {

            // Frame start.
            if (outputFrameHead == NONE && PipeWriter.tryWriteFragment(imagePipeOut, ImageSchema.MSG_FRAMESTART_1)) {

                // Write start data to the pipe.
                PipeWriter.writeInt(imagePipeOut, ImageSchema.MSG_FRAMESTART_1_FIELD_WIDTH_101, cropWidth);
                PipeWriter.writeInt(imagePipeOut, ImageSchema.MSG_FRAMESTART_1_FIELD_HEIGHT_201, cropHeight);
                PipeWriter.writeLong(imagePipeOut, ImageSchema.MSG_FRAMESTART_1_FIELD_TIMESTAMP_301, System.currentTimeMillis());
                PipeWriter.writeInt(imagePipeOut, ImageSchema.MSG_FRAMESTART_1_FIELD_FRAMEBYTES_401, outputFrame.length);
                PipeWriter.writeInt(imagePipeOut, ImageSchema.MSG_FRAMESTART_1_FIELD_BITSPERPIXEL_501, 24);
                PipeWriter.writeBytes(imagePipeOut, ImageSchema.MSG_FRAMESTART_1_FIELD_ENCODING_601, LinuxImageCaptureStage.OUTPUT_ENCODING);
                PipeWriter.publishWrites(imagePipeOut);
                outputFrameHead = 0;

                // Reset the output frame.
                // This must be done because some positions in the frame may not be written to,
                // and thus should appear empty (black).
                Arrays.fill(outputFrame, (byte) 0x00);

                // Skew the input frame into our output frame.
                shearRotateFrame(rotationAlpha, rotationBeta, inputFrame, inputFrameWidth, inputFrameHeight, outputFrame, cropWidth, cropHeight);
            }

            // Write output frame data.
            if (outputFrameHead >= 0 && PipeWriter.tryWriteFragment(imagePipeOut, ImageSchema.MSG_FRAMECHUNK_2)) {

                // Write bytes.
                PipeWriter.writeBytes(imagePipeOut, ImageSchema.MSG_FRAMECHUNK_2_FIELD_ROWBYTES_102, outputFrame, outputFrameHead, outputFrameRowSize);
                PipeWriter.publishWrites(imagePipeOut);

                // Progress head.
                outputFrameHead += outputFrameRowSize;

                // If the head exceeds the size of the frame bytes, we're done writing.
                if (outputFrameHead >= outputFrame.length) {
                    outputFrameHead = NONE;
                }
            }
        }
    }

}

class ImageRotationStageTest {

    public static final int SOURCE_WIDTH;
    public static final int SOURCE_ROW_SIZE;
    public static final int SOURCE_HEIGHT;
    public static final int ROTATION_WIDTH;
    public static final int ROTATION_ROW_SIZE;
    public static final int ROTATION_HEIGHT;
    public static final long[] ROTATION_ANGLE;
    public GraphManager gm;
    public Pipe<ImageSchema> imageInputPipe;
    public Pipe<ImageRotationSchema> imageRotationInputPipe;
    public Pipe<ImageSchema> rotationOutputPipe;
    public RotationPipeReaderState rotationPipeReaderState;

    @Test
    @Ignore
    public void shouldDownscaleImagesTest() {


        int iterations = 5;
        int j = iterations;
        while (--j>=0) {

            setup();

            // Create test scheduler.
            NonThreadScheduler scheduler = new NonThreadScheduler(gm);
            scheduler.startup();

            // Put a rotation request on the pipe.
            if (PipeWriter.tryWriteFragment(imageRotationInputPipe, ImageRotationSchema.MSG_ROTATIONREQUEST_1)) {
                PipeWriter.writeLong(imageRotationInputPipe, ImageRotationSchema.MSG_ROTATIONREQUEST_1_FIELD_NUMERATOR_101, ROTATION_ANGLE[0]);
                PipeWriter.writeLong(imageRotationInputPipe, ImageRotationSchema.MSG_ROTATIONREQUEST_1_FIELD_DENOMINATOR_201, ROTATION_ANGLE[1]);
                PipeWriter.writeLong(imageRotationInputPipe, ImageRotationSchema.MSG_ROTATIONREQUEST_1_FIELD_TIMESTAMP_301,Long.MAX_VALUE);
                PipeWriter.publishWrites(imageRotationInputPipe);
            }

            // Run until we write a frame.
            while (rotationPipeReaderState.framesProcessed <= 0) {

                // Run scheduler pipe.
                scheduler.run();

                // Read from pipe.
                readFromRotationPipe(rotationOutputPipe, rotationPipeReaderState, LinuxImageCaptureStage.OUTPUT_ENCODING);
            }

            // Verify the frame is valid.
            try {
                Path filePath = Paths.get("src", "test", "images", "cat1-320-240.rotated");
                byte[] fileBytes = Files.readAllBytes(filePath);
                Assert.assertTrue(Arrays.equals(fileBytes, rotationPipeReaderState.currentFrame));
            } catch (IOException e) {
                Assert.fail(e.getMessage());
            }
        }
    }

}