-
Notifications
You must be signed in to change notification settings - Fork 172
brianyin/ajs-310-play-local-audio-file-utility #788
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 8 commits
602b54a
9691e67
2559754
15bc8a1
7b3aeb8
49a3a93
cd01c43
2e69045
d2d6774
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| --- | ||
| '@livekit/agents': patch | ||
| --- | ||
|
|
||
| Add utility to play local audio file to livekit |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,6 @@ | ||
| // SPDX-FileCopyrightText: 2025 LiveKit, Inc. | ||
| // | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| import { audioFramesFromFile, loopAudioFramesFromFile } from './utils.js'; | ||
|
|
||
| export { audioFramesFromFile, loopAudioFramesFromFile }; |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,137 @@ | ||
| // SPDX-FileCopyrightText: 2025 LiveKit, Inc. | ||
| // | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| import ffmpegInstaller from '@ffmpeg-installer/ffmpeg'; | ||
| import type { AudioFrame } from '@livekit/rtc-node'; | ||
| import ffmpeg from 'fluent-ffmpeg'; | ||
| import type { ReadableStream } from 'node:stream/web'; | ||
| import { AudioByteStream } from '../audio.js'; | ||
| import { log } from '../log.js'; | ||
| import { createStreamChannel } from '../stream/stream_channel.js'; | ||
|
|
||
| ffmpeg.setFfmpegPath(ffmpegInstaller.path); | ||
|
|
||
| export interface AudioStreamDecoderOptions { | ||
| sampleRate?: number; | ||
| numChannels?: number; | ||
| /** | ||
| * Audio format hint (e.g., 'mp3', 'ogg', 'wav', 'opus') | ||
| * If not provided, FFmpeg will auto-detect | ||
| */ | ||
| format?: string; | ||
| abortSignal?: AbortSignal; | ||
| } | ||
|
|
||
| /** | ||
| * 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: AudioStreamDecoderOptions = {}, | ||
| ): 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: AudioStreamDecoderOptions = {}, | ||
| ): 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'); | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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(); | ||
| }); | ||
|
Comment on lines
+130
to
+141
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. add tests to make sure buffered read after close still reads correct values |
||
|
|
||
| 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); | ||
| }); | ||
| }); | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| // SPDX-FileCopyrightText: 2025 LiveKit, Inc. | ||
| // | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| import { type JobContext, WorkerOptions, cli, codecs, defineAgent, log } 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 codecs.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) })); | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
same set of flags as in python