tikal.high_level package

Submodules

tikal.high_level.toy_cache module

Persistent storage for toy model name mappings.

This module provides the ToyCache class, which maintains a persistent mapping between bluetooth_names and model_names. This allows you to remember which model each discovered toy is between sessions, so the user does not need to re-select models every time. The cache is stored as a JSON file on the disk.

Note

The cache file is created if it doesn’t exist. ToyCache does not raise exceptions. Encountered errors are logged, and ToyCache fails silently. ToyHub (part of the High-Level-API) uses ToyCache internally.

class tikal.high_level.toy_cache.ToyCache(cache_path: Path, default_model: str, logger_name: str)[source]

Bases: object

Persistent storage mapping Bluetooth names to toy model names.

Maintains a JSON file on the disk that maps Bluetooth device names (e.g., “LVS-A123”) to their corresponding model names (e.g., “Nora”). This eliminates the need for users to manually identify toys every time they connect.

The cache is loaded on initialization. Encountered errors are logged, and ToyCache fails silently.

Parameters:

  • cache_path – Path to the JSON cache file. If the path is empty (=``Path()``), the cache operates with no persistence.

  • default_model – Default model name to return when a Bluetooth name is not found in the cache.

  • logger_name – Name of the logger to use. Use empty string for root logger.

Example:

from pathlib import Path
from toy_cache import ToyCache

cache = ToyCache(
    cache_path=Path("./toys.json"),
    default_model="unknown",
    logger_name="myapp"
)

# First time: toy not in cache
model = cache.get_model_name("LVS-A123")  # Returns "unknown"

# User selects a model, you can update the cache
cache.update({"LVS-A123": "Nora"})

# Next time: toy is in cache
model = cache.get_model_name("LVS-A123")  # Returns "Nora"
get_model_name(bluetooth_name: str) → str[source]

Retrieve the cached model name for a Bluetooth device.

Looks up the model name associated with the given Bluetooth name. If the name is not found in the cache, it returns the default model name specified during initialization.

Parameters:

bluetooth_name – Bluetooth device name to look up (e.g., “LVS-A123”). This is obtained from ToyData.name during discovery.

Returns:

The cached model name (e.g., “Nora”) if found, otherwise the default model name.

Return type:

str

Example:

# Check if the toy is cached
model = cache.get_model_name("LVS-A123")
if model == default_model_name:
    print("Unknown toy")
else:
    print(f"Known toy: {model}")
update(updates: dict[str, str]) → None[source]

Update cache entries and persist to disk.

Merges the provided updates into the cache, overwriting existing entries and adding new ones. The updated cache is immediately written to the disk.

Parameters:

updates – Dictionary mapping Bluetooth names (keys) to model names (values)

Example:

# Update single entry
cache.update({"LVS-A123": "Nora"})

# Update multiple entries
cache.update({
    "LVS-B456": "Lush",
    "LVS-C789": "Max"
})
# Change existing model
cache.update({"LVS-A123": "Ridge"})  # Overwrites "Nora"

Note

Fails silently if disk I/O errors occur. Errors are logged and do not raise exceptions. The in-memory cache is always updated even if writing to disk fails. If the cache was initialized with an empty path (Path()), no disk write is attempted.

tikal.high_level.toy_controller module

Part of the High-level API: Provides representations of toys.

This module wraps the low-level Toy in synchronous methods and adds advanced features: - Synchronous API: All methods are synchronous (non-async), making them easy to use from regular Python code. Commands are queued and executed asynchronously by ToyHub. - Pattern Playback: Set time-based patterns that automatically control toy intensities. - Pause/Block States: Temporarily halt toy actions while maintaining the pattern state. - Callback Support: Optional callbacks provide feedback when a command completes.

This module provides:

Note

You should not instantiate controllers. They are created for you by :class: ToyHub The ToyHub manages the background communication loop that processes queued commands and handles pattern playback.

tikal.high_level.toy_controller.CONTROLLER_BY_BRAND: dict[str, type[ToyController]] = {'Lovense': <class 'tikal.high_level.toy_controller.LovenseController'>, 'MockEstimToys': <class 'tikal.high_level.toy_controller.MockEstimController'>}

Maps a toy’s brand (toy.brand) to its high-level controller class. Register a brand’s controller here when adding support for a new brand.

class tikal.high_level.toy_controller.LovenseController(toy: LovenseToy, logger_name: str)[source]

Bases: ToyController

High-level controller for Lovense toys.

Extends the low-level Lovense class with synchronous methods, command queueing, and pattern playback capabilities.

Parameters:

  • toy – Low-level Lovense instance.

  • logger_name – Name of the logger to use. Use empty string for root logger.

Example:

# Connection (see :class ToyHub)
controllers = hub.connect_toys_blocking(discovered_toys)
toy = controllers[0]  # LovenseController instance

# Manual control
toy.intensity1(15)  # Primary capability
toy.intensity2(10)  # Secondary capability

# Pattern control
pattern = [
    (1000, 10, 5),
    (500, 0, 0),
    (1000, 20, 10),
]
toy.set_pattern(pattern, wraparound=True)

# Pause/resume
toy.toggle_pause()

# Get battery level
toy.get_battery_level(lambda lvl: print(f"Battery: {lvl}%"))

# Advanced features
if toy.change_rotate_direction_available():
    toy.change_rotate_direction()

Note

This class should not be instantiated directly. Use ToyHub’s connection methods to get controller instances.

get_information(callback: Callable[[dict[str, str]], None]) → None[source]

Gather detailed information about the toy.

The following information is gathered (Key, Value pairs of the dictionary):

  • ‘Battery Level’: Battery percentage (e.g., “75%”)

  • ‘Status’: Status code (“2” for normal)

  • ‘Batch number’: Manufacturing batch (e.g., “241015”)

  • ‘Bluetooth Name’: BLE device name (e.g., “LVS-Z36D”)

  • ‘Device type’: Device info (e.g., “C:11:ADDRESS”)

Parameters:

callback – Callback invoked with a dictionary containing toy information. Keys describe the Information type, values contain the information.

Example:

def show_info(info):
    print("Toy Information:")
    for key, value in info.items():
        print(f"{key}:{value}")
toy.get_information(show_info)
class tikal.high_level.toy_controller.MockEstimController(toy: MockEstimToy, logger_name: str)[source]

Bases: ToyController

High-level controller for the fictional MockEstimToys brand.

Wraps a MockEstimToy with the same synchronous, queued, pattern-capable interface as every other controller. Used to explore the High-Level API without real hardware.

Parameters:

  • toy – Low-level MockEstimToy instance.

  • logger_name – Name of the logger to use. Use empty string for root logger.

Note

This class should not be instantiated directly. Use ToyHub’s connection methods to get controller instances.

get_information(callback: Callable[[dict[str, str]], None]) → None[source]

Gather information about the toy.

The following information is gathered (Key, Value pairs of the dictionary):

  • ‘Battery level’: Battery percentage (e.g., “77%”)

  • ‘Name’: Human-readable device name (e.g., “Thunder1”)

  • ‘Brand’: Brand name (“MockEstimToys”)

  • ‘Model’: Model name (e.g., “Thunder”)

Parameters:

callback – Callback invoked with a dictionary containing toy information. Keys describe the Information type, values contain the information.

class tikal.high_level.toy_controller.ToyController(toy: Toy, logger_name: str)[source]

Bases: ABC

Abstract base class for high-level toy control.

Extends the low-level Toy interface with synchronous methods, command queueing, and pattern playback capabilities.

Example:

# Get controller from ToyHub (see :class: ToyHub for that)
controllers = hub.connect_toys_blocking(discovered_toys)
toy = controllers[0]

