Skip to content

SignalRGB Client API Reference

This page provides detailed API documentation for both the synchronous SignalRGBClient class and the asynchronous AsyncSignalRGBClient class, which are the main interfaces for interacting with the SignalRGB API.

SignalRGBClient

signalrgb.client.SignalRGBClient

Client for interacting with the SignalRGB API.

This class provides methods to interact with the SignalRGB API, allowing users to retrieve, apply, and manage lighting effects and layouts.

This class is a wrapper around the AsyncSignalRGBClient that provides a synchronous interface to the API. Most methods simply delegate to the async client.

Source code in signalrgb/client.py
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
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
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
class SignalRGBClient:
    """Client for interacting with the SignalRGB API.

    This class provides methods to interact with the SignalRGB API, allowing users
    to retrieve, apply, and manage lighting effects and layouts.

    This class is a wrapper around the AsyncSignalRGBClient that provides a synchronous
    interface to the API. Most methods simply delegate to the async client.
    """

    def __init__(self, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, timeout: float = DEFAULT_TIMEOUT):
        """Initialize the SignalRGBClient.

        Args:
            host: The host of the SignalRGB API. Defaults to 'localhost'.
            port: The port of the SignalRGB API. Defaults to DEFAULT_PORT.
            timeout: The timeout for API requests in seconds. Defaults to 10.0.

        Example:
            >>> client = SignalRGBClient()
            >>> client = SignalRGBClient("192.168.1.100", 8080, 5.0)
        """
        self._base_url = f"http://{host}:{port}"
        self._timeout = timeout
        self._effects_cache: list[Effect] | None = None
        # For backward compatibility, we still maintain the session
        self._session = requests.Session()
        # Create an AsyncSignalRGBClient for internal use
        self._async_client = AsyncSignalRGBClient(host, port, timeout)
        # Create and manage an event loop for running async code
        self._loop = asyncio.new_event_loop()

    def __enter__(self) -> SignalRGBClient:
        """Context manager entry."""
        return self

    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        """Context manager exit."""
        self._session.close()
        try:
            # Close the async client if the loop is still running
            if not self._loop.is_closed():
                self._loop.run_until_complete(self._async_client.aclose())
                self._loop.close()
        except RuntimeError:
            # The loop might already be closed, just ignore
            pass

    def _run_async(self, coro: Awaitable[T]) -> T:
        """Run an asynchronous coroutine in a synchronous context.

        Args:
            coro: The coroutine to run.

        Returns:
            The result of the coroutine with preserved type.
        """
        try:
            if self._loop.is_closed():
                self._loop = asyncio.new_event_loop()
            return self._loop.run_until_complete(coro)
        except RuntimeError as e:
            # If we get a runtime error about the event loop, create a new one
            if "This event loop is already running" in str(e):
                # Create a new event loop for this thread
                new_loop = asyncio.new_event_loop()
                asyncio.set_event_loop(new_loop)
                self._loop = new_loop
                return self._loop.run_until_complete(coro)
            raise

    # Using a standard method with cached property pattern instead of lru_cache on method
    def _get_effects_cached(self) -> list[Effect]:
        """Internal method to get effects with caching."""
        return self._run_async(self._async_client.get_effects_cached())

    def get_effects(self) -> list[Effect]:
        """List available effects.

        Returns:
            List[Effect]: A list of available effects.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an error retrieving the effects.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> effects = client.get_effects()
            >>> print(f"Found {len(effects)} effects")
        """
        return self._run_async(self._async_client.get_effects())

    def get_effect(self, effect_id: str) -> Effect:
        """Get details of a specific effect.

        Args:
            effect_id (str): The ID of the effect to retrieve.

        Returns:
            Effect: The requested effect.

        Raises:
            NotFoundError: If the effect with the given ID is not found.
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> effect = client.get_effect("example_effect_id")
            >>> print(f"Effect name: {effect.attributes.name}")
        """
        return self._run_async(self._async_client.get_effect(effect_id))

    def get_effect_by_name(self, effect_name: str) -> Effect:
        """Get details of a specific effect by name.

        Args:
            effect_name: The name of the effect to retrieve.

        Returns:
            Effect: The requested effect.

        Raises:
            NotFoundError: If the effect with the given name is not found.
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> effect = client.get_effect_by_name("Rainbow Wave")
            >>> print(f"Effect ID: {effect.id}")
        """
        return self._run_async(self._async_client.get_effect_by_name(effect_name))

    @property
    def current_effect(self) -> Effect:
        """Get the current effect.

        Returns:
            Effect: The currently active effect.

        Raises:
            APIError: If there's an error retrieving the current effect.
        """
        return self.get_current_effect()

    def get_current_effect(self) -> Effect:
        """Get the current effect.

        Returns:
            Effect: The currently active effect.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> current_effect = client.get_current_effect()
            >>> print(f"Current effect: {current_effect.attributes.name}")
        """
        return self._run_async(self._async_client.get_current_effect())

    def _get_current_state(self) -> Any:  # Using Any for CurrentStateHolder
        """Get the current state of the SignalRGB instance.

        Returns:
            CurrentStateHolder: The current state of the SignalRGB instance.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.
        """
        return self._run_async(self._async_client.get_current_state())

    @property
    def brightness(self) -> int:
        """Get or set the current brightness level.

        Returns:
            int: The current brightness level (0-100).

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> print(f"Current brightness: {client.brightness}")
            >>> client.brightness = 75
            >>> print(f"New brightness: {client.brightness}")
        """
        return self._run_async(self._async_client.get_brightness())

    @brightness.setter
    def brightness(self, value: int) -> None:
        self._run_async(self._async_client.set_brightness(value))

    @property
    def enabled(self) -> bool:
        """Get or set the current enabled state of the canvas.

        Returns:
            bool: The current enabled state.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> print(f"Canvas enabled: {client.enabled}")
            >>> client.enabled = False
            >>> print(f"Canvas now disabled: {not client.enabled}")
        """
        return self._run_async(self._async_client.get_enabled())

    @enabled.setter
    def enabled(self, value: bool) -> None:
        self._run_async(self._async_client.set_enabled(value))

    def apply_effect(self, effect_id: str) -> None:
        """Apply an effect.

        Args:
            effect_id (str): The ID of the effect to apply.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> client.apply_effect("example_effect_id")
            >>> print("Effect applied successfully")
        """
        self._run_async(self._async_client.apply_effect(effect_id))

    def apply_effect_by_name(self, effect_name: str) -> None:
        """Apply an effect by name.

        Args:
            effect_name: The name of the effect to apply.

        Raises:
            NotFoundError: If the effect with the given name is not found.
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> client.apply_effect_by_name("Rainbow Wave")
            >>> print("Effect applied successfully")
        """
        self._run_async(self._async_client.apply_effect_by_name(effect_name))

    def get_effect_presets(self, effect_id: str) -> list[EffectPreset]:
        """Get presets for a specific effect.

        Args:
            effect_id (str): The ID of the effect to retrieve presets for.

        Returns:
            List[EffectPreset]: A list of effect presets.

        Raises:
            NotFoundError: If the effect with the given ID is not found.
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> presets = client.get_effect_presets("example_effect_id")
            >>> for preset in presets:
            ...     print(f"Preset ID: {preset.id}, Name: {preset.name}")
        """
        return self._run_async(self._async_client.get_effect_presets(effect_id))

    def apply_effect_preset(self, effect_id: str, preset_id: str) -> None:
        """Apply a preset for a specific effect.

        Args:
            effect_id (str): The ID of the effect to apply the preset to.
            preset_id (str): The ID of the preset to apply.

        Raises:
            NotFoundError: If the effect with the given ID is not found.
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> client.apply_effect_preset("example_effect_id", "My Fancy Preset 1")
            >>> print("Preset applied successfully")
        """
        self._run_async(self._async_client.apply_effect_preset(effect_id, preset_id))

    def get_next_effect(self) -> Effect | None:
        """Get information about the next effect in history.

        Returns:
            Optional[Effect]: The next effect if available, None otherwise.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> next_effect = client.get_next_effect()
            >>> if next_effect:
            ...     print(f"Next effect: {next_effect.attributes.name}")
            ... else:
            ...     print("No next effect available")
        """
        return self._run_async(self._async_client.get_next_effect())

    def apply_next_effect(self) -> Effect:
        """Apply the next effect in history or a random effect if there's no next effect.

        Returns:
            Effect: The newly applied effect.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> new_effect = client.apply_next_effect()
            >>> print(f"Applied effect: {new_effect.attributes.name}")
        """
        return self._run_async(self._async_client.apply_next_effect())

    def get_previous_effect(self) -> Effect | None:
        """Get information about the previous effect in history.

        Returns:
            Optional[Effect]: The previous effect if available, None otherwise.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> prev_effect = client.get_previous_effect()
            >>> if prev_effect:
            ...     print(f"Previous effect: {prev_effect.attributes.name}")
            ... else:
            ...     print("No previous effect available")
        """
        return self._run_async(self._async_client.get_previous_effect())

    def apply_previous_effect(self) -> Effect:
        """Apply the previous effect in history.

        Returns:
            Effect: The newly applied effect.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> new_effect = client.apply_previous_effect()
            >>> print(f"Applied effect: {new_effect.attributes.name}")
        """
        return self._run_async(self._async_client.apply_previous_effect())

    def apply_random_effect(self) -> Effect:
        """Apply a random effect.

        Returns:
            Effect: The newly applied random effect.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> random_effect = client.apply_random_effect()
            >>> print(f"Applied random effect: {random_effect.attributes.name}")
        """
        return self._run_async(self._async_client.apply_random_effect())

    @property
    def current_layout(self) -> Layout:
        """Get the current layout.

        Returns:
            Layout: The currently active layout.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> current_layout = client.current_layout
            >>> print(f"Current layout: {current_layout.id}")
        """
        return self._run_async(self._async_client.get_current_layout())

    @current_layout.setter
    def current_layout(self, layout_id: str) -> None:
        """Set the current layout.

        Args:
            layout_id: The ID of the layout to set as current.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> client.current_layout = "My Layout 1"
            >>> print(f"New current layout: {client.current_layout.id}")
        """
        self._run_async(self._async_client.set_current_layout(layout_id))

    def get_layouts(self) -> list[Layout]:
        """Get all available layouts.

        Returns:
            List[Layout]: A list of all available layouts.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> client = SignalRGBClient()
            >>> layouts = client.get_layouts()
            >>> for layout in layouts:
            ...     print(f"Layout: {layout.id}")
        """
        return self._run_async(self._async_client.get_layouts())

    def refresh_effects(self) -> None:
        """Refresh the cached effects.

        This method clears the cache for the get_effects method, forcing a fresh
        retrieval of effects on the next call.

        Example:
            >>> client = SignalRGBClient()
            >>> client.refresh_effects()
            >>> fresh_effects = client.get_effects()
        """
        self._run_async(self._async_client.refresh_effects())

    def __repr__(self) -> str:
        return f"SignalRGBClient(base_url='{self._base_url}')"

