tikal.websocket package

Submodules

tikal.websocket.cli module

WebSocket JSON-based server that exposes _ToyHub to connected clients. Offers an alternative to the Low-Level / High-Level API defined by the tikal library. Here information is exchanged via a websocket. Significantly harder to use than the Low-Level / High-Level APIs but offers some advantages: - Process separation - Service can be used in applications written in other programming languages (assuming websockets are supported) - Multiple clients can modify the same state (experimental, untested)

This module is the entry point to the ToyServer command-line-interface.

tikal.websocket.cli.main() → None

Entry point for the ToyServer command-line interface.

Parses command-line arguments, configures logging, constructs a ToyServer instance. ToyServer shuts down automatically if no client is connected for 3 seconds.

Command-line arguments:

–host: Host to bind to (default: localhost). –port: Port to listen on (default: 8142). –toy-cache-path: Path to the toy-cache file (default: ./data/toy_cache.json). –mock-toys: Use a software mock instead of real Bluetooth hardware. –log-path: Filepath to write logs to (default: ./data/tikal_ws.log). –log-level: Logging verbosity: DEBUG, INFO, WARNING, or ERROR (default: INFO).

tikal.websocket.toy_server module

WebSocket JSON-based server that exposes _ToyHub to connected clients. Offers an alternative to the Low-Level / High-Level API defined by the tikal library. Here information is exchanged via a websocket. Significantly harder to use than the Low-Level / High-Level APIs but offers some advantages:

• Process separation

• Service can be used in applications written in other programming languages (assuming websockets are supported)

• Multiple clients can modify the same state (experimental, untested)

Protocol

Request (client -> server):

{"request": "some_command", "id": "some_id", "data": {...}}

Response (server -> client):

{"reply": "some_command", "id": "some_id", "success": true,  "data": { ... }}
{"reply": "some_command", "id": "some_id", "success": false, "data": {"error": "...", "message": "...", ...}}

Event (server -> all clients / scan subscribers):

{"event": "some_event", "success": true,  "data": {...}}
{"event": "some_event", "success": false, "data": {"error": "...", "message": "..."}}

The success field lets you branch between error handling / normal operation without having to inspect data.

Examples:

Request:

{
    "request": "get_battery",
    "id": "some_id",
    "data": {"toy_id": "some_toy_id"}
}

Response:

{
    "reply": "get_battery",
    "id": "some_id",
    "success": True,
    "data": {"battery": 85, "toy_id": "some_toy_id"}
}

Error response:

{
    "reply": "get_battery",
    "id": "some_id",
    "success": False,
    "data": {"error": "UnknownToyError", "message": "Unable to execute 'get_battery' on 'some_toy_id'. Please add the toy first."}
}

Event:

{
    "event": "on_status_change",
    "success": True,
    "data": {"toy_id": "some_toy_id", "status": "RECONNECTING"}
}

Architecture

Each command is described by a CommandEntry dataclass that bundles:

• Request_model : Pydantic model that validates the incoming data object.

• Response_model : Pydantic model that validates (and serializes) the outgoing data.

• Handler : async callable(hub: _ToyHub, data: req_model) -> dict that performs the actual work and returns the raw result dict.

_handle_message is a generic dispatcher: validate -> look up entry -> validate inner data -> call handler -> validate response -> send. The commands start_scan / stop_scan are a special case as they require a reference to the per-client WebSocket connection. They are handled by dedicated methods flagged via CommandEntry.is_scan.

class tikal.websocket.toy_server.CommandEntry(req_model: type[BaseModel], resp_model: type[BaseModel], handler: Callable, is_scan: bool = False, is_shutdown: bool = False, is_heartbeat: bool = False, is_limit: bool = False)

Bases: object

Descriptor for a single command.

Bundles everything the dispatcher needs to handle a command: A Pydantic model to validate the incoming data payload, a Pydantic model to validate and serialize the outgoing result, and the async handler that performs the actual work.

req_model

The Pydantic model used to validate (and coerce) the data field of the incoming RequestEnvelope.

Type:

type[pydantic.main.BaseModel]

resp_model

The Pydantic model used to validate the dict returned by the handler before it is serialized into the ResponseEnvelope.

Type:

type[pydantic.main.BaseModel]

handler

Async callable with the signature (hub: _ToyHub, data: req_model) -> dict that performs the command and returns a result dict.

Type:

Callable

is_scan

If True, the command is a scan subscription command (start_scan / stop_scan) and routed to _handle_scan instead of the generic dispatcher. Handler is unused in that case.

Type:

bool

is_shutdown

If True, the command is a shutdown command and routed to _handle_shutdown instead of the generic dispatcher. Handler is unused in that case.

Type:

bool

is_heartbeat

