DG-Lab 终端

DGLabClient

DGLabClient(client_id: UUID4 = None, target_id: UUID4 = None)

Bases: ABC

DG-Lab 终端基础类

Parameters:

Name	Type	Description	Default
client_id	UUID4	终端 ID	None
target_id	UUID4	App ID	None

Source code in pydglab_ws/client/base.py

def __init__(
        self,
        client_id: UUID4 = None,
        target_id: UUID4 = None
):
    self._client_id: Optional[UUID4] = client_id
    self._target_id: Optional[UUID4] = target_id
    self._message_type_to_handler: Dict[
        MessageType,
        Callable[[WebSocketMessage], Any]
    ] = {
        MessageType.MSG: self._handle_msg,
        MessageType.BREAK: self._handle_break,
        MessageType.HEARTBEAT: self._handle_heartbeat
    }

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 的绑定

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 字符串，如果需要二维码图像需要自行从返回的文本进行生成

Source code in pydglab_ws/client/base.py

def get_qrcode(self, uri: str) -> Optional[str]:
    """
    终端二维码，二维码图像需要自行生成

    :param uri: WebSocket 服务端 URI，例如：``ws://107.47.91.92:4567``
        （注意末尾不能有 ``/``）
    :return: URL 字符串，如果需要二维码图像需要自行从返回的文本进行生成
    """
    if uri is None or self.not_registered:
        return None
    return dg_lab_client_qrcode(uri, self._client_id)

register async

register()

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

Source code in pydglab_ws/client/base.py

async def register(self):
    """
    从 WebSocket 服务端中获取 ``client_id`` 并保存
    """
    while self.not_registered:
        message = await self._recv()
        if message.type == MessageType.BIND and message.message == MessageDataHead.TARGET_ID:
            self._client_id = message.client_id

ensure_bind async

ensure_bind()

确保终端已完成与 App 的绑定

Source code in pydglab_ws/client/base.py

async def ensure_bind(self):
    """确保终端已完成与 App 的绑定"""
    while True:
        if self.not_registered:
            await self.register()
        elif self.not_bind:
            await self.bind()
        else:
            break

bind async

bind() -> RetCode

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

Returns:

Type	Description
RetCode	响应码

Source code in pydglab_ws/client/base.py

async def bind(self) -> RetCode:
    """
    等待与 DG-Lab App 的关系绑定，并保存 ``target_id``
    :return: 响应码
    """
    while self.not_bind:
        message = await self._recv_owned()
        if message.type == MessageType.BIND and isinstance(message.message, RetCode):
            if message.message == RetCode.SUCCESS:
                self._target_id = message.target_id
            return message.message

rebind async

rebind() -> RetCode

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

Returns:

Type	Description
RetCode	响应码

Source code in pydglab_ws/client/base.py

async def rebind(self) -> RetCode:
    """
    清除 ``target_id``，重新等待与 DG-Lab App 的关系绑定，适合 App 断开连接后调用
    :return: 响应码
    """
    self._target_id = None
    while self.not_bind:
        message = await self._recv_owned()
        if message.type == MessageType.BIND and isinstance(message.message, RetCode):
            if message.message == RetCode.SUCCESS:
                self._target_id = message.target_id
            return message.message

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

Source code in pydglab_ws/client/base.py

async def recv_data(self) -> Union[StrengthData, FeedbackButton, RetCode]:
    """
    获取 WebSocket 服务端的数据

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

    :return: 可能为 **强度数据** - [`StrengthData`][pydglab_ws.models.StrengthData]、 \
        **App 反馈数据** - [`FeedbackButton`][pydglab_ws.enums.FeedbackButton]、 \
        **心跳** - [`RetCode.SUCCESS`][pydglab_ws.enums.RetCode.SUCCESS]、 \
        **App 断开连接** - [`RetCode.CLIENT_DISCONNECTED`][pydglab_ws.enums.RetCode.CLIENT_DISCONNECTED]、\
        **消息长度大于 1950** - [`RetCode.MESSAGE_TOO_LONG`][pydglab_ws.enums.RetCode.MESSAGE_TOO_LONG]
    :raise InvalidStrengthData: [`InvalidStrengthData`][pydglab_ws.exceptions.InvalidStrengthData]
    :raise InvalidFeedbackData: [`InvalidFeedbackData`][pydglab_ws.exceptions.InvalidFeedbackData]
    """
    await self.ensure_bind()
    while True:
        message = await self._recv_owned()
        handler = self._message_type_to_handler.get(message.type)
        if handler and (result := handler(message)) is not None:
            return result

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

Source code in pydglab_ws/client/base.py

async def data_generator(
        self,
        *targets: _DataType,
) -> AsyncGenerator[_DataType, Any]:
    """
    强度数据异步生成器

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

    示例：
    ```python3
    async for data in client.data_generator(StrengthData, FeedbackButton):
        print(f"Got data from App: {data}")
    ```
    :param targets: 目标类型，只有为目标类型的数据会被返回，为空即默认值时则不进行限制
    :return: 可能为 **强度数据** - [`StrengthData`][pydglab_ws.models.StrengthData]、 \
        **App 反馈数据** - [`FeedbackButton`][pydglab_ws.enums.FeedbackButton] \
        、**心跳** - ``RetCode.SUCCESS``、**App 断开连接** - ``RetCode.CLIENT_DISCONNECTED``
    """
    while True:
        data = await self.recv_data()
        if not targets or type(data) in targets:
            yield data

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

Source code in pydglab_ws/client/base.py

async def set_strength(
        self,
        channel: Channel,
        operation_type: StrengthOperationType,
        value: int
):
    """
    设置强度

    :param channel: 通道选择
    :param operation_type: 强度变化模式
    :param value: 强度数值，范围在 [0, 200]
    """
    await self.ensure_bind()
    await self._send_owned(
        MessageType.MSG,
        dump_strength_operation(channel, operation_type, value)
    )

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

Source code in pydglab_ws/client/base.py

async def add_pulses(
        self,
        channel: Channel,
        *pulses: PulseOperation
):
    """
    下发波形数据

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

    :param channel: 通道选择
    :param pulses: 波形操作数据，最大长度为 100
    :raise InvalidPulseOperation: [`InvalidPulseOperation`][pydglab_ws.exceptions.InvalidPulseOperation]
    :raise PulseDataTooLong: 波形操作数据过长，最大长度应为 [`PULSE_DATA_MAX_LENGTH`][pydglab_ws.utils.PULSE_DATA_MAX_LENGTH]
    """
    await self.ensure_bind()
    await self._send_owned(
        MessageType.MSG,
        dump_add_pulses(channel, *pulses)
    )

clear_pulses async

clear_pulses(channel: Channel)

清空波形队列

Parameters:

Name	Type	Description	Default
channel	Channel	通道选择	required

Raises:

Type	Description
InvalidPulseOperation	InvalidPulseOperation

Source code in pydglab_ws/client/base.py

async def clear_pulses(self, channel: Channel):
    """
    清空波形队列

    :param channel: 通道选择
    :raise InvalidPulseOperation: [`InvalidPulseOperation`][pydglab_ws.exceptions.InvalidPulseOperation]
    """
    await self.ensure_bind()
    await self._send_owned(
        MessageType.MSG,
        dump_clear_pulses(channel)
    )