Skip to content

Audio Recorder ​

允许在 Flet 应用中录制音频。

它基于 record Flutter 包。

平台支持 ​

平台WindowsmacOSLinuxiOSAndroidWeb
支持✅✅✅✅✅✅

用法 ​

要使用 AudioRecorder 服务,请将 flet-audio-recorder 包添加到你的项目依赖中:

uv

bash
uv add flet-audio-recorder

pip

bash
pip install flet-audio-recorder  [^1]

[^1]: 之后,你还需要手动把这个包添加到 requirements.txt 或 pyproject.toml 中。

依赖要求 ​

下面几节展示了各平台所需的配置。

Android ​

访问麦克风所需的配置:

bash
flet build apk \
  --android-permissions android.permission.RECORD_AUDIO=true \
  --android-permissions android.permission.MODIFY_AUDIO_SETTINGS=true
toml
[tool.flet.android.permission]
"android.permission.RECORD_AUDIO" = true
"android.permission.MODIFY_AUDIO_SETTINGS" = true

INFO

保存录音不需要存储权限:写入应用自己的目录(参见 StoragePaths)始终是允许的, 保存到其他位置则应通过 FilePicker.save_file(), 它会使用系统选择器。改为请求 WRITE_EXTERNAL_STORAGE 或 READ_EXTERNAL_STORAGE 可能会导致你的应用被 Google Play 拒绝。

另请参阅:

iOS ​

访问麦克风所需的配置:

bash
flet build ipa \
  --info-plist NSMicrophoneUsageDescription="Some message to describe why you need this permission..."
toml
[tool.flet.ios.info]
NSMicrophoneUsageDescription = "Some message to describe why you need this permission..."

另请参阅:

macOS ​

访问麦克风所需的配置:

bash
flet build macos \
  --info-plist NSMicrophoneUsageDescription="Some message to describe why you need this permission..." \
  --macos-entitlements com.apple.security.device.audio-input=true
toml
[tool.flet.macos.info]
NSMicrophoneUsageDescription = "Some message to describe why you need this permission..."

[tool.flet.macos.entitlement]
"com.apple.security.device.audio-input" = true

另请参阅:

Linux ​

需要以下依赖(它们通常在你的系统上已可安装):

  • parecord:用于音频输入。
  • pactl:用于获取可用设备等工具方法。
  • ffmpeg:用于编码和输出。

在 Ubuntu 24.04.3 LTS 上,你可以使用以下命令安装它们:

bash
sudo apt install pulseaudio-utils ffmpeg

跨平台 ​

此外,你也可以使用我们预定义的跨平台 microphone权限包:

bash
flet build <target_platform> --permissions microphone
toml
[tool.flet]
permissions = ["microphone"]

示例 ​

python
import flet as ft
import flet_audio_recorder as far


def main(page: ft.Page):
    page.horizontal_alignment = ft.CrossAxisAlignment.CENTER
    page.appbar = ft.AppBar(title=ft.Text("Audio Recorder"), center_title=True)

    path = "test-audio-file.wav"

    def show_snackbar(message):
        page.show_dialog(ft.SnackBar(ft.Text(message)))

    async def handle_recording_start(e: ft.Event[ft.Button]):
        show_snackbar("Starting recording...")
        await recorder.start_recording(output_path=path)

    async def handle_recording_stop(e: ft.Event[ft.Button]):
        output_path = await recorder.stop_recording()
        show_snackbar(f"Stopped recording. Output Path: {output_path}")
        if page.web and output_path is not None:
            await ft.UrlLauncher().launch_url(output_path)

    async def handle_list_devices(e: ft.Event[ft.Button]):
        o = await recorder.get_input_devices()
        show_snackbar(f"Input Devices: {', '.join([f'{d.id}:{d.label}' for d in o])}")

    async def handle_has_permission(e: ft.Event[ft.Button]):
        try:
            status = await recorder.has_permission()
            show_snackbar(f"Audio Recording Permission status: {status}")
        except Exception as e:
            show_snackbar(f"Error checking permission: {e}")

    async def handle_pause(e: ft.Event[ft.Button]):
        print(f"isRecording: {await recorder.is_recording()}")
        if await recorder.is_recording():
            await recorder.pause_recording()

    async def handle_resume(e: ft.Event[ft.Button]):
        print(f"isPaused: {await recorder.is_paused()}")
        if await recorder.is_paused():
            await recorder.resume_recording()

    async def handle_audio_encoder_test(e: ft.Event[ft.Button]):
        print(await recorder.is_supported_encoder(far.AudioEncoder.WAV))

    recorder = far.AudioRecorder(
        configuration=far.AudioRecorderConfiguration(encoder=far.AudioEncoder.WAV),
        on_state_change=lambda e: print(f"State Changed: {e.data}"),
    )

    page.add(
        ft.SafeArea(
            content=ft.Column(
                controls=[
                    ft.Button(
                        content="Start Audio Recorder", on_click=handle_recording_start
                    ),
                    ft.Button(
                        content="Stop Audio Recorder", on_click=handle_recording_stop
                    ),
                    ft.Button(content="List Devices", on_click=handle_list_devices),
                    ft.Button(content="Pause Recording", on_click=handle_pause),
                    ft.Button(content="Resume Recording", on_click=handle_resume),
                    ft.Button(
                        content="WAV Encoder Support",
                        on_click=handle_audio_encoder_test,
                    ),
                    ft.Button(
                        content="Get Audio Recording Permission Status",
                        on_click=handle_has_permission,
                    ),
                ],
            ),
        )
    )