If True, the command is a heartbeat command (enable_heartbeat / heartbeat) and routed to _handle_heartbeat.

Type:

bool

is_limit

If True, the command is a limit command (set_intensity1_limit / set_intensity2_limit) and routed to _handle_limit.

Type:

bool

handler: Callable

is_heartbeat: bool = False

is_limit: bool = False

is_scan: bool = False

is_shutdown: bool = False

req_model: type[BaseModel]

resp_model: type[BaseModel]

class tikal.websocket.toy_server.ToyServer(toy_cache_path: Path = PosixPath('.'), host: str = 'localhost', port: int = 8142, idle_shutdown_delay: float = 3.0, mock_toys: bool = False, log_name: str = 'tikal_ws')

Bases: object

WebSocket server that wraps a _ToyHub instance.

The server owns the _ToyHub and wires itself up as all of its callbacks. Create it, then await self.serve() to start accepting connections.

async serve() → None

Start the WebSocket server and block until it shuts itself down.

tikal.websocket.toy_server_models module

Contains data models and payload structures used for the client-server communication of ToyServer

This module defines the primary data models for handling request and response envelopes, event envelopes, error payloads, and specific request payload structures for commands related to toys management. Each model is built with Pydantic for data validation and serialization purposes.

class tikal.websocket.toy_server_models.AckData(*, ack: bool = True, toy_id: str | None = None)

Bases: BaseModel

Generic acknowledgement payload returned by commands that have no meaningful data to return.

ack

Always True

Type:

bool

toy_id

Identifier of the affected toy, if applicable.

Type:

str | None

ack: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

toy_id: str | None

class tikal.websocket.toy_server_models.AddRequestData(*, toy_id: str, model_name: str)

Bases: BaseModel

Payload for the add command.

toy_id

Unique identifier of the toy to connect (e.g., its Bluetooth address). The toy must have been discovered by an active scan first.

Type:

str

model_name

Model name to assign (case-insensitive). Must be a valid model for the toy’s brand. Use the “get_brands” command for the full list

Type:

str

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_name: str

toy_id: str

class tikal.websocket.toy_server_models.BatteryResponseData(*, battery: int | None, toy_id: str)

Bases: BaseModel

Response payload for get_battery.

battery

Battery level in the range 0–100, or None if the toy has no battery.

Type:

int | None

toy_id

Unique identifier of the queried toy.

Type:

str

battery: int | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

toy_id: str

class tikal.websocket.toy_server_models.BrandsData(*, brands: dict[str, list[str]])

Bases: BaseModel

Response payload for get_brands.

brands

Mapping of brand name -> list of supported model names, e.g. {“Lovense”: [“Gush”, “Solace”, …]}.

Type:

dict[str, list[str]]

brands: dict[str, list[str]]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class tikal.websocket.toy_server_models.ConnectionStatusResponseData(*, connection_status: str, toy_id: str)

Bases: BaseModel

Response payload for get_connection_status.

connection_status

Current connection status: “connected”, “reconnecting”, “lost”, or “powered_off”.

Type:

str

toy_id

Unique identifier of the queried toy.

Type:

str

connection_status: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

toy_id: str

class tikal.websocket.toy_server_models.DirectCommandData(*, toy_id: str, command: str)

Bases: BaseModel

Payload for the direct_command command. Useful to access functionality not exposed by the server.

toy_id

Unique identifier of the target toy.

Type:

str

command

Raw command string to send directly to the toy over BLE.

Type:

str

Note

Do not use this to change the tracked state (e.g., intensities) as it bypasses the server’s state tracking.

command: str

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

toy_id: str

class tikal.websocket.toy_server_models.DirectCommandResponseData(*, response: str, toy_id: str)

Bases: BaseModel

Response payload for direct_command.

response

Raw response string returned by the toy, or an empty string if the command could not be delivered.

Type:

str

toy_id

Unique identifier of the toy that was commanded.

Type:

str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

response: str

toy_id: str

class tikal.websocket.toy_server_models.ErrorData(*, error: str, message: str, traceback: str | None = None, toy_id: str | None = None, model_name: str | None = None, brand: str | None = None)

Bases: BaseModel

The error payload returned by the server when an error occurs.

error

Name of the error.

Type:

str

message

Human-readable error message.

Type:

str

traceback

Traceback of the error if available.

Type:

str | None

toy_id

Toy ID that caused the error, if applicable.

Type:

str | None

model_name

Model name that caused the error, if applicable.

Type:

str | None

brand

Brand of the toy that caused the error, if applicable.

Type:

str | None

brand: str | None

error: str

message: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_name: str | None

toy_id: str | None

traceback: str | None

class tikal.websocket.toy_server_models.EventEnvelope(*, event: str, success: bool, data: dict)

Bases: BaseModel