brightness property writable

Get or set the current brightness level.

Returns:

Name Type Description
int int

The current brightness level (0-100).

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() print(f"Current brightness: {client.brightness}") client.brightness = 75 print(f"New brightness: {client.brightness}")

current_effect property

Get the current effect.

Returns:

Name Type Description
Effect Effect

The currently active effect.

Raises:

Type Description
APIError

If there's an error retrieving the current effect.

current_layout property writable

Get the current layout.

Returns:

Name Type Description
Layout Layout

The currently active layout.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() current_layout = client.current_layout print(f"Current layout: {current_layout.id}")

enabled property writable

Get or set the current enabled state of the canvas.

Returns:

Name Type Description
bool bool

The current enabled state.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() print(f"Canvas enabled: {client.enabled}") client.enabled = False print(f"Canvas now disabled: {not client.enabled}")

__enter__()

Context manager entry.

Source code in signalrgb/client.py
def __enter__(self) -> SignalRGBClient:
    """Context manager entry."""
    return self

__exit__(exc_type, exc_val, exc_tb)

Context manager exit.

Source code in signalrgb/client.py
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    """Context manager exit."""
    self._session.close()
    try:
        # Close the async client if the loop is still running
        if not self._loop.is_closed():
            self._loop.run_until_complete(self._async_client.aclose())
            self._loop.close()
    except RuntimeError:
        # The loop might already be closed, just ignore
        pass

__init__(host=DEFAULT_HOST, port=DEFAULT_PORT, timeout=DEFAULT_TIMEOUT)

Initialize the SignalRGBClient.

Parameters:

Name Type Description Default
host str

The host of the SignalRGB API. Defaults to 'localhost'.

DEFAULT_HOST
port int

The port of the SignalRGB API. Defaults to DEFAULT_PORT.

DEFAULT_PORT
timeout float

The timeout for API requests in seconds. Defaults to 10.0.

DEFAULT_TIMEOUT
Example

client = SignalRGBClient() client = SignalRGBClient("192.168.1.100", 8080, 5.0)

Source code in signalrgb/client.py
def __init__(self, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, timeout: float = DEFAULT_TIMEOUT):
    """Initialize the SignalRGBClient.

    Args:
        host: The host of the SignalRGB API. Defaults to 'localhost'.
        port: The port of the SignalRGB API. Defaults to DEFAULT_PORT.
        timeout: The timeout for API requests in seconds. Defaults to 10.0.

    Example:
        >>> client = SignalRGBClient()
        >>> client = SignalRGBClient("192.168.1.100", 8080, 5.0)
    """
    self._base_url = f"http://{host}:{port}"
    self._timeout = timeout
    self._effects_cache: list[Effect] | None = None
    # For backward compatibility, we still maintain the session
    self._session = requests.Session()
    # Create an AsyncSignalRGBClient for internal use
    self._async_client = AsyncSignalRGBClient(host, port, timeout)
    # Create and manage an event loop for running async code
    self._loop = asyncio.new_event_loop()

_get_current_state()

Get the current state of the SignalRGB instance.

Returns:

Name Type Description
CurrentStateHolder Any

The current state of the SignalRGB instance.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Source code in signalrgb/client.py
def _get_current_state(self) -> Any:  # Using Any for CurrentStateHolder
    """Get the current state of the SignalRGB instance.

    Returns:
        CurrentStateHolder: The current state of the SignalRGB instance.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.
    """
    return self._run_async(self._async_client.get_current_state())

_get_effects_cached()

Internal method to get effects with caching.

Source code in signalrgb/client.py
def _get_effects_cached(self) -> list[Effect]:
    """Internal method to get effects with caching."""
    return self._run_async(self._async_client.get_effects_cached())

_run_async(coro)

Run an asynchronous coroutine in a synchronous context.

Parameters:

Name Type Description Default
coro Awaitable[T]

The coroutine to run.

required

Returns:

Type Description
T

The result of the coroutine with preserved type.

Source code in signalrgb/client.py
def _run_async(self, coro: Awaitable[T]) -> T:
    """Run an asynchronous coroutine in a synchronous context.

    Args:
        coro: The coroutine to run.

    Returns:
        The result of the coroutine with preserved type.
    """
    try:
        if self._loop.is_closed():
            self._loop = asyncio.new_event_loop()
        return self._loop.run_until_complete(coro)
    except RuntimeError as e:
        # If we get a runtime error about the event loop, create a new one
        if "This event loop is already running" in str(e):
            # Create a new event loop for this thread
            new_loop = asyncio.new_event_loop()
            asyncio.set_event_loop(new_loop)
            self._loop = new_loop
            return self._loop.run_until_complete(coro)
        raise

apply_effect(effect_id)

Apply an effect.

Parameters:

Name Type Description Default
effect_id str

The ID of the effect to apply.

required

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() client.apply_effect("example_effect_id") print("Effect applied successfully")

Source code in signalrgb/client.py
def apply_effect(self, effect_id: str) -> None:
    """Apply an effect.

    Args:
        effect_id (str): The ID of the effect to apply.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> client.apply_effect("example_effect_id")
        >>> print("Effect applied successfully")
    """
    self._run_async(self._async_client.apply_effect(effect_id))

apply_effect_by_name(effect_name)

Apply an effect by name.

Parameters:

Name Type Description Default
effect_name str

The name of the effect to apply.

required

Raises:

Type Description
NotFoundError

If the effect with the given name is not found.

ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() client.apply_effect_by_name("Rainbow Wave") print("Effect applied successfully")

Source code in signalrgb/client.py
def apply_effect_by_name(self, effect_name: str) -> None:
    """Apply an effect by name.

    Args:
        effect_name: The name of the effect to apply.

    Raises:
        NotFoundError: If the effect with the given name is not found.
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> client.apply_effect_by_name("Rainbow Wave")
        >>> print("Effect applied successfully")
    """
    self._run_async(self._async_client.apply_effect_by_name(effect_name))

apply_effect_preset(effect_id, preset_id)

Apply a preset for a specific effect.

Parameters:

Name Type Description Default
effect_id str

The ID of the effect to apply the preset to.

required
preset_id str

The ID of the preset to apply.

required

Raises:

Type Description
NotFoundError

If the effect with the given ID is not found.

ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() client.apply_effect_preset("example_effect_id", "My Fancy Preset 1") print("Preset applied successfully")

Source code in signalrgb/client.py
def apply_effect_preset(self, effect_id: str, preset_id: str) -> None:
    """Apply a preset for a specific effect.

    Args:
        effect_id (str): The ID of the effect to apply the preset to.
        preset_id (str): The ID of the preset to apply.

    Raises:
        NotFoundError: If the effect with the given ID is not found.
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> client.apply_effect_preset("example_effect_id", "My Fancy Preset 1")
        >>> print("Preset applied successfully")
    """
    self._run_async(self._async_client.apply_effect_preset(effect_id, preset_id))

apply_next_effect()

Apply the next effect in history or a random effect if there's no next effect.

