tikal.low_level package

Subpackages

• tikal.low_level.brands package
  • Subpackages
    • tikal.low_level.brands.lovense package
      • Submodules
      • tikal.low_level.brands.lovense.data module
      • tikal.low_level.brands.lovense.handler module
      • tikal.low_level.brands.lovense.toy module
      • Module contents
  • Module contents

Submodules

tikal.low_level.brand_handler module

Part of the Low-Level API: the abstract contract every BLE toy brand implements.

A brand handler encapsulates everything brand-specific about BLE discovery and connection: - Recognizing the brand from a BLE advertisement - Creating the appropriate ToyData - Establishing a connection and returning a ready-to-use Toy instance

To add a new brand, create a subpackage under tikal/low_level/brands/ with a BLEBrandHandler subclass and register it in tikal/low_level/brands/__init__.py. The BLEConnectionBuilder instantiates every registered handler via the standard constructor defined here, so a new handler only needs to implement the four abstract methods.

class tikal.low_level.brand_handler.BLEBrandHandler(on_disconnect: ~typing.Callable[[str], ~typing.Any], on_power_off: ~typing.Callable[[str], ~typing.Any], logger_name: str = 'tikal', client_class: ~typing.Type[~bleak.BleakClient] = <class 'bleak.BleakClient'>)

Bases: ABC

Abstract interface for handling a specific brand of BLE toy.

Each concrete implementation provides the logic for: - Recognizing the brand from BLE advertisements - Creating the appropriate ToyData - Establishing a connection and returning a ready-to-use Toy instance

Subclasses are instantiated by the BLEConnectionBuilder through the constructor below. A subclass only needs to implement the four abstract methods.

Parameters:

• on_disconnect – Callback called when a toy disconnects unexpectedly. Receives the toy’s toy_id.

• on_power_off – Callback called when the user powers off a toy via its physical button. Receives the toy id (address).

• logger_name – Name for the logger used by this handler. Defaults to ‘tikal’.

• client_class – BLE client class to use. Defaults to BleakClient. Can be overridden for testing.

abstractmethod async create_toy(toy_data: ToyData, device: BLEDevice) → Toy

Connect to the toy described by toy_data using the provided BLEDevice.

This method handles brand-specific connection steps (UUID resolution, notification setup, etc.) and returns a connected Toy instance.

Raises:

• ValidationError – The toy_data contains an invalid model name.

• ConnectionError – The BLE connection, UUID resolution, or notification setup failed.

abstractmethod static create_toy_data(device: BLEDevice) → ToyData

Create a brand-specific ToyData object from a BLE device.

abstractmethod static handles_device(device: BLEDevice) → bool

Return True if the BLE device belongs to this brand.

abstractmethod static handles_toy(toy_data: ToyData) → bool

Return True if the given ToyData belongs to this brand.

tikal.low_level.connection_builder module

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

This module provides: - ConnectionBuilder: The transport-agnostic entry point. Composes every per-transport connection builder

(BLEConnectionBuilder and the fictional MockConnectionBuilder) and presents a single, unified discovery/connection API. This is the class the higher layers (and new code) should use.

• BLEConnectionBuilder: Entry point for all BLE toys. Manages BLE scanning (one‑shot and continuous) and creation of Toys.

Brand-specific logic lives behind BLEBrandHandler (brand_handler.py); concrete handlers and their registration live under tikal/low_level/brands/. The builder gets its handlers from that registry via build_handlers.

Adding a brand that speaks a new transport (e.g., USB) only requires writing the transport’s own connection builder and adding it to ConnectionBuilder’s internal list; every consumer keeps using ConnectionBuilder unchanged.

Example:

def handle_disconnect(transport: Transport):
    print(f"Toy at {transport.toy_id} disconnected unexpectedly")

def handle_power_off(address: str):
    print(f"Toy at {address} was powered off")

builder = ConnectionBuilder(
    on_disconnect=handle_disconnect,
    on_power_off=handle_power_off,
    logger_name="tikal"
)

class tikal.low_level.connection_builder.BLEConnectionBuilder(on_disconnect: ~typing.Callable[[str], ~typing.Any], on_power_off: ~typing.Callable[[str], ~typing.Any], logger_name: str, scanner_class: ~typing.Type[~bleak.BleakScanner] = <class 'bleak.BleakScanner'>, client_class: ~typing.Type[~bleak.BleakClient] = <class 'bleak.BleakClient'>)

Bases: object

async create_toy(to_connect: ToyData) → Toy | BaseException

Create a Toy instance from discovery data.

Parameters:

to_connect – ToyData object with a valid model_name. For Lovense valid model names are in LOVENSE_TOY_NAMES.keys() of module toy_data Instances of ToyData are created with a prior call to discover_toys() and the model_name must be set by you.

Returns:

• KeyError: toy address was not found in the cache (i.e., discover_toys() was not called first)

• StaleDeviceError: Subclass of ConnectionError: Device was discovered, but has since become stale. Retrieve a new snapshot i.e., via discover_toys()

• ConnectionError: BLE connection or notification setup failed, e.g., the toy may have become unavailable

• 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

• RuntimeError: Developer error. I did not specify a handler for this subclass of ToyData. Should never happen.

Return type:

connected Toy instance on success, or a BaseException on failure. Possible exceptions

Example::

toys = await builder.discover_toys(5.0) # Discover toys # Set model names (e.g., from user input) toys[0].model_name = “Nora” result = await builder.create_toys(toys[0]) # Connect print(f”Connected: {isinstance(result, Toy)}”)

async create_toys(to_connect: list[ToyData]) → list[Toy | BaseException]

Create Toy instances from discovery data.

Parameters:

to_connect – List of ToyData objects with valid model_names. For Lovense valid model names are in LOVENSE_TOY_NAMES.keys() of module toy_data Instances of ToyData are created by discover_toys() or retrieve_continuous() and model_names must be set by you.

Returns:

• KeyError: toy address was not found in the cache (i.e., discover_toys() was not called first)

• StaleDeviceError: Subclass of ConnectionError: Device was discovered, but has since become stale. Retrieve a new snapshot i.e., via discover_toys()

• ConnectionError: BLE connection or notification setup failed, e.g., the toy may have become unavailable

• 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

• RuntimeError: Developer error. I did not specify a handler for this subclass of ToyData. Should never happen.

The order of results matches the order of the input list.

Return type:

List where each element is either a connected Toy instance or a BaseException for failed connections. Possible exceptions per element

Example::

toys = await builder.discover_toys(5.0) # Discover toys # Set model names (e.g., from user input) toys[0].model_name = “Nora” toys[1].model_name = “Lush” results = await builder.create_toys(toys) # Connect # Process results connected_toys = [r for r in results if isinstance(r, Toy)] failed = [r for r in results if isinstance(r, BaseException)]

async discover_toys(timeout: float = 10.0) → list[ToyData]

Scan for all available BLE toys.

This method caches discovered BLE devices internally. You should call this method before calling create_toys().

Parameters:

timeout – Scan time in seconds. Longer timeouts increase the chance of finding all nearby devices. Default is 10 seconds.

Raises:

• Exception – Any exception from BleakScanner.discover(), such as permission errors or Bluetooth adapter issues

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

Returns:

List of ToyData objects with brand‑appropriate types. This is a snapshot, not a continuous stream of updates.

Examples:

toys = await builder.discover_toys(timeout=5.0)
print(f"Found {len(toys)} Lovense devices")
for toy in toys:
    print(f"{toy.name} at {toy.toy_id}")

handles_toy(toy_data: ToyData) → bool

Return True if one of this builder’s brand handlers recognizes toy_data.

Used by ConnectionBuilder to route a ToyData to the builder that can connect it.

async retrieve_continuous() → list[ToyData]

Retrieve the current snapshot of discovered Toys. Needs a continuous scan to be running (else always return the empty list) Use start_continuous() to start the continuous scan.

If an error occurred during continuous scanning and is still relevant (=no call to stop_continuous() was made afterward), the first call to this method raises that exception. Subsequent calls return an empty list. Note that such exceptions stop the continuous scan.

Raises:

Any exception that occurred during continuous scanning as described above. –

Returns:

List of ToyData objects with brand‑appropriate types. The list is empty if continuous discovery is not running. This is a snapshot, not a continuous stream of updates.

async start_continuous(on_update: Callable[[list[ToyData] | Exception], Any] | None = None) → None

Start continuous background discovery of Toys.

Continuously scan for new toys. New toys are added to the internal cache and stale toys removed. Use meth: retrieve_continuous to get the current snapshot. Use meth: stop_continuous to stop the background scan. The method is idempotent. Calls to this method will be ignored if a continuous scan is already running.

Parameters:

on_update – Optional callback. Called with a snapshot of all discovered Toys whenever the internal cache changes. Called with an exception if the continuous scan encounters an error. Can be seen as a push version of retrieve_continuous(). The first call to on_update reflects the clearing of the internal cache, meaning you’ll always get an empty list first. This serves to ensure any cache you might have made is cleaned too.

Raises:

Exception – Any exception raised by the BLE scanner (e.g., permission errors).

async stop_continuous() → None

Stop the continuous background discovery.

