mpeg2video をライブ再生させたい

ffmpeg.sh

ffmpeg -f mpegts -i - -map 0:v:0 -map 0:a:0 -c:v copy -c:a copy \
  -fflags +nobuffer -flags +low_delay -max_delay 0 -f mpegts -

index.html

<video id="video" autoplay controls></video>

<script src="./decoder_wasm/libffmpeg.js"></script>

<script>
  (async () => {
    await new Promise((resolve) => setTimeout(resolve, 1000)); // なんか初期化に時間がかかるので待つ

    const fetcher = new Worker('worker-extractor.js');
    const decoder = new Worker('worker-decoder.js');

    const video = document.getElementById('video');
  
    const videoTrackGenerator = new MediaStreamTrackGenerator({ kind: 'video' });
    const audioTrackGenerator = new MediaStreamTrackGenerator({ kind: 'audio' });
  
    const videoTrackGeneratorWriter = videoTrackGenerator.writable.getWriter();
    const audioTrackGeneratorWriter = audioTrackGenerator.writable.getWriter();

    const mediaStream = new MediaStream();
    mediaStream.addTrack(videoTrackGenerator);
    mediaStream.addTrack(audioTrackGenerator);

    const videoCallback = Module.addFunction((addr_y, addr_u, addr_v, stride_y, stride_u, stride_v, width, height, pts) => {
      const out_y = HEAP8.subarray(addr_y, addr_y + stride_y * height);
      const out_u = HEAP8.subarray(addr_u, addr_u + (stride_u * height) / 2);
      const out_v = HEAP8.subarray(addr_v, addr_v + (stride_v * height) / 2);
      const buf_y = new Uint8Array(out_y);
      const buf_u = new Uint8Array(out_u);
      const buf_v = new Uint8Array(out_v);
      const data = new Uint8Array(buf_y.byteLength + buf_u.byteLength + buf_v.byteLength);
      data.set(buf_y, 0);
      data.set(buf_u, buf_y.byteLength);
      data.set(buf_v, buf_y.byteLength + buf_u.byteLength);

      const videoFrame = new VideoFrame(data, {
        format: 'I420',
        codedWidth: width,
        codedHeight: height,
        timestamp: pts / 90 * 1000,
      })
      videoTrackGeneratorWriter.write(videoFrame);
      videoFrame.close();
    }, 'viiiiiiiii');

    const decoder_result = Module._openDecoder(2, videoCallback, 3);
  
    fetcher.onmessage = (e) => {
      if (e.data.type === 'video') {
        const { data,  pts } = e.data.data;

        const cacheBuffer = Module._malloc(data.byteLength);
        Module.HEAP8.set(data, cacheBuffer);
        Module._decodeData(cacheBuffer, data.byteLength, pts);
        Module._free(cacheBuffer);
      } else if (e.data.type === 'audio') {
        const encodedAudioChunk = e.data.encodedAudioChunk;
        decoder.postMessage({ type: 'audio', encodedAudioChunk });
      }
    };

    decoder.onmessage = (e) => {
      if (e.data.type === 'video') {
        // pass
      } else if(e.data.type === 'audio') {
        const audioFrame = e.data.audioFrame;
        audioTrackGeneratorWriter.write(audioFrame);
        audioFrame.close();
      }
    }
  
    video.srcObject = mediaStream;
  })();
  </script>

live.js

const { Writable } = require('stream');
const express = require('express');
const cors = require('cors');

const port = 3000;

const app = express();

app.use(cors());

peocess.stdin.pipe(new Wriable({
  write(chunk, encoding, callback) { callback(); }
}));

app.get('/live', (req, res, next) => {
  process.stdin.pipe(res);
});

app.listen(port);

MPEG2VIDEOをライブ再生したい.md

これは何?

Chrome M94 で追加された以下のAPIを使った TS (mpeg2video + aac) の再生実験

• WebCodecs
• Insertable Stream for MediaStreamTrack

mpeg2video のデコードは別途、mpeg2videoをデコードできる decoder_wasm を利用した.

• https://github.com/tsukumijima/decoder_wasm/tree/mpeg2video/dist.
  この dist にあるものは decoder_wasm ディレクトリを掘って参照している。

使い方

max_delay はリップシンクのため 0 推奨, データストリームは入れない事を推奨で、live.js に ffmpeg.sh の内容を読み込ませる。
それをクライアント側で参照すると、TSを WebRTC と同レベルの低遅延で再生できる。　それだけ。

仕組み

• TS の demux した PES パケットを WebCodecs で追加されたデコーダに渡すと VideoFrame/AudioFrame が手に入る
• Insertable Stream for MediaStreamTrack で追加された API に渡すと VideoFrame/AudioFrame が再生される
• VideoFrame として I420 をそのまま入れられるので YUV-RGB 変換をスキップして渡せる

制限

• MediaStreamTrack は timestamp での同期を取らないので、リップシンクは自分で合わせる必要がある
• 受信chunkが1フレームずつではない場合の平滑化を行うなら、独自に実装する必要がある
• ソフトウェアデコーダーなので少しCPUを使う CPU

server.js

const express = require('express')
const cors = require('cors')
const app = express()

const port = 3000;

app.use(express.static('./', {
  setHeaders: function(res, path) {
    res.setHeader('Cross-Origin-Embedder-Policy', 'require-corp');
    res.setHeader('Cross-Origin-Opener-Policy', 'same-origin');
  }
}));

app.listen(port, () => {
  console.log(`Live Streamer listening at http://localhost:${port}`)
})

worker-decoder.js

