API Reference

This section contains the complete API documentation for the Gecko IoT Client.

Main Client

class gecko_iot_client.GeckoIotClient(idd, transporter, config_timeout=5.0)

Bases: object

Main client for interacting with Gecko IoT devices.

The GeckoIotClient provides a high-level interface for connecting to and controlling Gecko IoT devices through various transport protocols (e.g., MQTT via AWS IoT). It handles device configuration, state management, and zone control.

Parameters:

• idd (str) – Unique identifier for the device/client

• transporter (AbstractTransporter) – Transport layer implementation for communication

• config_timeout (float) – Maximum time to wait for configuration loading in seconds (default: 30.0)

Example

>>> from gecko_iot_client import GeckoIotClient
>>> from gecko_iot_client.transporters.mqtt import MqttTransporter
>>>
>>> transporter = MqttTransporter(
...     endpoint="your-endpoint.amazonaws.com",
...     certificate_path="cert.pem.crt",
...     private_key_path="private.pem.key",
...     ca_file_path="AmazonRootCA1.pem"
... )
>>>
>>> with GeckoIotClient("device-123", transporter) as client:
...     zones = client.get_zones()
...     print(f"Found {len(zones)} zone types")

__init__(idd, transporter, config_timeout=5.0)

Parameters:

• idd (str)

• transporter (AbstractTransporter)

• config_timeout (float)

connect()

Establish connection to the device and initialize configuration.

This method sets up event handlers for configuration and state changes, connects to the transport layer, and automatically loads the device configuration.

Raises:

Exception – If connection or configuration loading fails

__enter__()

Enter the context manager.

Automatically calls connect() when entering the context.

Returns:

Self for use in with statement

Return type:

GeckoIotClient

__exit__(exc_type, exc_val, exc_tb)

Exit the context manager.

Automatically calls disconnect() when exiting the context.

Parameters:

• exc_type – Exception type if an exception occurred

• exc_val – Exception value if an exception occurred

• exc_tb – Exception traceback if an exception occurred

property is_connected: bool

Check if the client is currently connected.

This checks both MQTT connectivity and device shadow connectivity status.

Returns:

True if fully connected (MQTT + gateway + vessel), False otherwise

Return type:

bool

disconnect()

Disconnect from the device and clean up resources.

This method properly closes the transport connection and performs any necessary cleanup.

property connectivity_status: ConnectivityStatus

Get current connectivity status.

Returns:

Current connectivity including MQTT, gateway, and vessel status

Return type:

ConnectivityStatus

property operation_mode_controller: OperationModeController

Get the operation mode controller for read/write operations.

Returns:

Controller for operation mode functionality

Return type:

OperationModeController

property operation_mode_status: OperationModeController

Get current operation mode status (legacy property - use operation_mode_controller instead).

Returns:

Current operation mode controller

Return type:

OperationModeController

on(channel, callback)

Register a callback for a specific event channel.

Parameters:

• channel (EventChannel) – The event channel to listen to

• callback (Callable) – Function to call when the event occurs

Return type:

None

off(channel, callback)

Unregister a callback from a specific event channel.

Parameters:

• channel (EventChannel) – The event channel to stop listening to

• callback (Callable) – The callback function to remove

Return type:

None

get_zones()

Return the parsed zones organized by type.

Return type:

Dict[ZoneType, List[AbstractZone]]

Returns:

Dict mapping zone types to lists of zones of that type. Returns a copy to prevent external modification.

get_zones_by_type(zone_type)

Return all zones of a specific type.

Parameters:

zone_type (ZoneType) – The type of zones to retrieve

Return type:

List[AbstractZone]

Returns:

List of zones matching the specified type

get_zone_by_id_and_type(zone_type, zone_id)

Find and return a zone by its type and ID.

Parameters:

• zone_type (ZoneType) – The type of the zone

• zone_id (str) – The zone ID to search for

Return type:

AbstractZone

Returns:

The zone with matching type and ID

Raises:

ValueError – If no zone found with the specified type and ID

on_zone_update(callback)

Register callback for zone updates (legacy method).

This method is maintained for backward compatibility. New code should use: client.on(EventChannel.ZONE_UPDATE, callback)

Parameters:

callback (Callable[[Dict[ZoneType, List[AbstractZone]]], None]) – Function that takes a dictionary of zones organized by type

register_zone_callbacks()

Register callbacks for zone monitoring (legacy method).

This method is maintained for backward compatibility. For new code, use the event system: client.on(EventChannel.ZONE_UPDATE, callback)

setup_zone_control()

Set up zone and feature control functionality for publishing desired state updates.

Return type:

None

list_zones()

Get a simple list of all zones with basic info.

Return type:

List[Dict[str, Any]]

Returns:

List of dictionaries with zone information (id, name, type, has_control)

class gecko_iot_client.AbstractTransporter

Bases: object

Abstract base class for transport layer implementations.

This class defines the interface for communicating with Gecko IoT devices through various protocols (MQTT, HTTP, etc.). Concrete implementations handle the protocol-specific details while providing a unified interface for the client.

connect()

Establish connection to the transport medium.

Raises:

NotImplementedError – Must be implemented by subclasses

disconnect()

Close connection and clean up resources.

Raises:

NotImplementedError – Must be implemented by subclasses

on_state_change(callback)

Register callback for state change notifications.

Parameters:

callback – Function to call when state changes occur

Raises:

NotImplementedError – Must be implemented by subclasses

change_state(new_state)

Request a state change on the device.

Parameters:

new_state – The new state to apply

Raises:

NotImplementedError – Must be implemented by subclasses

load_configuration(timeout=30.0)

Load device configuration from the transport medium.

Parameters:

timeout (float) – Maximum time to wait for configuration response in seconds

Raises:

NotImplementedError – Must be implemented by subclasses

on_configuration_loaded(callback)

Register callback for configuration load events.

Parameters:

callback – Function to call when configuration is loaded

Raises:

NotImplementedError – Must be implemented by subclasses

load_state()

Load current device state from the transport medium.

Raises:

NotImplementedError – Must be implemented by subclasses

on_state_loaded(callback)

Register callback for state load events.

Parameters:

callback – Function to call when state is loaded

Raises:

NotImplementedError – Must be implemented by subclasses

on_connectivity_change(callback)

Register callback for connectivity changes.

Parameters:

callback – Function to call when connectivity status changes. Callback should accept a boolean (True=connected, False=disconnected).

Raises:

NotImplementedError – Must be implemented by subclasses

is_connected()

Check if the transport is currently connected.

Returns:

True if connected, False otherwise

Return type:

bool

Raises:

NotImplementedError – Must be implemented by subclasses

publish_desired_state(desired_state)

Publish desired state updates to the transport medium.

Parameters:

desired_state (Dict[str, Any]) – Dictionary of desired state structure to publish

Return type:

Future

Returns:

Future that resolves when update is published

Note

The actual transport implementation (MQTT, HTTP, etc.) handles the protocol-specific details like topics, payloads, field naming. This method is transport-agnostic and doesn’t impose any business logic about zones, features, or other domain concepts.

publish_batch_desired_state(zone_updates)

Publish desired state updates for multiple zones in a single operation.

Parameters:

zone_updates (Dict[str, Dict[str, Dict[str, Any]]]) – Nested dict {zone_type: {zone_id: updates}}

Return type:

Future

Returns:

Future that resolves when batch update is published

exception gecko_iot_client.ConfigurationTimeoutError

Bases: ConfigurationError

Raised when configuration request times out.

class gecko_iot_client.AbstractZone(id, zone_type, config, name=None)

Bases: object

Base zone class with change callback functionality

Parameters:

• id (str)

• zone_type (ZoneType)

• config (Any)

• name (Optional[str])

__init__(id, zone_type, config, name=None)

Initialize zone with required fields.