Returns:

Name Type Description
Effect Effect

The newly applied effect.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() new_effect = client.apply_next_effect() print(f"Applied effect: {new_effect.attributes.name}")

Source code in signalrgb/client.py
def apply_next_effect(self) -> Effect:
    """Apply the next effect in history or a random effect if there's no next effect.

    Returns:
        Effect: The newly applied effect.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> new_effect = client.apply_next_effect()
        >>> print(f"Applied effect: {new_effect.attributes.name}")
    """
    return self._run_async(self._async_client.apply_next_effect())

apply_previous_effect()

Apply the previous effect in history.

Returns:

Name Type Description
Effect Effect

The newly applied effect.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() new_effect = client.apply_previous_effect() print(f"Applied effect: {new_effect.attributes.name}")

Source code in signalrgb/client.py
def apply_previous_effect(self) -> Effect:
    """Apply the previous effect in history.

    Returns:
        Effect: The newly applied effect.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> new_effect = client.apply_previous_effect()
        >>> print(f"Applied effect: {new_effect.attributes.name}")
    """
    return self._run_async(self._async_client.apply_previous_effect())

apply_random_effect()

Apply a random effect.

Returns:

Name Type Description
Effect Effect

The newly applied random effect.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() random_effect = client.apply_random_effect() print(f"Applied random effect: {random_effect.attributes.name}")

Source code in signalrgb/client.py
def apply_random_effect(self) -> Effect:
    """Apply a random effect.

    Returns:
        Effect: The newly applied random effect.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> random_effect = client.apply_random_effect()
        >>> print(f"Applied random effect: {random_effect.attributes.name}")
    """
    return self._run_async(self._async_client.apply_random_effect())

get_current_effect()

Get the current effect.

Returns:

Name Type Description
Effect Effect

The currently active effect.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() current_effect = client.get_current_effect() print(f"Current effect: {current_effect.attributes.name}")

Source code in signalrgb/client.py
def get_current_effect(self) -> Effect:
    """Get the current effect.

    Returns:
        Effect: The currently active effect.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> current_effect = client.get_current_effect()
        >>> print(f"Current effect: {current_effect.attributes.name}")
    """
    return self._run_async(self._async_client.get_current_effect())

get_effect(effect_id)

Get details of a specific effect.

Parameters:

Name Type Description Default
effect_id str

The ID of the effect to retrieve.

required

Returns:

Name Type Description
Effect Effect

The requested effect.

Raises:

Type Description
NotFoundError

If the effect with the given ID is not found.

ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() effect = client.get_effect("example_effect_id") print(f"Effect name: {effect.attributes.name}")

Source code in signalrgb/client.py
def get_effect(self, effect_id: str) -> Effect:
    """Get details of a specific effect.

    Args:
        effect_id (str): The ID of the effect to retrieve.

    Returns:
        Effect: The requested effect.

    Raises:
        NotFoundError: If the effect with the given ID is not found.
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> effect = client.get_effect("example_effect_id")
        >>> print(f"Effect name: {effect.attributes.name}")
    """
    return self._run_async(self._async_client.get_effect(effect_id))

get_effect_by_name(effect_name)

Get details of a specific effect by name.

Parameters:

Name Type Description Default
effect_name str

The name of the effect to retrieve.

required

Returns:

Name Type Description
Effect Effect

The requested effect.

Raises:

Type Description
NotFoundError

If the effect with the given name is not found.

ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() effect = client.get_effect_by_name("Rainbow Wave") print(f"Effect ID: {effect.id}")

Source code in signalrgb/client.py
def get_effect_by_name(self, effect_name: str) -> Effect:
    """Get details of a specific effect by name.

    Args:
        effect_name: The name of the effect to retrieve.

    Returns:
        Effect: The requested effect.

    Raises:
        NotFoundError: If the effect with the given name is not found.
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> effect = client.get_effect_by_name("Rainbow Wave")
        >>> print(f"Effect ID: {effect.id}")
    """
    return self._run_async(self._async_client.get_effect_by_name(effect_name))

get_effect_presets(effect_id)

Get presets for a specific effect.

Parameters:

Name Type Description Default
effect_id str

The ID of the effect to retrieve presets for.

required

Returns:

Type Description
list[EffectPreset]

List[EffectPreset]: A list of effect presets.

Raises:

Type Description
NotFoundError

If the effect with the given ID is not found.

ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() presets = client.get_effect_presets("example_effect_id") for preset in presets: ... print(f"Preset ID: {preset.id}, Name: {preset.name}")

Source code in signalrgb/client.py
def get_effect_presets(self, effect_id: str) -> list[EffectPreset]:
    """Get presets for a specific effect.

    Args:
        effect_id (str): The ID of the effect to retrieve presets for.

    Returns:
        List[EffectPreset]: A list of effect presets.

    Raises:
        NotFoundError: If the effect with the given ID is not found.
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> presets = client.get_effect_presets("example_effect_id")
        >>> for preset in presets:
        ...     print(f"Preset ID: {preset.id}, Name: {preset.name}")
    """
    return self._run_async(self._async_client.get_effect_presets(effect_id))

get_effects()

List available effects.

Returns:

Type Description
list[Effect]

List[Effect]: A list of available effects.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an error retrieving the effects.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() effects = client.get_effects() print(f"Found {len(effects)} effects")

Source code in signalrgb/client.py
def get_effects(self) -> list[Effect]:
    """List available effects.

    Returns:
        List[Effect]: A list of available effects.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an error retrieving the effects.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> effects = client.get_effects()
        >>> print(f"Found {len(effects)} effects")
    """
    return self._run_async(self._async_client.get_effects())

get_layouts()

Get all available layouts.

Returns:

Type Description
list[Layout]

List[Layout]: A list of all available layouts.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() layouts = client.get_layouts() for layout in layouts: ... print(f"Layout: {layout.id}")

Source code in signalrgb/client.py
def get_layouts(self) -> list[Layout]:
    """Get all available layouts.

    Returns:
        List[Layout]: A list of all available layouts.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> layouts = client.get_layouts()
        >>> for layout in layouts:
        ...     print(f"Layout: {layout.id}")
    """
    return self._run_async(self._async_client.get_layouts())

get_next_effect()

Get information about the next effect in history.

Returns:

Type Description
Effect | None

Optional[Effect]: The next effect if available, None otherwise.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() next_effect = client.get_next_effect() if next_effect: ... print(f"Next effect: {next_effect.attributes.name}") ... else: ... print("No next effect available")

Source code in signalrgb/client.py
def get_next_effect(self) -> Effect | None:
    """Get information about the next effect in history.

    Returns:
        Optional[Effect]: The next effect if available, None otherwise.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> next_effect = client.get_next_effect()
        >>> if next_effect:
        ...     print(f"Next effect: {next_effect.attributes.name}")
        ... else:
        ...     print("No next effect available")
    """
    return self._run_async(self._async_client.get_next_effect())

get_previous_effect()

Get information about the previous effect in history.

Returns:

Type Description
Effect | None

Optional[Effect]: The previous effect if available, None otherwise.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

client = SignalRGBClient() prev_effect = client.get_previous_effect() if prev_effect: ... print(f"Previous effect: {prev_effect.attributes.name}") ... else: ... print("No previous effect available")

Source code in signalrgb/client.py
def get_previous_effect(self) -> Effect | None:
    """Get information about the previous effect in history.

    Returns:
        Optional[Effect]: The previous effect if available, None otherwise.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> client = SignalRGBClient()
        >>> prev_effect = client.get_previous_effect()
        >>> if prev_effect:
        ...     print(f"Previous effect: {prev_effect.attributes.name}")
        ... else:
        ...     print("No previous effect available")
    """
    return self._run_async(self._async_client.get_previous_effect())

refresh_effects()

Refresh the cached effects.

This method clears the cache for the get_effects method, forcing a fresh retrieval of effects on the next call.

Example

client = SignalRGBClient() client.refresh_effects() fresh_effects = client.get_effects()

Source code in signalrgb/client.py
def refresh_effects(self) -> None:
    """Refresh the cached effects.

    This method clears the cache for the get_effects method, forcing a fresh
    retrieval of effects on the next call.

    Example:
        >>> client = SignalRGBClient()
        >>> client.refresh_effects()
        >>> fresh_effects = client.get_effects()
    """
    self._run_async(self._async_client.refresh_effects())

AsyncSignalRGBClient

signalrgb.async_client.AsyncSignalRGBClient

Async client for interacting with the SignalRGB API.

This class provides asynchronous methods to interact with the SignalRGB API, allowing users to retrieve, apply, and manage lighting effects and layouts.

