主题
Audio Recorder
允许在 Flet 应用中录制音频。
它基于 record Flutter 包。
平台支持
| 平台 | Windows | macOS | Linux | iOS | Android | Web |
|---|---|---|---|---|---|---|
| 支持 | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
用法
要使用 AudioRecorder 服务,请将 flet-audio-recorder 包添加到你的项目依赖中:
uv
bash
uv add flet-audio-recorderpip
bash
pip install flet-audio-recorder [^1][^1]: 之后,你还需要手动把这个包添加到 requirements.txt 或 pyproject.toml 中。
依赖要求
下面几节展示了各平台所需的配置。
Android
访问麦克风所需的配置:
android.permission.RECORD_AUDIO:允许录音。android.permission.MODIFY_AUDIO_SETTINGS(可选):允许使用蓝牙通话设备,例如耳机/耳塞。
bash
flet build apk \
--android-permissions android.permission.RECORD_AUDIO=true \
--android-permissions android.permission.MODIFY_AUDIO_SETTINGS=truetoml
[tool.flet.android.permission]
"android.permission.RECORD_AUDIO" = true
"android.permission.MODIFY_AUDIO_SETTINGS" = trueINFO
保存录音不需要存储权限:写入应用自己的目录(参见 StoragePaths)始终是允许的, 保存到其他位置则应通过 FilePicker.save_file(), 它会使用系统选择器。改为请求 WRITE_EXTERNAL_STORAGE 或 READ_EXTERNAL_STORAGE 可能会导致你的应用被 Google Play 拒绝。
另请参阅:
iOS
访问麦克风所需的配置:
NSMicrophoneUsageDescription:录制音频所必需。
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
访问麦克风所需的配置:
NSMicrophoneUsageDescription:允许录音。com.apple.security.device.audio-input:允许使用内置麦克风录音,并通过 Core Audio 访问音频输入。
bash
flet build macos \
--info-plist NSMicrophoneUsageDescription="Some message to describe why you need this permission..." \
--macos-entitlements com.apple.security.device.audio-input=truetoml
[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 microphonetoml
[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
一个允许你从设备录制音频的控件。
该控件可以使用不同的音频编码器录制音频,还允许 配置各种音频录制参数,例如 噪声抑制、回声消除等。
属性
| 名称 | 类型 | 默认值 | 说明 |
|---|---|---|---|
configuration | AudioRecorderConfiguration | AudioRecorderConfiguration(() | 音频录制器的默认配置。 |
data | Any | skip_ | 任意类型的自定义数据。 |
key | Optional['KeyValue'] | None | 用于在更新之间保持控件标识稳定的键。 |
只读属性
| 名称 | 类型 | 默认值 | 说明 |
|---|---|---|---|
parent | Optional['BaseControl'] | — | 此控件的直接祖先(父级)。 默认为 None,只有在此控件被挂载(加入页面树)后才有值。 Page 控件(页面树的根)是个例外 —— 它的 parent 始终为 None。 |
page | 'Page' | — | 此控件所属的页面。 |
事件
| 名称 | 类型 | 默认值 | 说明 |
|---|---|---|---|
on_state_change | Optional[ft.EventHandler[AudioRecorderStateChangeEvent]] | None | 录音状态改变时调用。 |
on_upload | Optional[ft.EventHandler[AudioRecorderUploadEvent]] | None | 有流式上传进度或错误时调用。 |
on_stream | Optional[ft.EventHandler[AudioRecorderStreamEvent]] | None | 有原始 PCM16BITS 录音分块可用时调用。 |
方法
| 名称 | 类型 | 默认值 | 说明 |
|---|---|---|---|
start_recording | bool | — | 开始录制音频并保存到文件或进行流式传输。 如果既不使用 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_recording | bool | — | 检查音频录制器当前是否正在录音。 返回: 如果录制器正在录音则返回 True,否则返回 False。 |
stop_recording | Optional[str] | — | 停止录音,并可选地返回录音位置。 返回: 保存录音的本地文件路径;在 Web 上为 Blob URL;进行流式传输时(即设置了 upload 或 on_stream 时)为 None。 |
cancel_recording | — | — | 取消当前录音。 |
resume_recording | — | — | 恢复已暂停的录音。 |
pause_recording | — | — | 暂停正在进行的录音。 |
is_paused | bool | — | 检查音频录制器当前是否已暂停。 返回: 如果录制器已暂停则返回 True,否则返回 False。 |
is_supported_encoder | bool | — | 检查录制器是否支持给定的音频编码器。 参数: encoder:要检查的音频编码器。 返回: 如果支持该编码器则返回 True,否则返回 False。 |
get_input_devices | list[InputDevice] | — | 获取可用于录音的输入设备。 返回: 可用输入设备的列表。 |
has_permission | bool | — | 检查应用是否有录音权限,如需要会发起请求。 返回: 如果权限已授予,或在请求后获得授予,则返回 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 始终来自框架触发的事件,所以当此方法运行时,通道在两端都已存在。 |
update | None | — | 为此控件请求一次 UI 更新。 在更改控件状态或属性后调用。控件必须已挂载到页面且未被标记为冻结。 |