Parameters:

• id (str) – Unique identifier for the zone

• zone_type (ZoneType) – Type of zone (flow, temperature control, or lighting)

• config (Any) – Configuration dictionary for the zone

• name (Optional[str]) – Optional human-readable name for the zone

set_publish_callback(callback)

Set the callback function for publishing desired state updates.

Parameters:

callback (Callable[[str, str, Dict[str, Any]], None]) – Function that takes (zone_type, zone_id, updates) and handles publishing

Return type:

None

classmethod register_zone_type(zone_type)

Decorator to register a zone class with its type.

Parameters:

zone_type (ZoneType) – The zone type to register

Return type:

Callable[[type[AbstractZone]], type[AbstractZone]]

Returns:

Decorator function that registers the zone class

classmethod from_config(zone_id, config, zone_type)

Create a zone instance from configuration data.

Parameters:

• zone_id (str) – Unique identifier for the zone

• config (Dict[str, Any]) – Configuration dictionary

• zone_type (ZoneType) – Type of zone to create

Return type:

AbstractZone

Returns:

An instance of the appropriate zone class

classmethod from_state_dict(state_dict)

Deserialize a zone from a complete state dictionary.

Parameters:

state_dict (Dict[str, Any]) – Dictionary with zone state and metadata

Return type:

AbstractZone

Returns:

An instance of the appropriate zone class

update_from_config(config)

Update zone from configuration data (structure, limits, capabilities).

Parameters:

config (Dict[str, Any]) – Configuration dictionary with zone setup values

Return type:

None

update_from_state(state)

Update zone from runtime state data.

Parameters:

state (Dict[str, Any]) – State dictionary with current values

Return type:

None

to_config()

Serialize zone to configuration dictionary.

Return type:

Dict[str, Any]

Returns:

Dictionary with zone configuration

to_state_dict()

Serialize zone to complete state dictionary.

Return type:

Dict[str, Any]

Returns:

Dictionary with zone state and metadata

class gecko_iot_client.ZoneType(*values)

Bases: Enum

Enumeration of available zone types in Gecko IoT devices.

FLOW_ZONE = 'flow'

TEMPERATURE_CONTROL_ZONE = 'temperatureControl'

LIGHTING_ZONE = 'lighting'

class gecko_iot_client.TemperatureControlZone(zone_id, config)

Bases: AbstractZone

State representation for temperature control zone v1 with validation

Parameters:

• zone_id (str)

• config (Dict[str, Any])

__init__(zone_id, config)

Initialize TemperatureControlZone with zone_id and config.

Parameters:

• zone_id (str) – Unique identifier for the temperature control zone

• config (Dict[str, Any]) – Configuration dictionary with temperature control settings

Raises:

ValueError – If temperature values are outside valid range

property status: TemperatureControlZoneStatus | None

Get the current status of the temperature control zone.

property temperature: float | None

Get the current temperature in Celsius.

property mode: TemperatureControlMode | None

Get the current temperature control mode.

property target_temperature: float | None

Get the target temperature set point in Celsius.

property min_temperature_set_point_c_value: float | None

Get the minimum allowed temperature set point in Celsius.

property max_temperature_set_point_c_value: float | None

Get the maximum allowed temperature set point in Celsius.

__str__()

String representation of the TemperatureControlZone.

Return type:

str

set_target_temperature(temperature)

Set target temperature with validation against configured limits.

Parameters:

temperature (float) – Target temperature in Celsius

Raises:

ValueError – If temperature limits not configured or temperature outside range

Return type:

None

get_temperature_state()

Get the current temperature state as a simple dictionary.

Return type:

Dict[str, Any]

Returns:

Dictionary with current temperature, target, status, and eco mode

update_from_state(state)

Update temperature control zone from runtime state.

Parameters:

state (Dict[str, Any]) – State dictionary with current values

Return type:

None

ZONE_TYPE = 'temperatureControl'

class gecko_iot_client.FlowZone(zone_id, config)

Bases: AbstractZone

State representation for flow zone v1 with validation

Parameters:

• zone_id (str)

• config (FlowConfiguration)

__init__(zone_id, config)

Initialize FlowZone with zone_id and config.

Parameters:

• zone_id (str) – Unique identifier for the flow zone

• config (FlowConfiguration) – Configuration dictionary with flow zone settings

property speed_config: SpeedConfig | None

Get speed configuration if it exists and is properly structured.

Returns:

SpeedConfig dictionary or None if not available

property initiators: List[FlowZoneInitiator] | None

Get the list of active initiators for this flow zone.

Returns:

List of FlowZoneInitiator enums or None

property type: FlowZoneType

Get the type of the flow zone.

Returns:

FlowZoneType enum value

property capabilities: List[FlowZoneCapabilities]

Get the capabilities of the flow zone.

Returns:

List of FlowZoneCapabilities enums

property presets: List[FlowZonePreset]

Get the speed presets for the flow zone, if supported.

Returns:

List of FlowZonePreset objects with name and speed

get_flow_state()

Get the current flow state as a simple dictionary.

Return type:

Dict[str, Any]

Returns:

Dictionary with active status, speed, and initiator information

set_speed(speed, active=True)

Set flow speed with validation and optional active state.

Parameters:

• speed (float) – Speed value to set (percentage)

• active (Optional[bool]) – Whether to activate the zone (default: True)

Raises:

ValueError – If speed is outside valid range

Return type:

None

activate()

Activate this flow zone.

Return type:

None

deactivate()

Deactivate this flow zone.

Raises:

RuntimeError – If zone has active non-user initiators

Return type:

None

ZONE_TYPE = 'flow'

class gecko_iot_client.LightingZone(zone_id, config)

Bases: AbstractZone

State representation for lighting zone v1 with validation

Parameters:

• zone_id (str)

• config (Dict[str, Any])

__init__(zone_id, config)

Initialize LightingZone with zone_id and config.

Parameters:

• zone_id (str) – Unique identifier for the lighting zone

• config (Dict[str, Any]) – Configuration dictionary with lighting zone settings

get_lighting_state()

Get the current lighting state as a simple dictionary.

Return type:

Dict[str, Any]

Returns:

Dictionary with active status, color, and effect information

set_color(r, g, b, i=None)

Set lighting color.

Parameters:

• r (int) – Red component (0-255)

• g (int) – Green component (0-255)

• b (int) – Blue component (0-255)

• i (Optional[int]) – Optional intensity component (0-255)

Raises:

ValueError – If any component is outside the 0-255 range

Return type:

None

update_from_state(state)

Update lighting zone from runtime state with special handling for RGBI.

Parameters:

state (Dict[str, Any]) – State dictionary with current values

Return type:

None

set_effect(effect_name)

Set lighting effect with validation.

Parameters:

effect_name (str) – Name of the effect to set (1-50 characters)

Raises:

ValueError – If effect name is not between 1 and 50 characters

Return type:

None

activate()

Activate this lighting zone.

Return type:

None

deactivate()

Deactivate this lighting zone.

Return type:

None

ZONE_TYPE = 'lighting'

class gecko_iot_client.EventChannel(*values)

Bases: Enum

Event channels for different types of notifications.

CONNECTIVITY_UPDATE = 'connectivity_update'

OPERATION_MODE_UPDATE = 'operation_mode_update'

ZONE_UPDATE = 'zone_update'

SENSOR_UPDATE = 'sensor_update'

STATE_UPDATE = 'state_update'

CONFIGURATION_UPDATE = 'configuration_update'

class gecko_iot_client.EventEmitter

Bases: object

Event emitter for managing callbacks across different channels.

This provides a unified way to handle different types of events that can occur in the Gecko IoT client.

__init__()

on(channel, callback)

Register a callback for a specific event channel.

Parameters:

• channel (EventChannel) – The event channel to listen to

