Python API 说明

TiRTC Python SDK 的 RTC API 位于根 tirtc package。Client 是服务调用方和本地资源树的 owner，可以创建多个 Connection；Output 由应用独立创建和持有。

完整调用流程见 Python SDK 接入和 Python RTC API Sample。

Stream ID 约定

本页所有 stream_id 参数按同一套规则使用：取值范围是 0..15，标识同一条连接、一个发送方向中的一路流。同一方向、同一路流的设备端发送编号，必须与 Python Output 绑定、订阅或接收配置一致；同一发送方向内，不同音视频流使用不同编号，音频和视频不能共用编号。相反方向使用独立的编号空间，可以复用相同数值。

配套 API Samples 和 DevTools CLI 使用 10 表示设备端音频、11 表示设备端视频、14 表示客户端对讲音频。这些只是便于直接联调的示例值，不是 SDK 保留或固定的编号。实际项目可以使用其他有效值，但要同步修改对应方向的发送端和接收端。

Client 配置与鉴权

@dataclass(frozen=True, slots=True)
class ClientOptions:
    app_id: str
    cache_dir: str | os.PathLike[str]
    endpoint: str | None = None
    console_log_enabled: bool = False

class Client:
    def __init__(
        self,
        options: ClientOptions,
        *,
        access_key_id: str | None = None,
        access_key_secret: str | None = None,
    ) -> None: ...
    def create_connection(
        self,
        *,
        on_state_changed=None,
        on_command=None,
        on_stream_message=None,
    ) -> Connection: ...
    def upload_logs(self) -> str: ...
    def close(self) -> None: ...

cache_dir 必须是绝对路径。ClientOptions 不保存凭据；Access Key ID 和 Secret Key ID 只作为 Client 构造参数传入。省略两项时使用 External Token 模式，同时提供两项时使用 Access Key 模式。只提供一项、传入空字符串或 NUL 会在 Native 副作用前抛出 ValueError。

同一进程同时只能有一个 RTC Client。Access Key 与 External Token 模式不能并存；前一个 Client 完整关闭后可以改用另一种模式。RTC 与 Ti Cloud Storage Client 必须共享 cache 目录和 console 设置。

close() 幂等。它会拒绝新资源和操作，取消在途 Connect，Detach 仍绑定的 RTC Output，关闭所有 Connection，并等待已经接受的 callback 退出。调用已关闭 Client 的 upload_logs() 抛出 ClosedError。

Connection

class Connection:
    @property
    def state(self) -> ConnectionState: ...
    def connect(
        self,
        device_id: str,
        *,
        token: str | None = None,
        timeout: float = 30,
    ) -> None: ...
    def disconnect(self) -> None: ...
    def send_command(
        self, command_id: int, data: bytes | bytearray | memoryview
    ) -> None: ...
    def send_stream_message(
        self,
        stream_id: int,
        timestamp: timedelta,
        data: bytes | bytearray | memoryview,
    ) -> None: ...
    def subscribe_audio(self, stream_id: int) -> None: ...
    def unsubscribe_audio(self, stream_id: int) -> None: ...
    def subscribe_video(self, stream_id: int) -> None: ...
    def unsubscribe_video(self, stream_id: int) -> None: ...
    def request_video_keyframe(self, stream_id: int) -> None: ...
    def start_recording(
        self, *, video_stream_id: int, audio_stream_id: int | None = None
    ) -> RecordingTask: ...
    def start_raw_dump(self, options: RawDumpOptions) -> RawDump: ...
    def close(self) -> None: ...

Connection 只能由 Client.create_connection() 创建。Access Key Client 调用 connect(device_id)；External Token Client 调用 connect(device_id, token=token)。混用鉴权参数会在 Native 副作用和状态变化前抛出 ValueError。

connect() 是有界同步调用，成功返回时 state is ConnectionState.CONNECTED，此时可以立即订阅。当前调用的 CONNECTING、CONNECTED、初始失败和取消不会重复交给 on_state_changed；成功后的断线和状态变化才进入 callback。