Use :meth: start_continuous to start continuous background discover. After stopping, retrieve_continuous() will return an empty tuple. The method is idempotent. Does nothing if continuous discovery is not running.

class tikal.low_level.connection_builder.ConnectionBuilder(on_disconnect: ~typing.Callable[[str], ~typing.Any], on_power_off: ~typing.Callable[[str], ~typing.Any], logger_name: str, bluetooth_scanner: ~typing.Type[~bleak.BleakScanner] = <class 'bleak.BleakScanner'>, bluetooth_client: ~typing.Type[~bleak.BleakClient] = <class 'bleak.BleakClient'>, mock_toys: bool = False)

Bases: object

Transport-agnostic entry point of the Low-Level API.

A ConnectionBuilder composes one connection builder per supported transport (BLEConnectionBuilder for real BLE toys and MockConnectionBuilder for the fictional MockEstimToys brand) and exposes the exact same discovery/connection API as a single builder. Discovery results from every transport are merged into one list, and connection requests are routed to the builder that owns the relevant ToyData.

Adding a brand on a new transport only means: write that transport’s connection builder and append it to self._builders below. Every consumer of ConnectionBuilder keeps working unchanged.

Parameters:

• on_disconnect – Callback invoked when a toy disconnects unexpectedly. Receives the toy’s toy_id. Not called for intentional disconnects.

• on_power_off – Callback invoked when the user powers off a toy via the physical power button. Receives the toy id.

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

• 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.

async create_toy(to_connect: ToyData) → Toy | BaseException

Create a single Toy instance from discovery data, routing it to the transport that owns it.

See BLEConnectionBuilder.create_toy() for arguments and the result/exception types.

Returns:

A connected Toy on success, or a BaseException on failure. Returns RuntimeError if no transport recognizes the given ToyData (developer error; should never happen).

async create_toys(to_connect: list[ToyData]) → list[Toy | BaseException]

Create Toy instances from discovery data, routing each to the transport that owns it.

See BLEConnectionBuilder.create_toys() for arguments and the per-element result/exception types.

Returns:

List where each element is a connected Toy or a BaseException. Order matches the input list.

async discover_toys(timeout: float = 10.0) → list[ToyData]

Scan every transport for available toys and return the combined result.

Delegates to each underlying connection builder and concatenates their results. See BLEConnectionBuilder.discover_toys() for the per-transport behavior, arguments, and raised exceptions.

Parameters:

timeout – Scan time in seconds, passed to every transport.

Returns:

Combined list of ToyData across all transports. A snapshot, not a continuous stream.

async retrieve_continuous() → list[ToyData]

Retrieve the current merged snapshot of discovered Toys from every transport.

Needs a continuous scan to be running (see start_continuous()); otherwise returns the empty list. If a transport has a pending continuous-scan exception, the first call re-raises it (see BLEConnectionBuilder.retrieve_continuous()).

Returns:

Combined list of ToyData across all transports.

async start_continuous(on_update: Callable[[list[ToyData] | Exception], Any] | None = None) → None

Start continuous background discovery across every transport.

Each transport scans independently; their snapshots are merged so on_update always receives a single list spanning all transports. See BLEConnectionBuilder.start_continuous() for the general contract.

Parameters:

on_update – Optional callback. Called with a merged snapshot of all discovered Toys whenever any transport’s view changes, or with an exception if a transport’s scan errors.

async stop_continuous() → None

Stop continuous background discovery on every transport.

Idempotent. Does nothing for transports that are not currently scanning.

exception tikal.low_level.connection_builder.StaleDeviceError

Bases: ConnectionError

Raised when attempting to connect to a device that was at one point discovered but is no longer available.

tikal.low_level.toy module

Part of the Low Level API: the abstract Toy interface.

This module defines the brand-agnostic base class for communicating with toy devices: - Toy: Abstract base class defining the toy communication interface - UnexpectedToyResponse: raised when a toy returns an unexpected reply

Concrete, brand-specific implementations live in tikal/low_level/brands/ (e.g. tikal.low_level.brands.lovense.LovenseToy). You are not meant to instantiate these classes directly. ConnectionBuilder establishes connections to toys and returns instances of Toy

Example:

# After connecting via BLEConnectionBuilder
toy = connected_toys[0]

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

# Check battery
battery = await toy.get_battery_level()
print(f"Battery: {battery}%")

# Disconnect when done
await toy.disconnect()

class tikal.low_level.toy.Toy(transport: Transport, model_name: str, logger_name: str)

Bases: ABC

Abstract base class representing a low-level toy.

Responsible for handling the communication with physical toys. Provides functions for sending commands to the toy and handles its responses. Each toy brand implements this interface with brand-specific protocol details. You are not meant to instantiate these classes directly. ConnectionBuilder establishes connections to toys and returns instances of Toy

Parameters:

• transport – Connected Transport instance.

• model_name – Model name of the toy (e.g., “Gush”, “Nora”).

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

abstract property brand: str

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

abstractmethod async change_rotation_direction() → bool

Change rotation direction.

For toys with rotation capability (e.g., Nora, Ridge), this toggles the rotation direction. Returns True and does nothing if the toy does not support rotation.

Returns:

True if the toy acknowledged the command or if rotation is not supported, False if the command failed.

Example:

# Reverse rotation direction
await toy.rotate_change_direction()

abstract 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_rotation_direction_available:
    await toy.change_rotation_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}”)”)

abstractmethod async direct_command(command: str, timeout: float = 3.0) → str | None

Send any command directly to the toy.

This method allows sending commands that are not implemented by the library.

Parameters:

• command – Command string in UTF-8 format

• timeout – Response timeout in seconds. Defaults to 3.0.

Returns:

Response string from the toy, or None if timeout or error occurred.

abstractmethod async disconnect() → None

Disconnect from the device.

Stops all toy actions, disables notifications, and closes the BLE connection. This method should always be called before the toy object is destroyed to ensure proper cleanup. The method does not raise any exceptions.

abstractmethod async get_battery_level() → int | None

Retrieve the battery level of the connected device.

Returns:

Battery level as a percentage (0-100), or None if an error occurred, the command timed out, or the toy has no battery.

Example:

battery = await toy.get_battery_level()
if battery is not None:
    print(f"Battery: {battery}%")
else:
    print("Failed to read battery level")

abstractmethod async intensity1(level: int) → bool

Set the primary capability of the toy to a specified level.

The primary capability varies by toy model (e.g., vibration for Gush, thrusting for Solace).

Parameters:

level – Intensity level. The valid range is 0 - self.max_intensity. Values outside this range are clamped.

Returns:

True if the toy acknowledged the command, False otherwise.

Example:

# Set primary capability to medium intensity
success = await toy.intensity1(10)
if not success:
    print("Command failed or timed out")

abstractmethod async intensity2(level: int) → bool

Set the secondary capability of the toy to a specified level.

The secondary capability varies by toy model (e.g., Depth for Solace, air pump for Max). Not all toys have a secondary capability. Returns true and does nothing if the toy has no secondary capability.

Parameters:

level – Intensity level. The valid range is 0 - self.max_intensity. Values outside this range are clamped.

Returns:

True if the toy acknowledged the command or the toy does not have a secondary capability, False otherwise.

Example:

# Set secondary capability to low intensity
await toy.intensity2(5)

abstract 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_connected: bool

Check if the toy is currently connected.

Returns:

True if connected, False otherwise.

abstract 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.intensity_max_value
await toy.intensity1(max_val)  # Set to maximum

property model_name: str

Returns the model name of the toy (e.g., “Nora”, “Lush”).

property name: str

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

abstract property recommended_min_interval: int

Get the recommended minimum interval between intensity changes, in milliseconds.

This is a per-model suggestion, primarily useful for pattern playback (the smallest sensible segment length). This is a recommendation, and you are free to ignore it.

Returns:

Recommended minimum interval between intensity changes in milliseconds.

Return type:

int

abstractmethod async reconnect() → bool

Attempts to reconnect to the toy to after an unintentional disconnect.

Does nothing if already connected. Use only after an unintended disconnect (ConnectionError). If you disconnected via disconnect, use the ConnectionBuilder instead.

Returns:

True if the reconnection was successful, False otherwise.

abstractmethod async set_model_name(model_name: str) → None

Set the model name of the toy.

This method validates and updates the toy’s model name. The model name determines which commands are available and how they’re interpreted.

Parameters:

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

Raises:

• 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.

abstractmethod async stop() → bool

Stop all toy actions by setting all intensities to zero.

Returns:

True if successful, False if either intensity command failed.

Example:

await toy.stop()

abstractmethod async strict_change_rotation_direction() → bool

Similar to :meth: rotate_change_direction, but this method raises an exception if the command fails. :raises UnexpectedToyResponse: The toys’ response was unexpected, e.g. “ERROR” instead of “OK”. :raises ConnectionError: Command could not be sent, or the toy did not respond within an appropriate timeout.

Returns:

True if the toy supports rotation, False otherwise.

abstractmethod async strict_direct_command(command: str, timeout: float = 3.0) → str

Similar to :meth: direct_command, but this method raises an exception if the command fails.

Raises:

ConnectionError – Command could not be sent, or the toy did not respond within the provided timeout.

Returns:

response string from the toy

abstractmethod async strict_disconnect() → None

