Basic Examples

Below is a complete cross-platform example for enumerating all connected StreamDock devices and performing basic operations. These examples work correctly on Windows, macOS, and Linux systems.

Basic Device Operations

from StreamDock.DeviceManager import DeviceManager
import threading
import time

def main():
    # Create device manager
    manager = DeviceManager()

    # Start device listening thread (for hot-plug detection)
    listen_thread = threading.Thread(target=manager.listen)
    listen_thread.daemon = True  # Set as daemon thread
    listen_thread.start()

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

    # Operate each device
    for device in devices:
        print(f"Operating device: {device.getPath()}")

        try:
            # Open device
            device.open()

            # Initialize device (wake screen, set brightness, clear icons)
            device.init()

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

            device.set_key_callback(key_callback)

            # Set background image (ensure image file exists)
            try:
                device.set_touchscreen_image("bg.jpg")
                print("Background image set successfully")
            except Exception as e:
                print(f"Failed to set background image: {e}")

            time.sleep(1)

            # Set key icons
            try:
                device.set_key_image(1, "icon1.jpg")
                device.set_key_image(2, "icon2.jpg")
                print("Key icons set successfully")
            except Exception as e:
                print(f"Failed to set key icons: {e}")

            time.sleep(2)

            # Clear key icon
            device.clearIcon(1)
            print("Key 1 icon cleared")

            time.sleep(1)

            # Clear all key icons
            device.clearAllIcon()
            print("All key icons cleared")

            # Refresh display
            device.refresh()

            time.sleep(1)

            # LED control example (if device supports)
            if hasattr(device.feature_option, 'hasRGBLed') and device.feature_option.hasRGBLed:
                try:
                    # Set LED color to blue
                    device.set_led_color(0, 0, 255)
                    print("LED color set to blue")

                    time.sleep(2)

                    # Set LED brightness to 50%
                    device.set_led_brightness(50)
                    print("LED brightness set to 50%")

                    time.sleep(2)

                    # Reset LED effect
                    device.reset_led_effect()
                    print("LED effect reset")

                except Exception as e:
                    print(f"LED control failed: {e}")
            else:
                print("Device does not support LED functionality")

            time.sleep(1)

            time.sleep(1)

        except Exception as e:
            print(f"Error operating device: {e}")

        finally:
            # Close device
            try:
                device.close()
                print("Device closed")
            except Exception as e:
                print(f"Error closing device: {e}")

        time.sleep(1)

    print("All device operations completed")

if __name__ == "__main__":
    main()

Advanced Example: Device Controller

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

        # Start device listening
        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"""
        current_devices = self.manager.enumerate()
        current_paths = {device.getPath() for device in current_devices}
        known_paths = {device.getPath() for device in self.devices}

        # Add new devices
        for device in current_devices:
            if device.getPath() not in known_paths:
                self._setup_device(device)
                self.devices.append(device)
                print(f"New device connected: {device.getPath()}")

        # Remove disconnected devices
        self.devices = [device for device in self.devices
                       if device.getPath() in current_paths]

        print(f"Currently connected device count: {len(self.devices)}")

    def _setup_device(self, device):
        """Setup 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"Error setting up device {device.getPath()}: {e}")

    def _on_key_press(self, device, key, state):
        """Handle key events"""
        action = "pressed" if state else "released"
        print(f"Device {device.getPath()} key {key} {action}")

        if state:  # When key is pressed
            self._handle_key_press(device, key)

    def _handle_key_press(self, device, key):
        """Handle key press event"""
        try:
            # Create a simple response image
            image = create_image(device, background='red')
            draw = ImageDraw.Draw(image)

            # Draw key number
            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)

            # Convert and set image
            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")

            # Restore original icon after 1 second
            threading.Timer(1.0, self._restore_key, args=[device, key]).start()

        except Exception as e:
            print(f"Error handling key {key}: {e}")

    def _restore_key(self, device, key):
        """Restore key icon"""
        try:
            device.clearIcon(key)
            # Delete temporary file
            import os
            try:
                os.remove(f"temp_key_{key}.jpg")
            except:
                pass
        except Exception as e:
            print(f"Error restoring key {key}: {e}")

    def _create_initial_icons(self, device):
        """Create initial icons"""
        try:
            # Create simple number icons for each key
            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)

                # Convert and set image
                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 {device.getPath()} initial icons created")

        except Exception as e:
            print(f"Error creating initial icons: {e}")

    def run(self):
        """Run controller"""
        print("StreamDock controller started, press Ctrl+C to exit")

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

        # Clean up resources
        self._cleanup()

    def _cleanup(self):
        """Clean up resources"""
        print("Cleaning up resources...")

        # Close all devices
        for device in self.devices:
            try:
                device.close()
                print(f"Device {device.getPath()} closed")
            except Exception as e:
                print(f"Error closing device {device.getPath()}: {e}")

        # Delete temporary files
        import os
        for key in range(1, 16):
            for prefix in ["temp", "initial"]:
                try:
                    os.remove(f"{prefix}_key_{key}.jpg")
                except:
                    pass

        print("Resource cleanup completed")

# Use controller
if __name__ == "__main__":
    controller = StreamDockController()
    controller.run()

Async Example

import asyncio
from StreamDock.DeviceManager import DeviceManager

async def async_key_handler(device, key, state):
    """Async key event handler"""
    action = "pressed" if state else "released"
    print(f"Async processing: Device {device.getPath()} key {key} {action}")

    if state:
        # Simulate async operation
        await asyncio.sleep(0.5)
        print(f"Async operation completed: Key {key}")

async def main():
    """Async main function"""
    manager = DeviceManager()
    devices = manager.enumerate()

    print(f"Found {len(devices)} devices")

    for device in devices:
        device.open()
        device.init()

        # Set async callback
        device.set_key_callback_async(async_key_handler)

        print(f"Device {device.getPath()} async callback set")

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

    # Cleanup
    for device in devices:
        device.close()

if __name__ == "__main__":
    asyncio.run(main())

Notes

1. Error Handling: All examples include appropriate error handling to ensure programs don't crash when device operations fail.

2. Resource Management: Remember to call close() method to release resources after using devices.

3. Thread Safety: When using in multi-threaded environments, pay attention to adding appropriate synchronization mechanisms.

4. File Paths: Ensure image files referenced in examples exist, or modify to actual existing file paths.

5. Device Compatibility: Different device models may have different key counts and screen sizes, please adjust code according to actual device.