Source code in signalrgb/async_client.py
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
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
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
class AsyncSignalRGBClient:
    """Async client for interacting with the SignalRGB API.

    This class provides asynchronous methods to interact with the SignalRGB API, allowing users
    to retrieve, apply, and manage lighting effects and layouts.
    """

    def __init__(self, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, timeout: float = DEFAULT_TIMEOUT):
        """Initialize the AsyncSignalRGBClient.

        Args:
            host: The host of the SignalRGB API. Defaults to 'localhost'.
            port: The port of the SignalRGB API. Defaults to DEFAULT_PORT.
            timeout: The timeout for API requests in seconds. Defaults to 10.0.

        Example:
            >>> client = AsyncSignalRGBClient()
            >>> client = AsyncSignalRGBClient("192.168.1.100", 8080, 5.0)
        """
        self._base_url = f"http://{host}:{port}"
        self._timeout = timeout
        self._effects_cache: list[Effect] | None = None
        self._client = httpx.AsyncClient(timeout=self._timeout)

    async def __aenter__(self) -> AsyncSignalRGBClient:
        """Async context manager entry."""
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        """Async context manager exit."""
        if self._client:
            await self._client.aclose()

    @property
    def client(self) -> httpx.AsyncClient:
        """Get the HTTPX client instance."""
        return self._client

    @asynccontextmanager
    async def _request_context(self, method: str, endpoint: str, **kwargs: Any) -> AsyncIterator[dict[str, Any]]:
        """Async context manager for making API requests.

        This method handles common exception cases and debug logging.

        Args:
            method: The HTTP method to use for the request.
            endpoint: The API endpoint to request.
            **kwargs: Additional arguments to pass to the request.

        Yields:
            The JSON response from the API.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.
        """
        url = f"{self._base_url}{endpoint}"
        debug = os.getenv(DEBUG_ENV_VAR, "0") == "1"

        if debug:
            pass

        try:
            response = await self.client.request(method, url, **kwargs)
            response.raise_for_status()

            if debug:
                pass

            yield response.json()
        except httpx.ConnectError as e:
            raise ConnectionError(f"Failed to connect to SignalRGB API: {e}", Error(title=str(e))) from e
        except httpx.TimeoutException as e:
            raise ConnectionError("Request timed out", Error(title="Request Timeout")) from e
        except httpx.HTTPStatusError as e:
            # Create error object based on response
            error: Error
            if e.response is not None and hasattr(e.response, "json"):
                try:
                    json_data = e.response.json()
                    if "errors" in json_data and json_data["errors"] and isinstance(json_data["errors"], list):
                        error = Error.from_dict(json_data["errors"][0])
                    else:
                        error = Error(title=str(e))
                except Exception:  # noqa: BLE001
                    error = Error(title=str(e))
            else:
                error = Error(title=str(e))
            raise APIError(f"HTTP error occurred: {e}", error) from e
        except httpx.RequestError as e:
            error = Error(title=str(e))
            raise APIError(f"An error occurred while making the request: {e}", error) from e
        except (ValueError, TypeError) as e:
            # More specific exceptions instead of catching generic Exception
            raise SignalRGBError(f"An unexpected error occurred: {e}") from e

    async def _get_effects_cached(self) -> list[Effect]:
        """Internal method to get effects with caching."""
        if self._effects_cache is None:
            async with self._request_context("GET", f"{LIGHTING_V1}/effects") as data:
                response = EffectListResponse.from_dict(data)
                self._ensure_response_ok(response)
                effects = response.data
                if effects is None or effects.items is None:
                    raise APIError("No effects data in the response")
                self._effects_cache = effects.items
        return self._effects_cache

    async def get_effects(self) -> list[Effect]:
        """List available effects asynchronously.

        Returns:
            List[Effect]: A list of available effects.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an error retrieving the effects.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     effects = await client.get_effects()
            >>>     print(f"Found {len(effects)} effects")
        """
        return await self._get_effects_cached()

    async def get_effect(self, effect_id: str) -> Effect:
        """Get details of a specific effect asynchronously.

        Args:
            effect_id (str): The ID of the effect to retrieve.

        Returns:
            Effect: The requested effect.

        Raises:
            NotFoundError: If the effect with the given ID is not found.
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     effect = await client.get_effect("example_effect_id")
            >>>     print(f"Effect name: {effect.attributes.name}")
        """
        try:
            async with self._request_context("GET", f"{LIGHTING_V1}/effects/{effect_id}") as data:
                response = EffectDetailsResponse.from_dict(data)
                self._ensure_response_ok(response)
                if response.data is None:
                    raise APIError("No effect data in the response")
                return response.data
        except APIError as e:
            if e.error and e.error.code == "not_found":
                raise NotFoundError(f"Effect with ID '{effect_id}' not found", e.error) from e
            raise

    async def get_effect_by_name(self, effect_name: str) -> Effect:
        """Get details of a specific effect by name asynchronously.

        Args:
            effect_name: The name of the effect to retrieve.

        Returns:
            Effect: The requested effect.

        Raises:
            NotFoundError: If the effect with the given name is not found.
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     effect = await client.get_effect_by_name("Rainbow Wave")
            >>>     print(f"Effect ID: {effect.id}")
        """
        effects = await self.get_effects()
        effect = next((e for e in effects if e.attributes.name == effect_name), None)
        if effect is None:
            raise NotFoundError(f"Effect '{effect_name}' not found")
        return await self.get_effect(effect.id)

    async def get_current_effect(self) -> Effect:
        """Get the current effect asynchronously.

        Returns:
            Effect: The currently active effect.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     current_effect = await client.get_current_effect()
            >>>     print(f"Current effect: {current_effect.attributes.name}")
        """
        state = await self._get_current_state()
        if state.attributes is None:
            raise APIError("No current effect data in the response")
        return await self.get_effect(state.id)

    async def _get_current_state(self) -> CurrentStateHolder:
        """Get the current state of the SignalRGB instance asynchronously.

        Returns:
            CurrentStateHolder: The current state of the SignalRGB instance.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.
        """
        async with self._request_context("GET", LIGHTING_V1) as data:
            response = CurrentStateResponse.from_dict(data)
            self._ensure_response_ok(response)
            if response.data is None:
                raise APIError("No current state data in the response")
            return response.data

    async def get_brightness(self) -> int:
        """Get the current brightness level asynchronously.

        Returns:
            int: The current brightness level (0-100).

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     brightness = await client.get_brightness()
            >>>     print(f"Current brightness: {brightness}")
        """
        state = await self._get_current_state()
        return state.attributes.global_brightness

    async def set_brightness(self, value: int) -> None:
        """Set the brightness level asynchronously.

        Args:
            value: The brightness level to set (0-100).

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     await client.set_brightness(75)
            >>>     print("Brightness set to 75%")
        """
        async with self._request_context(
            "PATCH",
            f"{LIGHTING_V1}/global_brightness",
            json={"global_brightness": value},
        ):
            pass

    async def get_enabled(self) -> bool:
        """Get the current enabled state of the canvas asynchronously.

        Returns:
            bool: The current enabled state.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     enabled = await client.get_enabled()
            >>>     print(f"Canvas enabled: {enabled}")
        """
        state = await self._get_current_state()
        return state.attributes.enabled

    async def set_enabled(self, value: bool) -> None:
        """Set the enabled state of the canvas asynchronously.

        Args:
            value: Whether to enable the canvas.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     await client.set_enabled(False)
            >>>     print("Canvas disabled")
        """
        async with self._request_context("PATCH", f"{LIGHTING_V1}/enabled", json={"enabled": value}):
            pass

    async def apply_effect(self, effect_id: str) -> None:
        """Apply an effect asynchronously.

        Args:
            effect_id (str): The ID of the effect to apply.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     await client.apply_effect("example_effect_id")
            >>>     print("Effect applied successfully")
        """
        async with self._request_context("POST", f"{LIGHTING_V1}/effects/{effect_id}/apply") as data:
            response = SignalRGBResponse.from_dict(data)
            self._ensure_response_ok(response)

    async def apply_effect_by_name(self, effect_name: str) -> None:
        """Apply an effect by name asynchronously.

        Args:
            effect_name: The name of the effect to apply.

        Raises:
            NotFoundError: If the effect with the given name is not found.
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     await client.apply_effect_by_name("Rainbow Wave")
            >>>     print("Effect applied successfully")
        """
        effect = await self.get_effect_by_name(effect_name)
        apply_url = effect.links.apply
        if apply_url is None:
            # Fallback if apply link is missing
            await self.apply_effect(effect.id)
        else:
            async with self._request_context("POST", apply_url):
                pass

    async def get_effect_presets(self, effect_id: str) -> list[EffectPreset]:
        """Get presets for a specific effect asynchronously.

        Args:
            effect_id (str): The ID of the effect to retrieve presets for.

        Returns:
            List[EffectPreset]: A list of effect presets.

        Raises:
            NotFoundError: If the effect with the given ID is not found.
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     presets = await client.get_effect_presets("example_effect_id")
            >>>     for preset in presets:
            >>>         print(f"Preset ID: {preset.id}")
        """
        try:
            async with self._request_context("GET", f"{LIGHTING_V1}/effects/{effect_id}/presets") as data:
                response = EffectPresetListResponse.from_dict(data)
                self._ensure_response_ok(response)
                if response.data is None:
                    raise APIError("No preset data in the response")
                return response.data.items
        except APIError as e:
            if e.error and e.error.code == "not_found":
                raise NotFoundError(f"Effect with ID '{effect_id}' not found", e.error) from e
            raise

    async def apply_effect_preset(self, effect_id: str, preset_id: str) -> None:
        """Apply a preset for a specific effect asynchronously.

        Args:
            effect_id (str): The ID of the effect to apply the preset to.
            preset_id (str): The ID of the preset to apply.

        Raises:
            NotFoundError: If the effect with the given ID is not found.
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     await client.apply_effect_preset("example_effect_id", "My Fancy Preset 1")
            >>>     print("Preset applied successfully")
        """
        try:
            async with self._request_context(
                "PATCH",
                f"{LIGHTING_V1}/effects/{effect_id}/presets",
                json={"preset": preset_id},
            ) as data:
                response = EffectPresetResponse.from_dict(data)
                self._ensure_response_ok(response)
        except APIError as e:
            if e.error and e.error.code == "not_found":
                raise NotFoundError(
                    f"Effect with ID '{effect_id}' or preset '{preset_id}' not found",
                    e.error,
                ) from e
            raise

    async def get_next_effect(self) -> Effect | None:
        """Get information about the next effect in history asynchronously.

        Returns:
            Optional[Effect]: The next effect if available, None otherwise.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     next_effect = await client.get_next_effect()
            >>>     if next_effect:
            >>>         print(f"Next effect: {next_effect.attributes.name}")
            >>>     else:
            >>>         print("No next effect available")
        """
        try:
            async with self._request_context("GET", f"{LIGHTING_V1}/next") as data:
                response = EffectDetailsResponse.from_dict(data)
                self._ensure_response_ok(response)
                return response.data
        except APIError as e:
            if e.error and e.error.code == "409":
                return None
            raise

    async def apply_next_effect(self) -> Effect:
        """Apply the next effect in history or a random effect if there's no next effect asynchronously.

        Returns:
            Effect: The newly applied effect.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     new_effect = await client.apply_next_effect()
            >>>     print(f"Applied effect: {new_effect.attributes.name}")
        """
        async with self._request_context("POST", f"{LIGHTING_V1}/next") as data:
            response = EffectDetailsResponse.from_dict(data)
            self._ensure_response_ok(response)
            if response.data is None:
                raise APIError("No effect data in the response")
            return response.data

    async def get_previous_effect(self) -> Effect | None:
        """Get information about the previous effect in history asynchronously.

        Returns:
            Optional[Effect]: The previous effect if available, None otherwise.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     prev_effect = await client.get_previous_effect()
            >>>     if prev_effect:
            >>>         print(f"Previous effect: {prev_effect.attributes.name}")
            >>>     else:
            >>>         print("No previous effect available")
        """
        try:
            async with self._request_context("GET", f"{LIGHTING_V1}/previous") as data:
                response = EffectDetailsResponse.from_dict(data)
                self._ensure_response_ok(response)
                return response.data
        except APIError as e:
            if e.error and e.error.code == "409":
                return None
            raise

    async def apply_previous_effect(self) -> Effect:
        """Apply the previous effect in history asynchronously.

        Returns:
            Effect: The newly applied effect.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     new_effect = await client.apply_previous_effect()
            >>>     print(f"Applied effect: {new_effect.attributes.name}")
        """
        async with self._request_context("POST", f"{LIGHTING_V1}/previous") as data:
            response = EffectDetailsResponse.from_dict(data)
            self._ensure_response_ok(response)
            if response.data is None:
                raise APIError("No effect data in the response")
            return response.data

    async def apply_random_effect(self) -> Effect:
        """Apply a random effect asynchronously.

        Returns:
            Effect: The newly applied random effect.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     random_effect = await client.apply_random_effect()
            >>>     print(f"Applied random effect: {random_effect.attributes.name}")
        """
        async with self._request_context("POST", f"{LIGHTING_V1}/shuffle") as data:
            response = EffectDetailsResponse.from_dict(data)
            self._ensure_response_ok(response)
            if response.data is None:
                raise APIError("No effect data in the response")
            return response.data

    async def get_current_layout(self) -> Layout:
        """Get the current layout asynchronously.

        Returns:
            Layout: The currently active layout.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     current_layout = await client.get_current_layout()
            >>>     print(f"Current layout: {current_layout.id}")
        """
        async with self._request_context("GET", f"{SCENES_V1}/current_layout") as data:
            response = CurrentLayoutResponse.from_dict(data)
            self._ensure_response_ok(response)
            if response.data is None or response.data.current_layout is None:
                raise APIError("No current layout data in the response")
            return response.data.current_layout

    async def set_current_layout(self, layout_id: str) -> None:
        """Set the current layout asynchronously.

        Args:
            layout_id: The ID of the layout to set as current.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     await client.set_current_layout("My Layout 1")
            >>>     current = await client.get_current_layout()
            >>>     print(f"New current layout: {current.id}")
        """
        async with self._request_context("PATCH", f"{SCENES_V1}/current_layout", json={"layout": layout_id}) as data:
            response = CurrentLayoutResponse.from_dict(data)
            self._ensure_response_ok(response)
            if response.data is None or response.data.current_layout is None:
                raise APIError("No current layout data in the response")
            if response.data.current_layout.id != layout_id:
                raise APIError(f"Failed to set layout to '{layout_id}'")

    async def get_layouts(self) -> list[Layout]:
        """Get all available layouts asynchronously.

        Returns:
            List[Layout]: A list of all available layouts.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     layouts = await client.get_layouts()
            >>>     for layout in layouts:
            >>>         print(f"Layout: {layout.id}")
        """
        async with self._request_context("GET", f"{SCENES_V1}/layouts") as data:
            response = LayoutListResponse.from_dict(data)
            self._ensure_response_ok(response)
            if "data" not in data or "items" not in data["data"]:
                raise APIError("No layouts data in the response")
            return [Layout.from_dict(item) for item in data["data"]["items"]]

    def _ensure_response_ok(self, response: SignalRGBResponse) -> None:
        """Ensure the response status is 'ok'.

        Args:
            response: The response to check.

        Raises:
            APIError: If the response status is not 'ok'.
        """
        if response.status != "ok":
            error = response.errors[0] if response.errors else None
            raise APIError(f"API returned non-OK status: {response.status}", error)

    async def refresh_effects(self) -> None:
        """Refresh the cached effects asynchronously.

        This method clears the cache for the get_effects method, forcing a fresh
        retrieval of effects on the next call.

        Example:
            >>> async with AsyncSignalRGBClient() as client:
            >>>     await client.refresh_effects()
            >>>     fresh_effects = await client.get_effects()
        """
        self._effects_cache = None

    def __repr__(self) -> str:
        return f"AsyncSignalRGBClient(base_url='{self._base_url}')"

    async def aclose(self) -> None:
        """Close the async client explicitly.

        This method ensures resources are properly released when the client is no longer needed.
        """
        if self._client:
            await self._client.aclose()

    async def get_effects_cached(self) -> list[Effect]:
        """Get effects with caching.

        Returns cached effects if available, otherwise retrieves them from the API.

        Returns:
            List[Effect]: A list of available effects.
        """
        return await self._get_effects_cached()

    async def get_current_state(self) -> CurrentStateHolder:
        """Get the current state of the SignalRGB instance.

        Returns:
            CurrentStateHolder: The current state of the SignalRGB instance.

        Raises:
            ConnectionError: If there's a connection error.
            APIError: If there's an API error.
            SignalRGBError: For any other unexpected errors.
        """
        return await self._get_current_state()

