Python 绑定 Rust 媒体云 AI 工作器 SDK

基于 mcai_worker_sdk。

构建

要构建 rust 应用程序

cargo build

测试

要运行单元测试，您必须构建提供的工人示例（请参阅上面的构建部分）。

cargo test

使用方法

此工作器使用 PyO3 crate 加载 Python 文件并执行它。

要实现工作器，必须创建一个包含描述工作器元数据的 pyproject.toml 文件。它至少包含 project 和 build-system 部分。

示例：（最小配置）

[project]
name = "my_python_worker"
version = "1.2.3"
description = "My Python worker"
license = { text = "MIT" }

[build-system]
requires = []

Python 工作器参数必须作为一个 Python 类 的 静态 属性。每个参数类型都必须明确设置。

示例

import typing

class MyWorkerParameters:
    a_parameter: int
    another_parameter: typing.Optional[str] = None

Python 工作器本身必须定义为一个 Python 类，实现一些方法

• get_parameters_type() -> typing.Type (静态)
  • 检索工作器参数类
• init() (静态)
  • 初始化工作器进程（可选）
• process(handle_callback, parameters, job_id) -> dict (静态) 与 parameters 实例的工作器参数类
  • 执行工作进程并返回作业结果。

如果启用了 media 功能，必须实现以下函数

• init_process(stream_handler, format_context, parameters) -> list (静态)，其中 parameters 是工作进程参数类的实例
  • 初始化媒体工作进程并返回一个 GenericStreamDescriptor 的列表
• process_frames(job_id, stream_index, frames) -> dict (静态)
  • 处理一些输入的音频/视频帧并返回作业结果。
• process_ebu_ttml_live(job_id, stream_index, ttml_contents) -> dict (静态)
  • 处理一些输入的 EBU TTML 帧，并返回作业结果。
• ending_process() (静态)
  • 结束媒体工作进程

注意：当启用 media 功能时，不调用 process 函数。

有关更多详细信息，请参阅提供的 worker.py 和 media_worker.py 示例。

设置 PYTHON_WORKER_FILENAME 环境变量以指定您的 Python 工作进程路径。否则，默认加载 worker.py 文件。

运行示例

简单工作进程

RUST_LOG=debug \
SOURCE_ORDERS="examples/message.json" \
PYTHON_WORKER_FILENAME="worker.py" \
SOURCE_PATH="README.md" \
DESTINATION_PATH="README.md.out" \
cargo run

媒体工作进程

首先设置媒体文件名

export SOURCE_PATH="/folder/filename.ext"

然后使用以下参数运行 SDK

RUST_LOG=debug \
SOURCE_ORDERS="examples/message.json" \
PYTHON_WORKER_FILENAME="media_worker.py" \
DESTINATION_PATH="results.json" \
cargo run --features media