• callback (Callable) – Function to call when the event occurs

Return type:

None

off(channel, callback)

Unregister a callback from a specific event channel.

Parameters:

• channel (EventChannel) – The event channel to stop listening to

• callback (Callable) – The callback function to remove

Return type:

None

emit(channel, data=None)

Emit an event to all registered callbacks for a channel.

Parameters:

• channel (EventChannel) – The event channel to emit to

• data (Any) – Optional data to pass to the callbacks

Return type:

None

clear(channel=None)

Clear callbacks for a specific channel or all channels.

Parameters:

channel (EventChannel | None) – Optional specific channel to clear. If None, clears all.

Return type:

None

class gecko_iot_client.ConnectivityStatus(transport_connected=False, gateway_status='UNKNOWN', vessel_status='UNKNOWN')

Bases: object

Data structure for connectivity status information.

Parameters:

• transport_connected (bool)

• gateway_status (str)

• vessel_status (str)

__init__(transport_connected=False, gateway_status='UNKNOWN', vessel_status='UNKNOWN')

Parameters:

• transport_connected (bool)

• gateway_status (str)

• vessel_status (str)

classmethod from_state_data(state_data, transport_connected=False)

Create ConnectivityStatus from device shadow state data.

Parameters:

• state_data (Dict[str, Any]) – The device shadow state data from AWS IoT

• transport_connected (bool) – Current transport connection status

Return type:

ConnectivityStatus

Returns:

ConnectivityStatus object with extracted connectivity info

update_from_state_data(state_data)

Update connectivity status from state data.

Parameters:

state_data (Dict[str, Any]) – The device shadow state data

Return type:

bool

Returns:

True if any connectivity aspect changed, False otherwise

update_transport_status(connected)

Update transport connection status.

Parameters:

connected (bool) – New transport connection status

Return type:

bool

Returns:

True if transport status changed, False otherwise

property is_fully_connected: bool

Check if the device is fully connected.

Returns:

True if transport is connected and both gateway and vessel are in good state

to_dict()

Convert to dictionary representation.

Return type:

Dict[str, Any]

class gecko_iot_client.OperationMode(*values)

Bases: Enum

Enum for operation modes (watercare modes).

AWAY = 0

STANDARD = 1

SAVINGS = 2

SUPER_SAVINGS = 3

WEEKENDER = 4

OTHER = 5

classmethod from_value(value)

Convert a value to OperationMode enum.

Parameters:

value (Any) – The value to convert (int, str, or OperationMode)

Return type:

OperationMode

Returns:

OperationMode enum value, defaults to OTHER for unknown values

class gecko_iot_client.OperationModeStatus(operation_mode=OperationMode.OTHER)

Bases: object

Data structure for operation mode (watercare) status information.

Parameters:

operation_mode (OperationMode)

__init__(operation_mode=OperationMode.OTHER)

Initialize operation mode status.

Parameters:

operation_mode (OperationMode) – Current operation mode

classmethod from_state_data(state_data)

Create OperationModeStatus from device shadow state data.

Parameters:

state_data (Dict[str, Any]) – The device shadow state data from AWS IoT

Return type:

OperationModeStatus

Returns:

OperationModeStatus object with extracted operation mode info

update_from_state_data(state_data)

Update operation mode status from state data.

Parameters:

state_data (Dict[str, Any]) – The device shadow state data

Return type:

bool

Returns:

True if operation mode changed, False otherwise

property mode_name: str

Get the human-readable name of the current operation mode.

Returns:

String representation of the operation mode

property is_energy_saving: bool

Check if the current mode is an energy-saving mode.

Returns:

True if mode is SAVINGS, SUPER_SAVINGS, or AWAY

to_dict()

Convert to dictionary representation.

Return type:

Dict[str, Any]

class gecko_iot_client.OperationModeController

Bases: object

Controller for operation mode (watercare mode) with read/write capabilities.

Similar to zone classes, this handles both state parsing and control commands for the operation mode functionality.

__init__()

Initialize operation mode controller with default values.

set_publish_callback(callback)

Set the callback function for publishing desired state updates.

Parameters:

callback (Callable[[str, Dict[str, Any]], None]) – Function that takes (feature_name, updates) and handles publishing

Return type:

None

classmethod from_state_data(state_data)

Create OperationModeController from device shadow state data.

Parameters:

state_data (Dict[str, Any]) – The device shadow state data from AWS IoT

Return type:

OperationModeController

Returns:

OperationModeController object with extracted operation mode info

update_from_state_data(state_data)

Update operation mode status from state data.

Parameters:

state_data (Dict[str, Any]) – The device shadow state data

Return type:

bool

Returns:

True if operation mode changed, False otherwise

property mode_name: str

Get the human-readable name of the current operation mode.

Returns:

String representation of the operation mode

property is_energy_saving: bool

Check if the current mode is an energy-saving mode.

Returns:

True if mode is SAVINGS, SUPER_SAVINGS, or AWAY

set_mode(mode)

Set the operation mode.

Parameters:

mode (OperationMode) – The operation mode to set

Raises:

ValueError – If mode is not an OperationMode enum

Return type:

None

set_mode_by_name(mode_name)

Set the operation mode by name.

Parameters:

mode_name (str) – String name of the mode (case-insensitive)

Raises:

ValueError – If mode name is not recognized

Return type:

None

set_mode_by_value(mode_value)

Set the operation mode by numeric value.

Parameters:

mode_value (int) – Numeric value of the mode (0-5)

Return type:

None

to_dict()

Convert to dictionary representation.

Return type:

Dict[str, Any]

Returns:

Dictionary with operation mode information

class gecko_iot_client.GeckoApiClient(websession, api_url=API_BASE_URL, auth0_url=AUTH0_BASE_URL)

Bases: ABC

Provide Gecko authentication tied to an OAuth2 based config entry.

Parameters:

• websession (ClientSession)

• api_url (str)

• auth0_url (str)

__init__(websession, api_url=API_BASE_URL, auth0_url=AUTH0_BASE_URL)

Initialize Gecko auth.

Parameters:

• websession (ClientSession) – aiohttp ClientSession for making HTTP requests

• api_url (str) – Base URL for Gecko API (default: production API)

• auth0_url (str) – Base URL for Auth0 authentication (default: production Auth0)

abstractmethod async async_get_access_token()

Return a valid access token for the Gecko API.

This method must be implemented by subclasses to provide OAuth2 token management.

Return type:

str

Returns:

Valid access token string

async async_get_user_id()

Get user information from Auth0 or Gecko API.

Return type:

dict[str, Any]

Returns:

User ID (sub claim) from Auth0

Raises:

ValueError – If user ID not found in Auth0 response

async async_request(method, endpoint, **kwargs)

Make an authenticated request to the Gecko API.

Parameters:

• method (str) – HTTP method (GET, POST, etc.)

• endpoint (str) – API endpoint path (e.g., “/v4/accounts/123/vessels”)

• **kwargs (Any) – Additional arguments to pass to aiohttp request

Return type:

Any

Returns:

JSON response from the API

Raises:

aiohttp.ClientResponseError – If the request fails

async async_get_vessels(account_id)

Get available vessels for the account.

Parameters:

account_id (str) – Account ID to fetch vessels for

Return type:

list[dict[str, Any]]

Returns:

List of vessel dictionaries

async async_get_user_info(user_id)

Get user information from Gecko API.

Parameters:

user_id (str) – User ID to fetch information for

Return type:

dict[str, Any]

Returns:

User information dictionary

async async_get_monitor_livestream(monitor_id)

Get MQTT livestream connection details for a monitor.

Parameters:

monitor_id (str) – Monitor ID to get livestream details for

Return type:

dict[str, Any]

Returns:

Dictionary with MQTT connection details including endpoint and credentials

Models Package

