Skip to content

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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
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.
    """

    @staticmethod
    def scan(timeout: float = 5.0):
        """Scan for compatible LED devices synchronously."""
        return scan_devices_sync(timeout=timeout)

    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
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
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
222
223
224
225
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
227
228
229
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
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
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
137
138
139
140
141
142
143
144
145
146
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 ROM data on the device.

Source code in src/pypixelcolor/client.py
119
# Dynamically add command methods to AsyncClient

connect()

Connect to the BLE device and retrieve device info.

Source code in src/pypixelcolor/client.py
 99
100
101
    DeviceInfo object with device specifications.

Raises:

delete(n)

Delete a specific slot by its index.

Source code in src/pypixelcolor/client.py
122
123
for cmd_name, cmd_func in COMMANDS.items():
    setattr(AsyncClient, cmd_name, _create_async_method(cmd_name, cmd_func))

disconnect()

Disconnect from the BLE device.

Source code in src/pypixelcolor/client.py
103
104
"""
return self._session.get_device_info()

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
107
108
109
110
111
112
113
    """Ensure the client is connected."""
    if not self._connected:
        raise RuntimeError("Client not connected. Call connect() first")

async def __aenter__(self):
    """Async context manager entry."""
    await self.connect()

scan(timeout=5.0) staticmethod

Scan for compatible LED devices synchronously.

Source code in src/pypixelcolor/client.py
132
133
134
135
@staticmethod
def scan(timeout: float = 5.0):
    """Scan for compatible LED devices synchronously."""
    return scan_devices_sync(timeout=timeout)

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
125
126
class Client:
    """Synchronous client for controlling the LED matrix via BLE.

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
128
129
130
This is a synchronous wrapper around AsyncClient that handles the event loop
automatically for simpler usage in non-async code.
"""

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

Send a text to the device with configurable parameters.

Source code in src/pypixelcolor/client.py
131
132
@staticmethod
def scan(timeout: float = 5.0):

set_brightness(level)

Set the brightness of the device.

Source code in src/pypixelcolor/client.py
134
135
"""Scan for compatible LED devices synchronously."""
return scan_devices_sync(timeout=timeout)

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
137
138
def __init__(self, address: str):
    """Initialize the Client.

set_fun_mode(enable=False)

Enable or disable fun mode.

Source code in src/pypixelcolor/client.py
140
141
142
Args:
    address: Bluetooth device address (e.g., "1D:6B:5E:B5:A5:54")
"""

set_orientation(orientation=0)

Set the orientation of the device.

Source code in src/pypixelcolor/client.py
143
144
145
self._async_client = AsyncClient(address)
self._loop: Optional[asyncio.AbstractEventLoop] = None
self._loop_thread = None

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

Defines the color of a specific pixel.

Source code in src/pypixelcolor/client.py
146
147
148
    self._setup_loop()

def _setup_loop(self):

set_power(on=True)

Set the power state of the device.

Source code in src/pypixelcolor/client.py
149
150
"""Set up a persistent event loop in a separate thread."""
import threading

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
152
153
154
def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

set_rhythm_mode_2(style=0, t=0)

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

Source code in src/pypixelcolor/client.py
155
156
self._loop = asyncio.new_event_loop()
self._loop_thread = threading.Thread(target=start_loop, args=(self._loop,), daemon=True)

set_schedule(hour, minute, on, days=None, slot=0)

Set an automated on/off schedule by time and day of week.

Source code in src/pypixelcolor/client.py
158
159
160
self._loop_thread.start()

# Register cleanup handler to ensure disconnection on exit

set_scores(score_p1=0, score_p2=0)

Set scoreboard scores for player 1 and player 2.

Source code in src/pypixelcolor/client.py
161
162
163
    atexit.register(self._cleanup_on_exit)

def _cleanup_on_exit(self):

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

Set the device time.

Source code in src/pypixelcolor/client.py
164
165
166
"""Cleanup handler for atexit."""
try:
    if self._async_client._connected:

set_timer(action)

Set timer action (start, pause, stop).

Source code in src/pypixelcolor/client.py
167
168
169
        self.disconnect()
except Exception:
    pass

show_slot(number)

Shows the specified slot on the device.

Source code in src/pypixelcolor/client.py
170
171
finally:
    self._cleanup_loop()

version()

Get the client library version.

Source code in src/pypixelcolor/client.py
115
116
async def __aexit__(self, exc_type, exc_val, exc_tb):
    """Async context manager exit."""