client property

Get the HTTPX client instance.

__aenter__() async

Async context manager entry.

Source code in signalrgb/async_client.py
async def __aenter__(self) -> AsyncSignalRGBClient:
    """Async context manager entry."""
    return self

__aexit__(exc_type, exc_val, exc_tb) async

Async context manager exit.

Source code in signalrgb/async_client.py
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    """Async context manager exit."""
    if self._client:
        await self._client.aclose()

__init__(host=DEFAULT_HOST, port=DEFAULT_PORT, timeout=DEFAULT_TIMEOUT)

Initialize the AsyncSignalRGBClient.

Parameters:

Name Type Description Default
host str

The host of the SignalRGB API. Defaults to 'localhost'.

DEFAULT_HOST
port int

The port of the SignalRGB API. Defaults to DEFAULT_PORT.

DEFAULT_PORT
timeout float

The timeout for API requests in seconds. Defaults to 10.0.

DEFAULT_TIMEOUT
Example

client = AsyncSignalRGBClient() client = AsyncSignalRGBClient("192.168.1.100", 8080, 5.0)

Source code in signalrgb/async_client.py
def __init__(self, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, timeout: float = DEFAULT_TIMEOUT):
    """Initialize the AsyncSignalRGBClient.

    Args:
        host: The host of the SignalRGB API. Defaults to 'localhost'.
        port: The port of the SignalRGB API. Defaults to DEFAULT_PORT.
        timeout: The timeout for API requests in seconds. Defaults to 10.0.

    Example:
        >>> client = AsyncSignalRGBClient()
        >>> client = AsyncSignalRGBClient("192.168.1.100", 8080, 5.0)
    """
    self._base_url = f"http://{host}:{port}"
    self._timeout = timeout
    self._effects_cache: list[Effect] | None = None
    self._client = httpx.AsyncClient(timeout=self._timeout)

