DeviceManager

Central device manager. An instance of this class must be created to detect and use StreamDock devices. DeviceManager provides cross-platform device detection and management functionality, supporting Windows, macOS, and Linux systems.

Overview

DeviceManager is the core component of the StreamDock SDK, responsible for device enumeration, detection, and listening. It identifies connected StreamDock devices through USB Vendor ID and Product ID, and supports device hot-plug detection.

Cross-Platform Support

DeviceManager uses different underlying transport methods depending on the operating system:

• Windows: Uses hidapi library for HID device communication
• macOS: Uses system native IOKit framework
• Linux: Uses pyudev and libusb for device management

Constructor

DeviceManager(transport=None)

Creates a new StreamDock DeviceManager instance for detecting connected StreamDock devices.

Parameters:

• transport (optional): Transport layer instance, defaults to LibUSBHIDAPI

Example:

from StreamDock.DeviceManager import DeviceManager

# Use default transport layer
manager = DeviceManager()

# Use custom transport layer
# custom_transport = CustomTransport()
# manager = DeviceManager(transport=custom_transport)

Methods

enumerate()

enumerate()

Detects connected StreamDock devices.

Return Value:

• list: List containing all connected StreamDock devices

How it works:

1. Iterates through all known product ID lists
2. Uses transport layer to enumerate devices for each VID/PID combination
3. Creates corresponding device class instances for each found device
4. Returns list of all device instances

Example:

manager = DeviceManager()
streamdocks = manager.enumerate()
print(f"Found {len(streamdocks)} StreamDock devices")

for device in streamdocks:
    print(f"Device path: {device.getPath()}")
    print(f"Device type: {device.DECK_TYPE}")

listen()

listen()

Listens for device connection and disconnection events. This method blocks the current thread, recommended to run in a separate thread.

How it works:

1. Uses pyudev to listen for USB subsystem events
2. Filters out 'add' and 'remove' events
3. Parses device Vendor ID and Product ID
4. For matching devices, creates corresponding device instances or removes from list

Example:

import threading
from StreamDock.DeviceManager import DeviceManager

manager = DeviceManager()

# Start listening in separate thread
t = threading.Thread(target=manager.listen)
t.daemon = True  # Set as daemon thread, automatically ends when main program exits
t.start()

print("Starting to listen for device plug/unplug events...")

# Main thread continues other tasks
try:
    while True:
        # Periodically check device list
        devices = manager.enumerate()
        print(f"Currently connected device count: {len(devices)}")
        time.sleep(5)
except KeyboardInterrupt:
    print("Stop listening")

Usage Examples

Basic Usage

import threading
import time
from StreamDock.DeviceManager import DeviceManager

# Create device manager
manager = DeviceManager()

# Start device listening thread
listen_thread = threading.Thread(target=manager.listen)
listen_thread.daemon = True
listen_thread.start()

# Enumerate currently connected devices
devices = manager.enumerate()
print(f"Found {len(devices)} devices")

# Operate each device
for device in devices:
    device.open()
    device.init()  # Initialize device
    print(f"Device {device.getPath()} initialized")

    # Set key callback
    def key_callback(device, key, state):
        print(f"Device {device.getPath()} key {key} {'pressed' if state else 'released'}")

    device.set_key_callback(key_callback)

# Keep program running
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    print("Program exit")

Advanced Usage

import threading
import time
from StreamDock.DeviceManager import DeviceManager

class StreamDockController:
    def __init__(self):
        self.manager = DeviceManager()
        self.devices = []
        self.running = True

        # Start listening thread
        self.listen_thread = threading.Thread(target=self._listen_devices)
        self.listen_thread.daemon = True
        self.listen_thread.start()

        # Initial device enumeration
        self._refresh_devices()

    def _listen_devices(self):
        """Listen for device plug/unplug events"""
        self.manager.listen()

    def _refresh_devices(self):
        """Refresh device list"""
        self.devices = self.manager.enumerate()
        print(f"Currently connected device count: {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 {device.getPath()} initialized")

    def _on_key_press(self, device, key, state):
        """Key event handler"""
        print(f"Device {device.getPath()} key {key} {'pressed' if state else 'released'}")

    def run(self):
        """Run controller"""
        try:
            while self.running:
                # Periodically refresh device list
                self._refresh_devices()
                time.sleep(5)
        except KeyboardInterrupt:
            print("Stop controller")
            self.running = False

# Use controller
controller = StreamDockController()
controller.run()

Notes

1. Thread Safety: DeviceManager is not thread-safe. If used in multi-threaded environments, appropriate synchronization mechanisms need to be added.
2. Resource Management: After using devices, remember to call close() method to release resources.
3. Hot-plugging: The listen() method blocks the current thread, recommended to run in a separate thread.
4. Device Permissions: On Linux systems, may need to configure udev rules to gain access to USB devices.
5. Device Recognition: Devices are identified through VID/PID, ensure device drivers are properly loaded.