Models package for GeckoIotClient.

class gecko_iot_client.models.TemperatureControlZoneStatus(*values)

Bases: Enum

Enumeration of temperature control zone status values.

IDLE = 0

HEATING = 1

COOLING = 2

INVALID = 3

HEAT_PUMP_HEATING = 4

HEAT_PUMP_AND_HEATER_HEATING = 5

HEAT_PUMP_COOLING = 6

HEAT_PUMP_DEFROSTING = 7

HEAT_PUMP_ERROR = 8

property is_heating: bool

Check if the zone is currently in a heating state.

Returns:

True if status indicates heating is active

class gecko_iot_client.models.TemperatureControlMode(eco=False)

Bases: object

Temperature control mode configuration with eco mode support.

Parameters:

eco (bool)

__init__(eco=False)

Initialize temperature control mode.

Parameters:

eco (bool) – Whether eco mode is enabled (default: False)

class gecko_iot_client.models.FlowZoneInitiator(*values)

Bases: Enum

Enum for flow zone initiators

USER_DEMAND = 'UD'

CHECKFLOW = 'CF'

PURGE = 'PU'

FILTRATION = 'FI'

HEATING = 'HT'

COOLDOWN = 'CD'

HEAT_PUMP = 'HTP'

class gecko_iot_client.models.FlowZoneCapabilities(*values)

Bases: Enum

Enum for flow zone capabilities

SUPPORTS_SPEED_PRESETS = 'supports_speed_presets'

SUPPORTS_SPEED_PERCENTAGE = 'supports_speed_percentage'

SUPPORTS_TURN_ON = 'supports_turn_on'

SUPPORTS_TURN_OFF = 'supports_turn_off'

class gecko_iot_client.models.FlowZonePreset(name, speed)

Bases: object

Preset configuration for flow zone speeds.

Parameters:

• name (str)

• speed (float)

name: str

speed: float

__init__(name, speed)

Parameters:

• name (str)

• speed (float)

class gecko_iot_client.models.RGB(r, g, b, i=None)

Bases: object

RGB color representation with optional intensity.

Parameters:

• r (int)

• g (int)

• b (int)

• i (Optional[int])

__init__(r, g, b, i=None)

Initialize RGB color with validation.

Parameters:

• r (int) – Red component (0-255)

• g (int) – Green component (0-255)

• b (int) – Blue component (0-255)

• i (Optional[int]) – Optional intensity component (0-255)

Raises:

ValueError – If any component is outside the 0-255 range

model_dump()

Convert to dictionary (replaces Pydantic’s model_dump).

Return type:

Dict[str, Any]

Returns:

Dictionary with r, g, b, and optionally i keys

class gecko_iot_client.models.ZoneConfigurationParser

Bases: object

Simple parser for zone configurations.

Converts raw zone configuration data into typed zone instances and applies runtime state updates to existing zones.

ZONE_TYPES = {'flow': ZoneType.FLOW_ZONE, 'lighting': ZoneType.LIGHTING_ZONE, 'temperatureControl': ZoneType.TEMPERATURE_CONTROL_ZONE}

ZONE_TYPE_TO_CLASS = {ZoneType.FLOW_ZONE: <class 'gecko_iot_client.models.flow_zone.FlowZone'>, ZoneType.LIGHTING_ZONE: <class 'gecko_iot_client.models.lighting_zone.LightingZone'>, ZoneType.TEMPERATURE_CONTROL_ZONE: <class 'gecko_iot_client.models.temperature_control_zone.TemperatureControlZone'>}

parse_zones_configuration(zones_config)

Parse zones configuration into zone instances.

Parameters:

zones_config (Dict[str, Any]) – Raw zone configuration dictionary

Return type:

Dict[ZoneType, List[AbstractZone]]

Returns:

Dictionary mapping zone types to lists of zone instances

apply_state_to_zones(zones, state_data)

Apply runtime state data to existing zone instances (not configuration).

Parameters:

• zones (Dict[ZoneType, List[AbstractZone]]) – Dictionary of existing zone instances organized by type

• state_data (Dict[str, Any]) – Device shadow state data containing runtime zone states

Return type:

None

Zone Types

class gecko_iot_client.models.abstract_zone.ZoneType(*values)

Bases: Enum

Enumeration of available zone types in Gecko IoT devices.

FLOW_ZONE = 'flow'

TEMPERATURE_CONTROL_ZONE = 'temperatureControl'

LIGHTING_ZONE = 'lighting'

class gecko_iot_client.models.abstract_zone.AbstractZone(id, zone_type, config, name=None)

Bases: object

Base zone class with change callback functionality

Parameters:

• id (str)

• zone_type (ZoneType)

• config (Any)

• name (Optional[str])

__init__(id, zone_type, config, name=None)

Initialize zone with required fields.

Parameters:

• id (str) – Unique identifier for the zone

• zone_type (ZoneType) – Type of zone (flow, temperature control, or lighting)

• config (Any) – Configuration dictionary for the zone

• name (Optional[str]) – Optional human-readable name for the zone

set_publish_callback(callback)

Set the callback function for publishing desired state updates.

Parameters:

callback (Callable[[str, str, Dict[str, Any]], None]) – Function that takes (zone_type, zone_id, updates) and handles publishing

Return type:

None

classmethod register_zone_type(zone_type)

Decorator to register a zone class with its type.

Parameters:

zone_type (ZoneType) – The zone type to register

Return type:

Callable[[type[AbstractZone]], type[AbstractZone]]

Returns:

Decorator function that registers the zone class

classmethod from_config(zone_id, config, zone_type)

Create a zone instance from configuration data.

Parameters:

• zone_id (str) – Unique identifier for the zone

• config (Dict[str, Any]) – Configuration dictionary

• zone_type (ZoneType) – Type of zone to create

Return type:

AbstractZone

Returns:

An instance of the appropriate zone class

classmethod from_state_dict(state_dict)

Deserialize a zone from a complete state dictionary.

Parameters:

state_dict (Dict[str, Any]) – Dictionary with zone state and metadata

Return type:

AbstractZone

Returns:

An instance of the appropriate zone class

update_from_config(config)

Update zone from configuration data (structure, limits, capabilities).

Parameters:

config (Dict[str, Any]) – Configuration dictionary with zone setup values

Return type:

None

update_from_state(state)

Update zone from runtime state data.

Parameters:

state (Dict[str, Any]) – State dictionary with current values

Return type:

None

to_config()

Serialize zone to configuration dictionary.

Return type:

Dict[str, Any]

Returns:

Dictionary with zone configuration

to_state_dict()

Serialize zone to complete state dictionary.

Return type:

Dict[str, Any]

Returns:

Dictionary with zone state and metadata

gecko_iot_client.models.abstract_zone.create_zone_logger(zone_name)

Create a logging callback for zone changes.

Parameters:

zone_name (str) – Name of the zone for logging

Return type:

Callable[[str, Any, Any], None]

Returns:

Callback function that logs attribute changes

gecko_iot_client.models.abstract_zone.create_validation_callback(attribute, validator, error_message='Invalid value')

Create a validation callback for specific attributes.

Parameters:

• attribute (str) – Name of the attribute to validate

• validator (Callable[[Any], bool]) – Function that returns True if value is valid

• error_message (str) – Message to display on validation failure

Return type:

Callable[[str, Any, Any], None]

Returns:

Callback function that validates attribute changes

Temperature control zone models for Gecko IoT devices.

class gecko_iot_client.models.temperature_control_zone.TemperatureControlZoneStatus(*values)

Bases: Enum

Enumeration of temperature control zone status values.

IDLE = 0

HEATING = 1

COOLING = 2

INVALID = 3

HEAT_PUMP_HEATING = 4

HEAT_PUMP_AND_HEATER_HEATING = 5