# Manually control the toy
toy.intensity1(15)  # Set primary capability to level 15
toy.intensity2(10)  # Set secondary capability to level 10

# Set a pattern (duration_ms, intensity1, intensity2)
pattern = [
    (1000, 10, 5),   # 1 second at intensity 10/5
    (500, 0, 0),     # 0.5 seconds off
    (1000, 20, 20),  # 1 second at max intensity
]
toy.set_pattern(pattern, wraparound=True)

# Pause/resume pattern
toy.toggle_pause()  # Pauses pattern, sets toys intensity levels to  0
toy.toggle_pause()  # Resumes pattern
Parameters:

  • toy – Low-level toy object (Toy instance) for BLE communication.

  • logger_name – Name of the logger to use. Use empty string for root logger.

Note

This class should not be instantiated directly. Use ToyHub’s connection methods to get controller instances.

property brand: str

Returns a human-readable identifier of the toy brand e.g., ‘Lovense’.

change_rotation_direction(callback: Callable[[bool], None] | None = None) → None[source]

Change rotation direction (if supported).

This method toggles the rotation direction for toys with rotation capability. Safe to call on all toys. Does nothing and returns True via callback if rotation is not supported.

Parameters:

callback – Optional callback invoked when command completes. Receives True if successful or not supported, False if failed.

Example:

toy.change_rotation_direction(callback=lambda ok: print("Direction changed" if ok else "Failed"))

Note

You can use property:change_rotation_direction_available to check support before calling.

property change_rotation_direction_available: bool

Check if the toy supports changing the rotation direction.

Returns:

True if the rotation direction can be changed, False otherwise.

Return type:

bool

Example:

if toy.change_rotate_direction_available:
    toy.change_rotate_direction()
property current_intensities: tuple[int, int]

Get the current intensity values for the toy’s capabilities.

Returns:

A tuple of (primary_intensity, secondary_intensity). The secondary intensity is always 0 if the toy has only one capability.

Return type:

tuple[int, int]

Example::

intensity1, intensity2 = toy.current_intensities print(f”Primary intensity: {intensity1}, Secondary intensity: {intensity2}”)”)

direct_command(command: str, callback: Callable[[str], None]) → None[source]

Send a raw command directly to the toy.

Use this for accessing toy features not exposed by the library. Requires knowledge of the toy’s protocol.

Parameters:

  • command – Command string in the toy’s protocol format (e.g., “DeviceType”).

  • callback – Callback invoked with the toy’s response string. This callback is required (not optional).

Example:

def handle_response(response):
    print(f"Device type response: {response}")
    # Example: "C:11:0082059AD3BD"
toy.direct_command("DeviceType", callback=handle_response)
get_battery_level(callback: Callable[[int | None], None]) → None[source]

Retrieve the toy’s battery level.

Parameters:

callback – Callback invoked with battery level (0-100%) or None if unavailable. Unlike most methods, here providing a callback is required (not optional).

Example:

def show_battery(level):
    if level is not None:
        print(f"Battery: {level}%")
    else:
        print("Battery unavailable")
toy.get_battery_level(show_battery)

Note

You can provide a callback to ToyHub as well. If you do so, ToyHub queries battery levels regularly and invokes the hub’s battery callback. This method serves as an alternative to querying the battery level

abstractmethod get_information(callback: Callable[[dict[str, str]], None]) → None[source]

Gather detailed information about the toy.

The Information gathered depends on the toy, but might include the following dictionary keys:

  • ‘Battery Level’: Battery percentage (e.g., “75%”)

  • ‘Status’: Status code(e.g., “2” for normal)

  • ‘Batch number’: Manufacturing batch (e.g., “241015”)

  • ‘Bluetooth Name’: BLE device name (e.g., “LVS-Z36D”)

  • ‘Device type’: Device info (e.g., “C:11:ADDRESS”)

Parameters:

callback – Callback invoked with a dictionary containing toy information. Keys describe the Information type, values contain the information.

Example:

def show_info(info):
    print("Toy Information:")
    for key, value in info.items():
        print(f"{key}:{value}")
toy.get_information(show_info)
get_pattern_data() → tuple[list[tuple[int, int, int]], bool, bool, float][source]

Gets the complete pattern state for visualization.

Pattern state consists of: - pattern: List of tuples (duration, intensity1, intensity2) defining the pattern segments. - wraparound: Whether the pattern repeats from the beginning after completing the last segment. If False, both Intensities are 0 after the last segment. - is_paused: Whether the pattern is currently paused. Paused patterns do not advance - elapsed_time: Time elapsed since the start of the pattern or last wraparound in ms

Returns:

(pattern, wraparound, is_paused, elapsed_time)

Return type:

tuple

get_pattern_time() → float[source]

Get elapsed time in the current pattern (Time spent paused does not count toward elapsed time).

Returns:

Time elapsed in milliseconds since the pattern start or last wraparound. Returns 0.0 if no pattern is set

Return type:

float

Example:

elapsed = toy.get_pattern_time()
print(f"Pattern position: {elapsed}ms")
get_pattern_values(pattern_time: float) → tuple[int, int][source]

Get intensity values at a specific time in the pattern.

Parameters:

pattern_time – Time position in the pattern (milliseconds).

Returns:

(intensity1, intensity2) values at that time.

Return type:

tuple[int, int]

Note

For wraparound patterns, time is taken modulo the total pattern duration. For non-wraparound patterns, returns (0, 0) after the pattern completes.

intensity1(level: int, callback: Callable[[bool], None] | None = None) → None[source]

Set the intensity of the primary capability.

Commands are queued and executed asynchronously by ToyHub (every 50ms). If a pattern is active and not paused, calling this method pauses the pattern to avoid conflicts.

Parameters:

  • level – Intensity level. The Valid range is [0, self.max_intensity]. Values outside the range are clamped.

  • callback – Optional callback is invoked when the command completes. Receives True if successful, False if blocked or failed.

Example:

toy.intensity1(15)  # Simple command

def on_complete(success):
    print("Succeeded:", success)

toy.intensity1(toy.max_intensity, callback=on_complete)  # With callback

Note

If the toy is blocked, the callback receives False immediately and no command is sent. If disconnected, the command is queued and sent upon reconnection.

intensity2(level: int, callback: Callable[[bool], None] | None = None) → None[source]

Set the intensity of the secondary capability.

Behavior is identical to intensity1() but controls the secondary capability (e.g., rotation, air pump) Safe to call on toys without a secondary capability (will return true but do nothing).

Parameters:

  • level – Intensity level. The valid range is [0, self.max_intensity]. Values outside the range are clamped.

  • callback – Optional callback is invoked when the command completes. Receives True if successful, False if blocked or failed.

Example:

toy.intensity2(toy.max_intensity // 2)  # Set secondary capability intensity to medium
property intensity_names: tuple[str, str | None]

Get the display names for the toy’s capabilities.

Returns:

A tuple of (primary_name, secondary_name). The secondary name is None if the toy has only one capability.

Return type:

tuple[str, str | None]

Example:

names = toy.intensity_names
print(f"Primary: {names[0]}")  # example: Vibration
if names[1]:
    print(f"Secondary: {names[1]}")  # example: Rotation
property is_blocked: bool

Check if the toy is currently blocked.

When blocked, all intensity commands (manual and pattern-based) are rejected. Toy’s intensities are forced to 0.

Returns:

True if blocked, False otherwise.

Return type:

bool

property is_connected: bool

Check if the toy is currently connected.

When disconnected, commands are queued but not sent. Upon reconnection, queued commands are processed.

Returns:

True if connected, False otherwise.

Return type:

bool

property is_paused: bool

Check if pattern playback is currently paused.

When paused, the pattern timer stops advancing and toy intensities are set to zero. Manual commands can override the intensity levels

Returns:

True if paused, False otherwise.

Return type:

bool

property max_intensity: int

Get the maximum intensity value for this toy.

Returns:

Maximum intensity value (e.g., 20 for Lovense toys).

Return type:

int

Example:

max_val = toy.max_intensity
await toy.intensity1(max_val)  # Set to maximum
property model_name: str

Get the model name of the toy.

Returns:

Model name (e.g., “Nora”, “Lush”).

Return type:

str

property name: str

Returns a human-readable identifier of the toy e.g., Bluetooth name.

property pattern_version: int

Each time the pattern state changes, the version number is incremented.

Returns:

current version number.

async process_communication() → None[source]

Process queued commands and pattern playback (internal use only)

This method is called periodically by the ToyHub to execute queued commands and maintain pattern playback.

Warning

This is an internal method used by ToyHub and not meant to be used by you.

set_blocked(block: bool) → None[source]

Set the block state.

When blocked:

  • All intensity commands are rejected (return False via callback)

  • Toy intensities are forced to zero

  • Pattern continues advancing but doesn’t control the toy

  • Pause state is cleared if active (toy cannot be paused and blocked at the same time)

Parameters:

block – If true will be blocked, if false will be unblocked.

set_model_name(model_name: str, callback: Callable[[str | None], None] | None = None) → None[source]

Set the model name of the toy.

The model name determines which commands are available and how they’re interpreted. Like every other controller command, this is queued and executed asynchronously by ToyHub (validating the model against the toy involves sending commands), so the result is delivered via the optional callback rather than raised.

Parameters:

  • model_name – New model name. Must be a valid model for this toy’s brand.

  • callback – Optional callback invoked when the command completes. Receives the toy’s new model name on success, or None if the update failed (e.g. an invalid model name).

Example:

# Correct a model that was set incorrectly while connecting
toy.set_model_name("Nora", callback=lambda name: print(f"Model is now {name}"))

Note

For a blocking call that surfaces validation errors directly, use ToyHub.update_model_name() instead.

set_pattern(pattern: list[tuple[int, int, int]], wraparound: bool = True, reset_time: bool = True) → None[source]

Set a time-based pattern for automatic toy control.

Patterns are lists of segments. Each segment is a tuple of (duration_ms, intensity1, intensity2) where: - duration_ms: How long this segment lasts (milliseconds) - intensity1: Primary capability intensity (0-max) - intensity2: Secondary capability intensity (0-max) The maximum possible intensity can be looked up via intensity_max_value(). An empty list clears the pattern.

Parameters:

  • pattern – List of (duration_ms, intensity1, intensity2) tuples

  • wraparound – If True, the pattern loops indefinitely. If False, the pattern stops after one playthrough.

  • reset_time – If True, restart the pattern from the beginning. If False, maintain the current position in the pattern.

Example:

# Simple pulse pattern
pattern = [
    (500, 10, 0),   # 0.5s at intensity 10
    (500, 0, 0),    # 0.5s off
]
toy.set_pattern(pattern, wraparound=True)

# Clear pattern
toy.set_pattern([])

Note

Manual intensity commands automatically pause pattern playback to avoid conflicts. Call toggle_pause() to resume the pattern.

set_paused(pause: bool) → None[source]

Set the pattern playback pause state.

When paused: - If a pattern is active, it stops advancing. - Toy intensities are set to zero, but manual commands can override this. - Block state is cleared if active (toy cannot be paused and blocked at the same time)

Parameters:

pause – If true will be paused, if false will be unpaused.

stop(callback: Callable[[bool], None] | None = None) → None[source]

Stop all toy actions (set all intensities to zero).

If a pattern is active and not paused, this method pauses the pattern.

Parameters:

callback – Optional callback invoked when command completes. Receives True if successful, False otherwise.

Example:

toy.stop()
# With confirmation
toy.stop(callback=lambda ok: print("Stopped" if ok else "Failed"))
toggle_block() → bool[source]

Toggle block state.

When blocked: - All intensity commands are rejected (return False via callback) - Toy intensities are forced to zero - Pattern continues advancing but doesn’t control the toy - Pause state is cleared if active (toy cannot be paused and blocked at the same time)

Returns:

True if now blocked, False if now unblocked.

Return type:

bool

Example:

# Block all toy commands
is_blocked = toy.toggle_block()

# Try to control (will fail)
toy.intensity1(10, callback=lambda success: print(success))  # False

# Unblock
is_blocked = toy.toggle_block()
toggle_pause() → bool[source]

Toggle pattern playback pause state.

When paused: - If a pattern is active, it stops advancing. - Toy intensities are set to zero, but manual commands can override this. - Block state is cleared if active (toy cannot be paused and blocked at the same time)

Returns:

True if now paused, False if now unpaused.

Return type:

bool

Example:

# Pause pattern playback
is_paused = toy.toggle_pause()
print(f"Paused: {is_paused}")

# Resume
is_paused = toy.toggle_pause()
print(f"Paused: {is_paused}")
property toy: Toy

Get the underlying low-level toy object (internal use only)

Returns:

The low-level toy object.

Return type:

Toy

Warning

This is an internal method used by ToyHub and not meant to be used by you.

property toy_id: str

Get the unique identifier for this toy.

Returns:

Toy ID (typically the Bluetooth address).

Return type:

str

tikal.high_level.toy_hub module

Part of the High-Level API: Provides connection management for toy devices.

This module provides the ToyHub class, which serves as the entry point for all toy operations. The ToyHub manages: - Toy Discovery: Scanning for available devices via Bluetooth - Connection Management: Establishing and maintaining connections - Command Queueing: Processing commands from multiple toys concurrently - Pattern Playback: Managing time-based pattern execution - Battery Monitoring: Automatic periodic battery level updates - Reconnection: Automatic recovery from unexpected disconnects

Example

# Basic
hub = ToyHub()
toys = hub.discover_toys_blocking(5.0)
toys[0].model_name = "Lush"
controllers = hub.connect_toys_blocking(toys)
controllers[0].intensity1(15)
hub.shutdown()

# With callbacks
def on_error(exc, context, tb):
    print(f"Error {exc} while {context}. Traceback:{tb}")

def on_battery(levels):
    for toy_id, level in levels.items():
        print(f"Toy {toy_id} has battery ({level}%)")

hub = ToyHub(
    on_battery_update=on_battery,
    on_error=on_error,
    on_disconnect=lambda tid: print(f"{tid} disconnected"),
    on_reconnection_success=lambda tid: print(f"{tid} reconnected"),
    on_power_off=lambda tid: print(f"{tid} powered off"),
    logger_name="my_app",
    toy_cache_path=Path("./toys.json"),
    default_model="Please select a model"
)
class tikal.high_level.toy_hub.ToyHub(on_battery_update: ~typing.Callable[[dict[str, int | None]], ~typing.Any] | None = None, on_error: ~typing.Callable[[Exception, str, str], ~typing.Any] | None = None, on_disconnect: ~typing.Callable[[str], ~typing.Any] | None = None, on_reconnection_failure: ~typing.Callable[[str], ~typing.Any] | None = None, on_reconnection_success: ~typing.Callable[[str], ~typing.Any] | None = None, on_power_off: ~typing.Callable[[str], ~typing.Any] | None = None, logger_name: str = 'toy', toy_cache_path: ~pathlib.Path = PosixPath('.'), default_model: str = '', bluetooth_scanner: ~typing.Any = <class 'bleak.BleakScanner'>, bluetooth_client: ~typing.Any = <class 'bleak.BleakClient'>, mock_toys: bool = False)[source]

Bases: object

Central interface for toy communication and lifecycle management.

Part of the High-Level API: Handles discovery, connection, battery monitoring, and control of toys.

Parameters:

  • on_battery_update – Callback invoked when battery levels are updated (regularly). Receives dict mapping toy_id to battery level (int) or None if unavailable.

  • on_error – Callback invoked when critical errors occur. Receives (exception, context_message, traceback_string).

  • on_disconnect – Callback invoked when a toy disconnects unexpectedly. Receives toy_id. ToyHub automatically attempts reconnection.

  • on_reconnection_failure – Callback invoked when automatic reconnection fails. Receives toy_id.

  • on_reconnection_success – Callback invoked when automatic reconnection succeeds. Receives toy_id.

  • on_power_off – Callback invoked when a toy is powered off via its physical button. Receives toy_id.

  • logger_name – Name of the logger to use for logging messages.

  • toy_cache_path – Path to a file for caching toy model names. Allows automatic model name assignment on later discoveries.

  • default_model – Default model name to use if a toy isn’t in the cache.

  • bluetooth_scanner – BLE scanner class to use (defaults to BleakScanner). Can be overridden for testing.

  • bluetooth_client – BLE client class to use (defaults to BleakClient). Can be overridden for testing.

  • mock_toys – If True, use the MockEstimToys brand. Not part of the public API. This parameter may be removed without notice.

battery_update_callback(callback: Callable[[dict[str, int | None]], Any] | None) → None[source]

Set or update the battery update callback.

Parameters:

callback – New callback function or None to disable.

Example

def new_battery_handler(levels):
    print(f"Battery update: {levels}")
hub.battery_update_callback(new_battery_handler)
connect_toys_blocking(to_connect: list[ToyData], timeout: float = 30.0) → list[ToyController | BaseException][source]

Connect to specified toys synchronously (blocking call).

Attempts to connect to each toy in the list concurrently. Toys that connect successfully return ToyController instances; failed connections return exceptions.

Parameters:

  • to_connect – List of ToyData objects with a valid model_name set. Must have been discovered first.

  • timeout – Maximum time to wait for all connections in seconds.

Returns:

Each element is either a connected ToyController or an exception.

Order matches the input list.

Return type:

list[ToyController | BaseException]

Example

# Discover and connect
toys = hub.discover_toys_blocking(5.0)

# Set model names (required!)
toys[0].model_name = "Nora"
toys[1].model_name = "Lush"

# Connect
results = hub.connect_toys_blocking(toys, timeout=30.0)

# Process results
controllers = []
for i, result in enumerate(results):
    if isinstance(result, BaseException):
        print(f"Failed to connect to {toys[i].name}: {result}")
    else:
        print(f"Connected: {result.model_name}")
        controllers.append(result)
connect_toys_callback(to_connect: list[ToyData], on_connected: Callable[[list[ToyController | BaseException]], Any], timeout: float = 30.0) → None[source]

Connect to specified toys with a callback (non-blocking).

Starts connections in the background and returns immediately. The callback is invoked when all connection attempts are complete.

Parameters:

  • to_connect – List of ToyData objects with a valid model_name set.

  • on_connected – Callback invoked with a list of controllers or exceptions. Order matches the input list.

  • timeout – Maximum time to wait for all connections in seconds.

Example

def handle_connection(results):
    for result in results:
        if isinstance(result, BaseException):
            print(f"Connection failed: {result}")
        else:
            print(f"Connected: {result.model_name}")

hub.connect_toys_callback(toys, handle_connection, timeout=30.0)
disconnect_callback(callback: Callable[[str], Any] | None) → None[source]

Set or update the disconnect callback.

Parameters:

callback – New callback function or None to disable.

disconnect_toys_blocking(to_disconnect: list[str], timeout: float = 10.0) → Sequence[BaseException | None][source]

Disconnect specified toys synchronously (blocking call).

Cleanly disconnects from the specified toys, stopping all actions and closing BLE connections.

Parameters:

  • to_disconnect – List of toy_ids (Bluetooth addresses) to disconnect.

  • timeout – Maximum time to wait for all disconnections in seconds.

Returns:

List where each element is either None (successful disconnect) or an exception (failed disconnect). Order matches the input list. Toys are still disconnected even if an exception occurs.

Return type:

list[BaseException | None]

Example

# Disconnect specific toys
toy_ids = [controller.toy_id for controller in controllers]
results = hub.disconnect_toys_blocking(toy_ids, timeout=10.0)

# Check results
for toy_id, result in zip(toy_ids, results):
    if result is None:
        print(f"{toy_id} disconnected successfully")
    else:
        print(f"{toy_id} disconnect failed: {result}")
disconnect_toys_callback(to_disconnect: list[str], on_disconnected: Callable[[list[BaseException | None]], Any], timeout: float = 10.0) → None[source]

Disconnect specified toys with a callback (non-blocking).

Starts disconnections in the background and returns immediately. The callback is invoked when all disconnection attempts are complete.

Parameters:

  • to_disconnect – List of toy_ids to disconnect.

  • on_disconnected – Callback invoked with a list of exceptions (or None for successful disconnects). Toys are still disconnected even if an exception occurs. Order matches the input list.

  • timeout – Maximum time to wait for all disconnections in seconds.

Example

def handle_disconnects(results):
    success_count = sum(1 for r in results if r is None)
    print(f"{success_count}/{len(results)} disconnected successfully")

toy_ids = [c.toy_id for c in controllers]
hub.disconnect_toys_callback(toy_ids, handle_disconnects)
discover_toys_blocking(timeout: float = 10.0) → list[ToyData][source]

Discover available toys synchronously (blocking call).

Scans for nearby toys via Bluetooth and returns their discovery data. Model names are automatically filled from the cache if available.

Parameters:

timeout – Maximum scan duration in seconds. Longer timeouts may discover more devices but take longer.

Returns:

List of discovered toys. model_name set from cache if possible

Return type:

list[ToyData]

Raises:

  • TimeoutError – If discovery exceeds timeout * 2. Should not occur with BleakScanner

  • Exception – Any exception from the underlying BLE scanner.

  • RuntimeError – If a continuous scan is in progress. See meth: start_continuous_scan and meth: stop_continuous_scan

Example

toys = hub.discover_toys_blocking(timeout=10.0)

for toy in toys:
    print(f"Found: {toy.name}")
    if toy.model_name:
        print(f"Cached model: {toy.model_name}")
    else:
        print(f"Model unknown. Please set manually")
discover_toys_callback(on_discovered: Callable[[list[ToyData] | BaseException], None], timeout: float = 10.0) → None[source]

Discover available toys with a callback (non-blocking).

Starts discovery in the background and returns immediately. The callback is invoked when discovery completes.

Parameters:

  • on_discovered – Callback invoked with either a list of discovered toys or an exception if discovery failed.

  • timeout – Maximum scan duration in seconds.

Example

def handle_discovery(result):
    if isinstance(result, Exception):
        print(f"Discovery failed: {result}")
        return

    print(f"Found {len(result)} toys")
    for toy in result:
        print(toy.name)

hub.discover_toys_callback(handle_discovery, timeout=5.0)
error_callback(callback: Callable[[Exception, str, str], Any] | None) → None[source]

Set or update the error callback.

Parameters:

callback – New callback function or None to disable.

Example

def error_handler(exc, context, tb):
    print(f"Hub error {exc} while {context}. Traceback: {tb}")
hub.error_callback(error_handler)
property is_running: bool

Check if the communication loop is currently running.

Returns:

True if the loop is active, False otherwise.

Return type:

bool

Note

The loop starts automatically when toys are connected and stops when all toys are disconnected.

power_off_callback(callback: Callable[[str], Any] | None) → None[source]

Set or update the power-off callback.

Parameters:

callback – New callback function or None to disable.

reconnection_failure_callback(callback: Callable[[str], Any] | None) → None[source]

Set or update the reconnection failure callback.

Parameters:

callback – New callback function or None to disable.

reconnection_success_callback(callback: Callable[[str], Any] | None) → None[source]

Set or update the reconnection success callback.

Parameters:

callback – New callback function or None to disable.

shutdown() → None[source]

Stop the communication loop, disconnect all toys, and clean up resources.

This method should always be called before the program exits to ensure: - All toys are properly disconnected - The communication loop is stopped, and the async runner is shut down cleanly

Example

hub = ToyHub()
# ... use hub ...
hub.shutdown()

Note

After calling shutdown(), the ToyHub instance should not be reused. Create a new instance if you need to start working with toys again.

start_discovery(on_update: Callable[[list[ToyData]], Any]) → None[source]

Starts the discovery process for toys and updates the provided callback whenever new toy data is discovered or an error occurs.

Parameters:

on_update (Callable[[list[ToyData]], Any]) – Called with a list[ToyData] of ALL available toys whenever availability changes. ‘ALL’ includes toys that were previously discovered and are still available. Connected toys do not advertise and are not included. Invoked with an empty list if an exception occurs.

Note

Should any exception occur, it is provided to self.error_callback. Any exception stops discovery.

stop_discovery() → None[source]
update_model_name(toy_id: str, model_name: str) → ToyController | BaseException[source]

Update the model name for a connected toy.

Changes the toy’s model name, which affects which commands are available and how they’re interpreted

Parameters:

  • toy_id – Unique identifier of the toy to update.

  • model_name – New model name (must be valid for the toy’s brand).

Returns:

The updated controller if successful, or:

  • InvalidModelError: If model_name is not valid for this toy brand.

  • BadModelError: If the model_name is valid, but commands still fail. See BadModelError for details

  • ValueError if the toy_id is unknown.

Return type:

ToyController | BaseException

Example

# Correct a wrong model assignment
result = hub.update_model_name(toy_id, "Nora")
if isinstance(result, BaseException):
    print(f"Update failed: {result}")
else:
    print(f"Model updated to {result.model_name}")

Module contents

class tikal.high_level.LovenseController(toy: LovenseToy, logger_name: str)[source]

Bases: ToyController

High-level controller for Lovense toys.

Extends the low-level Lovense class with synchronous methods, command queueing, and pattern playback capabilities.

Parameters:

  • toy – Low-level Lovense instance.

  • logger_name – Name of the logger to use. Use empty string for root logger.

Example:

# Connection (see :class ToyHub)
controllers = hub.connect_toys_blocking(discovered_toys)
toy = controllers[0]  # LovenseController instance

# Manual control
toy.intensity1(15)  # Primary capability
toy.intensity2(10)  # Secondary capability

# Pattern control
pattern = [
    (1000, 10, 5),
    (500, 0, 0),
    (1000, 20, 10),
]
toy.set_pattern(pattern, wraparound=True)

# Pause/resume
toy.toggle_pause()

# Get battery level
toy.get_battery_level(lambda lvl: print(f"Battery: {lvl}%"))

# Advanced features
if toy.change_rotate_direction_available():
    toy.change_rotate_direction()

Note

This class should not be instantiated directly. Use ToyHub’s connection methods to get controller instances.

get_information(callback: Callable[[dict[str, str]], None]) → None[source]

Gather detailed information about the toy.

The following information is gathered (Key, Value pairs of the dictionary):

  • ‘Battery Level’: Battery percentage (e.g., “75%”)

  • ‘Status’: Status code (“2” for normal)

  • ‘Batch number’: Manufacturing batch (e.g., “241015”)

  • ‘Bluetooth Name’: BLE device name (e.g., “LVS-Z36D”)

  • ‘Device type’: Device info (e.g., “C:11:ADDRESS”)

Parameters:

callback – Callback invoked with a dictionary containing toy information. Keys describe the Information type, values contain the information.

Example:

def show_info(info):
    print("Toy Information:")
    for key, value in info.items():
        print(f"{key}:{value}")
toy.get_information(show_info)
class tikal.high_level.ToyCache(cache_path: Path, default_model: str, logger_name: str)[source]

Bases: object

Persistent storage mapping Bluetooth names to toy model names.

Maintains a JSON file on the disk that maps Bluetooth device names (e.g., “LVS-A123”) to their corresponding model names (e.g., “Nora”). This eliminates the need for users to manually identify toys every time they connect.

The cache is loaded on initialization. Encountered errors are logged, and ToyCache fails silently.

Parameters:

  • cache_path – Path to the JSON cache file. If the path is empty (=``Path()``), the cache operates with no persistence.

  • default_model – Default model name to return when a Bluetooth name is not found in the cache.

  • logger_name – Name of the logger to use. Use empty string for root logger.

Example:

from pathlib import Path
from toy_cache import ToyCache

cache = ToyCache(
    cache_path=Path("./toys.json"),
    default_model="unknown",
    logger_name="myapp"
)

# First time: toy not in cache
model = cache.get_model_name("LVS-A123")  # Returns "unknown"

# User selects a model, you can update the cache
cache.update({"LVS-A123": "Nora"})

# Next time: toy is in cache
model = cache.get_model_name("LVS-A123")  # Returns "Nora"
get_model_name(bluetooth_name: str) → str[source]

Retrieve the cached model name for a Bluetooth device.

Looks up the model name associated with the given Bluetooth name. If the name is not found in the cache, it returns the default model name specified during initialization.

Parameters:

bluetooth_name – Bluetooth device name to look up (e.g., “LVS-A123”). This is obtained from ToyData.name during discovery.

Returns:

The cached model name (e.g., “Nora”) if found, otherwise the default model name.

Return type:

str

Example:

# Check if the toy is cached
model = cache.get_model_name("LVS-A123")
if model == default_model_name:
    print("Unknown toy")
else:
    print(f"Known toy: {model}")
update(updates: dict[str, str]) → None[source]

Update cache entries and persist to disk.

Merges the provided updates into the cache, overwriting existing entries and adding new ones. The updated cache is immediately written to the disk.

Parameters:

updates – Dictionary mapping Bluetooth names (keys) to model names (values)

Example:

# Update single entry
cache.update({"LVS-A123": "Nora"})

# Update multiple entries
cache.update({
    "LVS-B456": "Lush",
    "LVS-C789": "Max"
})
# Change existing model
cache.update({"LVS-A123": "Ridge"})  # Overwrites "Nora"

Note

Fails silently if disk I/O errors occur. Errors are logged and do not raise exceptions. The in-memory cache is always updated even if writing to disk fails. If the cache was initialized with an empty path (Path()), no disk write is attempted.

class tikal.high_level.ToyController(toy: Toy, logger_name: str)[source]

Bases: ABC

Abstract base class for high-level toy control.

Extends the low-level Toy interface with synchronous methods, command queueing, and pattern playback capabilities.

Example:

# Get controller from ToyHub (see :class: ToyHub for that)
controllers = hub.connect_toys_blocking(discovered_toys)
toy = controllers[0]

# Manually control the toy
toy.intensity1(15)  # Set primary capability to level 15
toy.intensity2(10)  # Set secondary capability to level 10

# Set a pattern (duration_ms, intensity1, intensity2)
pattern = [
    (1000, 10, 5),   # 1 second at intensity 10/5
    (500, 0, 0),     # 0.5 seconds off
    (1000, 20, 20),  # 1 second at max intensity
]
toy.set_pattern(pattern, wraparound=True)

# Pause/resume pattern
toy.toggle_pause()  # Pauses pattern, sets toys intensity levels to  0
toy.toggle_pause()  # Resumes pattern
Parameters:

  • toy – Low-level toy object (Toy instance) for BLE communication.

  • logger_name – Name of the logger to use. Use empty string for root logger.

Note

This class should not be instantiated directly. Use ToyHub’s connection methods to get controller instances.

property brand: str

Returns a human-readable identifier of the toy brand e.g., ‘Lovense’.

change_rotation_direction(callback: Callable[[bool], None] | None = None) → None[source]

Change rotation direction (if supported).

This method toggles the rotation direction for toys with rotation capability. Safe to call on all toys. Does nothing and returns True via callback if rotation is not supported.

Parameters:

callback – Optional callback invoked when command completes. Receives True if successful or not supported, False if failed.

Example:

toy.change_rotation_direction(callback=lambda ok: print("Direction changed" if ok else "Failed"))

Note

You can use property:change_rotation_direction_available to check support before calling.

property change_rotation_direction_available: bool

Check if the toy supports changing the rotation direction.

Returns:

True if the rotation direction can be changed, False otherwise.

Return type:

bool

Example:

if toy.change_rotate_direction_available:
    toy.change_rotate_direction()
property current_intensities: tuple[int, int]

Get the current intensity values for the toy’s capabilities.

Returns:

A tuple of (primary_intensity, secondary_intensity). The secondary intensity is always 0 if the toy has only one capability.

Return type:

tuple[int, int]

Example::

intensity1, intensity2 = toy.current_intensities print(f”Primary intensity: {intensity1}, Secondary intensity: {intensity2}”)”)

