CreatorCreator
首页
快速上手
  • 中文简体
  • English
首页
快速上手
  • 中文简体
  • English
  • Creator

    • 快速上手
  • Stream Dock

    • 插件SDK
    • 图标包

      • 概述
      • 术语
      • 创建图标包
      • 图标规格
    • 场景配置

      • 什么是场景配置?
      • 导出
      • 使用案例
  • mPanel

    • 插件SDK
  • Cpp SDK

    • 概述
    • 依赖安装
    • 设备管理器
    • StreamDock 基类
    • 通信传输
    • 示例
    • API参考
    • 源代码
  • Python SDK

    • 概述
    • 依赖安装
    • 设备管理器
    • StreamDock 基类
    • API 参考
    • 示例
    • 源代码
  • Windows-WebSocket SDK

    • 概述
    • 快速上手
    • 发送的事件
    • 接收的事件
    • 源代码
  • 支持

    • 帮助和错误报告

DeviceManager

中央设备管理器,必须创建此类的实例才能检测和使用 StreamDock 设备。DeviceManager 提供了跨平台的设备检测和管理功能,支持 Windows、macOS 和 Linux 系统。

概述

DeviceManager 是 StreamDock SDK 的核心组件,负责设备的枚举、检测和监听。它通过 USB Vendor ID 和 Product ID 识别连接的 StreamDock 设备,并支持设备的热插拔检测。

跨平台支持

DeviceManager 根据不同的操作系统使用不同的底层传输方式:

  • Windows: 使用 hidapi 库进行 HID 设备通信
  • macOS: 使用系统原生的 IOKit 框架
  • Linux: 使用 pyudev 和 libusb 进行设备管理

构造函数

DeviceManager(transport=None)

创建新的 StreamDock DeviceManager 实例,用于检测连接的 StreamDock 设备。

参数:

  • transport (可选): 传输层实例,默认使用 LibUSBHIDAPI

示例:

from StreamDock.DeviceManager import DeviceManager

# 使用默认传输层
manager = DeviceManager()

# 使用自定义传输层
# custom_transport = CustomTransport()
# manager = DeviceManager(transport=custom_transport)

方法

enumerate()

enumerate()

检测连接的 StreamDock 设备。

返回值:

  • list: 包含所有连接的 StreamDock 设备的列表

工作原理:

  1. 遍历所有已知的产品 ID 列表
  2. 使用传输层枚举每个 VID/PID 组合的设备
  3. 为每个找到的设备创建相应的设备类实例
  4. 返回所有设备实例的列表

示例:

manager = DeviceManager()
streamdocks = manager.enumerate()
print(f"找到 {len(streamdocks)} 个 StreamDock 设备")

for device in streamdocks:
    print(f"设备路径: {device.getPath()}")
    print(f"设备类型: {device.DECK_TYPE}")

listen()

listen()

监听设备连接和断开事件。此方法会阻塞当前线程,建议在单独的线程中运行。

工作原理:

  1. 使用 pyudev 监听 USB 子系统事件
  2. 过滤出 'add' 和 'remove' 事件
  3. 解析设备的 Vendor ID 和 Product ID
  4. 对于匹配的设备,创建相应的设备实例或从列表中移除

示例:

import threading
from StreamDock.DeviceManager import DeviceManager

manager = DeviceManager()

# 在单独的线程中启动监听
t = threading.Thread(target=manager.listen)
t.daemon = True  # 设置为守护线程,主程序退出时自动结束
t.start()

print("开始监听设备插拔事件...")

# 主线程继续执行其他任务
try:
    while True:
        # 定期检查设备列表
        devices = manager.enumerate()
        print(f"当前连接的设备数量: {len(devices)}")
        time.sleep(5)
except KeyboardInterrupt:
    print("停止监听")

使用示例

基本使用

import threading
import time
from StreamDock.DeviceManager import DeviceManager

# 创建设备管理器
manager = DeviceManager()

# 启动设备监听线程
listen_thread = threading.Thread(target=manager.listen)
listen_thread.daemon = True
listen_thread.start()

# 枚举当前连接的设备
devices = manager.enumerate()
print(f"找到 {len(devices)} 个设备")

# 操作每个设备
for device in devices:
    device.open()
    device.init()  # 初始化设备
    print(f"设备 {device.getPath()} 已初始化")
  
    # 设置按键回调
    def key_callback(device, key, state):
        print(f"设备 {device.getPath()} 按键 {key} {'按下' if state else '释放'}")
  
    device.set_key_callback(key_callback)

# 保持程序运行
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    print("程序退出")

高级使用

import threading
import time
from StreamDock.DeviceManager import DeviceManager

class StreamDockController:
    def __init__(self):
        self.manager = DeviceManager()
        self.devices = []
        self.running = True
    
        # 启动监听线程
        self.listen_thread = threading.Thread(target=self._listen_devices)
        self.listen_thread.daemon = True
        self.listen_thread.start()
    
        # 初始枚举设备
        self._refresh_devices()
  
    def _listen_devices(self):
        """监听设备插拔事件"""
        self.manager.listen()
  
    def _refresh_devices(self):
        """刷新设备列表"""
        self.devices = self.manager.enumerate()
        print(f"当前连接的设备数量: {len(self.devices)}")
    
        for device in self.devices:
            if not hasattr(device, '_initialized'):
                device.open()
                device.init()
                device.set_key_callback(self._on_key_press)
                device._initialized = True
                print(f"设备 {device.getPath()} 已初始化")
  
    def _on_key_press(self, device, key, state):
        """按键事件处理"""
        print(f"设备 {device.getPath()} 按键 {key} {'按下' if state else '释放'}")
  
    def run(self):
        """运行控制器"""
        try:
            while self.running:
                # 定期刷新设备列表
                self._refresh_devices()
                time.sleep(5)
        except KeyboardInterrupt:
            print("停止控制器")
            self.running = False

# 使用控制器
controller = StreamDockController()
controller.run()

注意事项

  1. 线程安全: DeviceManager 不是线程安全的,如果在多线程环境中使用,需要添加适当的同步机制。
  2. 资源管理: 使用完设备后,记得调用 close() 方法释放资源。
  3. 热插拔: listen() 方法会阻塞当前线程,建议在单独的线程中运行。
  4. 设备权限: 在 Linux 系统上,可能需要配置 udev 规则以获得访问 USB 设备的权限。
  5. 设备识别: 设备是通过 VID/PID 识别的,确保设备驱动程序正确加载。
Last Updated:
Contributors: JKWTCN
Prev
依赖安装
Next
StreamDock 基类