_ensure_response_ok(response)

Ensure the response status is 'ok'.

Parameters:

Name Type Description Default
response SignalRGBResponse

The response to check.

required

Raises:

Type Description
APIError

If the response status is not 'ok'.

Source code in signalrgb/async_client.py
def _ensure_response_ok(self, response: SignalRGBResponse) -> None:
    """Ensure the response status is 'ok'.

    Args:
        response: The response to check.

    Raises:
        APIError: If the response status is not 'ok'.
    """
    if response.status != "ok":
        error = response.errors[0] if response.errors else None
        raise APIError(f"API returned non-OK status: {response.status}", error)

_get_current_state() async

Get the current state of the SignalRGB instance asynchronously.

Returns:

Name Type Description
CurrentStateHolder CurrentStateHolder

The current state of the SignalRGB instance.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Source code in signalrgb/async_client.py
async def _get_current_state(self) -> CurrentStateHolder:
    """Get the current state of the SignalRGB instance asynchronously.

    Returns:
        CurrentStateHolder: The current state of the SignalRGB instance.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.
    """
    async with self._request_context("GET", LIGHTING_V1) as data:
        response = CurrentStateResponse.from_dict(data)
        self._ensure_response_ok(response)
        if response.data is None:
            raise APIError("No current state data in the response")
        return response.data

_get_effects_cached() async

Internal method to get effects with caching.

Source code in signalrgb/async_client.py
async def _get_effects_cached(self) -> list[Effect]:
    """Internal method to get effects with caching."""
    if self._effects_cache is None:
        async with self._request_context("GET", f"{LIGHTING_V1}/effects") as data:
            response = EffectListResponse.from_dict(data)
            self._ensure_response_ok(response)
            effects = response.data
            if effects is None or effects.items is None:
                raise APIError("No effects data in the response")
            self._effects_cache = effects.items
    return self._effects_cache

_request_context(method, endpoint, **kwargs) async

Async context manager for making API requests.

This method handles common exception cases and debug logging.

Parameters:

Name Type Description Default
method str

The HTTP method to use for the request.

required
endpoint str

The API endpoint to request.

required
**kwargs Any

Additional arguments to pass to the request.

{}

Yields:

Type Description
AsyncIterator[dict[str, Any]]

The JSON response from the API.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Source code in signalrgb/async_client.py
@asynccontextmanager
async def _request_context(self, method: str, endpoint: str, **kwargs: Any) -> AsyncIterator[dict[str, Any]]:
    """Async context manager for making API requests.

    This method handles common exception cases and debug logging.

    Args:
        method: The HTTP method to use for the request.
        endpoint: The API endpoint to request.
        **kwargs: Additional arguments to pass to the request.

    Yields:
        The JSON response from the API.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.
    """
    url = f"{self._base_url}{endpoint}"
    debug = os.getenv(DEBUG_ENV_VAR, "0") == "1"

    if debug:
        pass

    try:
        response = await self.client.request(method, url, **kwargs)
        response.raise_for_status()

        if debug:
            pass

        yield response.json()
    except httpx.ConnectError as e:
        raise ConnectionError(f"Failed to connect to SignalRGB API: {e}", Error(title=str(e))) from e
    except httpx.TimeoutException as e:
        raise ConnectionError("Request timed out", Error(title="Request Timeout")) from e
    except httpx.HTTPStatusError as e:
        # Create error object based on response
        error: Error
        if e.response is not None and hasattr(e.response, "json"):
            try:
                json_data = e.response.json()
                if "errors" in json_data and json_data["errors"] and isinstance(json_data["errors"], list):
                    error = Error.from_dict(json_data["errors"][0])
                else:
                    error = Error(title=str(e))
            except Exception:  # noqa: BLE001
                error = Error(title=str(e))
        else:
            error = Error(title=str(e))
        raise APIError(f"HTTP error occurred: {e}", error) from e
    except httpx.RequestError as e:
        error = Error(title=str(e))
        raise APIError(f"An error occurred while making the request: {e}", error) from e
    except (ValueError, TypeError) as e:
        # More specific exceptions instead of catching generic Exception
        raise SignalRGBError(f"An unexpected error occurred: {e}") from e

aclose() async

Close the async client explicitly.

This method ensures resources are properly released when the client is no longer needed.

Source code in signalrgb/async_client.py
async def aclose(self) -> None:
    """Close the async client explicitly.

    This method ensures resources are properly released when the client is no longer needed.
    """
    if self._client:
        await self._client.aclose()

apply_effect(effect_id) async

Apply an effect asynchronously.

Parameters:

Name Type Description Default
effect_id str

The ID of the effect to apply.

required

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: await client.apply_effect("example_effect_id") print("Effect applied successfully")

Source code in signalrgb/async_client.py
async def apply_effect(self, effect_id: str) -> None:
    """Apply an effect asynchronously.

    Args:
        effect_id (str): The ID of the effect to apply.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     await client.apply_effect("example_effect_id")
        >>>     print("Effect applied successfully")
    """
    async with self._request_context("POST", f"{LIGHTING_V1}/effects/{effect_id}/apply") as data:
        response = SignalRGBResponse.from_dict(data)
        self._ensure_response_ok(response)

apply_effect_by_name(effect_name) async

Apply an effect by name asynchronously.

Parameters:

Name Type Description Default
effect_name str

The name of the effect to apply.

required

Raises:

Type Description
NotFoundError

If the effect with the given name is not found.

ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: await client.apply_effect_by_name("Rainbow Wave") print("Effect applied successfully")

Source code in signalrgb/async_client.py
async def apply_effect_by_name(self, effect_name: str) -> None:
    """Apply an effect by name asynchronously.

    Args:
        effect_name: The name of the effect to apply.

    Raises:
        NotFoundError: If the effect with the given name is not found.
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     await client.apply_effect_by_name("Rainbow Wave")
        >>>     print("Effect applied successfully")
    """
    effect = await self.get_effect_by_name(effect_name)
    apply_url = effect.links.apply
    if apply_url is None:
        # Fallback if apply link is missing
        await self.apply_effect(effect.id)
    else:
        async with self._request_context("POST", apply_url):
            pass

apply_effect_preset(effect_id, preset_id) async

Apply a preset for a specific effect asynchronously.

Parameters:

Name Type Description Default
effect_id str

The ID of the effect to apply the preset to.

required
preset_id str

The ID of the preset to apply.

required

Raises:

Type Description
NotFoundError

If the effect with the given ID is not found.

ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: await client.apply_effect_preset("example_effect_id", "My Fancy Preset 1") print("Preset applied successfully")

Source code in signalrgb/async_client.py
async def apply_effect_preset(self, effect_id: str, preset_id: str) -> None:
    """Apply a preset for a specific effect asynchronously.

    Args:
        effect_id (str): The ID of the effect to apply the preset to.
        preset_id (str): The ID of the preset to apply.

    Raises:
        NotFoundError: If the effect with the given ID is not found.
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     await client.apply_effect_preset("example_effect_id", "My Fancy Preset 1")
        >>>     print("Preset applied successfully")
    """
    try:
        async with self._request_context(
            "PATCH",
            f"{LIGHTING_V1}/effects/{effect_id}/presets",
            json={"preset": preset_id},
        ) as data:
            response = EffectPresetResponse.from_dict(data)
            self._ensure_response_ok(response)
    except APIError as e:
        if e.error and e.error.code == "not_found":
            raise NotFoundError(
                f"Effect with ID '{effect_id}' or preset '{preset_id}' not found",
                e.error,
            ) from e
        raise

apply_next_effect() async

Apply the next effect in history or a random effect if there's no next effect asynchronously.

Returns:

Name Type Description
Effect Effect

The newly applied effect.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: new_effect = await client.apply_next_effect() print(f"Applied effect: {new_effect.attributes.name}")

Source code in signalrgb/async_client.py
async def apply_next_effect(self) -> Effect:
    """Apply the next effect in history or a random effect if there's no next effect asynchronously.

    Returns:
        Effect: The newly applied effect.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     new_effect = await client.apply_next_effect()
        >>>     print(f"Applied effect: {new_effect.attributes.name}")
    """
    async with self._request_context("POST", f"{LIGHTING_V1}/next") as data:
        response = EffectDetailsResponse.from_dict(data)
        self._ensure_response_ok(response)
        if response.data is None:
            raise APIError("No effect data in the response")
        return response.data

apply_previous_effect() async

Apply the previous effect in history asynchronously.

Returns:

Name Type Description
Effect Effect

The newly applied effect.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: new_effect = await client.apply_previous_effect() print(f"Applied effect: {new_effect.attributes.name}")