direct_command(command: str, callback: Callable[[str], None]) → None[source]

Send a raw command directly to the toy.

Use this for accessing toy features not exposed by the library. Requires knowledge of the toy’s protocol.

Parameters:

  • command – Command string in the toy’s protocol format (e.g., “DeviceType”).

  • callback – Callback invoked with the toy’s response string. This callback is required (not optional).

Example:

def handle_response(response):
    print(f"Device type response: {response}")
    # Example: "C:11:0082059AD3BD"
toy.direct_command("DeviceType", callback=handle_response)
get_battery_level(callback: Callable[[int | None], None]) → None[source]

Retrieve the toy’s battery level.

Parameters:

callback – Callback invoked with battery level (0-100%) or None if unavailable. Unlike most methods, here providing a callback is required (not optional).

Example:

def show_battery(level):
    if level is not None:
        print(f"Battery: {level}%")
    else:
        print("Battery unavailable")
toy.get_battery_level(show_battery)

Note

You can provide a callback to ToyHub as well. If you do so, ToyHub queries battery levels regularly and invokes the hub’s battery callback. This method serves as an alternative to querying the battery level

abstractmethod get_information(callback: Callable[[dict[str, str]], None]) → None[source]

Gather detailed information about the toy.

The Information gathered depends on the toy, but might include the following dictionary keys:

  • ‘Battery Level’: Battery percentage (e.g., “75%”)

  • ‘Status’: Status code(e.g., “2” for normal)

  • ‘Batch number’: Manufacturing batch (e.g., “241015”)

  • ‘Bluetooth Name’: BLE device name (e.g., “LVS-Z36D”)

  • ‘Device type’: Device info (e.g., “C:11:ADDRESS”)

