基本示例
下面是一个完整的跨平台示例,用于枚举所有连接的 StreamDock 设备并进行基本操作。这些示例在 Windows、macOS 和 Linux 系统上都能正常运行。
基本设备操作
from StreamDock.DeviceManager import DeviceManager
import threading
import time
def main():
# 创建设备管理器
manager = DeviceManager()
# 启动设备监听线程(用于热插拔检测)
listen_thread = threading.Thread(target=manager.listen)
listen_thread.daemon = True # 设置为守护线程
listen_thread.start()
# 枚举当前连接的设备
devices = manager.enumerate()
print(f"找到 {len(devices)} 个 StreamDock 设备")
# 操作每个设备
for device in devices:
print(f"正在操作设备: {device.getPath()}")
try:
# 打开设备
device.open()
# 初始化设备(唤醒屏幕、设置亮度、清除图标)
device.init()
# 设置按键事件回调
def key_callback(device, key, state):
action = "按下" if state else "释放"
print(f"设备 {device.getPath()} 按键 {key} {action}")
device.set_key_callback(key_callback)
# 设置背景图片(确保图片文件存在)
try:
device.set_touchscreen_image("bg.jpg")
print("背景图片设置成功")
except Exception as e:
print(f"设置背景图片失败: {e}")
time.sleep(1)
# 设置按键图标
try:
device.set_key_image(1, "icon1.jpg")
device.set_key_image(2, "icon2.jpg")
print("按键图标设置成功")
except Exception as e:
print(f"设置按键图标失败: {e}")
time.sleep(2)
# 清除按键图标
device.clearIcon(1)
print("按键1图标已清除")
time.sleep(1)
# 清除所有按键图标
device.clearAllIcon()
print("所有按键图标已清除")
# 刷新显示
device.refresh()
time.sleep(1)
# LED 控制示例(如果设备支持)
if hasattr(device.feature_option, 'hasRGBLed') and device.feature_option.hasRGBLed:
try:
# 设置 LED 颜色为蓝色
device.set_led_color(0, 0, 255)
print("LED 颜色设置为蓝色")
time.sleep(2)
# 设置 LED 亮度为 50%
device.set_led_brightness(50)
print("LED 亮度设置为 50%")
time.sleep(2)
# 重置 LED 效果
device.reset_led_effect()
print("LED 效果已重置")
except Exception as e:
print(f"LED 控制失败: {e}")
else:
print("设备不支持 LED 功能")
time.sleep(1)
time.sleep(1)
except Exception as e:
print(f"操作设备时出错: {e}")
finally:
# 关闭设备
try:
device.close()
print("设备已关闭")
except Exception as e:
print(f"关闭设备时出错: {e}")
time.sleep(1)
print("所有设备操作完成")
if __name__ == "__main__":
main()
高级示例:设备控制器
from StreamDock.DeviceManager import DeviceManager
from StreamDock.ImageHelpers.PILHelper import create_image, to_native_key_format
import threading
import time
from PIL import Image, ImageDraw, ImageFont
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):
"""刷新设备列表"""
current_devices = self.manager.enumerate()
current_paths = {device.getPath() for device in current_devices}
known_paths = {device.getPath() for device in self.devices}
# 添加新设备
for device in current_devices:
if device.getPath() not in known_paths:
self._setup_device(device)
self.devices.append(device)
print(f"新设备已连接: {device.getPath()}")
# 移除断开的设备
self.devices = [device for device in self.devices
if device.getPath() in current_paths]
print(f"当前连接的设备数量: {len(self.devices)}")
def _setup_device(self, device):
"""设置设备"""
try:
device.open()
device.init()
device.set_key_callback(self._on_key_press)
self._create_initial_icons(device)
except Exception as e:
print(f"设置设备 {device.getPath()} 时出错: {e}")
def _on_key_press(self, device, key, state):
"""处理按键事件"""
action = "按下" if state else "释放"
print(f"设备 {device.getPath()} 按键 {key} {action}")
if state: # 按键按下时
self._handle_key_press(device, key)
def _handle_key_press(self, device, key):
"""处理按键按下事件"""
try:
# 创建一个简单的响应图像
image = create_image(device, background='red')
draw = ImageDraw.Draw(image)
# 绘制按键编号
try:
font = ImageFont.truetype("arial.ttf", 40)
except:
font = ImageFont.load_default()
text = str(key)
bbox = draw.textbbox((0, 0), text, font=font)
text_width = bbox[2] - bbox[0]
text_height = bbox[3] - bbox[1]
x = (image.width - text_width) // 2
y = (image.height - text_height) // 2
draw.text((x, y), text, fill='white', font=font)
# 转换并设置图像
native_image = to_native_key_format(device, image)
native_image.save(f"temp_key_{key}.jpg", "JPEG", quality=100)
device.set_key_image(key, f"temp_key_{key}.jpg")
# 1秒后恢复原始图标
threading.Timer(1.0, self._restore_key, args=[device, key]).start()
except Exception as e:
print(f"处理按键 {key} 时出错: {e}")
def _restore_key(self, device, key):
"""恢复按键图标"""
try:
device.clearIcon(key)
# 删除临时文件
import os
try:
os.remove(f"temp_key_{key}.jpg")
except:
pass
except Exception as e:
print(f"恢复按键 {key} 时出错: {e}")
def _create_initial_icons(self, device):
"""创建初始图标"""
try:
# 为每个按键创建简单的数字图标
for key in range(1, min(device.KEY_COUNT + 1, 16)):
image = create_image(device, background='blue')
draw = ImageDraw.Draw(image)
try:
font = ImageFont.truetype("arial.ttf", 40)
except:
font = ImageFont.load_default()
text = str(key)
bbox = draw.textbbox((0, 0), text, font=font)
text_width = bbox[2] - bbox[0]
text_height = bbox[3] - bbox[1]
x = (image.width - text_width) // 2
y = (image.height - text_height) // 2
draw.text((x, y), text, fill='white', font=font)
# 转换并设置图像
native_image = to_native_key_format(device, image)
native_image.save(f"initial_key_{key}.jpg", "JPEG", quality=100)
device.set_key_image(key, f"initial_key_{key}.jpg")
print(f"设备 {device.getPath()} 初始图标已创建")
except Exception as e:
print(f"创建初始图标时出错: {e}")
def run(self):
"""运行控制器"""
print("StreamDock 控制器已启动,按 Ctrl+C 退出")
try:
while self.running:
# 定期刷新设备列表
self._refresh_devices()
time.sleep(5)
except KeyboardInterrupt:
print("\n正在停止控制器...")
self.running = False
# 清理资源
self._cleanup()
def _cleanup(self):
"""清理资源"""
print("正在清理资源...")
# 关闭所有设备
for device in self.devices:
try:
device.close()
print(f"设备 {device.getPath()} 已关闭")
except Exception as e:
print(f"关闭设备 {device.getPath()} 时出错: {e}")
# 删除临时文件
import os
for key in range(1, 16):
for prefix in ["temp", "initial"]:
try:
os.remove(f"{prefix}_key_{key}.jpg")
except:
pass
print("资源清理完成")
# 使用控制器
if __name__ == "__main__":
controller = StreamDockController()
controller.run()
异步示例
import asyncio
from StreamDock.DeviceManager import DeviceManager
async def async_key_handler(device, key, state):
"""异步按键事件处理"""
action = "按下" if state else "释放"
print(f"异步处理: 设备 {device.getPath()} 按键 {key} {action}")
if state:
# 模拟异步操作
await asyncio.sleep(0.5)
print(f"异步操作完成: 按键 {key}")
async def main():
"""异步主函数"""
manager = DeviceManager()
devices = manager.enumerate()
print(f"找到 {len(devices)} 个设备")
for device in devices:
device.open()
device.init()
# 设置异步回调
device.set_key_callback_async(async_key_handler)
print(f"设备 {device.getPath()} 已设置异步回调")
# 保持程序运行
try:
while True:
await asyncio.sleep(1)
except KeyboardInterrupt:
print("程序退出")
# 清理
for device in devices:
device.close()
if __name__ == "__main__":
asyncio.run(main())
注意事项
错误处理: 所有示例都包含了适当的错误处理,确保程序在设备操作失败时不会崩溃。
资源管理: 使用完设备后记得调用
close()方法释放资源。线程安全: 在多线程环境中使用时,注意添加适当的同步机制。
文件路径: 确保示例中引用的图像文件存在,或者修改为实际存在的文件路径。
设备兼容性: 不同型号的设备可能有不同的按键数量和屏幕尺寸,请根据实际设备调整代码。
