终端基础类的属性和方法¶

DGLabClient 是终端基类，包含本地终端 DGLabLocalClient 和 WebSocket 终端 DGLabWSClient 共有的一些属性和方法

获取自身的各项信息¶

可获取终端自身的 ID、绑定的 App 的 ID、是否已从 WebSocket 服务端注册，即获取 ID（clientId）以及是否已和 DG-Lab App 绑定。

Info

ID 属性 client_id 和 target_id 可能为返回空，原因是终端未注册或还没有与 App 绑定。

可用属性¶

client_id `property` ¶

client_id: Optional[UUID4]

DG-Lab 终端 ID

target_id `property` ¶

target_id: Optional[UUID4]

DG-Lab App ID

not_registered `property` ¶

not_registered: bool

终端是否未注册

not_bind `property` ¶

not_bind: bool

终端是否未完成与 App 的绑定

示例¶

from pydglab_ws import DGLabWSConnect

async def main():
    async with DGLabWSConnect("ws://192.168.1.161:5678") as client:
        await client.bind()
        print(f"终端 {client.client_id} 已与 App {client.target_id} 成功绑定")
        ...

注册（获取终端 ID）¶

终端连接至 WebSocket 服务端后，需要先从服务端获取终端 ID（clientId），此处称为注册，不过一般情况下 PyDGLab-WS 会在异步上下文管理中完成该步骤，即以下用法：

使用连接器 DGLabWSConnect 创建 WebSocket 终端 DGLabWSClient
手动创建 WebSocket 终端 DGLabWSClient 对象，并在异步上下文管理（即 async with 中使用）
从 WebSocket 服务端 DGLabWSServer 创建本地终端 DGLabLocalClient

可用方法¶

register `async` ¶

register()

从 WebSocket 服务端中获取 client_id 并保存

获取二维码，绑定 DG-Lab App¶

从 WebSocket 服务端获取终端 ID（clientId）后，即可生成终端二维码， DG-Lab App 进入 Socket 被控界面后，扫码二维码即可绑定

Info

注意此处生成的二维码 仅仅是二维码内容/数据，二维码图像需要自行从返回的数据生成

Info

App 断开连接后，重新连接至服务端时 targetId 可能发生变化，因此最好是再调用一次 rebind 方法，重新等待绑定并更新 targetId

Warning

生成二维码后，需要调用绑定方法来等待 App 绑定，此时该异步事件会阻塞直到绑定完成

可用方法¶

get_qrcode ¶

get_qrcode(uri: str) -> Optional[str]

终端二维码，二维码图像需要自行生成

Parameters:

Name	Type	Description	Default
`uri`	`str`	WebSocket 服务端 URI，例如：`ws://107.47.91.92:4567` （注意末尾不能有 `/`）	required

Returns:

Type	Description
`Optional[str]`	URL 字符串，如果需要二维码图像需要自行从返回的文本进行生成

bind `async` ¶

bind() -> RetCode

等待与 DG-Lab App 的关系绑定，并保存 target_id

Returns:

Type	Description
`RetCode`	响应码

rebind `async` ¶

rebind() -> RetCode

清除 target_id，重新等待与 DG-Lab App 的关系绑定，适合 App 断开连接后调用

Returns:

Type	Description
`RetCode`	响应码

示例¶

from pydglab_ws import DGLabWSConnect

def print_qrcode(_: str):
    """输出二维码到终端界面"""
    ...

async def main():
    async with DGLabWSConnect("ws://192.168.1.161:5678") as client:
        # 获取二维码
        url = client.get_qrcode()
        print("请用 DG-Lab App 扫描二维码以连接")
        print_qrcode(url)

        # 等待绑定
        await client.bind()
        print(f"已与 App {client.target_id} 成功绑定")
        ...

获取 App 数据更新和服务端通知¶

可获取如下数据：

