Sync Client

pypixelcolor.client.Client

Synchronous client for controlling the LED matrix via BLE.

This is a synchronous wrapper around AsyncClient that handles the event loop automatically for simpler usage in non-async code.

Source code in src/pypixelcolor/client.py

class Client:
    """Synchronous client for controlling the LED matrix via BLE.

    This is a synchronous wrapper around AsyncClient that handles the event loop
    automatically for simpler usage in non-async code.
    """

    def __init__(self, address: str):
        """Initialize the Client.

        Args:
            address: Bluetooth device address (e.g., "1D:6B:5E:B5:A5:54")
        """
        self._async_client = AsyncClient(address)
        self._loop: Optional[asyncio.AbstractEventLoop] = None
        self._loop_thread = None
        self._setup_loop()

    def _setup_loop(self):
        """Set up a persistent event loop in a separate thread."""
        import threading

        def start_loop(loop):
            asyncio.set_event_loop(loop)
            loop.run_forever()

        self._loop = asyncio.new_event_loop()
        self._loop_thread = threading.Thread(target=start_loop, args=(self._loop,), daemon=True)
        self._loop_thread.start()

        # Register cleanup handler to ensure disconnection on exit
        atexit.register(self._cleanup_on_exit)

    def _cleanup_on_exit(self):
        """Cleanup handler for atexit."""
        try:
            if self._async_client._connected:
                self.disconnect()
        except Exception:
            pass
        finally:
            self._cleanup_loop()

    def _run_async(self, coro):
        """Run an async coroutine synchronously using the persistent loop."""
        if self._loop is None:
            raise RuntimeError("Event loop not initialized")

        future = asyncio.run_coroutine_threadsafe(coro, self._loop)
        return future.result()

    def __getattr__(self, name: str):
        """Delegate attribute access to async client, wrapping coroutines.

        This allows automatic synchronous wrapping of all async methods without
        explicit redefinition, reducing code duplication.
        """
        attr = getattr(self._async_client, name)

        # If it's a coroutine function, wrap it to run synchronously
        if asyncio.iscoroutinefunction(attr):
            def sync_wrapper(*args, **kwargs):
                return self._run_async(attr(*args, **kwargs))
            # Preserve function metadata for better IDE support
            sync_wrapper.__name__ = name
            sync_wrapper.__doc__ = attr.__doc__
            return sync_wrapper

        # Otherwise return the attribute as-is
        return attr

    def _cleanup_loop(self):
        """Clean up the event loop."""
        if self._loop and self._loop.is_running():
            self._loop.call_soon_threadsafe(self._loop.stop)

    def __del__(self):
        """Cleanup on deletion."""
        # Unregister atexit handler if object is deleted manually
        try:
            atexit.unregister(self._cleanup_on_exit)
        except Exception:
            pass

        try:
            if self._async_client._connected:
                self.disconnect()
        except Exception:
            pass
        finally:
            self._cleanup_loop()

    def __enter__(self):
        """Context manager entry."""
        self.connect()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit."""
        self.disconnect()

__del__()

Cleanup on deletion.

Source code in src/pypixelcolor/client.py

def __del__(self):
    """Cleanup on deletion."""
    # Unregister atexit handler if object is deleted manually
    try:
        atexit.unregister(self._cleanup_on_exit)
    except Exception:
        pass

    try:
        if self._async_client._connected:
            self.disconnect()
    except Exception:
        pass
    finally:
        self._cleanup_loop()

__enter__()

Context manager entry.

Source code in src/pypixelcolor/client.py

def __enter__(self):
    """Context manager entry."""
    self.connect()
    return self

__exit__(exc_type, exc_val, exc_tb)

Context manager exit.

Source code in src/pypixelcolor/client.py

def __exit__(self, exc_type, exc_val, exc_tb):
    """Context manager exit."""
    self.disconnect()

__getattr__(name)

Delegate attribute access to async client, wrapping coroutines.

This allows automatic synchronous wrapping of all async methods without explicit redefinition, reducing code duplication.

Source code in src/pypixelcolor/client.py

def __getattr__(self, name: str):
    """Delegate attribute access to async client, wrapping coroutines.

    This allows automatic synchronous wrapping of all async methods without
    explicit redefinition, reducing code duplication.
    """
    attr = getattr(self._async_client, name)

    # If it's a coroutine function, wrap it to run synchronously
    if asyncio.iscoroutinefunction(attr):
        def sync_wrapper(*args, **kwargs):
            return self._run_async(attr(*args, **kwargs))
        # Preserve function metadata for better IDE support
        sync_wrapper.__name__ = name
        sync_wrapper.__doc__ = attr.__doc__
        return sync_wrapper

    # Otherwise return the attribute as-is
    return attr

__init__(address)

Initialize the Client.

Parameters:

Name	Type	Description	Default
address	str	Bluetooth device address (e.g., "1D:6B:5E:B5:A5:54")	required

Source code in src/pypixelcolor/client.py

def __init__(self, address: str):
    """Initialize the Client.

    Args:
        address: Bluetooth device address (e.g., "1D:6B:5E:B5:A5:54")
    """
    self._async_client = AsyncClient(address)
    self._loop: Optional[asyncio.AbstractEventLoop] = None
    self._loop_thread = None
    self._setup_loop()

clear()

Clears the EEPROM.

Source code in src/pypixelcolor/client.py

async def __aexit__(self, exc_type, exc_val, exc_tb):
    """Async context manager exit."""
    await self.disconnect()

connect()

Connect to the BLE device and retrieve device info.

Source code in src/pypixelcolor/client.py

This is a simple getter for the cached data.

Returns:

delete(n)

Delete a specific screen by its index.

Source code in src/pypixelcolor/client.py

# Dynamically add command methods to AsyncClient

disconnect()

Disconnect from the BLE device.

Source code in src/pypixelcolor/client.py

Raises:
    RuntimeError: If not connected.

get_device_info()

Get cached device information.

Device info is automatically retrieved during connect(). This is a simple getter for the cached data.

Source code in src/pypixelcolor/client.py

    return self._session.get_device_info()

async def _ensure_connected(self) -> None:
    """Ensure the client is connected."""
    if not self._connected:
        raise RuntimeError("Client not connected. Call connect() first")

send_image(path, resize_method=ResizeMethod.CROP, device_info=None, save_slot=0)

Send an image or animation.

Source code in src/pypixelcolor/client.py

for cmd_name, cmd_func in COMMANDS.items():
    setattr(AsyncClient, cmd_name, _create_async_method(cmd_name, cmd_func))

send_image_hex(hex_string, file_extension, resize_method=ResizeMethod.CROP, device_info=None, save_slot=0)

Send an image or animation from a hexadecimal string.

Source code in src/pypixelcolor/client.py

class Client:
    """Synchronous client for controlling the LED matrix via BLE.

