brianyin/ajs-310-play-local-audio-file-utility (#788)

Author: toubatbrian

.changeset/hot-taxis-judge.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Add utility to play local audio file to livekit

REUSE.toml

@@ -36,3 +36,9 @@ SPDX-License-Identifier = "CC-BY-NC-SA-4.0"
 path = ["**/.gitattributes", "**.wav", "**/__snapshots__/**"]
 SPDX-FileCopyrightText = "2024 LiveKit, Inc."
 SPDX-License-Identifier = "Apache-2.0"
+
+# audio resources
+[[annotations]]
+path = ["agents/resources/*.ogg", "agents/resources/NOTICE"]
+SPDX-FileCopyrightText = "2024 LiveKit, Inc."
+SPDX-License-Identifier = "Apache-2.0"

agents/package.json

@@ -34,11 +34,14 @@
     "api:update": "api-extractor run --local --typescript-compiler-folder ../node_modules/typescript --verbose"
   },
   "devDependencies": {
-    "@livekit/rtc-node": "^0.13.12",
+    "@ffmpeg-installer/ffmpeg": "^1.1.0",
+    "@livekit/rtc-node": "^0.13.13",
     "@microsoft/api-extractor": "^7.35.0",
+    "@types/fluent-ffmpeg": "^2.1.28",
     "@types/json-schema": "^7.0.15",
     "@types/node": "^22.5.5",
     "@types/ws": "^8.5.10",
+    "fluent-ffmpeg": "^2.1.3",
     "tsup": "^8.4.0",
     "typescript": "^5.0.0"
   },
@@ -50,8 +53,8 @@
     "commander": "^12.0.0",
     "heap-js": "^2.6.0",
     "json-schema": "^0.4.0",
-    "openai": "^4.91.1",
     "livekit-server-sdk": "^2.13.3",
+    "openai": "^4.91.1",
     "pidusage": "^4.0.1",
     "pino": "^8.19.0",
     "pino-pretty": "^11.0.0",

agents/resources/NOTICE

@@ -0,0 +1,2 @@
+keyboard-typing.ogg by Anton -- https://freesound.org/s/137/ -- License: Attribution 4.0
+keyboard-typing2.opg by Anton -- https://freesound.org/s/137/ -- License: Attribution 4.0

agents/resources/keyboard-typing.ogg

@@ -1,10 +1,27 @@
 // SPDX-FileCopyrightText: 2024 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
+import ffmpegInstaller from '@ffmpeg-installer/ffmpeg';
 import { AudioFrame } from '@livekit/rtc-node';
+import ffmpeg from 'fluent-ffmpeg';
+import type { ReadableStream } from 'node:stream/web';
 import { log } from './log.js';
+import { createStreamChannel } from './stream/stream_channel.js';
 import type { AudioBuffer } from './utils.js';
 
+ffmpeg.setFfmpegPath(ffmpegInstaller.path);
+
+export interface AudioDecodeOptions {
+  sampleRate?: number;
+  numChannels?: number;
+  /**
+   * Audio format hint (e.g., 'mp3', 'ogg', 'wav', 'opus')
+   * If not provided, FFmpeg will auto-detect
+   */
+  format?: string;
+  abortSignal?: AbortSignal;
+}
+
 export function calculateAudioDurationSeconds(frame: AudioBuffer) {
   // TODO(AJS-102): use frame.durationMs once available in rtc-node
   return Array.isArray(frame)
@@ -72,3 +89,117 @@ export class AudioByteStream {
     return frames;
   }
 }