通道强度数据
App 触发的反馈按钮
App 返回的错误码
服务端发送的心跳
服务端发送的 App 断开连接通知

Info

data_generator 会不断从队列中读取消息并解析，队列为空时则会阻塞该异步事件，如果需要退出，可使用 break

Info

App 断开连接后，重新连接至服务端时 targetId 可能发生变化，因此最好是再调用一次 rebind 方法，重新等待绑定并更新 targetId

Danger

不能并发地读取 数据更新，因为消息一旦被读出，就会从队列中移除。另外 data_generator 在解析消息时，如果读出的消息不符合筛选条件，将会被丢弃，直到读到符合条件的消息

可用方法¶

recv_data `async` ¶

recv_data() -> Union[StrengthData, FeedbackButton, RetCode]

获取 WebSocket 服务端的数据

注意，获取到的是队列中最早的数据，可能不是最新的

Returns:

Type	Description
`Union[StrengthData, FeedbackButton, RetCode]`	可能为强度数据 - `StrengthData`、 App 反馈数据 - `FeedbackButton`、心跳 - `RetCode.SUCCESS`、 App 断开连接 - `RetCode.CLIENT_DISCONNECTED`、消息长度大于 1950 - `RetCode.MESSAGE_TOO_LONG`

Raises:

Type	Description
`InvalidStrengthData`	`InvalidStrengthData`
`InvalidFeedbackData`	`InvalidFeedbackData`

data_generator `async` ¶

data_generator(*targets: _DataType) -> AsyncGenerator[_DataType, Any]

强度数据异步生成器

注意，是从队列中最早的数据开始获取，可能不是最新的

示例：

async for data in client.data_generator(StrengthData, FeedbackButton):
    print(f"Got data from App: {data}")

Parameters:

Name	Type	Description	Default
`targets`	`_DataType`	目标类型，只有为目标类型的数据会被返回，为空即默认值时则不进行限制	`()`

Returns:

Type	Description
`AsyncGenerator[_DataType, Any]`	可能为强度数据 - `StrengthData`、 App 反馈数据 - `FeedbackButton` 、心跳 - `RetCode.SUCCESS`、App 断开连接 - `RetCode.CLIENT_DISCONNECTED`

示例¶

from pydglab_ws import DGLabWSConnect, StrengthData, FeedbackButton, RetCode

async def main():
    async with DGLabWSConnect("ws://192.168.1.161:5678") as client:
        ... # 完成了绑定
        async for data in client.data_generator():
            # 接收通道强度数据
            if isinstance(data, StrengthData):
                print(f"从 App 收到通道强度数据更新：{data}")

            # 接收 App 反馈按钮
            elif isinstance(data, FeedbackButton):
                print(f"App 触发了反馈按钮：{data.name}")

            # 接收 心跳 / App 断开通知
            elif data == RetCode.CLIENT_DISCONNECTED:
                print("App 已断开连接，你可以尝试重新扫码进行连接绑定")

设置通道强度¶

可设置郊狼 A 通道、B 通道的强度，并有几个变化模式可选。

可用方法¶

set_strength `async` ¶

set_strength(channel: Channel, operation_type: StrengthOperationType, value: int)

设置强度

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	通道选择	required
`operation_type`	`StrengthOperationType`	强度变化模式	required
`value`	`int`	强度数值，范围在 [0, 200]	required

示例¶

from pydglab_ws import DGLabWSConnect, Channel, StrengthOperationType

async def main():
    async with DGLabWSConnect("ws://192.168.1.161:5678") as client:
        ... # 完成了绑定
        await client.set_strength(
            Channel.A,
            StrengthOperationType.INCREASE,
            10
        )   # A 通道 增加 10 强度
        ...

下发波形数据¶

可向 App 下发波形操作，参数中的 *pulses 是位置参数，可传入多个 PulseOperation

Question