Envelope for server-initiated events broadcast to clients.

Unlike ResponseEnvelope, events are not correlated to a specific request and carry no id.

event

Event name (e.g. “on_status_change”, “on_scan_update”).

Type:

str

success

True for normal data events, False when the event carries error information.

Type:

bool

data

Event payload.

Type:

dict

data: dict

event: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

success: bool

class tikal.websocket.toy_server_models.GetAllResponseData(*, toy_id: str, name: str, model_name: str, brand: str, intensity_names: list[str], supports_rotation: bool, max_intensity: int, recommended_min_interval: int, battery: int | None = None, current_intensities: list[int], intensity_limits: list[int], is_blocked: bool, pattern_version: int, pattern: list[tuple[int, int, int]], wraparound: bool, is_paused: bool, elapsed: float, connection_status: str, **extra_data: Any)

Bases: BaseModel

Response payload for get_all. Combines state, info, and connection status.

battery: int | None

brand: str

connection_status: str

current_intensities: list[int]

elapsed: float

intensity_limits: list[int]

intensity_names: list[str]

is_blocked: bool

is_paused: bool

max_intensity: int

model_config = {'extra': 'allow'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_name: str

name: str

pattern: list[tuple[int, int, int]]

pattern_version: int

recommended_min_interval: int

supports_rotation: bool

toy_id: str

wraparound: bool

class tikal.websocket.toy_server_models.GetInfoData(*, toy_id: str, full: bool)

Bases: BaseModel

Payload for the get_info command.

toy_id

Unique identifier of the toy to query.

Type:

str

full

If False, return only the inexpensive in-memory fields (toy_id, name, model_name, brand, intensity_names, supports_rotation, max_intensity, battery) with no requests to the toy. If True, also request additional brand-specific information from the toy. Returns an empty dict if the command could not be delivered.

Type:

bool

full: bool

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

toy_id: str

class tikal.websocket.toy_server_models.HeartbeatEnableData(*, enable: bool)

Bases: BaseModel

Payload for the enable_heartbeat command.

enable

True to subscribe to the heartbeat watchdog, False to unsubscribe.

Type:

bool

enable: bool

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class tikal.websocket.toy_server_models.InfoResponseData(*, toy_id: str, name: str, model_name: str, brand: str, intensity_names: list[str], supports_rotation: bool, max_intensity: int, recommended_min_interval: int, **extra_data: Any)

Bases: BaseModel

Response payload for get_info. If full=True may contain additional brand-specific fields.

toy_id

Unique identifier of the queried toy.

Type:

str

name

Human-readable name of the toy (e.g., Bluetooth advertisement name).

Type:

str

model_name

Model name of the toy (e.g., “Gush”).

Type:

str

brand

Brand name of the toy (e.g., “Lovense”).

Type:

str

intensity_names

List of two human-readable strings. The second string is “None” if the toy only has one intensity.

Type:

list[str]

supports_rotation

True if the toy supports changing the rotation direction.

Type:

bool

max_intensity

Maximum intensity value supported by the toy.

Type:

int

recommended_min_interval

Recommended minimum interval between intensity changes, in milliseconds.

Type:

int

brand: str

intensity_names: list[str]

max_intensity: int

model_config = {'extra': 'allow'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_name: str

name: str

recommended_min_interval: int

supports_rotation: bool

toy_id: str

class tikal.websocket.toy_server_models.IntensityData(*, toy_id: str, intensity: int)

Bases: BaseModel

Payload for the intensity1 and intensity2 commands.

toy_id

Unique identifier of the toy to command.

Type:

str

intensity

Target intensity level (0 – max_intensity inclusive). Setting an intensity manually will automatically pause any active pattern;

Type:

int

intensity: int

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

toy_id: str

class tikal.websocket.toy_server_models.IntensityLimitData(*, toy_id: str, limit: int | None)

Bases: BaseModel

Payload for the set_intensity1_limit and set_intensity2_limit commands.

toy_id

Unique identifier of the toy to command.

Type:

str

limit

Maximum allowed intensity level (0 – max_intensity inclusive). Values above max_intensity are clamped. Send null to remove this client’s limit (the effective limit reverts to the minimum of the remaining clients’ limits, or max_intensity if none).

Type:

int | None

limit: int | None

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

toy_id: str

class tikal.websocket.toy_server_models.RequestEnvelope(*, request: str, id: str, data: dict[str, ~typing.Any]=<factory>)

Bases: BaseModel

Outer envelope for every client -> server message.

request

Command name (e.g. “add”, “get_toy_state”).

Type:

str

id

Caller-supplied token echoed back in the response id field.

Type:

str

data

Command-specific payload. Empty dict for commands that take no arguments.

Type:

dict[str, Any]

data: dict[str, Any]

id: str

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

request: str

class tikal.websocket.toy_server_models.ResponseEnvelope(*, reply: str, id: str, success: bool, data: dict[str, Any])

Bases: BaseModel

Outer envelope for every server -> client response.

reply

Echoes the request field of the originating request.

Type:

str

id

Echoes the id field of the originating request.

Type:

str

success

True if the command succeeded; False if it produced an error. Lets clients branch on a single key without inspecting data.

Type:

bool

data

Command-specific result payload on success, or an ErrorData-shaped dict on failure.

Type:

dict[str, Any]

data: dict[str, Any]

id: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

reply: str

success: bool

class tikal.websocket.toy_server_models.SetBlockedData(*, toy_id: str, block: bool)

Bases: BaseModel

Payload for the set_blocked command.

toy_id

Unique identifier of the toy to block or unblock.

Type:

str

block

True to force both intensities to zero regardless of any pattern or manual commands; False to restore normal operation. Blocking a paused toy clears the pause state (a toy cannot be both paused and blocked).

Type:

bool

block: bool

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

toy_id: str

class tikal.websocket.toy_server_models.SetModelData(*, toy_id: str, model_name: str)

Bases: BaseModel

Payload for the set_model command.

toy_id

Unique identifier of the already-added toy whose model assignment should change.

Type:

str

model_name

New model name to assign (case-insensitive). Must be valid for the toy’s brand.

Type:

str

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_name: str

toy_id: str

class tikal.websocket.toy_server_models.SetPatternData(*, toy_id: str, pattern: list[tuple[int, int, int]], wraparound: bool, reset_time: bool)

Bases: BaseModel

Payload for the set_pattern command.

toy_id

Identifier for the toy for which the pattern is being set.

Type:

str

pattern

The sequence of segments in the pattern, where each segment is (intensity1, intensity2, duration in ms)

Type:

list[tuple[int, int, int]]

wraparound

Determines if the pattern should wrap around when the end is reached.

Type:

bool

reset_time

Indicates whether the pattern should reset the time counter after being set.

Type:

bool

check_segments() → SetPatternData

Validate each segment of the pattern.

classmethod coerce_pattern(v: Any) → list[tuple[int, int, int]]

Accept either list-of-tuples or list-of-lists and normalize to list-of-tuples.

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pattern: list[tuple[int, int, int]]

reset_time: bool

toy_id: str

wraparound: bool

class tikal.websocket.toy_server_models.SetPausedData(*, toy_id: str, pause: bool)

Bases: BaseModel

Payload for the set_paused command.

toy_id

Unique identifier of the toy to pause or resume.

Type:

str

pause

True to pause pattern playback (toy stops, pattern timer freezes). False to resume from where it left off.

Type:

bool

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pause: bool

toy_id: str

class tikal.websocket.toy_server_models.ToyIdData(*, toy_id: str)

Bases: BaseModel

Single toy_id argument: Shared by the many commands that need nothing else.

toy_id

Unique identifier of the toy to perform the command on.

Type:

str

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

toy_id: str

class tikal.websocket.toy_server_models.ToyIdsData(*, toy_ids: list[str])

Bases: BaseModel

Response payload for get_toy_ids.

toy_ids

Snapshot list of all toy identifiers currently managed by _ToyHub.

Type:

list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

toy_ids: list[str]

class tikal.websocket.toy_server_models.ToyStateData(*, toy_id: str, current_intensities: list[int], intensity_limits: list[int], is_blocked: bool, pattern_version: int, pattern: list[tuple[int, int, int]], wraparound: bool, is_paused: bool, elapsed: float)

Bases: BaseModel

Response payload for get_toy_state and the on_toy_state_change event.

toy_id

Unique identifier of the toy.

Type:

str

current_intensities

Current intensity values as [intensity1, intensity2]. intensity2 is always 0 for single-intensity toys.

Type:

list[int]

intensity_limits

Current intensity limits as [limit1, limit2]. All intensity commands are clamped to these values.

Type:

list[int]

is_blocked

True when the toy is blocked (both intensities forced to zero).

Type:

bool

pattern_version

Increments each time the pattern state changes.

Type:

int

pattern

Active pattern as a list of (duration_ms, intensity1, intensity2) tuples.

Type:

list[tuple[int, int, int]]

wraparound

True if the pattern loops after the final segment; False if it stops.

Type:

bool

is_paused

True when pattern playback is paused.

Type:

bool

elapsed

Milliseconds elapsed since the start of the pattern or the last wraparound (Does not advance during pauses).

Type:

float

current_intensities: list[int]

elapsed: float

intensity_limits: list[int]

is_blocked: bool

is_paused: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pattern: list[tuple[int, int, int]]

pattern_version: int

toy_id: str

wraparound: bool

Module contents