Similar to :meth: disconnect, but this methode does raise an exception if the disconnect fails. The exception is only for logging. The toy is still disconnected in the error case.

Raises:

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

• UnexpectedToyResponse – The toys’ response was unexpected, e.g. “ERROR” instead of “OK”.

abstractmethod async strict_get_battery_level() → int | None

Similar to :meth: get_battery_level, but this method raises an exception if the command fails.

Raises:

• UnexpectedToyResponse – The toys’ response was unexpected.

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

Returns:

Battery level as a percentage (0-100), or None if the toy has no battery.

abstractmethod async strict_intensity1(level: int) → bool

Similar to :meth: intensity1, but this method raises an exception if the intensity command fails.

Raises:

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

• UnexpectedToyResponse – The toys’ response was unexpected, e.g. “ERROR” instead of “OK”.

Returns:

Always true

abstractmethod async strict_intensity2(level: int) → bool

Similar to :meth: intensity2, but this method raises an exception if the intensity command fails.

Raises:

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

• UnexpectedToyResponse – The toys’ response was unexpected, e.g. “ERROR” instead of “OK”.

Returns:

True if the toy supports a secondary capability, False otherwise.

abstractmethod async strict_reconnect() → bool

Similar to :meth: reconnect, but this methode does raise an exception if the reconnection fails.

Raises:

• ConnectionError – The reconnection failed.

• RuntimeError – The reconnection was attempted after intentionally disconnecting the toy.

Returns:

Always True.

abstractmethod async strict_stop() → bool

Similar to :meth: stop, but this method raises an exception if the stop command fails.

Raises:

• UnexpectedToyResponse – The toys’ response was unexpected, e.g. “ERROR” instead of “OK”.

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

Returns:

always True

property toy_id: str

Returns a unique identifier of the toy e.g., Bluetooth address.

exception tikal.low_level.toy.UnexpectedToyResponse

Bases: ConnectionError

Raised when the toy responds with an unexpected message.

tikal.low_level.toy_data module

Part of both the Low-Level and High-Level API: brand-agnostic data structures for toy device management.

This module defines the shared data classes used throughout the toy control system: - Exception classes for validation errors (raised if model_name is invalid) - ToyData returned by the connection builder after discovery - ToyCommands describing a toy model’s capabilities

Brand-specific model data (e.g., Lovense’s model -> command mapping in LOVENSE_TOY_NAMES) lives in the brand subpackages under tikal/low_level/brands/. The BRANDS mapping of brand name -> supported model names is built from the brand registry in tikal/low_level/brands/__init__.py.

exception tikal.low_level.toy_data.BadModelError

Bases: ValidationError

Exception raised when a model_name is valid, but its associated commands are not accepted by the toy.

This exception can mean two things:

1. A valid, but wrong model_name is being set

2. the commands being incorrect -> the Library does not handle this model correctly. Please contact the library maintainer in this case.

Can be raised during toy initialization or when setting a toy’s model name.

exception tikal.low_level.toy_data.InvalidModelError

Bases: ValidationError

Exception raised when a model_name is invalid.

This exception is raised when attempting to set a model name that is not recognized. Can be raised during toy initialization or when setting a toy’s model name.

Example

try:
    toy.set_model_name("InvalidModel")
except InvalidModelError as e:
    print(f"Invalid model: {e}")

class tikal.low_level.toy_data.ToyCommands(intensity1_name: str, intensity1_command: str, intensity2_name: str | None = None, intensity2_command: str | None = None)

Bases: object

Command configuration for a toy model.

Defines the available capabilities for a specific toy model. Provides names for user display and commands for protocol communication. You shouldn’t need to instantiate this class, but if you use LOVENSE_TOY_NAMES you will use instances of this class.

intensity1_name

Display name for the primary capability shown to users (e.g., “Vibration”, “Thrust”).

Type:

str

intensity1_command

Command string for the primary capability sent to the toy (e.g., “Vibrate”, “Thrusting”).

Type:

str

intensity2_name

Display name for the secondary capability, or None if the toy has only one capability (e.g., “Rotation”, “Air”).

Type:

str | None

intensity2_command

Command string for the secondary capability, or None if the toy has no secondary capability (e.g., “Rotate”, “Air:Level”).

Type:

str | None

Example

# Check capabilities
commands = LOVENSE_TOY_NAMES["Nora"]
print(f"{commands.intensity1_name}: {commands.intensity1_command}")
if commands.intensity2_name:
    print(f"{commands.intensity2_name}: {commands.intensity2_command}")

intensity1_command: str

intensity1_name: str

intensity2_command: str | None = None

intensity2_name: str | None = None

class tikal.low_level.toy_data.ToyData(_name: str, _toy_id: str, _model_name: str, _brand: str)

Bases: object

Base class for toy discovery data.

Contains the information needed to identify and connect to a toy device. You shouldn’t need to instantiate this class yourself. ConnectionBuilder.discover_toys() creates instances of this class for you.

Properties:

name: Read-only human-readable identifier for the toy. For Bluetooth toys, this is the Bluetooth name (e.g., “LVS-B12”). toy_id: Read-only unique identifier for the toy. For Bluetooth toys, this is the Bluetooth address (e.g., “DC:F5:05:A3:6D:1E”). model_name: Model name of the toy (e.g., “Lush”). For Lovense toys, this is empty and must be set manually. brand: Read-only brand name of the toy (e.g., “Lovense”).

Example

# Created during discovery
print(lovense_data.name)  # "LVS-Z36D"
print(lovense_data.toy_id)  # "DC:F5:05:A3:6D:1E"
print(lovense_data.model_name)  # ""

# User selects model
lovense_data.model_name = "Nora"

property brand: str

property model_name: str

property name: str

property toy_id: str

exception tikal.low_level.toy_data.ValidationError

Bases: Exception

Exception raised when a model_name is invalid.

This exception is raised when attempting to set a model name that is not recognized. Can be raised during toy initialization or when setting a toy’s model name.

Example

try:
    toy.set_model_name("InvalidModel")
except ValidationError as e:
    print(f"Invalid model: {e}")

tikal.low_level.transport module

Part of the low-level API: Provides the transport layer for toy communication. You are not meant to instantiate any of these classes yourself. Objects of these classes are created by the ConnectionBuilder as needed.

Decouples toy protocol logic from the underlying wire technology (BLE, USB, etc.). Concrete implementations wrap bleak or pyserial-asyncio-fast.

class tikal.low_level.transport.BleTransport(device: ~bleak.backends.device.BLEDevice, uuid_resolver: ~typing.Callable[[~bleak.BleakClient], ~typing.Awaitable[tuple[str, str]]], on_disconnect: ~typing.Callable[[str], None] | None = None, client_class: ~typing.Type[~bleak.BleakClient] = <class 'bleak.BleakClient'>)

Bases: Transport

Transport implementation for Bluetooth Low Energy via bleak.

Call connect() after construction! It establishes the BLE connection and resolves TX / RX UUIDs, via the supplied uuid_resolver

Parameters:

• device – The BLEDevice to connect to.

• uuid_resolver – Async callable that receives the connected BleakClient and returns (tx_uuid, rx_uuid). Invoked once inside connect() while the link is already open, so it can inspect the GATT services.

• on_disconnect – Optional callback is invoked with this self.toy_id when the underlying BLE connection drops unexpectedly. Not invoked for intentional disconnects via disconnect().

• client_class – BleakClient subclass to use. Defaults to BleakClient; override for testing.

async connect() → None

Do not call. The ConnectionBuilder will call this for you.

Opens the BLE connection and resolves TX / RX UUIDs. Must be called exactly once after construction and before any other method.

Raises:

ConnectionError – If the BLE connection fails or if UUID resolution raises.

async disconnect() → None

Close the BLE connection and release all resources. After this call is_connected returns False.

Raises:

ConnectionError – If the operation fails. Connection is still regarded as closed if this exception is raised.

property is_connected: bool

True while the underlying link is open, else False.

async reconnect() → None

Attempts to reconnect to the toy. Does nothing if already connected. Use only after an unintended disconnect (ConnectionError). If you disconnected via disconnect(), use the ConnectionBuilder instead.

Raises:

• RuntimeError – If called after an intentional disconnect.

• ConnectionError – If the reconnection attempt fails.

async send(data: bytes) → None

Send encoded data to the toy.

Parameters:

data – Encoded data to send to the toy.

Raises:

ConnectionError – If the transport is not connected or the operation fails.

async start_notify(callback: Callable[[bytes], None]) → None

Start receiving inbound data and invoke callback for each one.

Parameters:

callback – callback invoked with each inbound bytes payload.

Raises:

ConnectionError – If the transport is not connected or the operation fails.

class tikal.low_level.transport.MockTransport(toy_id: str, name: str, battery: int = 77, on_disconnect: Callable[[str], None] | None = None)

Bases: Transport

In-memory Transport for the fictional MockEstimToys brand.

Backed by no real hardware: it simulates a device that speaks a tiny line protocol (commands and responses are UTF-8 strings terminated by ;). On every send() it computes a canned response and feeds it straight back through the notification callback, so the Toy layer’s request/response flow works exactly as it would over BLE.

Used purely to explore the library’s architecture and to add a second brand/transport without a physical toy.

Parameters:

