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 设备的列表
工作原理:
- 遍历所有已知的产品 ID 列表
- 使用传输层枚举每个 VID/PID 组合的设备
- 为每个找到的设备创建相应的设备类实例
- 返回所有设备实例的列表
示例:
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()
监听设备连接和断开事件。此方法会阻塞当前线程,建议在单独的线程中运行。
工作原理:
- 使用 pyudev 监听 USB 子系统事件
- 过滤出 'add' 和 'remove' 事件
- 解析设备的 Vendor ID 和 Product ID
- 对于匹配的设备,创建相应的设备实例或从列表中移除
示例:
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()
注意事项
- 线程安全: DeviceManager 不是线程安全的,如果在多线程环境中使用,需要添加适当的同步机制。
- 资源管理: 使用完设备后,记得调用
close()方法释放资源。 - 热插拔:
listen()方法会阻塞当前线程,建议在单独的线程中运行。 - 设备权限: 在 Linux 系统上,可能需要配置 udev 规则以获得访问 USB 设备的权限。
- 设备识别: 设备是通过 VID/PID 识别的,确保设备驱动程序正确加载。