HEAT_PUMP_COOLING = 6

HEAT_PUMP_DEFROSTING = 7

HEAT_PUMP_ERROR = 8

property is_heating: bool

Check if the zone is currently in a heating state.

Returns:

True if status indicates heating is active

class gecko_iot_client.models.temperature_control_zone.TemperatureControlMode(eco=False)

Bases: object

Temperature control mode configuration with eco mode support.

Parameters:

eco (bool)

__init__(eco=False)

Initialize temperature control mode.

Parameters:

eco (bool) – Whether eco mode is enabled (default: False)

class gecko_iot_client.models.temperature_control_zone.TemperatureControlZone(zone_id, config)

Bases: AbstractZone

State representation for temperature control zone v1 with validation

Parameters:

• zone_id (str)

• config (Dict[str, Any])

__init__(zone_id, config)

Initialize TemperatureControlZone with zone_id and config.

Parameters:

• zone_id (str) – Unique identifier for the temperature control zone

• config (Dict[str, Any]) – Configuration dictionary with temperature control settings

Raises:

ValueError – If temperature values are outside valid range

status_: TemperatureControlZoneStatus | None

temperature_: float | None

mode_: TemperatureControlMode | None

set_point: float | None

property status: TemperatureControlZoneStatus | None

Get the current status of the temperature control zone.

property temperature: float | None

Get the current temperature in Celsius.

property mode: TemperatureControlMode | None

Get the current temperature control mode.

property target_temperature: float | None

Get the target temperature set point in Celsius.

property min_temperature_set_point_c_value: float | None

Get the minimum allowed temperature set point in Celsius.

property max_temperature_set_point_c_value: float | None

Get the maximum allowed temperature set point in Celsius.

__str__()

String representation of the TemperatureControlZone.

Return type:

str

set_target_temperature(temperature)

Set target temperature with validation against configured limits.

Parameters:

temperature (float) – Target temperature in Celsius

Raises:

ValueError – If temperature limits not configured or temperature outside range

Return type:

None

get_temperature_state()

Get the current temperature state as a simple dictionary.

Return type:

Dict[str, Any]

Returns:

Dictionary with current temperature, target, status, and eco mode

update_from_state(state)

Update temperature control zone from runtime state.

Parameters:

state (Dict[str, Any]) – State dictionary with current values

Return type:

None

ZONE_TYPE = 'temperatureControl'

Flow zone models and configurations for Gecko IoT devices.

class gecko_iot_client.models.flow_zone.SpeedConfig

Bases: TypedDict

Configuration for flow zone speed settings.

maximum: int

minimum: int

stepIncrement: int

class gecko_iot_client.models.flow_zone.FlowConfiguration

Bases: TypedDict

Configuration dictionary for flow zones.

name: str | None

pumps: List[str] | None

speed: SpeedConfig

class gecko_iot_client.models.flow_zone.FlowZoneCapabilities(*values)

Bases: Enum

Enum for flow zone capabilities

SUPPORTS_SPEED_PRESETS = 'supports_speed_presets'

SUPPORTS_SPEED_PERCENTAGE = 'supports_speed_percentage'

SUPPORTS_TURN_ON = 'supports_turn_on'

SUPPORTS_TURN_OFF = 'supports_turn_off'

class gecko_iot_client.models.flow_zone.FlowZoneInitiator(*values)

Bases: Enum

Enum for flow zone initiators

USER_DEMAND = 'UD'

CHECKFLOW = 'CF'

PURGE = 'PU'

FILTRATION = 'FI'

HEATING = 'HT'

COOLDOWN = 'CD'

HEAT_PUMP = 'HTP'

class gecko_iot_client.models.flow_zone.FlowZonePreset(name, speed)

Bases: object

Preset configuration for flow zone speeds.

Parameters:

• name (str)

• speed (float)

name: str

speed: float

__init__(name, speed)

Parameters:

• name (str)

• speed (float)

class gecko_iot_client.models.flow_zone.FlowZoneType(*values)

Bases: Enum

Types of flow zones available in Gecko IoT devices.

FLOW_ZONE = 'flow_zone'

WATERFALL_ZONE = 'waterfall_zone'

BLOWER_ZONE = 'blower_zone'

class gecko_iot_client.models.flow_zone.FlowZoneTypeProperties(format_name)

Bases: object

Properties for different flow zone types.

Parameters:

format_name (callable)

format_name: callable

__init__(format_name)

Parameters:

format_name (callable)

class gecko_iot_client.models.flow_zone.FlowZone(zone_id, config)

Bases: AbstractZone

State representation for flow zone v1 with validation

Parameters:

• zone_id (str)

• config (FlowConfiguration)

__init__(zone_id, config)

Initialize FlowZone with zone_id and config.

Parameters:

• zone_id (str) – Unique identifier for the flow zone

• config (FlowConfiguration) – Configuration dictionary with flow zone settings

active: bool | None

speed: float | None

initiators_: List[FlowZoneInitiator] | None

property speed_config: SpeedConfig | None

Get speed configuration if it exists and is properly structured.

Returns:

SpeedConfig dictionary or None if not available

property initiators: List[FlowZoneInitiator] | None

Get the list of active initiators for this flow zone.

Returns:

List of FlowZoneInitiator enums or None

property type: FlowZoneType

Get the type of the flow zone.

Returns:

FlowZoneType enum value

property capabilities: List[FlowZoneCapabilities]

Get the capabilities of the flow zone.

Returns:

List of FlowZoneCapabilities enums

property presets: List[FlowZonePreset]

Get the speed presets for the flow zone, if supported.

Returns:

List of FlowZonePreset objects with name and speed

get_flow_state()

Get the current flow state as a simple dictionary.

Return type:

Dict[str, Any]

Returns:

Dictionary with active status, speed, and initiator information

set_speed(speed, active=True)

Set flow speed with validation and optional active state.

Parameters:

• speed (float) – Speed value to set (percentage)

• active (Optional[bool]) – Whether to activate the zone (default: True)

Raises:

ValueError – If speed is outside valid range

Return type:

None

activate()

Activate this flow zone.

Return type:

None

deactivate()

Deactivate this flow zone.

Raises:

RuntimeError – If zone has active non-user initiators

Return type:

None

ZONE_TYPE = 'flow'

Lighting zone models for Gecko IoT devices.

class gecko_iot_client.models.lighting_zone.RGB(r, g, b, i=None)

Bases: object

RGB color representation with optional intensity.

Parameters:

• r (int)

• g (int)

• b (int)

• i (Optional[int])

__init__(r, g, b, i=None)

Initialize RGB color with validation.

Parameters:

• r (int) – Red component (0-255)

• g (int) – Green component (0-255)

• b (int) – Blue component (0-255)

• i (Optional[int]) – Optional intensity component (0-255)

Raises:

ValueError – If any component is outside the 0-255 range

model_dump()

Convert to dictionary (replaces Pydantic’s model_dump).

Return type:

Dict[str, Any]

Returns:

Dictionary with r, g, b, and optionally i keys

class gecko_iot_client.models.lighting_zone.LightingZone(zone_id, config)

Bases: AbstractZone

State representation for lighting zone v1 with validation

Parameters:

• zone_id (str)

• config (Dict[str, Any])

__init__(zone_id, config)

Initialize LightingZone with zone_id and config.

Parameters:

• zone_id (str) – Unique identifier for the lighting zone

• config (Dict[str, Any]) – Configuration dictionary with lighting zone settings

active: bool | None

rgbi: RGB | None

effect: str | None

get_lighting_state()

Get the current lighting state as a simple dictionary.

Return type:

Dict[str, Any]

Returns:

Dictionary with active status, color, and effect information

set_color(r, g, b, i=None)

Set lighting color.

Parameters:

• r (int) – Red component (0-255)

• g (int) – Green component (0-255)

