Skip to content

API.md

Latest commit

 

History

History
775 lines (572 loc) · 19.4 KB

API.md

File metadata and controls

775 lines (572 loc) · 19.4 KB
Raw

dglab-sdk Python API 参考

本文记录 dglab_kit Python 包全部公开接口。当前实现仅支持 DG-LAB WebSocket V4 协议。

1. 导入与约定

公开对象均可从包顶层导入：

from dglab_kit import (
    Channel,
    DglabDevice,
    DglabServer,
    DglabSocket,
)

接口约定：

  • 所有网络和设备控制方法都是异步方法，需要使用 await。
  • timeout、connect_timeout 和 response_timeout 使用秒。
  • duration_ms 使用毫秒。
  • Python 属性和参数使用 snake_case。
  • 协议业务字典中的 slotId、reqId、slotState 等字段保留 V4 名称。
  • 以下划线开头的方法属于包内实现，不是公共 API，也不保证兼容性。
  • V4 外层帧由 SDK 内部管理，不通过公共方法或事件暴露。

2. DglabSocket

DglabSocket 是 V4 控制端。它可以连接官方或自部署 relay，并管理多个 APP 和 设备实例。

2.1 构造函数

DglabSocket(
    url: str | None = None,
    *,
    protocols: str | Sequence[str] | None = None,
    connect_timeout: float = 8.0,
    response_timeout: float = 8.0,
    auto_request_devices: bool = True,
)
参数 说明
url V4 relay WebSocket 地址。普通控制端应传入该参数
protocols 可选 WebSocket 子协议或子协议列表
connect_timeout 等待连接完成的秒数，必须大于 0
response_timeout RPC 默认响应超时秒数，必须大于 0
auto_request_devices APP 接入后是否自动请求一次设备列表

省略 url 的模式只供包内 DglabServer 使用。

2.2 状态属性

属性 类型 说明
state SocketState 当前控制端状态
target_id `str None`
clients list[V4Client] 当前在线 APP 状态对象列表
devices list[DglabDevice] 跨全部在线 APP 的设备实例列表

clients 和 devices 每次访问都会返回新列表，修改列表本身不会修改 SDK 的内部 集合。APP ID 通过 client.client_id 获取。

2.3 查找 APP 和设备

socket.get_client(client_id: str) -> V4Client | None

返回指定 APP 的缓存对象；APP 不在线时返回 None。

socket.get_device(
    client_id: str,
    slot_id: str,
) -> DglabDevice | None

按 APP ID 和设备槽位查找实例。设备尚未发现、已移除或 APP 已断开时返回 None。

2.4 连接生命周期

await socket.connect() -> ConnectResult

连接 relay 并等待服务端完成初始化。返回：

{"target_id": "控制端 ID"}

重复调用时，如果已有有效连接结果，会直接返回当前结果；连接正在进行时会复用当前 等待过程。

await socket.disconnect(
    code: int = 1000,
    reason: str = "",
) -> None

断开连接、清空 APP 和设备状态，并终止所有未完成 RPC。code 和 reason 是 WebSocket 关闭信息。

2.5 事件监听方法

DglabSocket 继承事件分发能力。

socket.on(event: str, listener: Callable[..., Any]) -> DglabSocket

注册持久监听器。监听器可以是普通函数或异步函数。

socket.once(event: str, listener: Callable[..., Any]) -> DglabSocket

注册只执行一次的监听器。

socket.off(
    event: str,
    listener: Callable[..., Any] | None = None,
) -> DglabSocket

提供 listener 时删除指定监听器；省略时删除该事件的全部监听器。

socket.emit(event: str, *args: Any) -> bool

同步派发一个本地事件，并调度监听器返回的 awaitable。返回是否存在监听器。业务代码 通常只需使用 on()、once() 和 off()；emit() 不会向 APP 发送数据。

2.6 事件列表

事件 回调参数 说明
state (state, previous) SocketState 发生变化
open (event) WebSocket 传输已打开，event 可能为 None
close (event) 连接关闭，event 为 CloseEvent
error (error) SDK、网络或事件监听器错误
data (data, client_id) APP 上报的业务数据
action (action) APP 上报的 0-9 自定义动作，不包含来源 APP ID
device (patch, client_id) 单设备新增、移除或状态增量
devices (devices, client_id) 合并后的原始设备字典完整列表
device_instances (devices, client_id) 合并后的 DglabDevice 完整列表
client_attached (client_id) APP 已接入
client_disconnected (client_id) APP 已断开

device 的移除事件形如：

{"slotId": "slot-1", "removed": True}

action 不携带来源，需要来源时监听 data 并使用其 client_id 参数。

3. DglabServer

DglabServer 在当前 Python 进程中同时运行 V4 WebSocket server 和控制端。APP 直接连接该进程，不需要外部 relay。