• toy_id – Unique identifier for the fake device (e.g. "Thunder_ID").

• name – Human-readable device name (e.g. "Thunder1").

• battery – Battery percentage the device reports in response to a Battery command.

• on_disconnect – Optional callback invoked with toy_id on every disconnect() (intentional or not). Used by the MockConnectionBuilder to make the device discoverable again once it is no longer connected.

async disconnect() → None

Close the simulated link. After this call is_connected returns False.

property is_connected: bool

True while the underlying link is open, else False.

async reconnect() → None

Re-open the simulated link. Does nothing if already connected.

Raises:

RuntimeError – If called after an intentional disconnect().

async send(data: bytes) → None

Accept a command and immediately feed the simulated response back through the notification callback.

Raises:

ConnectionError – If the transport is not connected.

async start_notify(callback: Callable[[bytes], None]) → None

Register the callback that receives simulated responses.

Raises:

ConnectionError – If the transport is not connected.

class tikal.low_level.transport.Transport(toy_id: str, name: str)

Bases: ABC

Abstract transport layer for a connected toy.

Wraps the raw I/O operations so that Toy subclasses can send commands and receive notifications without knowing whether the underlying link is BLE, USB, or anything else.

A Transport is always created in a connected state. Connection setup is the responsibility of the ConnectionBuilder.

Parameters:

• toy_id – Unique identifier e.g., BLE: Bluetooth address string, USB: serial port path.

• name – Human-readable device name e.g., BLE: advertised name. USB: the literal string "usb-<serial port path>".

abstractmethod async disconnect() → None

Close the underlying connection and release all resources. After this call is_connected must return False.

Raises:

ConnectionError – If the operation fails.

abstract property is_connected: bool

True while the underlying link is open, else False.

property name: str

Returns a Human-readable device name.

abstractmethod async reconnect() → None

Attempts to reconnect to the toy. Does nothing if already connected.

abstractmethod async send(data: bytes) → None

Write raw bytes to the device.

Parameters:

data – Bytes to send (the Toy layer is responsible for framing).

Raises:

ConnectionError – If the transport is not connected or the operation fails.

abstractmethod async start_notify(callback: Callable[[bytes], None]) → None

Start receiving inbound data and invoke callback for each one.

Parameters:

callback – callback invoked with each inbound bytes payload.

Raises:

ConnectionError – If the transport is not connected or the operation fails.

property toy_id: str

returns a unique identifier (BLE address or USB port).

class tikal.low_level.transport.UsbTransport(port: str, baudrate: int, reader: StreamReader, writer: StreamWriter)

Bases: Transport

Transport implementation for USB serial via pyserial-asyncio-fast.

Parameters:

• port – Serial port path, e.g. "/dev/ttyUSB0" or "COM3".

• baudrate – Baud rate for the serial connection.

• reader – asyncio.StreamReader from serial_asyncio_fast.

• writer – asyncio.StreamWriter from serial_asyncio_fast.

async classmethod connect(port: str, baudrate: int) → UsbTransport

Open the serial port and return a connected UsbTransport.

async disconnect() → None

Close the serial connection and release all resources. After this call is_connected returns False.

Raises:

ConnectionError – If the operation fails. The connection is still regarded as closed if this exception is raised.

property is_connected: bool

True while the underlying link is open, else False.

async reconnect() → None

Attempts to reconnect to the toy. Does nothing if already connected. Use only after an unintended disconnect (ConnectionError). If you disconnected via disconnect(), use the ConnectionBuilder instead.

Raises:

• RuntimeError – If called after an intentional disconnect.

• ConnectionError – If the reconnection attempt fails.

async send(data: bytes) → None

Write raw bytes to the device.

Parameters:

data – Bytes to send (the Toy layer is responsible for framing).

Raises:

ConnectionError – If the transport is not connected or the operation fails.

async start_notify(callback: Callable[[bytes], None]) → None

Spawns a background task that reads lines from the serial port and invokes callback for each one.

Parameters:

callback – callback invoked with each inbound bytes payload.

Raises:

ConnectionError – If the transport is not connected or the operation fails.

Module contents

class tikal.low_level.BLEBrandHandler(on_disconnect: ~typing.Callable[[str], ~typing.Any], on_power_off: ~typing.Callable[[str], ~typing.Any], logger_name: str = 'tikal', client_class: ~typing.Type[~bleak.BleakClient] = <class 'bleak.BleakClient'>)

Bases: ABC

Abstract interface for handling a specific brand of BLE toy.

Each concrete implementation provides the logic for: - Recognizing the brand from BLE advertisements - Creating the appropriate ToyData - Establishing a connection and returning a ready-to-use Toy instance

Subclasses are instantiated by the BLEConnectionBuilder through the constructor below. A subclass only needs to implement the four abstract methods.

Parameters:

• on_disconnect – Callback called when a toy disconnects unexpectedly. Receives the toy’s toy_id.

• on_power_off – Callback called when the user powers off a toy via its physical button. Receives the toy id (address).

• logger_name – Name for the logger used by this handler. Defaults to ‘tikal’.

• client_class – BLE client class to use. Defaults to BleakClient. Can be overridden for testing.

abstractmethod async create_toy(toy_data: ToyData, device: BLEDevice) → Toy

Connect to the toy described by toy_data using the provided BLEDevice.

This method handles brand-specific connection steps (UUID resolution, notification setup, etc.) and returns a connected Toy instance.

Raises:

• ValidationError – The toy_data contains an invalid model name.

• ConnectionError – The BLE connection, UUID resolution, or notification setup failed.

abstractmethod static create_toy_data(device: BLEDevice) → ToyData

Create a brand-specific ToyData object from a BLE device.

abstractmethod static handles_device(device: BLEDevice) → bool

Return True if the BLE device belongs to this brand.

abstractmethod static handles_toy(toy_data: ToyData) → bool

Return True if the given ToyData belongs to this brand.

class tikal.low_level.BLEConnectionBuilder(on_disconnect: ~typing.Callable[[str], ~typing.Any], on_power_off: ~typing.Callable[[str], ~typing.Any], logger_name: str, scanner_class: ~typing.Type[~bleak.BleakScanner] = <class 'bleak.BleakScanner'>, client_class: ~typing.Type[~bleak.BleakClient] = <class 'bleak.BleakClient'>)

Bases: object

async create_toy(to_connect: ToyData) → Toy | BaseException

Create a Toy instance from discovery data.

Parameters:

to_connect – ToyData object with a valid model_name. For Lovense valid model names are in LOVENSE_TOY_NAMES.keys() of module toy_data Instances of ToyData are created with a prior call to discover_toys() and the model_name must be set by you.

Returns:

• KeyError: toy address was not found in the cache (i.e., discover_toys() was not called first)

• StaleDeviceError: Subclass of ConnectionError: Device was discovered, but has since become stale. Retrieve a new snapshot i.e., via discover_toys()

• ConnectionError: BLE connection or notification setup failed, e.g., the toy may have become unavailable

• 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

• RuntimeError: Developer error. I did not specify a handler for this subclass of ToyData. Should never happen.

Return type:

connected Toy instance on success, or a BaseException on failure. Possible exceptions

Example::

toys = await builder.discover_toys(5.0) # Discover toys # Set model names (e.g., from user input) toys[0].model_name = “Nora” result = await builder.create_toys(toys[0]) # Connect print(f”Connected: {isinstance(result, Toy)}”)

async create_toys(to_connect: list[ToyData]) → list[Toy | BaseException]

Create Toy instances from discovery data.

Parameters:

to_connect – List of ToyData objects with valid model_names. For Lovense valid model names are in LOVENSE_TOY_NAMES.keys() of module toy_data Instances of ToyData are created by discover_toys() or retrieve_continuous() and model_names must be set by you.

Returns:

• KeyError: toy address was not found in the cache (i.e., discover_toys() was not called first)

• StaleDeviceError: Subclass of ConnectionError: Device was discovered, but has since become stale. Retrieve a new snapshot i.e., via discover_toys()

• ConnectionError: BLE connection or notification setup failed, e.g., the toy may have become unavailable

• 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

• RuntimeError: Developer error. I did not specify a handler for this subclass of ToyData. Should never happen.

The order of results matches the order of the input list.

Return type:

List where each element is either a connected Toy instance or a BaseException for failed connections. Possible exceptions per element

Example::

toys = await builder.discover_toys(5.0) # Discover toys # Set model names (e.g., from user input) toys[0].model_name = “Nora” toys[1].model_name = “Lush” results = await builder.create_toys(toys) # Connect # Process results connected_toys = [r for r in results if isinstance(r, Toy)] failed = [r for r in results if isinstance(r, BaseException)]

async discover_toys(timeout: float = 10.0) → list[ToyData]

Scan for all available BLE toys.

This method caches discovered BLE devices internally. You should call this method before calling create_toys().

Parameters:

timeout – Scan time in seconds. Longer timeouts increase the chance of finding all nearby devices. Default is 10 seconds.

Raises:

• Exception – Any exception from BleakScanner.discover(), such as permission errors or Bluetooth adapter issues

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

Returns:

List of ToyData objects with brand‑appropriate types. This is a snapshot, not a continuous stream of updates.

Examples:

toys = await builder.discover_toys(timeout=5.0)
print(f"Found {len(toys)} Lovense devices")
for toy in toys:
    print(f"{toy.name} at {toy.toy_id}")

handles_toy(toy_data: ToyData) → bool

Return True if one of this builder’s brand handlers recognizes toy_data.

Used by ConnectionBuilder to route a ToyData to the builder that can connect it.

async retrieve_continuous() → list[ToyData]

Retrieve the current snapshot of discovered Toys. Needs a continuous scan to be running (else always return the empty list) Use start_continuous() to start the continuous scan.

If an error occurred during continuous scanning and is still relevant (=no call to stop_continuous() was made afterward), the first call to this method raises that exception. Subsequent calls return an empty list. Note that such exceptions stop the continuous scan.

Raises:

Any exception that occurred during continuous scanning as described above. –

Returns:

List of ToyData objects with brand‑appropriate types. The list is empty if continuous discovery is not running. This is a snapshot, not a continuous stream of updates.

async start_continuous(on_update: Callable[[list[ToyData] | Exception], Any] | None = None) → None

Start continuous background discovery of Toys.

Continuously scan for new toys. New toys are added to the internal cache and stale toys removed. Use meth: retrieve_continuous to get the current snapshot. Use meth: stop_continuous to stop the background scan. The method is idempotent. Calls to this method will be ignored if a continuous scan is already running.

Parameters:

on_update – Optional callback. Called with a snapshot of all discovered Toys whenever the internal cache changes. Called with an exception if the continuous scan encounters an error. Can be seen as a push version of retrieve_continuous(). The first call to on_update reflects the clearing of the internal cache, meaning you’ll always get an empty list first. This serves to ensure any cache you might have made is cleaned too.

Raises:

Exception – Any exception raised by the BLE scanner (e.g., permission errors).

async stop_continuous() → None

Stop the continuous background discovery.

Use :meth: start_continuous to start continuous background discover. After stopping, retrieve_continuous() will return an empty tuple. The method is idempotent. Does nothing if continuous discovery is not running.

exception tikal.low_level.BadModelError

Bases: ValidationError

Exception raised when a model_name is valid, but its associated commands are not accepted by the toy.

This exception can mean two things:

1. A valid, but wrong model_name is being set

2. the commands being incorrect -> the Library does not handle this model correctly. Please contact the library maintainer in this case.

Can be raised during toy initialization or when setting a toy’s model name.

class tikal.low_level.BleTransport(device: ~bleak.backends.device.BLEDevice, uuid_resolver: ~typing.Callable[[~bleak.BleakClient], ~typing.Awaitable[tuple[str, str]]], on_disconnect: ~typing.Callable[[str], None] | None = None, client_class: ~typing.Type[~bleak.BleakClient] = <class 'bleak.BleakClient'>)

Bases: Transport

Transport implementation for Bluetooth Low Energy via bleak.

Call connect() after construction! It establishes the BLE connection and resolves TX / RX UUIDs, via the supplied uuid_resolver

Parameters:

• device – The BLEDevice to connect to.

• uuid_resolver – Async callable that receives the connected BleakClient and returns (tx_uuid, rx_uuid). Invoked once inside connect() while the link is already open, so it can inspect the GATT services.

• on_disconnect – Optional callback is invoked with this self.toy_id when the underlying BLE connection drops unexpectedly. Not invoked for intentional disconnects via disconnect().

• client_class – BleakClient subclass to use. Defaults to BleakClient; override for testing.

async connect() → None

Do not call. The ConnectionBuilder will call this for you.

Opens the BLE connection and resolves TX / RX UUIDs. Must be called exactly once after construction and before any other method.

Raises:

ConnectionError – If the BLE connection fails or if UUID resolution raises.

async disconnect() → None

Close the BLE connection and release all resources. After this call is_connected returns False.

Raises:

ConnectionError – If the operation fails. Connection is still regarded as closed if this exception is raised.

property is_connected: bool

True while the underlying link is open, else False.

async reconnect() → None

Attempts to reconnect to the toy. Does nothing if already connected. Use only after an unintended disconnect (ConnectionError). If you disconnected via disconnect(), use the ConnectionBuilder instead.

Raises:

• RuntimeError – If called after an intentional disconnect.

• ConnectionError – If the reconnection attempt fails.

async send(data: bytes) → None

Send encoded data to the toy.

Parameters:

data – Encoded data to send to the toy.

Raises:

ConnectionError – If the transport is not connected or the operation fails.

async start_notify(callback: Callable[[bytes], None]) → None

Start receiving inbound data and invoke callback for each one.

Parameters:

callback – callback invoked with each inbound bytes payload.

Raises:

ConnectionError – If the transport is not connected or the operation fails.

class tikal.low_level.ConnectionBuilder(on_disconnect: ~typing.Callable[[str], ~typing.Any], on_power_off: ~typing.Callable[[str], ~typing.Any], logger_name: str, bluetooth_scanner: ~typing.Type[~bleak.BleakScanner] = <class 'bleak.BleakScanner'>, bluetooth_client: ~typing.Type[~bleak.BleakClient] = <class 'bleak.BleakClient'>, mock_toys: bool = False)

Bases: object

Transport-agnostic entry point of the Low-Level API.

A ConnectionBuilder composes one connection builder per supported transport (BLEConnectionBuilder for real BLE toys and MockConnectionBuilder for the fictional MockEstimToys brand) and exposes the exact same discovery/connection API as a single builder. Discovery results from every transport are merged into one list, and connection requests are routed to the builder that owns the relevant ToyData.

Adding a brand on a new transport only means: write that transport’s connection builder and append it to self._builders below. Every consumer of ConnectionBuilder keeps working unchanged.

Parameters:

• on_disconnect – Callback invoked when a toy disconnects unexpectedly. Receives the toy’s toy_id. Not called for intentional disconnects.

• on_power_off – Callback invoked when the user powers off a toy via the physical power button. Receives the toy id.

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

• 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.

async create_toy(to_connect: ToyData) → Toy | BaseException

Create a single Toy instance from discovery data, routing it to the transport that owns it.

See BLEConnectionBuilder.create_toy() for arguments and the result/exception types.

Returns:

A connected Toy on success, or a BaseException on failure. Returns RuntimeError if no transport recognizes the given ToyData (developer error; should never happen).

async create_toys(to_connect: list[ToyData]) → list[Toy | BaseException]

Create Toy instances from discovery data, routing each to the transport that owns it.

See BLEConnectionBuilder.create_toys() for arguments and the per-element result/exception types.

Returns:

List where each element is a connected Toy or a BaseException. Order matches the input list.

async discover_toys(timeout: float = 10.0) → list[ToyData]

Scan every transport for available toys and return the combined result.

Delegates to each underlying connection builder and concatenates their results. See BLEConnectionBuilder.discover_toys() for the per-transport behavior, arguments, and raised exceptions.

Parameters:

timeout – Scan time in seconds, passed to every transport.

Returns:

Combined list of ToyData across all transports. A snapshot, not a continuous stream.

async retrieve_continuous() → list[ToyData]

Retrieve the current merged snapshot of discovered Toys from every transport.

Needs a continuous scan to be running (see start_continuous()); otherwise returns the empty list. If a transport has a pending continuous-scan exception, the first call re-raises it (see BLEConnectionBuilder.retrieve_continuous()).

Returns:

Combined list of ToyData across all transports.

async start_continuous(on_update: Callable[[list[ToyData] | Exception], Any] | None = None) → None

Start continuous background discovery across every transport.

Each transport scans independently; their snapshots are merged so on_update always receives a single list spanning all transports. See BLEConnectionBuilder.start_continuous() for the general contract.

Parameters:

on_update – Optional callback. Called with a merged snapshot of all discovered Toys whenever any transport’s view changes, or with an exception if a transport’s scan errors.

async stop_continuous() → None

Stop continuous background discovery on every transport.

Idempotent. Does nothing for transports that are not currently scanning.

exception tikal.low_level.InvalidModelError

Bases: ValidationError

Exception raised when a model_name is invalid.

This exception is raised when attempting to set a model name that is not recognized. Can be raised during toy initialization or when setting a toy’s model name.

Example

try:
    toy.set_model_name("InvalidModel")
except InvalidModelError as e:
    print(f"Invalid model: {e}")

tikal.low_level.Lovense

Backwards-compatible alias for LovenseToy (the class was formerly named Lovense).

class tikal.low_level.LovenseToy(transport: BleTransport, model_name: str, on_power_off: Callable[[str], Any], logger_name: str)

Bases: Toy

Low-level representation of a Lovense BLE toy.

Implements the Lovense-specific protocol for communication with Lovense toys over Bluetooth Low Energy. Handles command formatting, response parsing, and Lovense-specific notifications (like power-off events). You are not meant to instantiate these classes directly. BLEConnectionBuilder establishes connections to toys and returns instances of LovenseToy

Parameters:

• transport – Transport layer handling the BLE communication.

• model_name – Model name (e.g., “Nora”, “Lush”). Must be a key in LOVENSE_TOY_NAMES.

• on_power_off – Callback invoked when the user powers off the toy via the physical power button. Receives the toy’s Bluetooth address as a string argument.

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

Example:

# Set intensity of the primary capability to maximum
await toy.intensity1(20)