• b (int) – Blue component (0-255)

• i (Optional[int]) – Optional intensity component (0-255)

Raises:

ValueError – If any component is outside the 0-255 range

Return type:

None

update_from_state(state)

Update lighting zone from runtime state with special handling for RGBI.

Parameters:

state (Dict[str, Any]) – State dictionary with current values

Return type:

None

set_effect(effect_name)

Set lighting effect with validation.

Parameters:

effect_name (str) – Name of the effect to set (1-50 characters)

Raises:

ValueError – If effect name is not between 1 and 50 characters

Return type:

None

activate()

Activate this lighting zone.

Return type:

None

deactivate()

Deactivate this lighting zone.

Return type:

None

ZONE_TYPE = 'lighting'

Configuration and State Management

Zone configuration parser module.

Simple parser for converting zone configurations into zone instances.

class gecko_iot_client.models.zone_parser.ZoneConfigurationParser

Bases: object

Simple parser for zone configurations.

Converts raw zone configuration data into typed zone instances and applies runtime state updates to existing zones.

ZONE_TYPES = {'flow': ZoneType.FLOW_ZONE, 'lighting': ZoneType.LIGHTING_ZONE, 'temperatureControl': ZoneType.TEMPERATURE_CONTROL_ZONE}

ZONE_TYPE_TO_CLASS = {ZoneType.FLOW_ZONE: <class 'gecko_iot_client.models.flow_zone.FlowZone'>, ZoneType.LIGHTING_ZONE: <class 'gecko_iot_client.models.lighting_zone.LightingZone'>, ZoneType.TEMPERATURE_CONTROL_ZONE: <class 'gecko_iot_client.models.temperature_control_zone.TemperatureControlZone'>}

parse_zones_configuration(zones_config)

Parse zones configuration into zone instances.

Parameters:

zones_config (Dict[str, Any]) – Raw zone configuration dictionary

Return type:

Dict[ZoneType, List[AbstractZone]]

Returns:

Dictionary mapping zone types to lists of zone instances

apply_state_to_zones(zones, state_data)

Apply runtime state data to existing zone instances (not configuration).

Parameters:

• zones (Dict[ZoneType, List[AbstractZone]]) – Dictionary of existing zone instances organized by type

• state_data (Dict[str, Any]) – Device shadow state data containing runtime zone states

Return type:

None

Operation mode (watercare) model for Gecko IoT devices.

class gecko_iot_client.models.operation_mode.OperationMode(*values)

Bases: Enum

Enum for operation modes (watercare modes).

AWAY = 0

STANDARD = 1

SAVINGS = 2

SUPER_SAVINGS = 3

WEEKENDER = 4

OTHER = 5

classmethod from_value(value)

Convert a value to OperationMode enum.

Parameters:

value (Any) – The value to convert (int, str, or OperationMode)

Return type:

OperationMode

Returns:

OperationMode enum value, defaults to OTHER for unknown values

class gecko_iot_client.models.operation_mode.OperationModeStatus(operation_mode=OperationMode.OTHER)

Bases: object

Data structure for operation mode (watercare) status information.

Parameters:

operation_mode (OperationMode)

__init__(operation_mode=OperationMode.OTHER)

Initialize operation mode status.

Parameters:

operation_mode (OperationMode) – Current operation mode

classmethod from_state_data(state_data)

Create OperationModeStatus from device shadow state data.

Parameters:

state_data (Dict[str, Any]) – The device shadow state data from AWS IoT

Return type:

OperationModeStatus

Returns:

OperationModeStatus object with extracted operation mode info

update_from_state_data(state_data)

Update operation mode status from state data.

Parameters:

state_data (Dict[str, Any]) – The device shadow state data

Return type:

bool

Returns:

True if operation mode changed, False otherwise

property mode_name: str

Get the human-readable name of the current operation mode.

Returns:

String representation of the operation mode

property is_energy_saving: bool

Check if the current mode is an energy-saving mode.

Returns:

True if mode is SAVINGS, SUPER_SAVINGS, or AWAY

to_dict()

Convert to dictionary representation.

Return type:

Dict[str, Any]

Connectivity status model for Gecko IoT devices.

class gecko_iot_client.models.connectivity.ConnectivityStatus(transport_connected=False, gateway_status='UNKNOWN', vessel_status='UNKNOWN')

Bases: object

Data structure for connectivity status information.

Parameters:

• transport_connected (bool)

• gateway_status (str)

• vessel_status (str)

__init__(transport_connected=False, gateway_status='UNKNOWN', vessel_status='UNKNOWN')

Parameters:

• transport_connected (bool)

• gateway_status (str)

• vessel_status (str)

classmethod from_state_data(state_data, transport_connected=False)

Create ConnectivityStatus from device shadow state data.

Parameters:

• state_data (Dict[str, Any]) – The device shadow state data from AWS IoT

• transport_connected (bool) – Current transport connection status

Return type:

ConnectivityStatus

Returns:

ConnectivityStatus object with extracted connectivity info

update_from_state_data(state_data)

Update connectivity status from state data.

Parameters:

state_data (Dict[str, Any]) – The device shadow state data

Return type:

bool

Returns:

True if any connectivity aspect changed, False otherwise

update_transport_status(connected)

Update transport connection status.

Parameters:

connected (bool) – New transport connection status

Return type:

bool

Returns:

True if transport status changed, False otherwise

property is_fully_connected: bool

Check if the device is fully connected.

Returns:

True if transport is connected and both gateway and vessel are in good state

to_dict()

Convert to dictionary representation.

Return type:

Dict[str, Any]

Event system for Gecko IoT Client notifications.

class gecko_iot_client.models.events.EventChannel(*values)

Bases: Enum

Event channels for different types of notifications.

CONNECTIVITY_UPDATE = 'connectivity_update'

OPERATION_MODE_UPDATE = 'operation_mode_update'

ZONE_UPDATE = 'zone_update'

SENSOR_UPDATE = 'sensor_update'

STATE_UPDATE = 'state_update'

CONFIGURATION_UPDATE = 'configuration_update'

class gecko_iot_client.models.events.EventEmitter

Bases: object

Event emitter for managing callbacks across different channels.

This provides a unified way to handle different types of events that can occur in the Gecko IoT client.

__init__()

on(channel, callback)

Register a callback for a specific event channel.

Parameters:

• channel (EventChannel) – The event channel to listen to

• callback (Callable) – Function to call when the event occurs

Return type:

None

off(channel, callback)

Unregister a callback from a specific event channel.

Parameters:

• channel (EventChannel) – The event channel to stop listening to

• callback (Callable) – The callback function to remove

Return type:

None

emit(channel, data=None)

Emit an event to all registered callbacks for a channel.

Parameters:

• channel (EventChannel) – The event channel to emit to

• data (Any) – Optional data to pass to the callbacks

Return type:

None

clear(channel=None)

Clear callbacks for a specific channel or all channels.

Parameters:

channel (EventChannel | None) – Optional specific channel to clear. If None, clears all.

Return type:

None

Transporters Package

class gecko_iot_client.transporters.AbstractTransporter

Bases: object

Abstract base class for transport layer implementations.

This class defines the interface for communicating with Gecko IoT devices through various protocols (MQTT, HTTP, etc.). Concrete implementations handle the protocol-specific details while providing a unified interface for the client.

connect()

Establish connection to the transport medium.

Raises:

NotImplementedError – Must be implemented by subclasses

disconnect()

Close connection and clean up resources.

Raises:

NotImplementedError – Must be implemented by subclasses

on_state_change(callback)

Register callback for state change notifications.

Parameters:

callback – Function to call when state changes occur

Raises:

NotImplementedError – Must be implemented by subclasses

change_state(new_state)

Request a state change on the device.

Parameters:

new_state – The new state to apply