3.1 构造函数

DglabServer(
    host: str = "0.0.0.0",
    port: int = 9998,
    *,
    path: str = "/",
    public_url: str | None = None,
    target_id: str | None = None,
    ssl_context: ssl.SSLContext | None = None,
    heartbeat_interval: float = 30.0,
    response_timeout: float = 8.0,
    auto_request_devices: bool = True,
)
参数 说明
host 监听网卡地址
port TCP 端口，0 表示自动选择可用端口
path 接受的 WebSocket 路径，会规范为绝对路径
public_url APP 实际可访问的公开 URL，用于 LAN、TLS 或反向代理
target_id 固定控制端 ID；省略时随机生成 8 位十六进制 ID
ssl_context 可选 TLS 上下文；提供后直接监听 wss://
heartbeat_interval 向 APP 发送业务心跳的秒数，必须大于 0
response_timeout 内置控制端的 RPC 默认超时秒数
auto_request_devices APP 接入后是否自动请求一次设备列表

port 必须位于 0-65535。

3.2 属性

属性 类型 说明
controller DglabSocket 内置完整控制端
target_id str APP 连接时必须指定的控制端 ID
host str 配置的监听地址
port int 实际监听端口；自动分配后返回真实端口
is_running bool server 是否已启动
server_url str 对外公布的基础 WebSocket URL
pairing_url str 已加入 tid 的 APP 配对 WebSocket URL
app_launch_url str 通过浏览器唤起 DG-LAB APP 的链接

未提供 public_url 且 host 是 0.0.0.0 或 :: 时，server_url 使用 127.0.0.1 作为展示地址。手机连接时通常应显式传入 LAN public_url。

3.3 生命周期

await server.start() -> DglabServer

启动监听并初始化 controller。重复调用不会重复启动。

await server.stop() -> None

通知并关闭全部 APP、停止监听，然后断开内置控制端。尚未启动时无操作。

await server.wait_closed() -> None

等待 stop() 完成。

await server.serve_forever() -> None

启动并持续运行；任务取消时自动执行 stop()。

推荐使用异步上下文：

async with DglabServer(port=8765) as server:
    print(server.pairing_url)
    await asyncio.Event().wait()

__aenter__() 调用 start() 并返回自身，__aexit__() 调用 stop()。

4. DglabDevice

DglabDevice 表示一个绑定到控制端的稳定设备槽位实例。相同 (client_id, slot_id) 在线期间保持同一个 Python 对象；状态更新会原地更新实例。

不建议业务代码直接构造该类，应通过 socket.get_device()、socket.devices 或 client.device_instances 获取。

4.1 属性

属性 类型 说明
client_id str 暴露该设备的 APP ID
slot_id str 设备槽位 ID
name str APP 报告的显示名称，缺失时为空字符串
device_type `DeviceType str`
props dict[str, Any] 最新设备属性副本
slot_state dict[str, Any] 最新槽位状态副本
has_device `bool None`
is_online bool 槽位是否仍属于在线 APP
raw dict[str, Any] 完整设备数据副本

props、slot_state 和 raw 都是深拷贝，修改返回对象不会改变 SDK 缓存。

设备移除或 APP 断开后，已有引用的 is_online 变为 False，控制方法抛出 ConnectionClosedError。

4.2 相对强度

await device.add_intensity(
    channel: Channel,
    value: int | float,
    *,
    priority: int | None = None,
    immediate: bool | None = None,
    timeout: float | None = None,
) -> JsonValue
await device.reduce_strength(
    channel: Channel,
    value: int | float,
    *,
    priority: int | None = None,
    immediate: bool | None = None,
    timeout: float | None = None,
) -> JsonValue

行为与 DglabSocket 对应方法相同，但自动携带 client_id 和 slot_id。

4.3 临时强度与归零

await device.set_temp_intensity(
    channel: Channel,
    value: int | float,
    duration_ms: int,
    *,
    priority: int | None = None,
    immediate: bool | None = None,
    timeout: float | None = None,
) -> JsonValue
await device.reset_intensity(
    channel: Channel,
    *,
    priority: int | None = None,
    immediate: bool | None = None,
    timeout: float | None = None,
) -> JsonValue

4.4 波形

await device.send_pulse(
    channel: Channel,
    duration_ms: int,
    frames: Sequence[str] | Sequence[Sequence[int]],
    *,
    version: int | None = None,
    sequence: int | None = None,
    priority: int | None = None,
    immediate: bool | None = None,
    timeout: float | None = None,
) -> JsonValue

4.5 清理任务

await device.clear(
    channel: Channel | None = None,
    *,
    timeout: float | None = None,
) -> JsonValue

省略 channel 时清理该设备全部任务；提供通道时只清理该通道任务。

