Add playback engine with A/V timestamping

- AnnexBSplitter incrementally splits the Annex-B stream into access
  units, holding partial slices across packet feeds.
- VideoDecoder exposes feedAccessUnit and the decoded frame rate.
- PlaybackEngine demuxes and decodes in lockstep, tagging video frames
  and audio chunks with 90 kHz PTS (audio from PES headers, video from
  PES anchors interpolated by decoder frame rate).
- CLI 'play' command runs the engine headlessly and reports A/V drift.
- Validated against Batman Begins: 95 frames in 4.0s, max drift -78.6 ms
  (video trails audio at stream start), avg frame duration 3833.6 ticks.
- Tests for AnnexBSplitter.
This commit is contained in:
2026-09-15 22:17:49 -04:00
parent 8267750207
commit 3768e50603
6 changed files with 440 additions and 52 deletions
@@ -0,0 +1,89 @@
// SPDX-License-Identifier: GPL-3.0-or-later
package dev.quoije.umd.media;
import java.util.Arrays;
/**
* Incrementally splits an Annex-B H.264 byte stream into access units.
*
* <p>An access unit is emitted only once it is known to be complete: a slice
* NAL (type 1 or 5) terminates an access unit, and its end is confirmed by the
* presence of the following start code. Incomplete trailing data is buffered
* until more bytes arrive or until {@link #drain()} is called.</p>
*/
public final class AnnexBSplitter {
private byte[] buffer = new byte[0];
/** Appends raw Annex-B bytes. */
public void feed(byte[] data) {
if (data == null || data.length == 0) {
return;
}
byte[] merged = Arrays.copyOf(buffer, buffer.length + data.length);
System.arraycopy(data, 0, merged, buffer.length, data.length);
buffer = merged;
}
/**
* Returns the next complete access unit, or {@code null} if none is
* available yet.
*/
public byte[] nextAccessUnit() {
int i = 0;
while (i < buffer.length) {
int startCodeLength = startCodeLengthAt(buffer, i);
if (startCodeLength > 0) {
int nalu = i + startCodeLength;
if (nalu < buffer.length) {
int type = buffer[nalu] & 0x1F;
if (type == 1 || type == 5) {
int end = findNextStartCode(buffer, nalu);
if (end >= buffer.length) {
// Slice may still be incomplete; wait for more data.
return null;
}
byte[] accessUnit = Arrays.copyOfRange(buffer, 0, end);
buffer = Arrays.copyOfRange(buffer, end, buffer.length);
return accessUnit;
}
}
i = nalu;
} else {
i++;
}
}
return null;
}
/** Returns any remaining buffered bytes, or {@code null} if empty. */
public byte[] drain() {
if (buffer.length == 0) {
return null;
}
byte[] tail = buffer;
buffer = new byte[0];
return tail;
}
private static int startCodeLengthAt(byte[] b, int i) {
if (i + 2 < b.length && b[i] == 0 && b[i + 1] == 0 && b[i + 2] == 1) {
return 3;
}
if (i + 3 < b.length && b[i] == 0 && b[i + 1] == 0 && b[i + 2] == 0 && b[i + 3] == 1) {
return 4;
}
return -1;
}
private static int findNextStartCode(byte[] b, int from) {
int j = from;
while (j < b.length) {
if (startCodeLengthAt(b, j) > 0) {
return j;
}
j++;
}
return b.length;
}
}
@@ -5,13 +5,13 @@ import org.bytedeco.ffmpeg.avcodec.AVCodec;
import org.bytedeco.ffmpeg.avcodec.AVCodecContext;
import org.bytedeco.ffmpeg.avcodec.AVPacket;
import org.bytedeco.ffmpeg.avutil.AVFrame;
import org.bytedeco.ffmpeg.avutil.AVRational;
import org.bytedeco.ffmpeg.global.avcodec;
import org.bytedeco.ffmpeg.global.avutil;
import org.bytedeco.javacpp.BytePointer;
import org.bytedeco.javacpp.PointerPointer;
import java.util.ArrayDeque;
import java.util.Arrays;
/** H.264 (Annex B) video decoder backed by FFmpeg. */
public final class VideoDecoder implements AutoCloseable {
@@ -47,59 +47,24 @@ public final class VideoDecoder implements AutoCloseable {
* produces one frame per packet.
*/
public void feedAnnexB(byte[] annexB) {
int accessUnitStart = 0;
int i = 0;
while (i < annexB.length) {
int startCodeLength = startCodeLengthAt(annexB, i);
if (startCodeLength > 0) {
int nalu = i + startCodeLength;
if (nalu < annexB.length) {
int type = annexB[nalu] & 0x1F;
if (type == 1 || type == 5) {
int end = findNextStartCode(annexB, nalu);
feedOne(Arrays.copyOfRange(annexB, accessUnitStart, end));
accessUnitStart = end;
i = end;
continue;
}
}
i = nalu;
} else {
i++;
}
AnnexBSplitter splitter = new AnnexBSplitter();
splitter.feed(annexB);
byte[] accessUnit;
while ((accessUnit = splitter.nextAccessUnit()) != null) {
feedAccessUnit(accessUnit);
}
if (accessUnitStart < annexB.length) {
feedOne(Arrays.copyOfRange(annexB, accessUnitStart, annexB.length));
byte[] tail = splitter.drain();
if (tail != null && tail.length > 0) {
feedAccessUnit(tail);
}
}
private static int startCodeLengthAt(byte[] b, int i) {
if (i + 2 < b.length && b[i] == 0 && b[i + 1] == 0 && b[i + 2] == 1) {
return 3;
}
if (i + 3 < b.length && b[i] == 0 && b[i + 1] == 0 && b[i + 2] == 0 && b[i + 3] == 1) {
return 4;
}
return -1;
}
private static int findNextStartCode(byte[] b, int from) {
int j = from;
while (j < b.length) {
if (startCodeLengthAt(b, j) > 0) {
return j;
}
j++;
}
return b.length;
}
private void feedOne(byte[] data) {
if (avcodec.av_new_packet(packet, data.length) < 0) {
/** Feeds one complete Annex-B access unit to the decoder. */
public void feedAccessUnit(byte[] accessUnit) {
if (avcodec.av_new_packet(packet, accessUnit.length) < 0) {
throw new IllegalStateException("av_new_packet failed");
}
packet.data().put(data);
packet.data().put(accessUnit);
int ret = avcodec.avcodec_send_packet(context, packet);
avcodec.av_packet_unref(packet);
@@ -147,6 +112,18 @@ public final class VideoDecoder implements AutoCloseable {
return frames.poll();
}
/**
* Returns the decoded frame rate reported by FFmpeg (from the SPS VUI),
* or 0.0 if not yet known.
*/
public double frameRate() {
AVRational fr = context.framerate();
if (fr != null && !fr.isNull() && fr.num() > 0 && fr.den() > 0) {
return (double) fr.num() / fr.den();
}
return 0.0;
}
@Override
public void close() {
avutil.av_frame_free(frame);
@@ -0,0 +1,226 @@
// SPDX-License-Identifier: GPL-3.0-or-later
package dev.quoije.umd.player;
import dev.quoije.umd.iso.UmdIsoReader;
import dev.quoije.umd.media.AnnexBSplitter;
import dev.quoije.umd.media.Atrac3pDecoder;
import dev.quoije.umd.media.VideoDecoder;
import dev.quoije.umd.mps.AudioFramer;
import dev.quoije.umd.mps.MpsDemuxer;
import java.io.IOException;
import java.util.ArrayDeque;
/**
* Decodes an MPS stream into timestamped video frames and PCM audio.
*
* <p>The engine reads MPEG-PS packets, routes video to the H.264 decoder and
* the selected audio substream to the ATRAC3+ decoder, and exposes the results
* through bounded pull queues. Timestamps are in the 90 kHz MPEG clock.</p>
*
* <p>Audio timestamps come directly from the PES headers (one ATRAC3+ frame
* is 2048 samples at 48 kHz = 3840 ticks). Video timestamps are anchored to
* the occasional PES PTS and interpolated using the frame rate reported by the
* decoder; most video packets carry no PTS.</p>
*/
public final class PlaybackEngine implements AutoCloseable {
public static final int MPEG_TIMESTAMP = 90000;
public static final int AUDIO_SAMPLE_RATE = 48000;
public static final int AUDIO_FRAME_SAMPLES = 2048;
public static final long AUDIO_FRAME_TICKS = (long) AUDIO_FRAME_SAMPLES * MPEG_TIMESTAMP / AUDIO_SAMPLE_RATE;
/** A decoded video frame tagged with its presentation timestamp. */
public record VideoFrame(long pts, int width, int height, int pixelFormat, byte[] yPlane, int yStride) {
}
/** A chunk of interleaved S16 PCM tagged with its presentation timestamp. */
public record AudioChunk(long pts, byte[] pcm) {
}
private final MpsDemuxer demuxer;
private final int audioSubstream;
private final VideoDecoder videoDecoder = new VideoDecoder();
private final AnnexBSplitter videoSplitter = new AnnexBSplitter();
private final AudioFramer audioFramer = new AudioFramer();
private Atrac3pDecoder audioDecoder;
private final ArrayDeque<VideoFrame> videoQueue = new ArrayDeque<>();
private final ArrayDeque<AudioChunk> audioQueue = new ArrayDeque<>();
private long pendingVideoPts = -1;
private long nextVideoPts = 0;
private double videoFrameDurationTicks = -1;
private long audioFirstPts = 0;
private boolean audioPtsSet = false;
private long audioFramesEmitted = 0;
private boolean eof = false;
private PlaybackEngine(UmdIsoReader iso, String streamName, int audioSubstream) throws IOException {
this.demuxer = MpsDemuxer.open(iso, streamName);
this.audioSubstream = audioSubstream;
}
/** Opens the given playlist stream for playback, using the given audio substream (0 or 1). */
public static PlaybackEngine open(UmdIsoReader iso, String streamName, int audioSubstream) throws IOException {
return new PlaybackEngine(iso, streamName, audioSubstream);
}
/**
* Decodes until the next video frame is available, or end of stream.
* Returns {@code null} at end of stream.
*/
public VideoFrame nextVideoFrame() throws IOException {
int budget = 1000;
while (videoQueue.isEmpty() && budget-- > 0) {
if (!routeOne()) {
break;
}
}
return videoQueue.poll();
}
/**
* Decodes until the next audio chunk is available, or end of stream.
* Returns {@code null} at end of stream.
*/
public AudioChunk nextAudioChunk() throws IOException {
int budget = 1000;
while (audioQueue.isEmpty() && budget-- > 0) {
if (!routeOne()) {
break;
}
}
return audioQueue.poll();
}
/** True once the stream has been fully decoded. */
public boolean isEof() {
return eof && videoQueue.isEmpty() && audioQueue.isEmpty();
}
/** Reads and routes one packet. Returns false at end of stream. */
private boolean routeOne() throws IOException {
if (eof) {
return false;
}
MpsDemuxer.Packet packet = demuxer.nextPacket();
if (packet == null) {
finish();
eof = true;
return false;
}
if (packet.isVideo()) {
handleVideo(packet);
} else if (packet.isAudio() && packet.channel() == audioSubstream) {
handleAudio(packet);
}
return true;
}
private void handleVideo(MpsDemuxer.Packet packet) {
if (packet.pts() >= 0) {
if (pendingVideoPts < 0) {
nextVideoPts = packet.pts();
}
pendingVideoPts = packet.pts();
}
videoSplitter.feed(packet.payload());
byte[] accessUnit;
while ((accessUnit = videoSplitter.nextAccessUnit()) != null) {
videoDecoder.feedAccessUnit(accessUnit);
VideoDecoder.Frame frame;
while ((frame = videoDecoder.receive()) != null) {
videoQueue.add(new VideoFrame(
assignVideoPts(), frame.width(), frame.height(),
frame.pixelFormat(), frame.yPlane(), frame.yStride()));
}
}
}
private long assignVideoPts() {
long pts;
if (pendingVideoPts >= 0) {
pts = pendingVideoPts;
pendingVideoPts = -1;
} else {
pts = nextVideoPts;
}
nextVideoPts = pts + Math.round(videoFrameDurationTicks());
return pts;
}
private double videoFrameDurationTicks() {
double frameRate = videoDecoder.frameRate();
if (frameRate > 0) {
videoFrameDurationTicks = MPEG_TIMESTAMP / frameRate;
} else if (videoFrameDurationTicks < 0) {
videoFrameDurationTicks = 3003.0; // provisional NTSC default
}
return videoFrameDurationTicks;
}
private void handleAudio(MpsDemuxer.Packet packet) {
if (packet.pts() >= 0) {
if (!audioPtsSet) {
audioFirstPts = packet.pts();
audioPtsSet = true;
}
long expected = audioFirstPts + audioFramesEmitted * AUDIO_FRAME_TICKS;
if (Math.abs(packet.pts() - expected) >= AUDIO_FRAME_TICKS) {
audioFirstPts = packet.pts() - audioFramesEmitted * AUDIO_FRAME_TICKS;
}
}
audioFramer.feed(packet.payload());
byte[] frame;
while ((frame = audioFramer.nextFrame()) != null) {
if (audioDecoder == null) {
audioDecoder = new Atrac3pDecoder(AUDIO_SAMPLE_RATE, 2, Math.max(688, frame.length));
}
long pts = audioFirstPts + audioFramesEmitted * AUDIO_FRAME_TICKS;
audioFramesEmitted++;
audioDecoder.feedFrame(frame);
byte[] pcm;
while ((pcm = audioDecoder.receive()) != null) {
audioQueue.add(new AudioChunk(pts, pcm));
}
}
}
private void finish() {
byte[] tail = videoSplitter.drain();
if (tail != null && tail.length > 0) {
videoDecoder.feedAccessUnit(tail);
}
videoDecoder.flush();
VideoDecoder.Frame frame;
while ((frame = videoDecoder.receive()) != null) {
videoQueue.add(new VideoFrame(
assignVideoPts(), frame.width(), frame.height(),
frame.pixelFormat(), frame.yPlane(), frame.yStride()));
}
if (audioDecoder != null) {
audioDecoder.flush();
byte[] pcm;
while ((pcm = audioDecoder.receive()) != null) {
audioQueue.add(new AudioChunk(
audioFirstPts + audioFramesEmitted * AUDIO_FRAME_TICKS, pcm));
}
}
}
@Override
public void close() {
videoDecoder.close();
if (audioDecoder != null) {
audioDecoder.close();
}
demuxer.close();
}
}
@@ -0,0 +1,47 @@
// SPDX-License-Identifier: GPL-3.0-or-later
package dev.quoije.umd.media;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
import static org.assertj.core.api.Assertions.assertThat;
class AnnexBSplitterTest {
@Test
void splitsCompleteAccessUnitsAndHoldsPartialTail() {
// SPS + slice1 + slice2 (slice2 has no following start code).
byte[] stream = {
0, 0, 0, 1, 0x67, 1, 2, 3, // SPS
0, 0, 0, 1, 0x41, 4, 5, 6, // slice (type 1)
0, 0, 0, 1, 0x41, 7, 8, 9, // slice (type 1), incomplete tail
};
AnnexBSplitter splitter = new AnnexBSplitter();
splitter.feed(stream);
assertThat(splitter.nextAccessUnit()).isEqualTo(Arrays.copyOfRange(stream, 0, 16));
assertThat(splitter.nextAccessUnit()).isNull();
assertThat(splitter.drain()).isEqualTo(Arrays.copyOfRange(stream, 16, 24));
}
@Test
void holdsIncompleteSliceAcrossFeeds() {
byte[] stream = {
0, 0, 0, 1, 0x67, 1, 2, 3,
0, 0, 0, 1, 0x41, 4, 5, 6,
0, 0, 0, 1, 0x41, 7, 8, 9,
};
AnnexBSplitter splitter = new AnnexBSplitter();
splitter.feed(Arrays.copyOfRange(stream, 0, 15));
// The slice NAL's end (the next start code) has not arrived yet.
assertThat(splitter.nextAccessUnit()).isNull();
splitter.feed(Arrays.copyOfRange(stream, 15, stream.length));
assertThat(splitter.nextAccessUnit()).isEqualTo(Arrays.copyOfRange(stream, 0, 16));
assertThat(splitter.nextAccessUnit()).isNull();
assertThat(splitter.drain()).isEqualTo(Arrays.copyOfRange(stream, 16, 24));
}
}