send_text(text, rainbow_mode=0, animation=0, save_slot=0, speed=80, color='ffffff', bg_color=None, font='CUSONG', char_height=None, device_info=None)

Send a text to the device with configurable parameters.

Source code in src/pypixelcolor/client.py

This is a synchronous wrapper around AsyncClient that handles the event loop
automatically for simpler usage in non-async code.
"""

set_brightness(level)

Set the brightness of the device.

Source code in src/pypixelcolor/client.py

def __init__(self, address: str):
    """Initialize the Client.

set_clock_mode(style=1, date='', show_date=True, format_24=True)

Set the clock mode of the device.

Source code in src/pypixelcolor/client.py

Args:
    address: Bluetooth device address (e.g., "1D:6B:5E:B5:A5:54")

set_fun_mode(enable=False)

Enable or disable fun mode.

Source code in src/pypixelcolor/client.py

"""
self._async_client = AsyncClient(address)
self._loop: Optional[asyncio.AbstractEventLoop] = None

set_orientation(orientation=0)

Set the orientation of the device.

Source code in src/pypixelcolor/client.py

self._loop_thread = None
self._setup_loop()

set_pixel(x, y, color, device_info=None)

Defines the color of a specific pixel.

Source code in src/pypixelcolor/client.py

def _setup_loop(self):
    """Set up a persistent event loop in a separate thread."""
    import threading

set_power(on=True)

Set the power state of the device.

Source code in src/pypixelcolor/client.py

def start_loop(loop):
    asyncio.set_event_loop(loop)

set_rhythm_mode(style=0, l1=0, l2=0, l3=0, l4=0, l5=0, l6=0, l7=0, l8=0, l9=0, l10=0, l11=0)

Set the rhythm mode of the device.

Source code in src/pypixelcolor/client.py

    loop.run_forever()

self._loop = asyncio.new_event_loop()

set_rhythm_mode_2(style=0, t=0)

Set the rhythm mode of the device (alternative version).

Source code in src/pypixelcolor/client.py

self._loop_thread = threading.Thread(target=start_loop, args=(self._loop,), daemon=True)
self._loop_thread.start()

set_time(hour=None, minute=None, second=None)

Set the device time.

Source code in src/pypixelcolor/client.py

# Register cleanup handler to ensure disconnection on exit
atexit.register(self._cleanup_on_exit)

show_slot(number)

Shows the specified slot on the device.

Source code in src/pypixelcolor/client.py

def _cleanup_on_exit(self):
    """Cleanup handler for atexit."""
    try:

version()

Get the client library version.

Source code in src/pypixelcolor/client.py

"""Async context manager entry."""
await self.connect()
return self