4.6 调试表示

repr(device)

返回包含 client_id、slot_id 和在线状态的简短字符串。

5. V4Client

V4Client 保存一个在线 APP 的设备状态。实例由 DglabSocket 创建，可通过 socket.get_client() 或 socket.clients 获取。

5.1 构造函数

V4Client(
    client_id: str,
    devices: list[dict[str, Any]] | None = None,
    *,
    controller: DglabSocket | None = None,
)

这是公开可构造类型，但普通业务不应自行构造。没有 controller 时不会创建可控制的 DglabDevice 实例。

5.2 属性

属性 类型 说明
client_id str APP ID
is_online bool APP 是否仍连接到当前控制端
devices list[dict[str, Any]] 合并后的原始设备字典列表
device_instances list[DglabDevice] 当前设备实例列表副本

建议将 devices 视为只读缓存；直接修改它不会自动同步设备实例。

5.3 查找

client.get_device(slot_id: str) -> dict[str, Any] | None

返回原始设备字典引用。

client.get_device_instance(slot_id: str) -> DglabDevice | None

返回设备实例。

5.4 APP 探测和设备发现

await client.ping(
    *,
    timeout: float | None = None,
) -> int | float

请求该 APP 响应，返回 APP 提供的往返延迟毫秒值。

await client.request_devices() -> dict[str, JsonValue]

强制刷新该 APP 的设备列表，同时更新 devices 和 device_instances。默认自动设备 发现开启时，APP 接入后 SDK 已自动调用一次；业务只在需要校准时手动调用。

5.5 APP 全部任务清理

await client.clear_all(
    *,
    timeout: float | None = None,
) -> JsonValue

清理该 APP 中全部设备的全部任务。设备或通道级清理使用 DglabDevice.clear()。

5.6 自定义业务 RPC

client.create_request(
    method: str,
    data: JsonValue = None,
) -> dict[str, JsonValue]

创建带新 reqId 的业务请求。data=None 时省略 data 字段。

await client.send(
    data: JsonValue,
    *,
    timeout: float | None = None,
) -> JsonValue

向该 APP 发送业务数据并等待匹配响应。

  • 字典数据会复制后加入 reqId，不会修改调用方对象。
  • 兼容输入字段 requestId 会转换为 reqId。
  • 非字典 JSON 值会作为 custom 业务请求发送。
  • APP 返回错误时抛出 ResponseError。
  • 等待超时时抛出 ResponseTimeoutError。

APP 断开后，is_online 变为 False，上述 APP 方法抛出 ConnectionClosedError。

5.7 高级状态维护方法

以下方法公开存在，但主要供 SDK 状态同步使用：

client.dispatch(data: object) -> bool

识别设备响应或事件并合并状态；产生设备变化时返回 True。

client.replace_devices(devices: object) -> None

替换完整设备列表，同步创建、更新或离线设备实例。

client.patch_devices(added: object, removed: object) -> None

合并新增设备并移除指定 slotId。

client.patch_slots(slots: object) -> None

递归合并设备 props 和 slotState 增量。

client.destroy() -> None

将全部设备实例标记为离线并清空缓存。

6. 枚举

6.1 SocketState

成员 值 说明
SocketState.IDLE "idle" 尚未连接
SocketState.CONNECTING "connecting" 正在连接
SocketState.WAITING_FOR_PEER "waiting_for_peer" 已连接，等待 APP
SocketState.PAIRED "paired" 至少一个 APP 已接入
SocketState.DISCONNECTED "disconnected" 已断开

SocketState 继承 StrEnum。

6.2 DeviceType

成员 值 设备
DeviceType.COYOTE_020 "COYOTE_020" 郊狼 2.0
DeviceType.COYOTE_030 "COYOTE_030" 郊狼 3.0
DeviceType.BMTR_1 "BMTR_1" 灵猫 1.0
DeviceType.OVC_1 "OVC_1" 负鼠 1.0

DeviceType 继承 StrEnum。未来未知类型由 DglabDevice.device_type 保留为字符串。

6.3 Channel

成员 值 说明
Channel.A 0 A 通道
Channel.B 1 B 通道

Channel 继承 IntEnum，可直接序列化为协议整数。

6.4 ActionType

成员 值 说明
ActionType.APPEND_PULSE_DATA 0 追加波形数据
ActionType.ADD_INTENSITY 3 相对增减强度
ActionType.SET_TEMP_INTENSITY 4 设置临时强度
ActionType.SET_INTENSITY 7 设置绝对强度，当前封装用于归零

通常无需直接使用 ActionType，设备控制方法会自动选择正确类型。

7. 数据类型

这些类型主要用于静态类型检查，运行时仍是普通字典和 JSON 值。

7.1 JsonValue

递归 JSON 类型别名：

JsonValue = (
    str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]
)