+
+/**
+ * Decode an audio file into AudioFrame instances
+ *
+ * @param filePath - Path to the audio file
+ * @param options - Decoding options
+ * @returns AsyncGenerator that yields AudioFrame objects
+ *
+ * @example
+ * ```typescript
+ * for await (const frame of audioFramesFromFile('audio.ogg', { sampleRate: 48000 })) {
+ *   console.log('Frame:', frame.samplesPerChannel, 'samples');
+ * }
+ * ```
+ */
+export function audioFramesFromFile(
+  filePath: string,
+  options: AudioDecodeOptions = {},
+): ReadableStream<AudioFrame> {
+  const sampleRate = options.sampleRate ?? 48000;
+  const numChannels = options.numChannels ?? 1;
+
+  const audioStream = new AudioByteStream(sampleRate, numChannels);
+  const channel = createStreamChannel<AudioFrame>();
+  const logger = log();
+
+  // TODO (Brian): decode WAV using a custom decoder instead of FFmpeg
+  const command = ffmpeg(filePath)
+    .inputOptions([
+      '-probesize',
+      '32',
+      '-analyzeduration',
+      '0',
+      '-fflags',
+      '+nobuffer+flush_packets',
+      '-flags',
+      'low_delay',
+    ])
+    .format('s16le') // signed 16-bit little-endian PCM to be consistent cross-platform
+    .audioChannels(numChannels)
+    .audioFrequency(sampleRate);
+
+  let commandRunning = true;
+
+  const onClose = () => {
+    logger.debug('Audio file playback aborted');
+
+    channel.close();
+    if (commandRunning) {
+      commandRunning = false;
+      command.kill('SIGKILL');
+    }
+  };
+
+  const outputStream = command.pipe();
+  options.abortSignal?.addEventListener('abort', onClose, { once: true });
+
+  outputStream.on('data', (chunk: Buffer) => {
+    const arrayBuffer = chunk.buffer.slice(
+      chunk.byteOffset,
+      chunk.byteOffset + chunk.byteLength,
+    ) as ArrayBuffer;
+
+    const frames = audioStream.write(arrayBuffer);
+    for (const frame of frames) {
+      channel.write(frame);
+    }
+  });
+
+  outputStream.on('end', () => {
+    const frames = audioStream.flush();
+    for (const frame of frames) {
+      channel.write(frame);
+    }
+    commandRunning = false;
+    channel.close();
+  });
+
+  outputStream.on('error', (err: Error) => {
+    logger.error(err);
+    commandRunning = false;
+    onClose();
+  });
+
+  return channel.stream();
+}
+
+/**
+ * Loop audio frames from a file indefinitely
+ *
+ * @param filePath - Path to the audio file
+ * @param options - Decoding options
+ * @returns AsyncGenerator that yields AudioFrame objects in an infinite loop
+ */
+export async function* loopAudioFramesFromFile(
+  filePath: string,
+  options: AudioDecodeOptions = {},
+): AsyncGenerator<AudioFrame, void, unknown> {
+  const frames: AudioFrame[] = [];
+  const logger = log();
+
+  for await (const frame of audioFramesFromFile(filePath, options)) {
+    frames.push(frame);
+    yield frame;
+  }
+
+  while (!options.abortSignal?.aborted) {
+    for (const frame of frames) {
+      yield frame;
+    }
+  }
+
+  logger.debug('Audio file playback loop finished');
+}

agents/resources/keyboard-typing2.ogg

@@ -126,4 +126,41 @@ describe('StreamChannel', () => {
     const nextResult = await reader.read();
     expect(nextResult.done).toBe(true);
   });
+
+  it('should gracefully handle close while read is pending', async () => {
+    const channel = createStreamChannel<string>();
+    const reader = channel.stream().getReader();
+
+    const readPromise = reader.read();
+
+    await channel.close();
+
+    const result = await readPromise;
+    expect(result.done).toBe(true);
+    expect(result.value).toBeUndefined();
+  });
+
+  it('should complete all pending reads when closed', async () => {
+    const channel = createStreamChannel<number>();
+    const reader = channel.stream().getReader();
+
+    const read1 = reader.read();
+    const read2 = reader.read();
+    const read3 = reader.read();
+
+    await channel.write(42);
+    await channel.write(43);
+    await channel.close();
+
+    const result1 = await read1;
+    expect(result1.done).toBe(false);
+    expect(result1.value).toBe(42);
+
+    const result2 = await read2;
+    expect(result2.done).toBe(false);
+    expect(result2.value).toBe(43);
+
+    const result3 = await read3;
+    expect(result3.done).toBe(true);
+  });
 });

agents/resources/office-ambience.ogg

@@ -0,0 +1,67 @@
+// SPDX-FileCopyrightText: 2025 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import {
+  type JobContext,
+  WorkerOptions,
+  cli,
+  defineAgent,
+  log,
+  loopAudioFramesFromFile,
+} from '@livekit/agents';
+import { AudioSource, LocalAudioTrack, TrackPublishOptions, TrackSource } from '@livekit/rtc-node';
+import { dirname, join } from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+export default defineAgent({
+  entry: async (ctx: JobContext) => {
+    const logger = log();
+
+    await ctx.connect();
+
+    logger.info('Playing audio file to LiveKit track...');
+
+    const audioSource = new AudioSource(48000, 1);
+
+    const track = LocalAudioTrack.createAudioTrack('background_audio', audioSource);
+
+    const publication = await ctx.room.localParticipant!.publishTrack(
+      track,
+      new TrackPublishOptions({
+        source: TrackSource.SOURCE_MICROPHONE,
+      }),
+    );
+
+    await publication.waitForSubscription();
+
+    logger.info(`Audio track published: ${publication?.sid}`);
+
+    const currentDir = dirname(fileURLToPath(import.meta.url));
+    const resourcesPath = join(currentDir, '../../agents/resources');
+    const audioFile = join(resourcesPath, 'office-ambience.ogg');
+
+    logger.info(`Playing: ${audioFile}`);
+
+    const abortController = new AbortController();
+
+    ctx.addShutdownCallback(async () => {
+      abortController.abort();
+    });
+
+    let frameCount = 0;
+    for await (const frame of loopAudioFramesFromFile(audioFile, {
+      sampleRate: 48000,
+      numChannels: 1,
+      abortSignal: abortController.signal,
+    })) {
+      await audioSource.captureFrame(frame);
+      frameCount++;
+
+      if (frameCount % 100 === 0) {
+        logger.info(`Played ${frameCount} frames (${(frameCount * 0.1).toFixed(1)}s)`);
+      }
+    }
+  },
+});
+
+cli.runApp(new WorkerOptions({ agent: fileURLToPath(import.meta.url) }));