Parameters:

callback – Callback invoked with a dictionary containing toy information. Keys describe the Information type, values contain the information.

Example:

def show_info(info):
    print("Toy Information:")
    for key, value in info.items():
        print(f"{key}:{value}")
toy.get_information(show_info)
get_pattern_data() → tuple[list[tuple[int, int, int]], bool, bool, float][source]

Gets the complete pattern state for visualization.

Pattern state consists of: - pattern: List of tuples (duration, intensity1, intensity2) defining the pattern segments. - wraparound: Whether the pattern repeats from the beginning after completing the last segment. If False, both Intensities are 0 after the last segment. - is_paused: Whether the pattern is currently paused. Paused patterns do not advance - elapsed_time: Time elapsed since the start of the pattern or last wraparound in ms

Returns:

(pattern, wraparound, is_paused, elapsed_time)

Return type:

tuple

get_pattern_time() → float[source]

Get elapsed time in the current pattern (Time spent paused does not count toward elapsed time).

Returns:

Time elapsed in milliseconds since the pattern start or last wraparound. Returns 0.0 if no pattern is set

Return type:

float

Example:

elapsed = toy.get_pattern_time()
print(f"Pattern position: {elapsed}ms")
get_pattern_values(pattern_time: float) → tuple[int, int][source]