Raises:

NotImplementedError – Must be implemented by subclasses

load_configuration(timeout=30.0)

Load device configuration from the transport medium.

Parameters:

timeout (float) – Maximum time to wait for configuration response in seconds

Raises:

NotImplementedError – Must be implemented by subclasses

on_configuration_loaded(callback)

Register callback for configuration load events.

Parameters:

callback – Function to call when configuration is loaded

Raises:

NotImplementedError – Must be implemented by subclasses

load_state()

Load current device state from the transport medium.

Raises:

NotImplementedError – Must be implemented by subclasses

on_state_loaded(callback)

Register callback for state load events.

Parameters:

callback – Function to call when state is loaded

Raises:

NotImplementedError – Must be implemented by subclasses

on_connectivity_change(callback)

Register callback for connectivity changes.

Parameters:

callback – Function to call when connectivity status changes. Callback should accept a boolean (True=connected, False=disconnected).

Raises:

NotImplementedError – Must be implemented by subclasses

is_connected()

Check if the transport is currently connected.

Returns:

True if connected, False otherwise

Return type:

bool

Raises:

NotImplementedError – Must be implemented by subclasses

publish_desired_state(desired_state)

Publish desired state updates to the transport medium.

Parameters:

desired_state (Dict[str, Any]) – Dictionary of desired state structure to publish

Return type:

Future

Returns:

Future that resolves when update is published

Note

The actual transport implementation (MQTT, HTTP, etc.) handles the protocol-specific details like topics, payloads, field naming. This method is transport-agnostic and doesn’t impose any business logic about zones, features, or other domain concepts.

publish_batch_desired_state(zone_updates)

Publish desired state updates for multiple zones in a single operation.

Parameters:

zone_updates (Dict[str, Dict[str, Dict[str, Any]]]) – Nested dict {zone_type: {zone_id: updates}}

Return type:

Future

Returns:

Future that resolves when batch update is published

MQTT Transport

MQTT transporter package for Gecko IoT devices.

This package provides MQTT5 connectivity for AWS IoT Core with JWT token authentication, automatic token refresh, and Gecko-specific shadow operations.

Main exports: - MqttTransporter: High-level Gecko IoT transporter (primary interface) - MqttClient: Low-level MQTT protocol client (for advanced use)

class gecko_iot_client.transporters.mqtt.MqttTransporter(broker_url, monitor_id, token_refresh_callback=None, token_refresh_buffer_seconds=300)

Bases: AbstractTransporter

Gecko-specific MQTT transporter.

Responsibilities: - Gecko topic structure (config, state, shadow) - Token refresh and expiration management - Configuration and state loading - AbstractTransporter interface implementation - Reconnection logic with token refresh

This class contains all Gecko IoT business logic and delegates MQTT protocol operations to MqttClient.

Parameters:

• broker_url (str)

• monitor_id (str)

• token_refresh_callback (Optional[Callable[[str], str]])

• token_refresh_buffer_seconds (int)

__init__(broker_url, monitor_id, token_refresh_callback=None, token_refresh_buffer_seconds=300)

Initialize MQTT transporter with Gecko-specific logic.

Parameters:

• broker_url (str) – WebSocket URL with embedded JWT token

• monitor_id (str) – Device monitor identifier

• token_refresh_callback (Optional[Callable[[str], str]]) – Function to get new broker URL with fresh token

• token_refresh_buffer_seconds (int) – Seconds before expiry to refresh token

connect(**kwargs)

Connect using preformatted WebSocket URL with expiration management.

disconnect()

Disconnect and cleanup.

is_connected()

Check if connected to broker.

Return type:

bool

update_broker_url(new_broker_url)

Update broker URL with a fresh token.

This is useful when reusing an existing connection that needs a token refresh. The method updates the broker URL and token manager without disconnecting.

Parameters:

new_broker_url (str) – New WebSocket URL with fresh JWT token

Return type:

None

load_configuration(timeout=30.0)

Load configuration from AWS IoT.

Parameters:

timeout (float)

load_state()

Load state from AWS IoT shadow.

publish_desired_state(desired_state)

Publish desired state update to AWS IoT shadow.

Parameters:

desired_state (Dict[str, Any])

Return type:

Future

publish_batch_desired_state(zone_updates)

Publish batch desired state updates for multiple zones.

Parameters:

zone_updates (Dict[str, Dict[str, Dict[str, Any]]])

Return type:

Future

on_configuration_loaded(callback)

Register config callback.

on_state_loaded(callback)

Register state callback.

on_state_change(callback)

Register state change callback.

on_connectivity_change(callback)

Register connectivity change callback.

change_state(new_state)

Change state (placeholder for interface compliance).

class gecko_iot_client.transporters.mqtt.MqttClient(on_connected=None, on_message=None)

Bases: object

Low-level MQTT client for AWS IoT Core.

Responsibilities: - Connection/disconnection to AWS IoT MQTT broker - Publishing messages - Subscribing to topics - Message routing to handlers - Connection lifecycle callbacks

This class handles pure MQTT protocol concerns without any Gecko-specific business logic.

Parameters:

• on_connected (Optional[Callable[[bool], None]])

• on_message (Optional[Callable[[str, str], None]])

__init__(on_connected=None, on_message=None)

Initialize MQTT client.

Parameters:

• on_connected (Optional[Callable[[bool], None]]) – Callback for connection status changes (bool: connected)

• on_message (Optional[Callable[[str, str], None]]) – Default callback for messages without specific handlers

is_connected()

Check if connected to broker.

Return type:

bool

Returns:

True if currently connected to MQTT broker

connect(broker_url, client_id, timeout=CONNECTION_TIMEOUT)

Connect to AWS IoT MQTT broker.

Parameters:

• broker_url (str) – WebSocket URL with embedded JWT token and auth params

• client_id (str) – Unique client identifier

• timeout (int) – Connection timeout in seconds

Return type:

None

disconnect()

Disconnect from broker (intentional disconnect).

Return type:

None

stop_for_refresh()

Stop client for token refresh (intentional disconnect).

This is different from disconnect() as it’s specifically for token refresh scenarios and clears the intentional flag after.

Return type:

None

publish(topic, payload, timeout=PUBLISH_TIMEOUT)

Publish message to a topic.

Parameters:

• topic (str) – MQTT topic

• payload (str) – Message payload (string)

• timeout (float) – Publish timeout in seconds

Return type:

Future

Returns:

Future for the publish operation

subscribe(topic, handler)

Subscribe to a topic with a message handler.

Parameters:

• topic (str) – MQTT topic to subscribe to

• handler (Callable[[str, str], None]) – Callback function(topic, payload)

Return type:

None

clear_intentional_disconnect_flag()

Clear the intentional disconnect flag after reconnection.

Return type:

None

Connection Management

Connection state manager for MQTT transporter.

class gecko_iot_client.transporters.connection_state.ConnectionState(*values)

Bases: Enum

MQTT connection states.

DISCONNECTED = 'disconnected'

CONNECTING = 'connecting'

CONNECTED = 'connected'

RECONNECTING = 'reconnecting'

CONNECTION_FAILED = 'connection_failed'

DISCONNECTING = 'disconnecting'

CREDENTIAL_REFRESH_FAILED = 'credential_refresh_failed'

class gecko_iot_client.transporters.connection_state.ConnectionEvent(state, timestamp, message=None, error=None)

Bases: object

Connection state change event.

Parameters:

• state (ConnectionState)

• timestamp (datetime)

• message (Optional[str])

• error (Optional[Exception])

state: ConnectionState

timestamp: datetime

message: str | None = None

error: Exception | None = None

__init__(state, timestamp, message=None, error=None)

Parameters:

• state (ConnectionState)

• timestamp (datetime)

• message (Optional[str])

• error (Optional[Exception])