if __name__ == "__main__":
    ft.run(main)

流式分块并保存/下载 {#stream-chunks-and-save-download} ​

在 Web 上,[AudioRecorder.stop_recording()][flet_audio_recorder.AudioRecorder.stop_recording] 返回浏览器本地的 Blob URL。当你的应用需要访问录制的字节数据时,请使用流式传输。

将 [AudioRecorderConfiguration.encoder][flet_audio_recorder.AudioRecorderConfiguration.encoder] 设置为 [AudioEncoder.PCM16BITS][flet_audio_recorder.AudioEncoder.PCM16BITS], 并处理 [AudioRecorder.on_stream][flet_audio_recorder.AudioRecorder.on_stream], 以便在 Python 中接收 [AudioRecorderStreamEvent.chunk][flet_audio_recorder.AudioRecorderStreamEvent.chunk] 字节。

[AudioEncoder.PCM16BITS][flet_audio_recorder.AudioEncoder.PCM16BITS] 编码的流在所有平台上都受支持。流分块是原始的 PCM16 字节,不能直接作为音频文件 播放。当目标位置需要可直接播放的录音时,请在 Python 中把这些字节封装到 WAV 之类的 容器里。

下面的示例会收集流式传输的分块,将 PCM16 字节封装到 WAV 容器中, 并把生成的字节传给 [FilePicker.save_file()][flet.FilePicker.save_file], 让用户可以保存或下载录音。

python
import io
import time
import wave

import flet as ft
import flet_audio_recorder as far

SAMPLE_RATE = 44100
CHANNELS = 1
BYTES_PER_SAMPLE = 2


def pcm_to_wav_bytes(pcm_data: bytes) -> bytes:
    wav_buffer = io.BytesIO()
    with wave.open(wav_buffer, "wb") as wav:
        wav.setnchannels(CHANNELS)
        wav.setsampwidth(BYTES_PER_SAMPLE)
        wav.setframerate(SAMPLE_RATE)
        wav.writeframes(pcm_data)
    return wav_buffer.getvalue()


def main(page: ft.Page):
    page.horizontal_alignment = ft.CrossAxisAlignment.CENTER

    buffer = bytearray()

    def show_snackbar(message: str):
        page.show_dialog(ft.SnackBar(content=message, duration=ft.Duration(seconds=5)))

    def handle_stream(e: far.AudioRecorderStreamEvent):
        buffer.extend(e.chunk)
        status.value = (
            f"Streaming chunk {e.sequence}; {e.bytes_streamed} bytes collected."
        )

    async def handle_recording_start(e: ft.Event[ft.Button]):
        if not await recorder.has_permission():
            show_snackbar("Microphone permission is required.")
            return

        buffer.clear()
        status.value = "Recording..."
        await recorder.start_recording(
            configuration=far.AudioRecorderConfiguration(
                encoder=far.AudioEncoder.PCM16BITS,
                sample_rate=SAMPLE_RATE,
                channels=CHANNELS,
            ),
        )

    async def handle_recording_stop(e: ft.Event[ft.Button]):
        await recorder.stop_recording()
        if not buffer:
            show_snackbar("Nothing was recorded.")
            return

        wav_bytes = pcm_to_wav_bytes(bytes(buffer))
        file_name = f"recording-{int(time.time())}.wav"
        file_path = await ft.FilePicker().save_file(
            file_name=file_name,
            file_type=ft.FilePickerFileType.CUSTOM,
            allowed_extensions=["wav"],
            src_bytes=wav_bytes,
        )

        if file_path and not (page.web or page.platform.is_mobile()):
            with open(file_path, "wb") as output:
                output.write(wav_bytes)

        if page.web:
            status.value = f"Downloaded {file_name} ({len(wav_bytes)} bytes)."
        else:
            status.value = f"Saved to: {file_path}" if file_path else "Save cancelled."
        show_snackbar(status.value)

    recorder = far.AudioRecorder(on_stream=handle_stream)

    page.add(
        ft.SafeArea(
            content=ft.Column(
                horizontal_alignment=ft.CrossAxisAlignment.CENTER,
                controls=[
                    ft.Text(
                        "Stream PCM16 audio chunks and save or download "
                        "them as a WAV file."
                    ),
                    ft.Button("Start recording", on_click=handle_recording_start),
                    ft.Button("Stop and save", on_click=handle_recording_stop),
                    status := ft.Text(),
                ],
            ),
        )
    )


if __name__ == "__main__":
    ft.run(main)

流式上传 ​

将 [AudioRecorderUploadSettings][flet_audio_recorder.AudioRecorderUploadSettings] 传给 [AudioRecorder.start_recording()][flet_audio_recorder.AudioRecorder.start_recording], 即可在录音过程中直接上传 [AudioEncoder.PCM16BITS][flet_audio_recorder.AudioEncoder.PCM16BITS] 录音字节。

上传的文件包含原始 PCM16 字节,因此使用 .pcm 扩展名是有意为之。 如果你需要创建可播放的 WAV 文件,可以参考 流式分块并保存/下载示例。

INFO

来自 [Page.get_upload_url()][flet.Page.get_upload_url] 的内置上传 URL 需要上传存储和签名密钥。请使用 upload_dir 运行,并设置 FLET_SECRET_KEY。

python
import time

import flet as ft
import flet_audio_recorder as far


def main(page: ft.Page):
    page.horizontal_alignment = ft.CrossAxisAlignment.CENTER

    def show_snackbar(message: str):
        page.show_dialog(ft.SnackBar(content=message, duration=ft.Duration(seconds=5)))

    def handle_upload(e: far.AudioRecorderUploadEvent):
        if e.error:
            status.value = f"Upload error: {e.error}"
        elif e.progress == 1:
            status.value = f"Upload complete: {e.bytes_uploaded or 0} bytes."
        else:
            status.value = f"Uploading: {e.bytes_uploaded or 0} bytes sent."

    async def handle_recording_start(e: ft.Event[ft.Button]):
        if not await recorder.has_permission():
            show_snackbar("Microphone permission is required.")
            return

        file_name = f"recordings/rec-{int(time.time())}.pcm"
        try:
            upload_url = page.get_upload_url(file_name=file_name, expires=600)
        except RuntimeError as ex:
            if "FLET_SECRET_KEY" not in str(ex):
                raise
            status.value = (
                "Uploads require a secret key. "
                "Set FLET_SECRET_KEY before running this app."
            )
            show_snackbar(status.value)
            return

        status.value = "Recording..."
        await recorder.start_recording(
            upload=far.AudioRecorderUploadSettings(
                upload_url=upload_url,
                file_name=file_name,
            ),
            configuration=far.AudioRecorderConfiguration(
                encoder=far.AudioEncoder.PCM16BITS,
                channels=1,
            ),
        )

    async def handle_recording_stop(e: ft.Event[ft.Button]):
        await recorder.stop_recording()
        show_snackbar(
            "Recording stopped. See 'uploads/recordings' folder for the recorded file."
        )

    recorder = far.AudioRecorder(on_upload=handle_upload)

    page.add(
        ft.SafeArea(
            content=ft.Column(
                horizontal_alignment=ft.CrossAxisAlignment.CENTER,
                controls=[
                    ft.Text("Record PCM16 audio and upload it as it streams."),
                    ft.Button("Start upload", on_click=handle_recording_start),
                    ft.Button("Stop recording", on_click=handle_recording_stop),
                    status := ft.Text(),
                ],
            ),
        )
    )


if __name__ == "__main__":
    ft.run(
        main,
        upload_dir="uploads",
    )

说明 ​

AudioRecorder 继承自:ft.Service, BaseControl

一个允许你从设备录制音频的控件。

该控件可以使用不同的音频编码器录制音频,还允许 配置各种音频录制参数,例如 噪声抑制、回声消除等。

属性 ​

名称类型默认值说明
configurationAudioRecorderConfigurationAudioRecorderConfiguration(()音频录制器的默认配置。
dataAnyskip_任意类型的自定义数据。
keyOptional['KeyValue']None用于在更新之间保持控件标识稳定的键。

只读属性 ​

名称类型默认值说明
parentOptional['BaseControl']—此控件的直接祖先(父级)。 默认为 None,只有在此控件被挂载(加入页面树)后才有值。 Page 控件(页面树的根)是个例外 —— 它的 parent 始终为 None。
page'Page'—此控件所属的页面。

事件 ​

名称类型默认值说明
on_state_changeOptional[ft.EventHandler[AudioRecorderStateChangeEvent]]None录音状态改变时调用。
on_uploadOptional[ft.EventHandler[AudioRecorderUploadEvent]]None有流式上传进度或错误时调用。
on_streamOptional[ft.EventHandler[AudioRecorderStreamEvent]]None有原始 PCM16BITS 录音分块可用时调用。

方法 ​

名称类型默认值说明
start_recordingbool—开始录制音频并保存到文件或进行流式传输。 如果既不使用 upload 也不使用 on_stream,则在 Web 以外的平台上必须提供 output_path。 进行流式传输时,请使用 PCM16BITS 作为编码器。此时,发出或上传的 chunk 包含原始 PCM16 数据。在某些用例中,如果输出必须能直接作为音频文件播放,可以把这些分块封装到 WAV 之类的容器中。 参数: output_path:音频保存的文件路径。如果不是在 Web 上,则必须指定。 configuration:音频录制器的配置。如果为 None,将使用 configuration。 upload:上传设置,用于把录音字节直接流式传输到某个目标,例如由 get_upload_url 返回的 URL。 返回: 如果录音成功开始则返回 True,否则返回 False。 抛出: ValueError:如果既未请求流式传输也未请求上传,且非 Web 平台上未提供 output_path。 ValueError:如果请求流式传输时使用的编码器不是 PCM16BITS。
is_recordingbool—检查音频录制器当前是否正在录音。 返回: 如果录制器正在录音则返回 True,否则返回 False。
stop_recordingOptional[str]—停止录音,并可选地返回录音位置。 返回: 保存录音的本地文件路径;在 Web 上为 Blob URL;进行流式传输时(即设置了 upload 或 on_stream 时)为 None。
cancel_recording——取消当前录音。
resume_recording——恢复已暂停的录音。
pause_recording——暂停正在进行的录音。
is_pausedbool—检查音频录制器当前是否已暂停。 返回: 如果录制器已暂停则返回 True,否则返回 False。
is_supported_encoderbool—检查录制器是否支持给定的音频编码器。 参数: encoder:要检查的音频编码器。 返回: 如果支持该编码器则返回 True,否则返回 False。
get_input_deviceslist[InputDevice]—获取可用于录音的输入设备。 返回: 可用输入设备的列表。
has_permissionbool—检查应用是否有录音权限,如需要会发起请求。 返回: 如果权限已授予,或在请求后获得授予,则返回 True;否则返回 False。
init———
is_isolated——返回此控件是否被标记为隔离。 隔离控件会被排除在父级驱动的更新遍历之外,并需要自行管理更新边界。
build——在控件初始化期间调用一次,用于定义其子控件。在此方法中 page 属性可用。
before_update——每当此控件被更新时都会调用此方法。 注意: 务必不要在此处调用或请求 update()。
before_event——在事件处理器执行之前拦截事件。 返回 False 可取消分发。返回 True 或 None 则继续正常的事件处理。
did_mount——在控件挂载到页面树之后调用。 重写此方法以启动需要已挂载页面的资源,例如订阅、定时器或服务监听器。
will_unmount——在控件从页面树移除之前调用。 重写此方法以释放 did_mount() 中创建的资源,例如订阅、定时器或外部句柄。
get_data_channel——解析为此 widget 在 Dart 侧分配的 [DataChannel]。模式: on_data_channel_open: Optional[ft.EventHandler[DataChannelOpenEvent]] = None def init(self): self.on_data_channel_open = self._on_open def _on_open(self, e): self._channel = self.get_data_channel(e.channel_id) 幂等 —— 底层的 Connection 会按 id 缓存 DataChannel,因此重复调用会返回同一实例。没有错误路径:该 id 始终来自框架触发的事件,所以当此方法运行时,通道在两端都已存在。
updateNone—为此控件请求一次 UI 更新。 在更改控件状态或属性后调用。控件必须已挂载到页面且未被标记为冻结。

关注我,学习大模型、大数据、Python等热门技术。