Amazon Kinesis Video Streams から動画画像をリアルタイム受信して画像処理をおこなう (Java)

Kinesis Video Streams の API 呼び出しは AWS SDK for Java を、ストリームから MKV を解析するためのライブラリとして別途 Amazon Kinesis Video Streams Parser Library が提供されているので、これらを使って処理する方法です。

参考) Amazon Kinesis Video Stream Parser Library

関連

• Amazon Kinesis Video Streams から動画を受信して画像処理をおこなう (Python)

AWS クライアントクラス (AWS SDK)

Kineis Video Streams は 3 つのカテゴリに分かれており、API によって異なる以下の Client クラスを使用します。ストリームに対する操作は API によって異なるデータエンドポイントを指定する必要があります。

• AmazonKinesisVideoClient
• AmazonKinesisVideoMediaClient
• AmazonKinesisVideoArchivedMediaClient

例) GetMedia API の呼び出し

Regions region = Regions.AP_NORTHEAST_1;
String streamName = "stream-test";

// AWS 認証情報
AWSCredentialsProvider credentialsProvider = new ProfileCredentialsProvider();

// AmazonKinesisVideoClient を生成
AmazonKinesisVideoClient amazonKinesisVideo = AmazonKinesisVideoClientBuilder.standard()
        .withRegion(region)
        .withCredentials(credentialsProvider)
        .build();

// AmazonKinesisVideoMediaClient を生成 : GetDataEndpoint API でデータエンドポイントを取得して指定する
String endPoint = amazonKinesisVideo.getDataEndpoint(new GetDataEndpointRequest().withAPIName(APIName.GET_MEDIA).withStreamName(streamName)).getDataEndpoint();
AmazonKinesisVideoMediaClient amazonKinesisVideoMedia = AmazonKinesisVideoMediaClientBuilder.standard()
        .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endPoint, region.getName()))
        .withCredentials(credentialsProvider)
        .build();

// GetMedia API 呼び出し
StartSelector startSelector = new StartSelector().withStartSelectorType(StartSelectorType.NOW);
GetMediaResult result = amazonKinesisVideoMedia.getMedia(new GetMediaRequest().withStreamName(streamName).withStartSelector(startSelector));

// ストリーム生成
InputStream inputStream = new InputStreamParserByteSource(result.getPayload());

StreamingMkvReader

ストリームから MKV エレメントを読み出すクラスです。Visitor クラス(後述)を指定することで読み出した MKV エレメントを処理します。

StreamingMkvReader mkvStreamReader = StreamingMkvReader.createDefault(inputStream);
mkvStreamReader.apply(visitor);

MkvElementVisitor

ストリームから読み出したセグメント、フレームを処理するクラスです。

• MkvElementVisitor
  • CompositeMkvElementVisitor
    • FrameRendererVisitor
    • FrameVisitor : フレーム+メタデータを処理する (フレーム単位)
    • FragmentMetadataVisitor : メタデータを処理する (フラグメント単位)
    • FragmentProgressTracker
    • OutputSegmentMerger : セグメントをマージして出力する
    • SimpleFrameVisitor : フレームを処理する(簡易版) (フレーム単位)
  • CopyVisitor
  • CountVisitor
  • ElementSizeAndOffsetVisitor
  • MkvChildElementCollector

CompositeMkvElementVisitor

複数の Visitor を束ねます。

CompositeMkvElementVisitor visitors = new CompositeMkvElementVisitor(visitor1, visitor2, ...);

OutputSegmentMerger

指定したストリームに出力します。

OutputSegmentMerger outputSegmentMerger = OutputSegmentMerger.createDefault(outputStream);

FrameVisitor

FrameVisitor.FrameProcessor を継承して処理を記述します。

class MyFrameProcessor implements FrameVisitor.FrameProcessor {
    @Override
    public void process(final Frame frame, final MkvTrackMetadata trackMetadata, final Optional<FragmentMetadata> fragmentMetadata) throws FrameProcessException {
        // TODO
    }
}

FrameVisitor frameVisitor = FrameVisitor.create(new MyFrameProcessor());

以下のデータが処理できます。

• Frame : フレームデータ
  • trackNumber : トラック番号 (1~)
  • timeCode : タイムコード (セグメント先頭からの時間 (ms))
  • keyFrame : キーフレームかどうか
  • invisible
  • discardable
  • lacing
  • frameData : フレームデータ
• MkvTrackMetadata : トラックに対するメタデータ
  • trackNumber : トラック番号 (1~)
  • trackUID [OPTIONAL]
  • trackName : kinesis_video
  • codecId : コーデック (例: "V_MPEG4/ISO/AVC")
  • codecName : コーデック名
  • pixelWidth [OPTIONAL] : 横ピクセル数
  • pixelHeight [OPTIONAL] : 縦ピクセル数
  • samplingFrequency [OPTIONAL]
  • channels [OPTIONAL]
  • bitDepth [OPTIONAL]
• FragmentMetadata : フラグメントに対するメタデータ
  • fragmentNumberString : フラグメント番号 (String)
  • serverSideTimestampMillis : サーバータイムスタンプ (UNIX EPOCH (ms))
  • producerSideTimestampMillis : プロデューサータイムスタンプ (UNIX EPOCH (ms))
  • fragmentNumber : フラグメント番号 (BigInteger)
  • success : 成否 (boolean)
  • errorId : エラー番号 (long)
  • errorCode : エラーコード (String)
  • millisBehindNow [OPTIONAL]
  • continuationToken [OPTIONAL]

FrameProcessor には以下のものが用意されています。

• H264FrameDecoder : H.264 デコード
• H264FrameRenderer
• H264BoundingBoxFrameRenderer

H264FrameRenderer

映像をウインドウに表示する Visitor

KinesisVideoFrameViewer frameViewer = new KinesisVideoFrameViewer(1280, 720);
frameViewer.setVisible(true);

H264FrameRenderer frameRenderer = H264FrameRenderer.create(frameViewer);
FrameVisitor frameRenderVisitor = FrameVisitor.create(frameRenderer);

H264FrameDecoder

H.264 デコードして処理したい場合はこちらを継承します。

class MyFrameDecodeProcessor extends H264FrameDecoder {
    @Override
    public void process(final Frame frame, final MkvTrackMetadata trackMetadata, final Optional<FragmentMetadata> fragmentMetadata) throws FrameProcessException {
        BufferedImage image = decodeH264Frame(frame, trackMetadata);
        // TODO
    }
}

FrameVisitor frameVisitor = FrameVisitor.create(new MyFrameDecodeProcessor());

以下のようにすれば JPEG で保存できます。

Date d = new Date(fragmentMetadata.get().getProducerSideTimestampMillis() + frame.getTimeCode());
String dt = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSXXX").format(d);
File outputFile = new File(String.format("output/%s.jpg", dt));
ImageIO.write(image, "jpg", outputfile);