class gecko_iot_client.transporters.connection_state.ConnectionStateManager(reconnect_enabled=True, max_reconnect_attempts=5, initial_retry_delay=1.0, max_retry_delay=60.0, backoff_multiplier=2.0)

Bases: object

Manages connection state and provides reconnection logic.

Parameters:

• reconnect_enabled (bool)

• max_reconnect_attempts (int)

• initial_retry_delay (float)

• max_retry_delay (float)

• backoff_multiplier (float)

__init__(reconnect_enabled=True, max_reconnect_attempts=5, initial_retry_delay=1.0, max_retry_delay=60.0, backoff_multiplier=2.0)

Initialize connection state manager.

Parameters:

• reconnect_enabled (bool) – Whether automatic reconnection is enabled

• max_reconnect_attempts (int) – Maximum number of reconnection attempts

• initial_retry_delay (float) – Initial delay between reconnection attempts (seconds)

• max_retry_delay (float) – Maximum delay between reconnection attempts (seconds)

• backoff_multiplier (float) – Multiplier for exponential backoff

add_state_change_callback(callback)

Add a callback to be notified of state changes.

Parameters:

callback (Callable[[ConnectionEvent], None])

remove_state_change_callback(callback)

Remove a state change callback.

Parameters:

callback (Callable[[ConnectionEvent], None])

set_reconnect_callback(callback)

Set the callback function to call when attempting reconnection.

Parameters:

callback (Callable[[], None])

get_current_state()

Get the current connection state.

Return type:

ConnectionState

is_connected()

Check if currently connected.

Return type:

bool

is_connecting()

Check if currently connecting or reconnecting.

Return type:

bool

set_state(new_state, message=None, error=None)

Set the connection state and notify callbacks.

Parameters:

• new_state (ConnectionState) – The new connection state

• message (str) – Optional message describing the state change

• error (Exception) – Optional error that caused the state change

get_event_history()

Get the connection event history.

Return type:

List[ConnectionEvent]

enable_reconnection()

Enable automatic reconnection.

disable_reconnection()

Disable automatic reconnection.

is_reconnection_enabled()

Check if automatic reconnection is enabled.

Return type:

bool

reset_reconnect_attempts()

Reset the reconnection attempt counter.

get_reconnect_attempts()

Get the current number of reconnection attempts.

Return type:

int

cleanup()

Clean up resources.

Token manager for handling AWS IoT token refresh and expiration.

class gecko_iot_client.transporters.token_manager.TokenManager(token_refresh_callback, refresh_threshold_minutes=15)

Bases: object

Manages AWS IoT tokens with automatic refresh capabilities.

Parameters:

• token_refresh_callback (Callable[[], dict])

• refresh_threshold_minutes (int)

__init__(token_refresh_callback, refresh_threshold_minutes=15)

Initialize the token manager.

Parameters:

• token_refresh_callback (Callable[[], dict]) – Function that returns new token data

• refresh_threshold_minutes (int) – How many minutes before expiry to refresh

add_refresh_listener(callback)

Add a callback to be notified when token is refreshed.

Parameters:

callback (Callable[[dict], None])

remove_refresh_listener(callback)

Remove a token refresh listener.

Parameters:

callback (Callable[[dict], None])

set_token(token_data)

Set the current token and start refresh monitoring.

Parameters:

token_data (dict) – Dictionary containing token and expiry information Expected keys: ‘access_token’, ‘expires_in’ or ‘expires_at’

get_current_token()

Get the current token data.

Return type:

Optional[dict]

is_token_valid()

Check if the current token is valid and not expired.

Return type:

bool

needs_refresh()

Check if token needs to be refreshed based on threshold.

Return type:

bool

refresh_token()

Manually refresh the token.

Return type:

dict

Returns:

New token data

Raises:

TokenRefreshError – If refresh fails

stop_refresh_monitoring()

Stop the background token refresh monitoring.

cleanup()

Clean up resources.

Configuration and Logging

Logging configuration for MQTT transporter.

class gecko_iot_client.transporters.logging_config.MqttTransporterLogger(name='gecko_iot_mqtt', level=logging.INFO, log_file=None, max_file_size=10 * 1024 * 1024, backup_count=5, include_console=True)

Bases: object

Custom logger configuration for MQTT transporter with structured logging.

Parameters:

• name (str)

• level (int)

• log_file (Optional[str])

• max_file_size (int)

• backup_count (int)

• include_console (bool)

__init__(name='gecko_iot_mqtt', level=logging.INFO, log_file=None, max_file_size=10 * 1024 * 1024, backup_count=5, include_console=True)

Initialize the logger configuration.

Parameters:

• name (str) – Logger name

• level (int) – Logging level

• log_file (Optional[str]) – Path to log file (optional)

• max_file_size (int) – Maximum size of log file before rotation

• backup_count (int) – Number of backup files to keep

• include_console (bool) – Whether to include console logging

get_logger()

Get the configured logger instance.

Return type:

Logger

add_connection_context(client_id, broker_url)

Add connection context to all log messages.

Parameters:

• client_id (str)

• broker_url (str)

class gecko_iot_client.transporters.logging_config.ConnectionEventLogger(base_logger)

Bases: object

Specialized logger for connection events with metrics tracking.

Parameters:

base_logger (Logger)

__init__(base_logger)

Parameters:

base_logger (Logger)

log_connection_attempt(broker_url, client_id)

Log a connection attempt.

Parameters:

• broker_url (str)

• client_id (str)

log_connection_success(client_id, duration_ms)

Log successful connection.

Parameters:

• client_id (str)

• duration_ms (float)

log_connection_failure(client_id, error, duration_ms)

Log connection failure.

Parameters:

• client_id (str)

• error (Exception)

• duration_ms (float)

log_disconnection(client_id, reason)

Log disconnection event.

Parameters:

• client_id (str)

• reason (str)

log_token_refresh(success, error=None)

Log token refresh event.

Parameters:

• success (bool)

• error (Optional[Exception])

log_reconnection_attempt(attempt, max_attempts, delay)

Log reconnection attempt.

Parameters:

• attempt (int)

• max_attempts (int)

• delay (float)

get_success_rate()

Calculate connection success rate.

Return type:

float

get_connection_stats()

Get connection statistics.

Return type:

dict

log_periodic_stats()

Log periodic connection statistics.

gecko_iot_client.transporters.logging_config.setup_mqtt_logging(log_level='INFO', log_file=None, enable_console=True)

Setup logging for MQTT transporter.

Parameters:

• log_level (str) – Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)

• log_file (Optional[str]) – Optional path to log file

• enable_console (bool) – Whether to enable console logging

Return type:

tuple[Logger, ConnectionEventLogger]

Returns:

Tuple of (main_logger, event_logger)

Exceptions

Custom exceptions for MQTT transporter operations.

exception gecko_iot_client.transporters.exceptions.MqttTransporterError

Bases: Exception

Base exception class for MQTT transporter errors.

exception gecko_iot_client.transporters.exceptions.ConnectionError

Bases: MqttTransporterError

Raised when connection to MQTT broker fails.

exception gecko_iot_client.transporters.exceptions.AuthenticationError

Bases: MqttTransporterError

Raised when authentication with MQTT broker fails.

exception gecko_iot_client.transporters.exceptions.TokenRefreshError

Bases: MqttTransporterError

Raised when token refresh operation fails.

exception gecko_iot_client.transporters.exceptions.DisconnectionError

Bases: MqttTransporterError

Raised when disconnection from MQTT broker fails.

exception gecko_iot_client.transporters.exceptions.ConfigurationError

Bases: MqttTransporterError

Raised when configuration is invalid or missing.

exception gecko_iot_client.transporters.exceptions.ConfigurationTimeoutError

Bases: ConfigurationError

Raised when configuration request times out.