Get intensity values at a specific time in the pattern.

Parameters:

pattern_time – Time position in the pattern (milliseconds).

Returns:

(intensity1, intensity2) values at that time.

Return type:

tuple[int, int]

Note

For wraparound patterns, time is taken modulo the total pattern duration. For non-wraparound patterns, returns (0, 0) after the pattern completes.

intensity1(level: int, callback: Callable[[bool], None] | None = None) → None[source]

Set the intensity of the primary capability.

Commands are queued and executed asynchronously by ToyHub (every 50ms). If a pattern is active and not paused, calling this method pauses the pattern to avoid conflicts.

Parameters:

  • level – Intensity level. The Valid range is [0, self.max_intensity]. Values outside the range are clamped.

  • callback – Optional callback is invoked when the command completes. Receives True if successful, False if blocked or failed.

Example:

toy.intensity1(15)  # Simple command

def on_complete(success):
    print("Succeeded:", success)

toy.intensity1(toy.max_intensity, callback=on_complete)  # With callback

Note

If the toy is blocked, the callback receives False immediately and no command is sent. If disconnected, the command is queued and sent upon reconnection.

intensity2(level: int, callback: Callable[[bool], None] | None = None) → None[source]

Set the intensity of the secondary capability.

Behavior is identical to intensity1() but controls the secondary capability (e.g., rotation, air pump) Safe to call on toys without a secondary capability (will return true but do nothing).

Parameters:

  • level – Intensity level. The valid range is [0, self.max_intensity]. Values outside the range are clamped.

  • callback – Optional callback is invoked when the command completes. Receives True if successful, False if blocked or failed.

Example:

toy.intensity2(toy.max_intensity // 2)  # Set secondary capability intensity to medium
property intensity_names: tuple[str, str | None]

Get the display names for the toy’s capabilities.

Returns:

A tuple of (primary_name, secondary_name). The secondary name is None if the toy has only one capability.

Return type:

tuple[str, str | None]

Example:

names = toy.intensity_names
print(f"Primary: {names[0]}")  # example: Vibration
if names[1]:
    print(f"Secondary: {names[1]}")  # example: Rotation
property is_blocked: bool

Check if the toy is currently blocked.

When blocked, all intensity commands (manual and pattern-based) are rejected. Toy’s intensities are forced to 0.

Returns:

True if blocked, False otherwise.

Return type:

bool

property is_connected: bool

Check if the toy is currently connected.

When disconnected, commands are queued but not sent. Upon reconnection, queued commands are processed.

Returns:

True if connected, False otherwise.

Return type:

bool

property is_paused: bool

Check if pattern playback is currently paused.

When paused, the pattern timer stops advancing and toy intensities are set to zero. Manual commands can override the intensity levels

Returns:

True if paused, False otherwise.

Return type:

bool

property max_intensity: int

Get the maximum intensity value for this toy.

Returns:

Maximum intensity value (e.g., 20 for Lovense toys).

Return type:

int

Example:

max_val = toy.max_intensity
await toy.intensity1(max_val)  # Set to maximum
property model_name: str

Get the model name of the toy.

Returns:

Model name (e.g., “Nora”, “Lush”).

Return type:

str

property name: str

Returns a human-readable identifier of the toy e.g., Bluetooth name.

property pattern_version: int

Each time the pattern state changes, the version number is incremented.

Returns:

current version number.

async process_communication() → None[source]

Process queued commands and pattern playback (internal use only)

This method is called periodically by the ToyHub to execute queued commands and maintain pattern playback.

Warning

This is an internal method used by ToyHub and not meant to be used by you.

set_blocked(block: bool) → None[source]

Set the block state.

When blocked:

  • All intensity commands are rejected (return False via callback)

  • Toy intensities are forced to zero

  • Pattern continues advancing but doesn’t control the toy

  • Pause state is cleared if active (toy cannot be paused and blocked at the same time)

Parameters:

block – If true will be blocked, if false will be unblocked.

set_model_name(model_name: str, callback: Callable[[str | None], None] | None = None) → None[source]

Set the model name of the toy.

The model name determines which commands are available and how they’re interpreted. Like every other controller command, this is queued and executed asynchronously by ToyHub (validating the model against the toy involves sending commands), so the result is delivered via the optional callback rather than raised.

Parameters:

  • model_name – New model name. Must be a valid model for this toy’s brand.

  • callback – Optional callback invoked when the command completes. Receives the toy’s new model name on success, or None if the update failed (e.g. an invalid model name).

Example:

# Correct a model that was set incorrectly while connecting
toy.set_model_name("Nora", callback=lambda name: print(f"Model is now {name}"))

Note

For a blocking call that surfaces validation errors directly, use ToyHub.update_model_name() instead.

set_pattern(pattern: list[tuple[int, int, int]], wraparound: bool = True, reset_time: bool = True) → None[source]

Set a time-based pattern for automatic toy control.

Patterns are lists of segments. Each segment is a tuple of (duration_ms, intensity1, intensity2) where: - duration_ms: How long this segment lasts (milliseconds) - intensity1: Primary capability intensity (0-max) - intensity2: Secondary capability intensity (0-max) The maximum possible intensity can be looked up via intensity_max_value(). An empty list clears the pattern.

Parameters:

  • pattern – List of (duration_ms, intensity1, intensity2) tuples

  • wraparound – If True, the pattern loops indefinitely. If False, the pattern stops after one playthrough.

  • reset_time – If True, restart the pattern from the beginning. If False, maintain the current position in the pattern.

Example:

# Simple pulse pattern
pattern = [
    (500, 10, 0),   # 0.5s at intensity 10
    (500, 0, 0),    # 0.5s off
]
toy.set_pattern(pattern, wraparound=True)

# Clear pattern
toy.set_pattern([])

Note

Manual intensity commands automatically pause pattern playback to avoid conflicts. Call toggle_pause() to resume the pattern.

set_paused(pause: bool) → None[source]

Set the pattern playback pause state.

When paused: - If a pattern is active, it stops advancing. - Toy intensities are set to zero, but manual commands can override this. - Block state is cleared if active (toy cannot be paused and blocked at the same time)

Parameters:

pause – If true will be paused, if false will be unpaused.

stop(callback: Callable[[bool], None] | None = None) → None[source]

Stop all toy actions (set all intensities to zero).

If a pattern is active and not paused, this method pauses the pattern.

Parameters:

callback – Optional callback invoked when command completes. Receives True if successful, False otherwise.

Example:

toy.stop()
# With confirmation
toy.stop(callback=lambda ok: print("Stopped" if ok else "Failed"))
toggle_block() → bool[source]

Toggle block state.

When blocked: - All intensity commands are rejected (return False via callback) - Toy intensities are forced to zero - Pattern continues advancing but doesn’t control the toy - Pause state is cleared if active (toy cannot be paused and blocked at the same time)

Returns:

True if now blocked, False if now unblocked.

Return type:

bool

Example:

# Block all toy commands
is_blocked = toy.toggle_block()

# Try to control (will fail)
toy.intensity1(10, callback=lambda success: print(success))  # False

# Unblock
is_blocked = toy.toggle_block()
toggle_pause() → bool[source]

Toggle pattern playback pause state.

When paused: - If a pattern is active, it stops advancing. - Toy intensities are set to zero, but manual commands can override this. - Block state is cleared if active (toy cannot be paused and blocked at the same time)

Returns:

True if now paused, False if now unpaused.

Return type:

bool

Example:

# Pause pattern playback
is_paused = toy.toggle_pause()
print(f"Paused: {is_paused}")

# Resume
is_paused = toy.toggle_pause()
print(f"Paused: {is_paused}")
property toy: Toy

Get the underlying low-level toy object (internal use only)

Returns:

The low-level toy object.

Return type:

Toy

Warning

This is an internal method used by ToyHub and not meant to be used by you.

property toy_id: str

Get the unique identifier for this toy.

Returns:

Toy ID (typically the Bluetooth address).

Return type:

str

class tikal.high_level.ToyHub(on_battery_update: ~typing.Callable[[dict[str, int | None]], ~typing.Any] | None = None, on_error: ~typing.Callable[[Exception, str, str], ~typing.Any] | None = None, on_disconnect: ~typing.Callable[[str], ~typing.Any] | None = None, on_reconnection_failure: ~typing.Callable[[str], ~typing.Any] | None = None, on_reconnection_success: ~typing.Callable[[str], ~typing.Any] | None = None, on_power_off: ~typing.Callable[[str], ~typing.Any] | None = None, logger_name: str = 'toy', toy_cache_path: ~pathlib.Path = PosixPath('.'), default_model: str = '', bluetooth_scanner: ~typing.Any = <class 'bleak.BleakScanner'>, bluetooth_client: ~typing.Any = <class 'bleak.BleakClient'>, mock_toys: bool = False)[source]

Bases: object

Central interface for toy communication and lifecycle management.

Part of the High-Level API: Handles discovery, connection, battery monitoring, and control of toys.

Parameters:

  • on_battery_update – Callback invoked when battery levels are updated (regularly). Receives dict mapping toy_id to battery level (int) or None if unavailable.

  • on_error – Callback invoked when critical errors occur. Receives (exception, context_message, traceback_string).

  • on_disconnect – Callback invoked when a toy disconnects unexpectedly. Receives toy_id. ToyHub automatically attempts reconnection.

  • on_reconnection_failure – Callback invoked when automatic reconnection fails. Receives toy_id.

  • on_reconnection_success – Callback invoked when automatic reconnection succeeds. Receives toy_id.

  • on_power_off – Callback invoked when a toy is powered off via its physical button. Receives toy_id.

  • logger_name – Name of the logger to use for logging messages.

  • toy_cache_path – Path to a file for caching toy model names. Allows automatic model name assignment on later discoveries.

  • default_model – Default model name to use if a toy isn’t in the cache.

  • bluetooth_scanner – BLE scanner class to use (defaults to BleakScanner). Can be overridden for testing.

  • bluetooth_client – BLE client class to use (defaults to BleakClient). Can be overridden for testing.

  • mock_toys – If True, use the MockEstimToys brand. Not part of the public API. This parameter may be removed without notice.

battery_update_callback(callback: Callable[[dict[str, int | None]], Any] | None) → None[source]

Set or update the battery update callback.

Parameters:

callback – New callback function or None to disable.

Example

def new_battery_handler(levels):
    print(f"Battery update: {levels}")
hub.battery_update_callback(new_battery_handler)
connect_toys_blocking(to_connect: list[ToyData], timeout: float = 30.0) → list[ToyController | BaseException][source]

Connect to specified toys synchronously (blocking call).

Attempts to connect to each toy in the list concurrently. Toys that connect successfully return ToyController instances; failed connections return exceptions.

Parameters:

  • to_connect – List of ToyData objects with a valid model_name set. Must have been discovered first.

  • timeout – Maximum time to wait for all connections in seconds.

Returns:

Each element is either a connected ToyController or an exception.

Order matches the input list.

Return type:

list[ToyController | BaseException]

Example

# Discover and connect
toys = hub.discover_toys_blocking(5.0)

# Set model names (required!)
toys[0].model_name = "Nora"
toys[1].model_name = "Lush"

# Connect
results = hub.connect_toys_blocking(toys, timeout=30.0)

# Process results
controllers = []
for i, result in enumerate(results):
    if isinstance(result, BaseException):
        print(f"Failed to connect to {toys[i].name}: {result}")
    else:
        print(f"Connected: {result.model_name}")
        controllers.append(result)
connect_toys_callback(to_connect: list[ToyData], on_connected: Callable[[list[ToyController | BaseException]], Any], timeout: float = 30.0) → None[source]

Connect to specified toys with a callback (non-blocking).

Starts connections in the background and returns immediately. The callback is invoked when all connection attempts are complete.

Parameters:

  • to_connect – List of ToyData objects with a valid model_name set.

  • on_connected – Callback invoked with a list of controllers or exceptions. Order matches the input list.

  • timeout – Maximum time to wait for all connections in seconds.

Example

def handle_connection(results):
    for result in results:
        if isinstance(result, BaseException):
            print(f"Connection failed: {result}")
        else:
            print(f"Connected: {result.model_name}")

hub.connect_toys_callback(toys, handle_connection, timeout=30.0)
disconnect_callback(callback: Callable[[str], Any] | None) → None[source]

Set or update the disconnect callback.

Parameters:

callback – New callback function or None to disable.

disconnect_toys_blocking(to_disconnect: list[str], timeout: float = 10.0) → Sequence[BaseException | None][source]

Disconnect specified toys synchronously (blocking call).

Cleanly disconnects from the specified toys, stopping all actions and closing BLE connections.

Parameters:

  • to_disconnect – List of toy_ids (Bluetooth addresses) to disconnect.

  • timeout – Maximum time to wait for all disconnections in seconds.

Returns:

List where each element is either None (successful disconnect) or an exception (failed disconnect). Order matches the input list. Toys are still disconnected even if an exception occurs.

Return type:

list[BaseException | None]

Example

# Disconnect specific toys
toy_ids = [controller.toy_id for controller in controllers]
results = hub.disconnect_toys_blocking(toy_ids, timeout=10.0)

# Check results
for toy_id, result in zip(toy_ids, results):
    if result is None:
        print(f"{toy_id} disconnected successfully")
    else:
        print(f"{toy_id} disconnect failed: {result}")
disconnect_toys_callback(to_disconnect: list[str], on_disconnected: Callable[[list[BaseException | None]], Any], timeout: float = 10.0) → None[source]

Disconnect specified toys with a callback (non-blocking).

Starts disconnections in the background and returns immediately. The callback is invoked when all disconnection attempts are complete.

Parameters:

  • to_disconnect – List of toy_ids to disconnect.

  • on_disconnected – Callback invoked with a list of exceptions (or None for successful disconnects). Toys are still disconnected even if an exception occurs. Order matches the input list.

  • timeout – Maximum time to wait for all disconnections in seconds.

Example

def handle_disconnects(results):
    success_count = sum(1 for r in results if r is None)
    print(f"{success_count}/{len(results)} disconnected successfully")

toy_ids = [c.toy_id for c in controllers]
hub.disconnect_toys_callback(toy_ids, handle_disconnects)
discover_toys_blocking(timeout: float = 10.0) → list[ToyData][source]

Discover available toys synchronously (blocking call).

Scans for nearby toys via Bluetooth and returns their discovery data. Model names are automatically filled from the cache if available.

Parameters:

timeout – Maximum scan duration in seconds. Longer timeouts may discover more devices but take longer.

Returns:

List of discovered toys. model_name set from cache if possible

Return type:

list[ToyData]

Raises:

  • TimeoutError – If discovery exceeds timeout * 2. Should not occur with BleakScanner

  • Exception – Any exception from the underlying BLE scanner.

  • RuntimeError – If a continuous scan is in progress. See meth: start_continuous_scan and meth: stop_continuous_scan

Example

toys = hub.discover_toys_blocking(timeout=10.0)

for toy in toys:
    print(f"Found: {toy.name}")
    if toy.model_name:
        print(f"Cached model: {toy.model_name}")
    else:
        print(f"Model unknown. Please set manually")
discover_toys_callback(on_discovered: Callable[[list[ToyData] | BaseException], None], timeout: float = 10.0) → None[source]

Discover available toys with a callback (non-blocking).

Starts discovery in the background and returns immediately. The callback is invoked when discovery completes.

Parameters:

  • on_discovered – Callback invoked with either a list of discovered toys or an exception if discovery failed.

  • timeout – Maximum scan duration in seconds.

Example

def handle_discovery(result):
    if isinstance(result, Exception):
        print(f"Discovery failed: {result}")
        return

    print(f"Found {len(result)} toys")
    for toy in result:
        print(toy.name)

hub.discover_toys_callback(handle_discovery, timeout=5.0)
error_callback(callback: Callable[[Exception, str, str], Any] | None) → None[source]

Set or update the error callback.

Parameters:

callback – New callback function or None to disable.

Example

def error_handler(exc, context, tb):
    print(f"Hub error {exc} while {context}. Traceback: {tb}")
hub.error_callback(error_handler)
property is_running: bool

Check if the communication loop is currently running.

Returns:

True if the loop is active, False otherwise.

Return type:

bool

Note

The loop starts automatically when toys are connected and stops when all toys are disconnected.

power_off_callback(callback: Callable[[str], Any] | None) → None[source]

Set or update the power-off callback.

Parameters:

callback – New callback function or None to disable.

reconnection_failure_callback(callback: Callable[[str], Any] | None) → None[source]

Set or update the reconnection failure callback.

Parameters:

callback – New callback function or None to disable.

reconnection_success_callback(callback: Callable[[str], Any] | None) → None[source]

Set or update the reconnection success callback.

Parameters:

callback – New callback function or None to disable.

shutdown() → None[source]

Stop the communication loop, disconnect all toys, and clean up resources.

This method should always be called before the program exits to ensure: - All toys are properly disconnected - The communication loop is stopped, and the async runner is shut down cleanly

Example

hub = ToyHub()
# ... use hub ...
hub.shutdown()

Note

After calling shutdown(), the ToyHub instance should not be reused. Create a new instance if you need to start working with toys again.

start_discovery(on_update: Callable[[list[ToyData]], Any]) → None[source]

Starts the discovery process for toys and updates the provided callback whenever new toy data is discovered or an error occurs.

Parameters:

on_update (Callable[[list[ToyData]], Any]) – Called with a list[ToyData] of ALL available toys whenever availability changes. ‘ALL’ includes toys that were previously discovered and are still available. Connected toys do not advertise and are not included. Invoked with an empty list if an exception occurs.

Note

Should any exception occur, it is provided to self.error_callback. Any exception stops discovery.

stop_discovery() → None[source]
update_model_name(toy_id: str, model_name: str) → ToyController | BaseException[source]

Update the model name for a connected toy.

Changes the toy’s model name, which affects which commands are available and how they’re interpreted

Parameters:

  • toy_id – Unique identifier of the toy to update.

  • model_name – New model name (must be valid for the toy’s brand).

Returns:

The updated controller if successful, or:

  • InvalidModelError: If model_name is not valid for this toy brand.

  • BadModelError: If the model_name is valid, but commands still fail. See BadModelError for details

  • ValueError if the toy_id is unknown.

Return type:

ToyController | BaseException

Example

# Correct a wrong model assignment
result = hub.update_model_name(toy_id, "Nora")
if isinstance(result, BaseException):
    print(f"Update failed: {result}")
else:
    print(f"Model updated to {result.model_name}")