timeout 接受有限正数秒值，最大为 120 秒。同一 Connection 的并发 Connect 抛出 InUseError。失败、取消或断开后，可以按原鉴权模式重连。在 TiRTC callback 内调用同步 Connect 会抛出 InUseError。

Command ID 位于 0x2001..0xffffffff；较低区间由底层协议保留。Stream Message 的 timestamp 是精确到整毫秒、位于 0..0xffffffff 毫秒的 timedelta。buffer 入参在同步调用返回前读完。

Output 与 Frame

根 package 提供 AudioOutput、VideoOutput、EncodedAudioOutput 和 EncodedVideoOutput。它们都提供 attach(connection, stream_id)、detach()、只读 state、幂等 close() 和 context manager；解码 VideoOutput 另外提供 take_snapshot()。

四类 Output 的第一个 on_frame 参数必填。Attach 只建立 binding，不转移 ownership。Connection 或 Client 关闭时会先 Detach Output 并等待已经接受的 callback 退出，Output 本身保持 open。应用最终负责关闭 Output。

解码 Output 可以配置：

@dataclass(frozen=True, slots=True)
class OutputBufferOptions:
    strategy: OutputBufferStrategy = OutputBufferStrategy.AUTOMATIC
    max_buffer_watermark: timedelta | None = None

AudioOutput 的 agc_level 与 ans_level 可取 disabled、low、medium、high。VideoOutput 的 decoder preference 可取 auto、software、hardware。

Frame 是 SDK 创建的不可变值，媒体数据是只读 memoryview。需要修改数据或让数据脱离 Frame 生命周期时，显式复制为 bytes(frame.data)。source_time 是 UTC-aware datetime 或 None。

同一对象的 callback 串行执行，不同对象的 callback 可能并发执行。每个 Video Output 最多排队 2 帧，每个 Audio Output 最多排队 64 帧；发生背压时只丢完整 Frame，下一帧设置 discontinuity=True。

原始音视频诊断采集

Connection 可以按 Stream ID 采集播放时收到的原始音视频数据，用于排查无声、卡顿、花屏或黑屏等问题：

options = tirtc.RawDumpOptions(
    audio_stream_ids=(10,),
    video_stream_ids=(11,),
)

with connection.start_raw_dump(options) as capture:
    # 复现问题后停止采集。
    archive = capture.stop()

两个 Stream ID 元组至少填写一个，取值范围都是 0..15。停止后检查 RawDumpArchive 的 capture_complete、stop_reason、unsaved_packet_count、unsaved_byte_count 和 empty，再调用 Client.upload_logs() 将采集文件随下一次日志上传提交。完整流程、限制和隐私要求见接入或使用客户端诊断能力。

Recording、Snapshot 与临时文件

Connection.start_recording() 返回 RecordingTask。RecordingTask.stop() 同步等待唯一终局；重复或并发调用会返回同一个 RecordingFile，或重抛同一个异常。

VideoOutput.take_snapshot() 返回 SnapshotFile。两个文件类型都提供只读 path: Path，Recording 另外提供 duration: timedelta。路径位于 SDK cache；需要长期保存时，先复制到应用目录，再调用幂等 delete()，或使用 context manager 删除临时源文件。已经返回给应用的文件可以在 Client 关闭后继续保存或删除。

Error、进程与释放

Python 可在进入 Native 前判断的类型和值域问题分别抛出 TypeError 与 ValueError。Native 失败抛出 TiRTCError 或稳定子类。调用方超时抛出 OperationTimeoutError(code=None)；未知 Native code 保留为普通 TiRTCError。

在当前 Client 或 descendant callback 中同步关闭 Client 会形成自等待，close() 会在 teardown 前抛出 InUseError。正常释放顺序是完成 RecordingTask、Detach/关闭 Output、Disconnect/关闭 Connection、关闭 Client；Client 级联关闭也能完成同一资源收口。

import tirtc 不创建线程、目录、网络连接或 Runtime 状态。父进程仅 import 后 fork 时，子进程可以创建自己的 Client；任一 Client 活跃后 fork，子进程首次调用 SDK 抛出 UnsupportedError(code=None, name="forked_process")，并且必须立即 exec 或 os._exit()。