(async () => {
  const videoDecoder = new VideoDecoder({
    output: (videoFrame) => {
      self.postMessage({ type: 'video', videoFrame });
      videoFrame.close();
    },
    error: () => {},
  })
  await videoDecoder.configure({
    codec: 'avc1.64001f',
    hardwareAcceleration: "prefer-hardware",
  });

  const audioDecoder = new AudioDecoder({
    output: (audioFrame) => {
      self.postMessage({ type: 'audio', audioFrame });
      audioFrame.close();
    },
    error: () => {},
  });
  await audioDecoder.configure({
    codec: 'mp4a.40.2',
    sampleRate: 48000,
    numberOfChannels: 2,
  });

  self.onmessage = (e) => {
    if (e.data.type === 'video') {
      const encodedVideoChunk = e.data.encodedVideoChunk;
      videoDecoder.decode(encodedVideoChunk);
    } else if (e.data.type === 'audio') {
      const encodedAudioChunk = e.data.encodedAudioChunk;
      audioDecoder.decode(encodedAudioChunk);
    }
  }
})();

worker-extractor.js

const liveUrl = `${配信URL}`;

(async () => {
  let remains = Uint8Array.from([]);
  
  let videoPESTotalLength = 0;
  let videoPESList = [];

  let audioPESTotalLength = 0;
  let audioPESList = [];

  const decodeVideo = new WritableStream({
    write(stream) {
      const chunk = new Uint8Array(remains.byteLength + stream.byteLength);
      chunk.set(remains, 0);
      chunk.set(stream, remains.byteLength);

      for (let index = 0; index < chunk.length;) {
        if (chunk[index] === 0x47 && index + 188 >= chunk.length) {
          remains = chunk.slice(index);
          break;
        } else if(chunk[index] !== 0x47) {
          index++;
          continue;
        }

        const pid = ((chunk[index + 1] & 0x1F) << 8) | chunk[index + 2];
        if (pid !== 0x100 && pid !== 0x101) { 
          index += 188;
          continue;
        }

        const end = index + 188;
        let payload_start = index + 4;

        const payload_unit_start_indicator = (chunk[index + 1] & 0x40) >>> 6;
        const adaptation_field_control = (chunk[index + 3] & 0x30) >>> 4;
        if (adaptation_field_control === 0x02 || adaptation_field_control === 0x03) {
          payload_start += 1 + chunk[index + 4];
        }

        if (pid === 0x100){
          if (payload_unit_start_indicator && videoPESTotalLength > 0) {
            const videoPES = new Uint8Array(videoPESTotalLength);
            for (let i = 0, offset = 0; i < videoPESList.length; i++) {
              videoPES.set(videoPESList[i], offset);
              offset += videoPESList[i].byteLength;
            }
            videoPESTotalLength = 0;
            videoPESList = [];

            const PTS_DTS_flags = (videoPES[7] & 0xC0) >>> 6;
            const PES_header_data_length = videoPES[8];
            const payload_start_index = 6 + 3 + PES_header_data_length;
            const data = videoPES.slice(payload_start_index);

            let pts = 0;
            if (PTS_DTS_flags === 0x02 || PTS_DTS_flags === 0x03) {
                pts = (videoPES[9] & 0x0E) * 536870912 + // 1 << 29
                        (videoPES[10] & 0xFF) * 4194304 + // 1 << 22
                        (videoPES[11] & 0xFE) * 16384 + // 1 << 14
                        (videoPES[12] & 0xFF) * 128 + // 1 << 7
                        (videoPES[13] & 0xFE) / 2;
            }

            self.postMessage({ type: 'video', data: {
              data: data,
              pts: pts
            }});
          } else if (!payload_unit_start_indicator && videoPESTotalLength === 0){
            index += 188;
            continue;
          }
          
          videoPESList.push(chunk.slice(payload_start, end));
          videoPESTotalLength += end - payload_start;
          index += 188;
        } else if (pid === 0x101){
          if (payload_unit_start_indicator && audioPESTotalLength > 0) {
            const audioPES = new Uint8Array(audioPESTotalLength);
            for (let i = 0, offset = 0; i < audioPESList.length; i++) {
              audioPES.set(audioPESList[i], offset);
              offset += audioPESList[i].byteLength;
            }
            audioPESTotalLength = 0;
            audioPESList = [];

            const PTS_DTS_flags = (audioPES[7] & 0xC0) >>> 6;
            const PES_header_data_length = audioPES[8];
            const payload_start_index = 6 + 3 + PES_header_data_length;
            const data = audioPES.slice(payload_start_index);

            let pts = 0;
            if (PTS_DTS_flags === 0x02 || PTS_DTS_flags === 0x03) {
                pts = (audioPES[9] & 0x0E) * 536870912 + // 1 << 29
                        (audioPES[10] & 0xFF) * 4194304 + // 1 << 22
                        (audioPES[11] & 0xFE) * 16384 + // 1 << 14
                        (audioPES[12] & 0xFF) * 128 + // 1 << 7
                        (audioPES[13] & 0xFE) / 2;
            }

            const encodedAudioChunk = new EncodedAudioChunk({
              type: 'key',
              timestamp: pts / 90 * 1000,
              data: data,
            });
            self.postMessage({ type: 'audio', encodedAudioChunk });
          } else if (!payload_unit_start_indicator && audioPESTotalLength === 0){
            index += 188;
            continue;
          }

          audioPESList.push(chunk.slice(payload_start, end));
          audioPESTotalLength += end - payload_start;
          index += 188;
        }
      }
    }
  }, {});

  const response = await fetch(liveUrl)
  response.body.pipeTo(decodeVideo);
})();