PulseOperation 波形操作 数据是由一个波形频率操作数据 WaveformFrequencyOperation 和一个波形强度操作数据 WaveformStrengthOperation 组合成的元组。波形频率操作数据和波形强度操作数据则都是由 4 个 int 整数组成的元组，代表了目标数值

可用方法¶

add_pulses `async` ¶

add_pulses(channel: Channel, *pulses: PulseOperation)

下发波形数据

每条波形数据代表了 100ms 的数据，所以若每次发送的数据有 10 条，那么就是 1s 的数据。由于网络有一定延时，若要保证波形输出的连续性，建议波形数据的发送间隔略微小于波形数据的时间长度 (< 1s)
数组最大长度为 100,也就是最多放置 10s 的数据，另外 App 中的波形队列最大长度为 500，即为 50s 的数据，若后接收到的数据无法全部放入波形队列，多余的部分会丢弃。所以谨慎考虑您的数据长度和数据发送间隔

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	通道选择	required
`pulses`	`PulseOperation`	波形操作数据，最大长度为 100	`()`

Raises:

Type	Description
`InvalidPulseOperation`	`InvalidPulseOperation`
`PulseDataTooLong`	波形操作数据过长，最大长度应为 `PULSE_DATA_MAX_LENGTH`

示例¶

from pydglab_ws import DGLabWSConnect, Channel

async def main():
    async with DGLabWSConnect("ws://192.168.1.161:5678") as client:
        ... # 完成了绑定
        await client.add_pulses(
            Channel.A,
            ((10, 10, 20, 30), (0, 5, 10, 50)), # 波形频率4条 {10,10,20,30}，波形强度4条 {0,5,10,50}
            ((10, 10, 20, 30), (0, 5, 10, 50)), # 同上
            ((10, 10, 20, 30), (0, 5, 10, 50))  # 同上
        )   # 向 A 通道波形队列增加了持续 300 毫秒的操作
        ...

清空波形队列¶

App 中的波形执行是基于波形队列，遵循先进先出的原则，并且队列可以缓存 500 条波形数据 (50s 的数据)。

当波形队列中还有尚未执行完的波形数据时，第三方终端希望立刻执行新的波形数据，则需要先将对应通道的波形队列执行清空操作后，再发送波形数据，即可实现立刻执行新的波形数据的需求。

来自DG-LAB 官方文档解释

可用方法¶

clear_pulses `async` ¶

clear_pulses(channel: Channel)

清空波形队列

Parameters:

Name	Type	Description	Default
`channel`	`Channel`	通道选择	required

Raises:

Type	Description
`InvalidPulseOperation`	`InvalidPulseOperation`

示例¶

from pydglab_ws import DGLabWSConnect, Channel

async def main():
    async with DGLabWSConnect("ws://192.168.1.161:5678") as client:
        ... # 完成了绑定
        await client.clear_pulses(Channel.A)    # 清空 A 通道波形队列
        ...

终端基础类的属性和方法¶

获取自身的各项信息¶

可用属性¶

client_id property ¶

target_id property ¶

not_registered property ¶

not_bind property ¶

示例¶

注册（获取终端 ID）¶

可用方法¶

register async ¶

获取二维码，绑定 DG-Lab App¶

可用方法¶

get_qrcode ¶

bind async ¶

rebind async ¶

示例¶

获取 App 数据更新和服务端通知¶

可用方法¶

recv_data async ¶

data_generator async ¶

示例¶

设置通道强度¶

可用方法¶

set_strength async ¶

示例¶

下发波形数据¶

可用方法¶

add_pulses async ¶

示例¶

清空波形队列¶

可用方法¶

clear_pulses async ¶

示例¶

client_id `property` ¶

target_id `property` ¶

not_registered `property` ¶

not_bind `property` ¶

register `async` ¶

bind `async` ¶

rebind `async` ¶

recv_data `async` ¶

data_generator `async` ¶

set_strength `async` ¶

add_pulses `async` ¶

clear_pulses `async` ¶