# Send a custom command
response = await toy.direct_command("DeviceType")
print(f"Device info: {response}")

property brand: str

Returns the brand of the toy. Always “Lovense” for this class

async change_rotation_direction() → bool

Change rotation direction for toys with rotation capability.

This method only affects toys that support rotation (Nora and Ridge). For other toys, it returns True immediately without sending any command.

Returns:

True if the toy acknowledged the command or if rotation is not supported, False if the command failed.

Example:

await toy.rotate_change_direction()  # Toggle rotation direction

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_rotation_direction_available:
    await toy.change_rotation_direction()

async direct_command(command: str, timeout: float = 3.0) → str | None

Send any command directly to the toy.

This method allows sending commands that are not implemented by the library.

Parameters:

• command – Command string in UTF-8 format. A semicolon terminator will be added if not present.

• timeout – Response timeout in seconds. Defaults to 3.0.

Returns:

Response string from the toy (with semicolon stripped), or None if timeout or error occurred.

Example:

# Query firmware version
response = await toy.direct_command("DeviceType")
# Response format: "C:11:0082059AD3BD"
# (C = device type, 11 = firmware version, address)

async disconnect() → None

Disconnect from the device.

Stops all toy actions, disables notifications, and closes the BLE connection. This is regarded as intentional disconnect and does not trigger an on_disconnect callback even if an on_disconnect callback was set (either during initialization or via set_on_disconnect). The method does not raise any exceptions.

Note

After calling this method, the toy object is unusable. To connect again, you will need to re-scan and connect using the ConnectionBuilder. Use the newly provided Lovense object by the ConnectionBuilder.

async get_batch_number() → str | None

Retrieve the production batch number.

The batch number appears to be in YYMMDD format, indicating the manufacturing date.

Returns:

Batch number string (e.g., “241015” for October 15, 2024), or None if an error occurred.

Example:

batch = await toy.get_batch_number()
if batch:
    print(f"Manufactured: 20{batch[:2]}/{batch[2:4]}/{batch[4:6]}")

async get_battery_level() → int | None

Retrieve the battery level as a percentage.

Returns:

Battery level (0-100%), or None if the command failed or timed out.

Example:

battery = await toy.get_battery_level()
if battery is not None:
    if battery < 20:
        print(f"Low battery: {battery}%")
    else:
        print(f"Battery: {battery}%")

Note

Lovense toys have a quirk where they prefix ‘s’ before the value if the toy was recently reconnected. This method strips the ‘s’ when present

async get_device_type() → str | None

Retrieve device type and firmware information.

Returns:

FirmwareVersion:Address” (e.g., “C:11:0082059AD3BD”), or None if an error occurred.

Return type:

String in format “DeviceType

Example:

info = await toy.get_device_type()
if info:
    parts = info.split(":")
    device_type = parts[0]
    firmware = parts[1]
    print(f"Device type: {device_type}, Firmware: {firmware}")

async get_status() → int | None

Retrieve the status code of the toy.

Returns:

Status code (2 = Normal operation), or None if an error occurred.

Example:

status = await toy.get_status()
if status == 2:
    print("Toy is operating normally")
else
    print(f"Unusual status code: {status}")

async intensity1(level: int) → bool

Set the primary capability to the specified level.

The primary capability depends on the toy model (e.g., vibration for Gush, thrusting for Solace)

Parameters:

level – Intensity level (0-20). Values outside this range are clamped.

Returns:

True if the toy acknowledged the command, False otherwise.

Example:

await toy.intensity1(20)  # Set primary capability to maximum

async intensity2(level: int) → bool

Set the secondary capability to the specified level.

The secondary capability depends on the toy model (e.g., Rotation for Nora, depth control for Solace). Not all toys have a secondary capability. Returns true immediately and does not send a command if the toy has no secondary capability.

Parameters:

level – Intensity level (0-20). Values outside the valid range are clamped.

Returns:

True if the toy acknowledged the command or if no secondary capability exists, False otherwise

Example:

await toy.intensity2(10)  # Set secondary capability to medium intensity

Note

For Max’s air pump, the level is automatically divided by 4 to convert from 0-20 scale to 0-5 scale.

property intensity_names: tuple[str, str | None]

Get display names for Lovense toy capabilities.

Returns:

(primary_name, secondary_name). Secondary name is None if the toy has only one capability.

Return type:

tuple[str, str | None]

Example:

names = toy.intensity_names
print(f"{names[0]}: intensity1")  # "Vibration: intensity1"
if names[1]:
    print(f"{names[1]}: intensity2")  # "Rotation: intensity2"

property max_intensity: int

Get maximum intensity value for Lovense toys.

Returns:

Always 20 for Lovense toys.

Return type:

int

Note

Some capabilities (like Max’s air pump) use different ranges. Those are automatically scaled for you.

async power_off() → bool

Turn off power to the toy.

This sends the PowerOff command which turns off the toy. The toy can then only be turned back on via the physical power button.

Returns:

True if successful, False otherwise.

Example:

await toy.power_off()

property recommended_min_interval: int

Get the recommended minimum interval between intensity changes (milliseconds).

Returns:

The Lovense per-model suggested minimum segment length.

Return type:

int

async reconnect() → bool

Attempts to reconnect to the toy to after an unintentional disconnect.

Does nothing if already connected. Use only after an unintended disconnect (ConnectionError). If you disconnected via disconnect, use the ConnectionBuilder instead.

Returns:

True if the reconnection was successful, False otherwise.

async set_model_name(model_name: str) → None

Set the model name of the toy.

Parameters:

model_name – New model name. Must be in LOVENSE_TOY_NAMES.keys() of module ToyData. Case Insensitive. Case Insensitive.

Raises:

• 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.

Example:

# Update model name in case it was set incorrectly while building the connection via the ConnectionBuilder
toy.set_model_name("Nora")

async start_notifications() → None

Start listening for messages from the Lovense toy.

This method is called by the ConnectionBuilder during connection setup and should not be called manually.

Raises:

ConnectionError – Failed to start notifications.

async stop() → bool

Stop all toy actions by setting all intensities to zero.

This method is a shortcut for intensity1(0) and intensity2(0)

Returns:

True if both commands succeeded, False if either failed.

Example:

await toy.stop()

async strict_change_rotation_direction() → bool

Similar to :meth: rotate_change_direction, but this method raises an exception if the command fails.

Raises:

• UnexpectedToyResponse – The toys’ response was unexpected, e.g. “ERROR” instead of “OK”.

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

Returns:

True if the toy supports rotation, False otherwise. Does not raise if the toy does not support rotation.

async strict_direct_command(command: str, timeout: float = 3.0) → str

Similar to :meth: direct_command, but this method raises an exception if the command fails.

Raises:

ConnectionError – Command could not be sent, or the toy did not respond within the provided timeout.

Returns:

response string from the toy

async strict_disconnect() → None

Similar to :meth: disconnect, but this methode does raise an exception if the disconnect fails. The exception is only for logging. The toy is still disconnected in the error case.

Raises:

ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

async strict_get_batch_number() → str

Similar to :meth: get_batch_number, but this method raises an exception if the command fails.

Raises:

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

• UnexpectedToyResponse – The toys’ response was unexpected.

Returns:

Batch number string (e.g., “241015” for October 15, 2024)

async strict_get_battery_level() → int

Similar to :meth: get_battery_level, but this method raises an exception if the command fails.

Raises:

• UnexpectedToyResponse – The toys’ response was unexpected.

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

Returns:

Battery level as a percentage (0-100).

async strict_get_device_type() → str

Similar to :meth: get_device_type, but this method raises an exception if the command fails.

Raises:

ConnectionError – Command could not be sent, or the toy did not respond within the provided timeout.

Returns:

FirmwareVersion:Address” (e.g., “C:11:0082059AD3BD”)

Return type:

String in format “DeviceType

Note

does not raise UnexpectedToyResponse as the response is not verified.

async strict_get_status() → int

Similar to :meth: get_status, but this method raises an exception if the command fails.

Raises:

• ConnectionError – Command could not be sent, or the toy did not respond within the provided timeout.

• UnexpectedToyResponse – The toys’ response was unexpected, in this case any string containing non-digit characters.

Returns:

Status code (2 = Normal operation)

async strict_intensity1(level: int) → bool

Similar to :meth: intensity1, but this method raises an exception if the intensity command fails.

Raises:

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

• UnexpectedToyResponse – The toys’ response was unexpected, e.g. “ERROR” instead of “OK”.

Returns:

Always true

async strict_intensity2(level: int) → bool

Similar to :meth: intensity2, but this method raises an exception if the intensity command fails.

Raises:

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

• UnexpectedToyResponse – The toys’ response was unexpected, e.g. “ERROR” instead of “OK”.

Returns:

True if the toy supports a secondary capability, False otherwise.

async strict_power_off() → bool

Similar to :meth: power_off, but this method raises an exception if the command fails.

Raises:

• UnexpectedToyResponse – The toys’ response was unexpected.

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

Returns:

Always True.

async strict_reconnect() → bool

Similar to :meth: reconnect, but this methode does raise an exception if the reconnection fails.

Raises:

• ConnectionError – The reconnection failed.

• RuntimeError – The reconnection was attempted after intentionally disconnecting the toy.