7.2 DeviceDescriptor

必填字段：

class DeviceDescriptor(TypedDict):
    slotId: str
    name: str
    type: str

7.3 DeviceInfo

继承 DeviceDescriptor，新增可选字段：

class DeviceInfo(DeviceDescriptor, total=False):
    props: dict[str, JsonValue]
    slotState: dict[str, JsonValue]

7.4 DevicesGetResult

class DevicesGetResult(TypedDict):
    devices: list[DeviceDescriptor]

7.5 RpcRequest

class RpcRequest(TypedDict):
    t: Literal["req"]
    reqId: str
    m: str
    data: NotRequired[JsonValue]

这是业务 RPC 数据，不是 V4 外层帧。

7.6 RpcResponse

class RpcResponse(TypedDict, total=False):
    t: Literal["resp"]
    reqId: str
    requestId: str
    result: JsonValue
    error: str

由于 total=False，静态类型上所有字段均可省略。标准响应应包含 t 和请求 ID， 成功时提供 result，失败时提供 error。

7.7 ConnectResult

class ConnectResult(TypedDict):
    target_id: str

7.8 CloseEvent

class CloseEvent(TypedDict):
    code: int
    reason: str
    was_clean: bool

7.9 EventName

事件名类型别名包含：

EventName = Literal[
    "state",
    "open",
    "close",
    "error",
    "data",
    "action",
    "device",
    "devices",
    "device_instances",
    "client_attached",
    "client_disconnected",
]

8. 异常

8.1 DglabError

DglabError(code: str, message: str)

所有 SDK 特定异常的基类。

属性 类型 说明
code str 稳定的机器可读错误码
str(error) str 可读错误消息

8.2 ConnectionClosedError

继承 DglabError。连接、APP 或设备槽位离线导致操作无法继续时抛出。

8.3 ResponseError

继承 DglabError。APP 返回业务错误时抛出。

8.4 ResponseTimeoutError

同时继承 DglabError 和内置 TimeoutError。RPC 未在期限内响应时抛出，因此既可 按 SDK 异常捕获，也可按超时异常捕获。

8.5 常见错误码

错误码 常见来源
socket-connect-timeout 等待连接初始化超时
socket-response-timeout RPC 响应超时
socket-v4-response APP 返回业务错误
socket-disconnected WebSocket 或 APP 断开
client-offline 通过已离线 APP 实例调用 APP 方法
device-offline 通过已离线设备实例控制
socket-send 当前没有可用传输连接
socket-request-replaced 同一 APP 重用了未完成请求 ID
socket-ping-timeout relay 协议探测超时
socket-idle-timeout relay 报告控制端空闲超时
socket-v4-server relay 返回服务端错误
client-not-found 内置 Server 找不到目标 APP

错误码集合可能随协议能力扩展；业务应保留 DglabError 兜底处理。

9. 内置波形

9.1 WaveformLabel

class WaveformLabel(TypedDict):
    cn: str
    en: str

9.2 Waveform

class Waveform(TypedDict):
    label: WaveformLabel
    raw: tuple[str, ...]

raw 可直接传给 send_pulse()。

9.3 CoyoteWaveform

CoyoteWaveform 是 StrEnum，包含 24 个成员：

EXTRUSTION, BUBBLE, RHYTHM, AIR_WAVES, DANCE, CLIMB,
SHADE, PULSE, BREATHING, TIDE, PULSATING, QUICK_RUB,
GRADUALRUB, HEARTBEAT, COMPRESS, RHYTHMIC, GRAINY, BOUNCY,
RIPPLE, RAINFALL, TEMPO_TAP, SIGNAL, TEASE_1, TEASE_2

成员值与成员名相同。

9.4 COYOTE_WAVEFORMS

COYOTE_WAVEFORMS: dict[CoyoteWaveform, Waveform]

包含全部郊狼内置波形：

frames = COYOTE_WAVEFORMS[CoyoteWaveform.BUBBLE]["raw"]

9.5 OvcWaveform

OvcWaveform 是 StrEnum，包含 20 个成员：

ALARM, PINPOINT, UNDERTOW, SOOTHE, TRAPEZOID, ENGINE_WAVE,
AFTERSHOCK, STEADY_HIT, CURVE_PRESS, BEAT, LOOP_SPREAD,
INTERMITTENT, SURF_WAVE, LOOP_PUSH, CANOPY, JOY_CLIMB,
DECEL, RANDOM, POP_CANDY, PHASE_SHIFT

成员值与成员名相同。

9.6 OVC_WAVEFORMS

OVC_WAVEFORMS: dict[OvcWaveform, Waveform]

包含全部负鼠内置波形。

10. 包版本

import dglab_kit

print(dglab_kit.__version__)

当前版本值为 "1.0.1"。