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

Kinesis Video Streams の API 呼び出しは [AWS SDK for Java](https://github.com/aws/aws-sdk-java) を、ストリームから MKV を解析するためのライブラリとして別途 [Amazon Kinesis Video Streams Parser Library](https://github.com/aws/amazon-kinesis-video-streams-parser-library) が提供されているので、これらを使って処理する方法です。

![概要](https://hirose-public.s3-ap-northeast-1.amazonaws.com/qiita/2022/11/24/34b53e89-2750-457a-9eca-98d4438c23ed.png)

参考) [Amazon Kinesis Video Stream Parser Library](https://docs.aws.amazon.com/kinesisvideostreams/latest/dg/parser-library.html)

## 関連

- [Amazon Kinesis Video Streams から動画を受信して画像処理をおこなう (Python)](https://qiita.com/yh1224/items/563f90c62fe7b177d102)

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

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

- AmazonKinesisVideoClient
- AmazonKinesisVideoMediaClient
- AmazonKinesisVideoArchivedMediaClient

例) GetMedia API の呼び出し

```java
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 エレメントを処理します。

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

## MkvElementVisitor

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

- [MkvElementVisitor](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/mkv/MkvElementVisitor.java)
  - [CompositeMkvElementVisitor](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/mkv/visitors/CompositeMkvElementVisitor.java)
    - [FrameRendererVisitor](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/utilities/FrameRendererVisitor.java)
    - [FrameVisitor](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/utilities/FrameVisitor.java) : フレーム+メタデータを処理する (フレーム単位)
    - [FragmentMetadataVisitor](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/utilities/FragmentMetadataVisitor.java) : メタデータを処理する (フラグメント単位)
    - [FragmentProgressTracker](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/utilities/consumer/FragmentProgressTracker.java)
    - [OutputSegmentMerger](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/utilities/OutputSegmentMerger.java) : セグメントをマージして出力する
    - [SimpleFrameVisitor](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/utilities/SimpleFrameVisitor.java) : フレームを処理する(簡易版) (フレーム単位)
  - [CopyVisitor](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/mkv/visitors/CopyVisitor.java)
  - [CountVisitor](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/mkv/visitors/CountVisitor.java)
  - [ElementSizeAndOffsetVisitor](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/mkv/visitors/ElementSizeAndOffsetVisitor.java)
  - [MkvChildElementCollector](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/utilities/MkvChildElementCollector.java)

### CompositeMkvElementVisitor

複数の Visitor を束ねます。

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

### OutputSegmentMerger

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

```java
OutputSegmentMerger outputSegmentMerger = OutputSegmentMerger.createDefault(outputStream);
```

### FrameVisitor

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

```java
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](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/mkv/Frame.java) : フレームデータ
  - trackNumber : トラック番号 (1~)
  - timeCode : タイムコード (セグメント先頭からの時間 (ms))
  - keyFrame : キーフレームかどうか
  - invisible
  - discardable
  - lacing
  - frameData : フレームデータ
- [MkvTrackMetadata](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/utilities/MkvTrackMetadata.java) : トラックに対するメタデータ
  - trackNumber : トラック番号 (1~)
  - trackUID [OPTIONAL]
  - trackName : kinesis_video
  - codecId : コーデック (例: "V_MPEG4/ISO/AVC")
  - codecName : コーデック名
  - pixelWidth [OPTIONAL] : 横ピクセル数
  - pixelHeight [OPTIONAL] : 縦ピクセル数
  - samplingFrequency [OPTIONAL]
  - channels [OPTIONAL]
  - bitDepth [OPTIONAL]
- [FragmentMetadata](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/utilities/FragmentMetadata.java) : フラグメントに対するメタデータ
  - fragmentNumberString : フラグメント番号 (String)
  - serverSideTimestampMillis : サーバータイムスタンプ (UNIX EPOCH (ms))
  - producerSideTimestampMillis : プロデューサータイムスタンプ (UNIX EPOCH (ms))
  - fragmentNumber : フラグメント番号 (BigInteger)
  - success : 成否 (boolean)
  - errorId : エラー番号 (long)
  - errorCode : エラーコード (String)
  - millisBehindNow [OPTIONAL]
  - continuationToken [OPTIONAL]

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

- [H264FrameDecoder](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/utilities/H264FrameDecoder.java) : H.264 デコード
- [H264FrameRenderer](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/utilities/H264FrameRenderer.java)
- [H264BoundingBoxFrameRenderer](https://github.com/aws/amazon-kinesis-video-streams-parser-library/blob/master/src/main/java/com/amazonaws/kinesisvideo/parser/utilities/H264BoundingBoxFrameRenderer.java)

#### H264FrameRenderer

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

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

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

#### H264FrameDecoder

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

```java
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 で保存できます。

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