Returns:

Always True.

async strict_stop() → bool

Similar to :meth: stop, but this method raises an exception if the stop command fails.

Raises:

• UnexpectedToyResponse – The toys’ response was unexpected, e.g. “ERROR” instead of “OK”.

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

Returns:

Always True.

class tikal.low_level.MockConnectionBuilder(on_disconnect: Callable[[str], Any], on_power_off: Callable[[str], Any], logger_name: str)

Bases: object

Connection builder for the fictional MockEstimToys brand.

Discovers a fixed set of fake devices and connects them via in-memory transports. Implements the same surface the composite ConnectionBuilder requires of every per-transport builder: discover_toys, start_continuous, stop_continuous, retrieve_continuous, handles_toy, and create_toy.

Parameters:

• on_disconnect – Callback invoked when a toy disconnects unexpectedly. Receives the toy id. Unused by the mock (the fake devices never drop on their own) but accepted for interface symmetry.

• on_power_off – Callback invoked when a toy reports a power-off. Receives the toy id.

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

async create_toy(to_connect: ToyData) → Toy | BaseException

Connect to a fake MockEstimToys device.

Returns:

• KeyError: the toy id is not one of the known fake devices.

• InvalidModelError: the model name is not valid for this brand.

• ConnectionError: connection or notification setup failed.

Return type:

A connected MockEstimToy on success, or a BaseException on failure

async discover_toys(timeout: float = 10.0) → list[ToyData]

Return the fixed set of fake MockEstimToys devices.

Parameters:

timeout – Ignored; there is no real scan. Present for interface compatibility.

Returns:

A fresh list of ToyData (model name pre-filled, since the fake devices self-identify).

handles_toy(toy_data: ToyData) → bool

Return True if the given ToyData belongs to the MockEstimToys brand.

async retrieve_continuous() → list[ToyData]

Return the current snapshot: the fake devices while scanning, otherwise an empty list.

async start_continuous(on_update: Callable[[list[ToyData] | Exception], Any] | None = None) → None

Start “continuous” discovery. The fake devices are reported immediately and then never change.

Parameters:

on_update – Optional callback. Called once with an empty list (to clear any cache) and then with the fake devices, mirroring the contract of the real builders.

async stop_continuous() → None

Stop “continuous” discovery. Idempotent.

class tikal.low_level.MockEstimToy(transport: MockTransport, model_name: str, on_power_off: Callable[[str], Any], logger_name: str)

Bases: Toy

Low-level representation of a fictional MockEstimToys device.

Implements the brand’s simple channel-based protocol over MockTransport. Thunder is a single-channel model; Lightning is a dual-channel model. You are not meant to instantiate this class directly.

Parameters:

• transport – Connected MockTransport instance.

• model_name – Model name. Must be a key in MOCK_ESTIM_TOY_NAMES (“Thunder” or “Lightning”).

• on_power_off – Callback invoked when the device reports a power-off. Receives the toy id.

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

property brand: str

Returns the brand of the toy. Always “MockEstimToys” for this class.

async change_rotation_direction() → bool

No-op: MockEstimToys do not support rotation. Returns True.

property change_rotation_direction_available: bool

MockEstimToys have no rotation capability.

async direct_command(command: str, timeout: float = 3.0) → str | None

Send any command directly to the device. Returns the response, or None on failure.

async disconnect() → None

Stop all actions and close the simulated connection. Does not raise.

async get_battery_level() → int | None

Retrieve the battery level as a percentage, or None on failure.

async intensity1(level: int) → bool

Set the primary channel. Returns True if the device acknowledged the command.

async intensity2(level: int) → bool

Set the secondary channel. Returns True (no-op) for single-channel models.

property intensity_names: tuple[str, str | None]

Get the display names for the device’s channels. Secondary is None for single-channel models.

property max_intensity: int

Maximum intensity value for MockEstimToys (0 - MAX_INTENSITY).

property recommended_min_interval: int

Recommended minimum interval between intensity changes (milliseconds).

async reconnect() → bool

Attempts to reconnect after an unintentional disconnect. Does nothing if already connected.

Returns:

True if the reconnection was successful, False otherwise.

async set_model_name(model_name: str) → None

Set the model name of the toy.

Parameters:

model_name – New model name. Must be in MOCK_ESTIM_TOY_NAMES (case-insensitive).

Raises:

• InvalidModelError – If model_name is not valid for this brand.

• BadModelError – If the model_name is valid, but commands still fail.

async start_notifications() → None

Start listening for messages from the device. Called by the MockConnectionBuilder during connection setup.

Raises:

ConnectionError – Failed to start notifications.

async stop() → bool

Set all channels to zero.

async strict_change_rotation_direction() → bool

No-op: MockEstimToys do not support rotation. Returns False.

async strict_direct_command(command: str, timeout: float = 3.0) → str

Like direct_command(), but raises on failure.

async strict_disconnect() → None

Like disconnect(), but raises if the disconnect fails. The toy is still disconnected.

async strict_get_battery_level() → int | None

Like get_battery_level(), but raises on failure.

async strict_intensity1(level: int) → bool

Like intensity1(), but raises on failure. Always returns True.

async strict_intensity2(level: int) → bool

Like intensity2(), but raises on failure. Returns False for single-channel models.

async strict_reconnect() → bool

Like reconnect(), but raises on failure.

async strict_stop() → bool

Like stop(), but raises on failure. Always returns True.

class tikal.low_level.MockTransport(toy_id: str, name: str, battery: int = 77, on_disconnect: Callable[[str], None] | None = None)

Bases: Transport

In-memory Transport for the fictional MockEstimToys brand.

Backed by no real hardware: it simulates a device that speaks a tiny line protocol (commands and responses are UTF-8 strings terminated by ;). On every send() it computes a canned response and feeds it straight back through the notification callback, so the Toy layer’s request/response flow works exactly as it would over BLE.

Used purely to explore the library’s architecture and to add a second brand/transport without a physical toy.

Parameters:

• toy_id – Unique identifier for the fake device (e.g. "Thunder_ID").

• name – Human-readable device name (e.g. "Thunder1").

• battery – Battery percentage the device reports in response to a Battery command.

• on_disconnect – Optional callback invoked with toy_id on every disconnect() (intentional or not). Used by the MockConnectionBuilder to make the device discoverable again once it is no longer connected.

async disconnect() → None

Close the simulated link. After this call is_connected returns False.

property is_connected: bool

True while the underlying link is open, else False.

async reconnect() → None

Re-open the simulated link. Does nothing if already connected.

Raises:

RuntimeError – If called after an intentional disconnect().

async send(data: bytes) → None

Accept a command and immediately feed the simulated response back through the notification callback.

Raises:

ConnectionError – If the transport is not connected.

async start_notify(callback: Callable[[bytes], None]) → None

Register the callback that receives simulated responses.

Raises:

ConnectionError – If the transport is not connected.

exception tikal.low_level.StaleDeviceError

Bases: ConnectionError

Raised when attempting to connect to a device that was at one point discovered but is no longer available.

class tikal.low_level.Toy(transport: Transport, model_name: str, logger_name: str)

Bases: ABC

Abstract base class representing a low-level toy.

Responsible for handling the communication with physical toys. Provides functions for sending commands to the toy and handles its responses. Each toy brand implements this interface with brand-specific protocol details. You are not meant to instantiate these classes directly. ConnectionBuilder establishes connections to toys and returns instances of Toy

Parameters:

• transport – Connected Transport instance.

• model_name – Model name of the toy (e.g., “Gush”, “Nora”).

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

abstract property brand: str

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

abstractmethod async change_rotation_direction() → bool

Change rotation direction.

For toys with rotation capability (e.g., Nora, Ridge), this toggles the rotation direction. Returns True and does nothing if the toy does not support rotation.

Returns:

True if the toy acknowledged the command or if rotation is not supported, False if the command failed.

Example:

# Reverse rotation direction
await toy.rotate_change_direction()

abstract 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_rotation_direction_available:
    await toy.change_rotation_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}”)”)

abstractmethod async direct_command(command: str, timeout: float = 3.0) → str | None

Send any command directly to the toy.

This method allows sending commands that are not implemented by the library.

Parameters:

• command – Command string in UTF-8 format

• timeout – Response timeout in seconds. Defaults to 3.0.

Returns:

Response string from the toy, or None if timeout or error occurred.

abstractmethod async disconnect() → None

Disconnect from the device.

Stops all toy actions, disables notifications, and closes the BLE connection. This method should always be called before the toy object is destroyed to ensure proper cleanup. The method does not raise any exceptions.

abstractmethod async get_battery_level() → int | None

Retrieve the battery level of the connected device.

Returns:

Battery level as a percentage (0-100), or None if an error occurred, the command timed out, or the toy has no battery.

Example:

battery = await toy.get_battery_level()
if battery is not None:
    print(f"Battery: {battery}%")
else:
    print("Failed to read battery level")

abstractmethod async intensity1(level: int) → bool

Set the primary capability of the toy to a specified level.

The primary capability varies by toy model (e.g., vibration for Gush, thrusting for Solace).

Parameters:

level – Intensity level. The valid range is 0 - self.max_intensity. Values outside this range are clamped.

Returns:

True if the toy acknowledged the command, False otherwise.

Example:

# Set primary capability to medium intensity
success = await toy.intensity1(10)
if not success:
    print("Command failed or timed out")

abstractmethod async intensity2(level: int) → bool

Set the secondary capability of the toy to a specified level.

The secondary capability varies by toy model (e.g., Depth for Solace, air pump for Max). Not all toys have a secondary capability. Returns true and does nothing if the toy has no secondary capability.

Parameters:

level – Intensity level. The valid range is 0 - self.max_intensity. Values outside this range are clamped.

Returns:

True if the toy acknowledged the command or the toy does not have a secondary capability, False otherwise.

Example:

# Set secondary capability to low intensity
await toy.intensity2(5)

abstract 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_connected: bool

Check if the toy is currently connected.

Returns:

True if connected, False otherwise.

abstract 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.intensity_max_value
await toy.intensity1(max_val)  # Set to maximum

property model_name: str

Returns the model name of the toy (e.g., “Nora”, “Lush”).

property name: str

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

abstract property recommended_min_interval: int

Get the recommended minimum interval between intensity changes, in milliseconds.

This is a per-model suggestion, primarily useful for pattern playback (the smallest sensible segment length). This is a recommendation, and you are free to ignore it.

Returns:

Recommended minimum interval between intensity changes in milliseconds.

Return type:

int

abstractmethod async reconnect() → bool

Attempts to reconnect to the toy to after an unintentional disconnect.

Does nothing if already connected. Use only after an unintended disconnect (ConnectionError). If you disconnected via disconnect, use the ConnectionBuilder instead.

Returns:

True if the reconnection was successful, False otherwise.

abstractmethod async set_model_name(model_name: str) → None

Set the model name of the toy.

This method validates and updates the toy’s model name. The model name determines which commands are available and how they’re interpreted.

Parameters:

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

Raises:

• 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.

abstractmethod async stop() → bool

Stop all toy actions by setting all intensities to zero.

Returns:

True if successful, False if either intensity command failed.

Example:

await toy.stop()

abstractmethod async strict_change_rotation_direction() → bool

Similar to :meth: rotate_change_direction, but this method raises an exception if the command fails. :raises UnexpectedToyResponse: The toys’ response was unexpected, e.g. “ERROR” instead of “OK”. :raises ConnectionError: Command could not be sent, or the toy did not respond within an appropriate timeout.

Returns:

True if the toy supports rotation, False otherwise.

abstractmethod async strict_direct_command(command: str, timeout: float = 3.0) → str

Similar to :meth: direct_command, but this method raises an exception if the command fails.

Raises:

ConnectionError – Command could not be sent, or the toy did not respond within the provided timeout.

Returns:

response string from the toy

abstractmethod async strict_disconnect() → None

Similar to :meth: disconnect, but this methode does raise an exception if the disconnect fails. The exception is only for logging. The toy is still disconnected in the error case.

Raises:

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

• UnexpectedToyResponse – The toys’ response was unexpected, e.g. “ERROR” instead of “OK”.

abstractmethod async strict_get_battery_level() → int | None

Similar to :meth: get_battery_level, but this method raises an exception if the command fails.

Raises:

• UnexpectedToyResponse – The toys’ response was unexpected.

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

Returns:

Battery level as a percentage (0-100), or None if the toy has no battery.

abstractmethod async strict_intensity1(level: int) → bool

Similar to :meth: intensity1, but this method raises an exception if the intensity command fails.

Raises:

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

• UnexpectedToyResponse – The toys’ response was unexpected, e.g. “ERROR” instead of “OK”.

Returns:

Always true

abstractmethod async strict_intensity2(level: int) → bool

Similar to :meth: intensity2, but this method raises an exception if the intensity command fails.

Raises:

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

• UnexpectedToyResponse – The toys’ response was unexpected, e.g. “ERROR” instead of “OK”.

Returns:

True if the toy supports a secondary capability, False otherwise.

abstractmethod async strict_reconnect() → bool

Similar to :meth: reconnect, but this methode does raise an exception if the reconnection fails.

Raises:

• ConnectionError – The reconnection failed.

• RuntimeError – The reconnection was attempted after intentionally disconnecting the toy.

Returns:

Always True.

abstractmethod async strict_stop() → bool

Similar to :meth: stop, but this method raises an exception if the stop command fails.

Raises:

• UnexpectedToyResponse – The toys’ response was unexpected, e.g. “ERROR” instead of “OK”.

• ConnectionError – Command could not be sent, or the toy did not respond within an appropriate timeout.

Returns:

always True

property toy_id: str

Returns a unique identifier of the toy e.g., Bluetooth address.

class tikal.low_level.ToyData(_name: str, _toy_id: str, _model_name: str, _brand: str)

Bases: object

Base class for toy discovery data.

Contains the information needed to identify and connect to a toy device. You shouldn’t need to instantiate this class yourself. ConnectionBuilder.discover_toys() creates instances of this class for you.

Properties:

name: Read-only human-readable identifier for the toy. For Bluetooth toys, this is the Bluetooth name (e.g., “LVS-B12”). toy_id: Read-only unique identifier for the toy. For Bluetooth toys, this is the Bluetooth address (e.g., “DC:F5:05:A3:6D:1E”). model_name: Model name of the toy (e.g., “Lush”). For Lovense toys, this is empty and must be set manually. brand: Read-only brand name of the toy (e.g., “Lovense”).

Example

# Created during discovery
print(lovense_data.name)  # "LVS-Z36D"
print(lovense_data.toy_id)  # "DC:F5:05:A3:6D:1E"
print(lovense_data.model_name)  # ""

# User selects model
lovense_data.model_name = "Nora"

property brand: str

property model_name: str

property name: str

property toy_id: str

class tikal.low_level.Transport(toy_id: str, name: str)

Bases: ABC

Abstract transport layer for a connected toy.

Wraps the raw I/O operations so that Toy subclasses can send commands and receive notifications without knowing whether the underlying link is BLE, USB, or anything else.

A Transport is always created in a connected state. Connection setup is the responsibility of the ConnectionBuilder.

Parameters:

• toy_id – Unique identifier e.g., BLE: Bluetooth address string, USB: serial port path.

• name – Human-readable device name e.g., BLE: advertised name. USB: the literal string "usb-<serial port path>".

abstractmethod async disconnect() → None

Close the underlying connection and release all resources. After this call is_connected must return False.

Raises:

ConnectionError – If the operation fails.

abstract property is_connected: bool

True while the underlying link is open, else False.

property name: str

Returns a Human-readable device name.

abstractmethod async reconnect() → None

Attempts to reconnect to the toy. Does nothing if already connected.

abstractmethod async send(data: bytes) → None

Write raw bytes to the device.

Parameters:

data – Bytes to send (the Toy layer is responsible for framing).

Raises:

ConnectionError – If the transport is not connected or the operation fails.

abstractmethod async start_notify(callback: Callable[[bytes], None]) → None

Start receiving inbound data and invoke callback for each one.

Parameters:

callback – callback invoked with each inbound bytes payload.

Raises:

ConnectionError – If the transport is not connected or the operation fails.

property toy_id: str

returns a unique identifier (BLE address or USB port).

exception tikal.low_level.UnexpectedToyResponse

Bases: ConnectionError

Raised when the toy responds with an unexpected message.

class tikal.low_level.UsbTransport(port: str, baudrate: int, reader: StreamReader, writer: StreamWriter)

Bases: Transport

Transport implementation for USB serial via pyserial-asyncio-fast.

Parameters:

• port – Serial port path, e.g. "/dev/ttyUSB0" or "COM3".

• baudrate – Baud rate for the serial connection.

• reader – asyncio.StreamReader from serial_asyncio_fast.

• writer – asyncio.StreamWriter from serial_asyncio_fast.

async classmethod connect(port: str, baudrate: int) → UsbTransport

Open the serial port and return a connected UsbTransport.

async disconnect() → None

Close the serial connection and release all resources. After this call is_connected returns False.

Raises:

ConnectionError – If the operation fails. The connection is still regarded as closed if this exception is raised.

property is_connected: bool

True while the underlying link is open, else False.

async reconnect() → None

Attempts to reconnect to the toy. Does nothing if already connected. Use only after an unintended disconnect (ConnectionError). If you disconnected via disconnect(), use the ConnectionBuilder instead.

Raises:

• RuntimeError – If called after an intentional disconnect.

• ConnectionError – If the reconnection attempt fails.

async send(data: bytes) → None

Write raw bytes to the device.

Parameters:

data – Bytes to send (the Toy layer is responsible for framing).

Raises:

ConnectionError – If the transport is not connected or the operation fails.

async start_notify(callback: Callable[[bytes], None]) → None

Spawns a background task that reads lines from the serial port and invokes callback for each one.

Parameters:

callback – callback invoked with each inbound bytes payload.

Raises:

ConnectionError – If the transport is not connected or the operation fails.

exception tikal.low_level.ValidationError

Bases: Exception

Exception raised when a model_name is invalid.

This exception is raised when attempting to set a model name that is not recognized. Can be raised during toy initialization or when setting a toy’s model name.

Example

try:
    toy.set_model_name("InvalidModel")
except ValidationError as e:
    print(f"Invalid model: {e}")