Source code in signalrgb/async_client.py
async def apply_previous_effect(self) -> Effect:
    """Apply the previous effect in history asynchronously.

    Returns:
        Effect: The newly applied effect.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     new_effect = await client.apply_previous_effect()
        >>>     print(f"Applied effect: {new_effect.attributes.name}")
    """
    async with self._request_context("POST", f"{LIGHTING_V1}/previous") as data:
        response = EffectDetailsResponse.from_dict(data)
        self._ensure_response_ok(response)
        if response.data is None:
            raise APIError("No effect data in the response")
        return response.data

apply_random_effect() async

Apply a random effect asynchronously.

Returns:

Name Type Description
Effect Effect

The newly applied random effect.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: random_effect = await client.apply_random_effect() print(f"Applied random effect: {random_effect.attributes.name}")

Source code in signalrgb/async_client.py
async def apply_random_effect(self) -> Effect:
    """Apply a random effect asynchronously.

    Returns:
        Effect: The newly applied random effect.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     random_effect = await client.apply_random_effect()
        >>>     print(f"Applied random effect: {random_effect.attributes.name}")
    """
    async with self._request_context("POST", f"{LIGHTING_V1}/shuffle") as data:
        response = EffectDetailsResponse.from_dict(data)
        self._ensure_response_ok(response)
        if response.data is None:
            raise APIError("No effect data in the response")
        return response.data

get_brightness() async

Get the current brightness level asynchronously.

Returns:

Name Type Description
int int

The current brightness level (0-100).

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: brightness = await client.get_brightness() print(f"Current brightness: {brightness}")

Source code in signalrgb/async_client.py
async def get_brightness(self) -> int:
    """Get the current brightness level asynchronously.

    Returns:
        int: The current brightness level (0-100).

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     brightness = await client.get_brightness()
        >>>     print(f"Current brightness: {brightness}")
    """
    state = await self._get_current_state()
    return state.attributes.global_brightness

get_current_effect() async

Get the current effect asynchronously.

Returns:

Name Type Description
Effect Effect

The currently active effect.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: current_effect = await client.get_current_effect() print(f"Current effect: {current_effect.attributes.name}")

Source code in signalrgb/async_client.py
async def get_current_effect(self) -> Effect:
    """Get the current effect asynchronously.

    Returns:
        Effect: The currently active effect.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     current_effect = await client.get_current_effect()
        >>>     print(f"Current effect: {current_effect.attributes.name}")
    """
    state = await self._get_current_state()
    if state.attributes is None:
        raise APIError("No current effect data in the response")
    return await self.get_effect(state.id)

get_current_layout() async

Get the current layout asynchronously.

Returns:

Name Type Description
Layout Layout

The currently active layout.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: current_layout = await client.get_current_layout() print(f"Current layout: {current_layout.id}")

Source code in signalrgb/async_client.py
async def get_current_layout(self) -> Layout:
    """Get the current layout asynchronously.

    Returns:
        Layout: The currently active layout.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     current_layout = await client.get_current_layout()
        >>>     print(f"Current layout: {current_layout.id}")
    """
    async with self._request_context("GET", f"{SCENES_V1}/current_layout") as data:
        response = CurrentLayoutResponse.from_dict(data)
        self._ensure_response_ok(response)
        if response.data is None or response.data.current_layout is None:
            raise APIError("No current layout data in the response")
        return response.data.current_layout

get_current_state() async

Get the current state of the SignalRGB instance.

Returns:

Name Type Description
CurrentStateHolder CurrentStateHolder

The current state of the SignalRGB instance.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Source code in signalrgb/async_client.py
async def get_current_state(self) -> CurrentStateHolder:
    """Get the current state of the SignalRGB instance.

    Returns:
        CurrentStateHolder: The current state of the SignalRGB instance.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.
    """
    return await self._get_current_state()

get_effect(effect_id) async

Get details of a specific effect asynchronously.

Parameters:

Name Type Description Default
effect_id str

The ID of the effect to retrieve.

required

Returns:

Name Type Description
Effect Effect

The requested effect.

Raises:

Type Description
NotFoundError

If the effect with the given ID is not found.

ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: effect = await client.get_effect("example_effect_id") print(f"Effect name: {effect.attributes.name}")

Source code in signalrgb/async_client.py
async def get_effect(self, effect_id: str) -> Effect:
    """Get details of a specific effect asynchronously.

    Args:
        effect_id (str): The ID of the effect to retrieve.

    Returns:
        Effect: The requested effect.

    Raises:
        NotFoundError: If the effect with the given ID is not found.
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     effect = await client.get_effect("example_effect_id")
        >>>     print(f"Effect name: {effect.attributes.name}")
    """
    try:
        async with self._request_context("GET", f"{LIGHTING_V1}/effects/{effect_id}") as data:
            response = EffectDetailsResponse.from_dict(data)
            self._ensure_response_ok(response)
            if response.data is None:
                raise APIError("No effect data in the response")
            return response.data
    except APIError as e:
        if e.error and e.error.code == "not_found":
            raise NotFoundError(f"Effect with ID '{effect_id}' not found", e.error) from e
        raise

get_effect_by_name(effect_name) async

Get details of a specific effect by name asynchronously.

Parameters:

Name Type Description Default
effect_name str

The name of the effect to retrieve.

required

Returns:

Name Type Description
Effect Effect

The requested effect.

Raises:

Type Description
NotFoundError

If the effect with the given name is not found.

ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: effect = await client.get_effect_by_name("Rainbow Wave") print(f"Effect ID: {effect.id}")

Source code in signalrgb/async_client.py
async def get_effect_by_name(self, effect_name: str) -> Effect:
    """Get details of a specific effect by name asynchronously.

    Args:
        effect_name: The name of the effect to retrieve.

    Returns:
        Effect: The requested effect.

    Raises:
        NotFoundError: If the effect with the given name is not found.
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     effect = await client.get_effect_by_name("Rainbow Wave")
        >>>     print(f"Effect ID: {effect.id}")
    """
    effects = await self.get_effects()
    effect = next((e for e in effects if e.attributes.name == effect_name), None)
    if effect is None:
        raise NotFoundError(f"Effect '{effect_name}' not found")
    return await self.get_effect(effect.id)

get_effect_presets(effect_id) async

Get presets for a specific effect asynchronously.

Parameters:

Name Type Description Default
effect_id str

The ID of the effect to retrieve presets for.

required

Returns:

Type Description
list[EffectPreset]

List[EffectPreset]: A list of effect presets.

Raises:

Type Description
NotFoundError

If the effect with the given ID is not found.

ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: presets = await client.get_effect_presets("example_effect_id") for preset in presets: print(f"Preset ID: {preset.id}")

Source code in signalrgb/async_client.py
async def get_effect_presets(self, effect_id: str) -> list[EffectPreset]:
    """Get presets for a specific effect asynchronously.

    Args:
        effect_id (str): The ID of the effect to retrieve presets for.

    Returns:
        List[EffectPreset]: A list of effect presets.

    Raises:
        NotFoundError: If the effect with the given ID is not found.
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     presets = await client.get_effect_presets("example_effect_id")
        >>>     for preset in presets:
        >>>         print(f"Preset ID: {preset.id}")
    """
    try:
        async with self._request_context("GET", f"{LIGHTING_V1}/effects/{effect_id}/presets") as data:
            response = EffectPresetListResponse.from_dict(data)
            self._ensure_response_ok(response)
            if response.data is None:
                raise APIError("No preset data in the response")
            return response.data.items
    except APIError as e:
        if e.error and e.error.code == "not_found":
            raise NotFoundError(f"Effect with ID '{effect_id}' not found", e.error) from e
        raise

get_effects() async

List available effects asynchronously.

Returns:

Type Description
list[Effect]

List[Effect]: A list of available effects.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an error retrieving the effects.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: effects = await client.get_effects() print(f"Found {len(effects)} effects")

Source code in signalrgb/async_client.py
async def get_effects(self) -> list[Effect]:
    """List available effects asynchronously.

    Returns:
        List[Effect]: A list of available effects.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an error retrieving the effects.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     effects = await client.get_effects()
        >>>     print(f"Found {len(effects)} effects")
    """
    return await self._get_effects_cached()

get_effects_cached() async

Get effects with caching.

Returns cached effects if available, otherwise retrieves them from the API.

Returns:

Type Description
list[Effect]

List[Effect]: A list of available effects.

Source code in signalrgb/async_client.py
async def get_effects_cached(self) -> list[Effect]:
    """Get effects with caching.

    Returns cached effects if available, otherwise retrieves them from the API.

    Returns:
        List[Effect]: A list of available effects.
    """
    return await self._get_effects_cached()

get_enabled() async

Get the current enabled state of the canvas asynchronously.

Returns:

Name Type Description
bool bool

The current enabled state.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: enabled = await client.get_enabled() print(f"Canvas enabled: {enabled}")

Source code in signalrgb/async_client.py
async def get_enabled(self) -> bool:
    """Get the current enabled state of the canvas asynchronously.

    Returns:
        bool: The current enabled state.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     enabled = await client.get_enabled()
        >>>     print(f"Canvas enabled: {enabled}")
    """
    state = await self._get_current_state()
    return state.attributes.enabled

get_layouts() async

Get all available layouts asynchronously.

Returns:

Type Description
list[Layout]

List[Layout]: A list of all available layouts.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: layouts = await client.get_layouts() for layout in layouts: print(f"Layout: {layout.id}")

Source code in signalrgb/async_client.py
async def get_layouts(self) -> list[Layout]:
    """Get all available layouts asynchronously.

    Returns:
        List[Layout]: A list of all available layouts.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     layouts = await client.get_layouts()
        >>>     for layout in layouts:
        >>>         print(f"Layout: {layout.id}")
    """
    async with self._request_context("GET", f"{SCENES_V1}/layouts") as data:
        response = LayoutListResponse.from_dict(data)
        self._ensure_response_ok(response)
        if "data" not in data or "items" not in data["data"]:
            raise APIError("No layouts data in the response")
        return [Layout.from_dict(item) for item in data["data"]["items"]]

get_next_effect() async

Get information about the next effect in history asynchronously.

Returns:

Type Description
Effect | None

Optional[Effect]: The next effect if available, None otherwise.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: next_effect = await client.get_next_effect() if next_effect: print(f"Next effect: {next_effect.attributes.name}") else: print("No next effect available")

Source code in signalrgb/async_client.py
async def get_next_effect(self) -> Effect | None:
    """Get information about the next effect in history asynchronously.

    Returns:
        Optional[Effect]: The next effect if available, None otherwise.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     next_effect = await client.get_next_effect()
        >>>     if next_effect:
        >>>         print(f"Next effect: {next_effect.attributes.name}")
        >>>     else:
        >>>         print("No next effect available")
    """
    try:
        async with self._request_context("GET", f"{LIGHTING_V1}/next") as data:
            response = EffectDetailsResponse.from_dict(data)
            self._ensure_response_ok(response)
            return response.data
    except APIError as e:
        if e.error and e.error.code == "409":
            return None
        raise

get_previous_effect() async

Get information about the previous effect in history asynchronously.

Returns:

Type Description
Effect | None

Optional[Effect]: The previous effect if available, None otherwise.

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: prev_effect = await client.get_previous_effect() if prev_effect: print(f"Previous effect: {prev_effect.attributes.name}") else: print("No previous effect available")

Source code in signalrgb/async_client.py
async def get_previous_effect(self) -> Effect | None:
    """Get information about the previous effect in history asynchronously.

    Returns:
        Optional[Effect]: The previous effect if available, None otherwise.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     prev_effect = await client.get_previous_effect()
        >>>     if prev_effect:
        >>>         print(f"Previous effect: {prev_effect.attributes.name}")
        >>>     else:
        >>>         print("No previous effect available")
    """
    try:
        async with self._request_context("GET", f"{LIGHTING_V1}/previous") as data:
            response = EffectDetailsResponse.from_dict(data)
            self._ensure_response_ok(response)
            return response.data
    except APIError as e:
        if e.error and e.error.code == "409":
            return None
        raise

refresh_effects() async

Refresh the cached effects asynchronously.

This method clears the cache for the get_effects method, forcing a fresh retrieval of effects on the next call.

Example

async with AsyncSignalRGBClient() as client: await client.refresh_effects() fresh_effects = await client.get_effects()

Source code in signalrgb/async_client.py
async def refresh_effects(self) -> None:
    """Refresh the cached effects asynchronously.

    This method clears the cache for the get_effects method, forcing a fresh
    retrieval of effects on the next call.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     await client.refresh_effects()
        >>>     fresh_effects = await client.get_effects()
    """
    self._effects_cache = None

set_brightness(value) async

Set the brightness level asynchronously.

Parameters:

Name Type Description Default
value int

The brightness level to set (0-100).

required

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: await client.set_brightness(75) print("Brightness set to 75%")

Source code in signalrgb/async_client.py
async def set_brightness(self, value: int) -> None:
    """Set the brightness level asynchronously.

    Args:
        value: The brightness level to set (0-100).

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     await client.set_brightness(75)
        >>>     print("Brightness set to 75%")
    """
    async with self._request_context(
        "PATCH",
        f"{LIGHTING_V1}/global_brightness",
        json={"global_brightness": value},
    ):
        pass

set_current_layout(layout_id) async

Set the current layout asynchronously.

Parameters:

Name Type Description Default
layout_id str

The ID of the layout to set as current.

required

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: await client.set_current_layout("My Layout 1") current = await client.get_current_layout() print(f"New current layout: {current.id}")

Source code in signalrgb/async_client.py
async def set_current_layout(self, layout_id: str) -> None:
    """Set the current layout asynchronously.

    Args:
        layout_id: The ID of the layout to set as current.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     await client.set_current_layout("My Layout 1")
        >>>     current = await client.get_current_layout()
        >>>     print(f"New current layout: {current.id}")
    """
    async with self._request_context("PATCH", f"{SCENES_V1}/current_layout", json={"layout": layout_id}) as data:
        response = CurrentLayoutResponse.from_dict(data)
        self._ensure_response_ok(response)
        if response.data is None or response.data.current_layout is None:
            raise APIError("No current layout data in the response")
        if response.data.current_layout.id != layout_id:
            raise APIError(f"Failed to set layout to '{layout_id}'")

set_enabled(value) async

Set the enabled state of the canvas asynchronously.

Parameters:

Name Type Description Default
value bool

Whether to enable the canvas.

required

Raises:

Type Description
ConnectionError

If there's a connection error.

APIError

If there's an API error.

SignalRGBError

For any other unexpected errors.

Example

async with AsyncSignalRGBClient() as client: await client.set_enabled(False) print("Canvas disabled")

Source code in signalrgb/async_client.py
async def set_enabled(self, value: bool) -> None:
    """Set the enabled state of the canvas asynchronously.

    Args:
        value: Whether to enable the canvas.

    Raises:
        ConnectionError: If there's a connection error.
        APIError: If there's an API error.
        SignalRGBError: For any other unexpected errors.

    Example:
        >>> async with AsyncSignalRGBClient() as client:
        >>>     await client.set_enabled(False)
        >>>     print("Canvas disabled")
    """
    async with self._request_context("PATCH", f"{LIGHTING_V1}/enabled", json={"enabled": value}):
        pass

Exceptions

The SignalRGB client defines several custom exceptions for error handling:

signalrgb.exceptions.SignalRGBException = SignalRGBError module-attribute

signalrgb.exceptions.ConnectionError

Bases: SignalRGBError

Exception raised for connection errors.

This exception is raised when there's an issue connecting to the SignalRGB API.

Source code in signalrgb/exceptions.py
class ConnectionError(SignalRGBError):
    """Exception raised for connection errors.

    This exception is raised when there's an issue connecting to the SignalRGB API.
    """

signalrgb.exceptions.APIError

Bases: SignalRGBError

Exception raised for API errors.

This exception is raised when the API returns an error response.

Source code in signalrgb/exceptions.py
class APIError(SignalRGBError):
    """Exception raised for API errors.

    This exception is raised when the API returns an error response.
    """

signalrgb.exceptions.NotFoundError

Bases: SignalRGBError

Exception raised when an item is not found.

This exception is raised when trying to retrieve or apply a non-existent effect, preset, or layout.

Source code in signalrgb/exceptions.py
class NotFoundError(SignalRGBError):
    """Exception raised when an item is not found.

    This exception is raised when trying to retrieve or apply a non-existent effect, preset, or layout.
    """

Usage Examples

Synchronous Client Example

from signalrgb import SignalRGBClient

# Initialize the client
client = SignalRGBClient(host="localhost", port=16038)

# Get all effects
effects = client.get_effects()
for effect in effects:
    print(f"Effect: {effect.attributes.name}")

# Apply an effect
client.apply_effect_by_name("Rainbow Wave")

# Get current effect
current_effect = client.get_current_effect()
print(f"Current effect: {current_effect.attributes.name}")

# Adjust brightness
client.brightness = 75
print(f"Brightness set to: {client.brightness}")

Asynchronous Client Example

import asyncio
from signalrgb import AsyncSignalRGBClient

async def main():
    # Initialize the client using a context manager
    async with AsyncSignalRGBClient(host="localhost", port=16038) as client:
        # Get all effects
        effects = await client.get_effects()
        for effect in effects:
            print(f"Effect: {effect.attributes.name}")

        # Apply an effect
        await client.apply_effect_by_name("Rainbow Wave")

        # Get current effect
        current_effect = await client.get_current_effect()
        print(f"Current effect: {current_effect.attributes.name}")

        # Adjust brightness
        await client.set_brightness(75)
        brightness = await client.get_brightness()
        print(f"Brightness set to: {brightness}")

# Run the async example
asyncio.run(main())

For more detailed usage examples, please refer to: - Python Library Usage for the synchronous client - Asynchronous Library Usage for the async client