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

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

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.

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

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

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

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

load_state()

Load current device state from the transport medium.

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

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

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

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

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

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.

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)

• zone_type (ZoneType)

• config (Any)

• name (Optional[str])

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

classmethod register_zone_type(zone_type)

Decorator to register a zone class with its type

Parameters:

zone_type (ZoneType)

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

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

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

class gecko_iot_client.ZoneType(*values)

Bases: Enum

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])

ZONE_TYPE = 'temperatureControl'

__init__(zone_id, config)

Initialize TemperatureControlZone with zone_id and config.

Parameters:

• zone_id (str)

• config (Dict[str, Any])

__str__()

String representation of the TemperatureControlZone.

Return type:

str

get_temperature_state()

Get the current temperature state as a simple dictionary.

Return type:

Dict[str, Any]

property max_temperature_set_point_c_value: float | None

property min_temperature_set_point_c_value: float | None

property mode: TemperatureControlMode | None

set_target_temperature(temperature)

Set target temperature with validation against configured limits.

Parameters:

temperature (float)

Return type:

None

property status: TemperatureControlZoneStatus | None

property target_temperature: float | None

property temperature: float | None

update_from_state(state)

Update temperature control zone from runtime state.

Parameters:

state (Dict[str, Any])

Return type:

None

class gecko_iot_client.FlowZone(zone_id, config)

Bases: AbstractZone

State representation for flow zone v1 with validation

Parameters:

• zone_id (str)

• config (FlowConfiguration)

ZONE_TYPE = 'flow'

__init__(zone_id, config)

Initialize FlowZone with zone_id and config.

Parameters:

• zone_id (str)

• config (FlowConfiguration)

activate()

Activate this zone.

Return type:

None

property capabilities: List[FlowZoneCapabilities]

Get the capabilities of the flow zone.

deactivate()

Deactivate this zone.

Return type:

None

get_flow_state()

Get the current flow state as a simple dictionary.

Return type:

Dict[str, Any]

property initiators: List[FlowZoneInitiator] | None

property presets: List[FlowZonePreset]

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

set_speed(speed, active=True)

Set flow speed with validation and optional active state.

Parameters:

• speed (float)

• active (Optional[bool])

Return type:

None

property speed_config: SpeedConfig | None

Get speed configuration if it exists and is properly structured.

property type: FlowZoneType

Get the type of the flow zone.

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])

ZONE_TYPE = 'lighting'

__init__(zone_id, config)

Initialize LightingZone with zone_id and config.

Parameters:

• zone_id (str)

• config (Dict[str, Any])

activate()

Activate this zone.

Return type:

None

deactivate()

Deactivate this zone.

Return type:

None

get_lighting_state()

Get the current lighting state as a simple dictionary.

Return type:

Dict[str, Any]

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

Set lighting color.

Parameters:

• r (int)

• g (int)

• b (int)

• i (Optional[int])

Return type:

None

set_effect(effect_name)

Set lighting effect with validation.

Parameters:

effect_name (str)

Return type:

None

update_from_state(state)

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

Parameters:

state (Dict[str, Any])

Return type:

None

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__()

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

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

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

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

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

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]

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

class gecko_iot_client.OperationMode(*values)

Bases: Enum

Enum for operation modes (watercare modes).

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

AWAY = 0

STANDARD = 1

SAVINGS = 2

SUPER_SAVINGS = 3

WEEKENDER = 4

OTHER = 5

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

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

property mode_name: str

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

Returns:

String representation of the operation mode

to_dict()

Convert to dictionary representation.

Return type:

Dict[str, Any]

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

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.

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

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

property mode_name: str

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

Returns:

String representation of the operation mode

set_mode(mode)

Set the operation mode.

Parameters:

mode (OperationMode) – The operation mode to set

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)

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

Return type:

None

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

to_dict()

Convert to dictionary representation.

Return type:

Dict[str, Any]

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

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)

• api_url (str)

• auth0_url (str)

abstractmethod async async_get_access_token()

Return a valid access token for the Gecko API.

Return type:

str

async async_get_monitor_livestream(monitor_id)

Get MQTT livestream connection details for a monitor.

Parameters:

monitor_id (str)

Return type:

dict[str, Any]

async async_get_user_id()

Get user information from Auth0 or Gecko API.

Return type:

dict[str, Any]

async async_get_user_info(user_id)

Parameters:

user_id (str)

Return type:

dict[str, Any]

async async_get_vessels(account_id)

Get available vessels for the account.

Parameters:

account_id (str)

Return type:

list[dict[str, Any]]

async async_request(method, endpoint, **kwargs)

Make an authenticated request to the Gecko API.

Parameters:

• method (str)

• endpoint (str)

• kwargs (Any)

Return type:

Any

Models Package

Models package for GeckoIotClient.

class gecko_iot_client.models.TemperatureControlZoneStatus(*values)

Bases: Enum

Enum for temperature control zone status

property is_heating: bool

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

class gecko_iot_client.models.TemperatureControlMode(eco=False)

Bases: object

Temperature control mode configuration

Parameters:

eco (bool)

__init__(eco=False)

Initialize temperature control mode.

Parameters:

eco (bool)

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

Parameters:

• name (str)

• speed (float)

__init__(name, speed)

Parameters:

• name (str)

• speed (float)

name: str

speed: float

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

Bases: object

RGB color representation

Parameters:

• r (int)

• g (int)

• b (int)

• i (Optional[int])

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

Initialize RGB color with validation.

Parameters:

• r (int)

• g (int)

• b (int)

• i (Optional[int])

model_dump()

Convert to dictionary (replaces Pydantic’s model_dump).

Return type:

Dict[str, Any]

class gecko_iot_client.models.ZoneConfigurationParser

Bases: object

Simple parser for zone configurations.

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'>}

apply_state_to_zones(zones, state_data)

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

Parameters:

• zones (Dict[ZoneType, List[AbstractZone]])

• state_data (Dict[str, Any])

Return type:

None

parse_zones_configuration(zones_config)

Parse zones configuration into zone instances.

Parameters:

zones_config (Dict[str, Any])

Return type:

Dict[ZoneType, List[AbstractZone]]

Zone Types

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

Bases: Enum

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)

• zone_type (ZoneType)

• config (Any)

• name (Optional[str])

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)

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)

Return type:

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

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)

• validator (Callable[[Any], bool])

• error_message (str)

Return type:

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

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

Bases: Enum

Enum for temperature control zone status

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

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

Bases: object

Temperature control mode configuration

Parameters:

eco (bool)

__init__(eco=False)

Initialize temperature control mode.

Parameters:

eco (bool)

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)

• config (Dict[str, Any])

status_: Optional[TemperatureControlZoneStatus]

temperature_: Optional[float]

mode_: Optional[TemperatureControlMode]

set_point: Optional[float]

property status: TemperatureControlZoneStatus | None

property temperature: float | None

property mode: TemperatureControlMode | None

property target_temperature: float | None

property min_temperature_set_point_c_value: float | None

property max_temperature_set_point_c_value: float | None

__str__()

String representation of the TemperatureControlZone.

Return type:

str

set_target_temperature(temperature)

Set target temperature with validation against configured limits.

Parameters:

temperature (float)

Return type:

None

get_temperature_state()

Get the current temperature state as a simple dictionary.

Return type:

Dict[str, Any]

update_from_state(state)

Update temperature control zone from runtime state.

Parameters:

state (Dict[str, Any])

Return type:

None

ZONE_TYPE = 'temperatureControl'

class gecko_iot_client.models.flow_zone.SpeedConfig

Bases: TypedDict

maximum: int

minimum: int

stepIncrement: int

class gecko_iot_client.models.flow_zone.FlowConfiguration

Bases: TypedDict

name: Optional[str]

pumps: Optional[List[str]]

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

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

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)

• config (FlowConfiguration)

active: Optional[bool]

speed: Optional[float]

initiators_: Optional[List[FlowZoneInitiator]]

property speed_config: SpeedConfig | None

Get speed configuration if it exists and is properly structured.

property initiators: List[FlowZoneInitiator] | None

property type: FlowZoneType

Get the type of the flow zone.

property capabilities: List[FlowZoneCapabilities]

Get the capabilities of the flow zone.

property presets: List[FlowZonePreset]

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

get_flow_state()

Get the current flow state as a simple dictionary.

Return type:

Dict[str, Any]

set_speed(speed, active=True)

Set flow speed with validation and optional active state.

Parameters:

• speed (float)

• active (Optional[bool])

Return type:

None

activate()

Activate this zone.

Return type:

None

deactivate()

Deactivate this zone.

Return type:

None

ZONE_TYPE = 'flow'

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

Bases: object

RGB color representation

Parameters:

• r (int)

• g (int)

• b (int)

• i (Optional[int])

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

Initialize RGB color with validation.

Parameters:

• r (int)

• g (int)

• b (int)

• i (Optional[int])

model_dump()

Convert to dictionary (replaces Pydantic’s model_dump).

Return type:

Dict[str, Any]

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)

• config (Dict[str, Any])

active: Optional[bool]

rgbi: Optional[RGB]

effect: Optional[str]

get_lighting_state()

Get the current lighting state as a simple dictionary.

Return type:

Dict[str, Any]

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

Set lighting color.

Parameters:

• r (int)

• g (int)

• b (int)

• i (Optional[int])

Return type:

None

update_from_state(state)

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

Parameters:

state (Dict[str, Any])

Return type:

None

set_effect(effect_name)

Set lighting effect with validation.

Parameters:

effect_name (str)

Return type:

None

activate()

Activate this zone.

Return type:

None

deactivate()

Deactivate this 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.

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])

Return type:

Dict[ZoneType, List[AbstractZone]]

apply_state_to_zones(zones, state_data)

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

Parameters:

• zones (Dict[ZoneType, List[AbstractZone]])

• state_data (Dict[str, Any])

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

change_state(new_state)

Change state (placeholder for interface compliance).

connect(**kwargs)

Connect using preformatted WebSocket URL with expiration management.

disconnect()

Disconnect and cleanup.

is_connected()

Check if connected to broker.

Return type:

bool

load_configuration(timeout=30.0)

Load configuration from AWS IoT.

Parameters:

timeout (float)

load_state()

Load state from AWS IoT shadow.

on_configuration_loaded(callback)

Register config callback.

on_connectivity_change(callback)

Register connectivity change callback.

on_state_change(callback)

Register state change callback.

on_state_loaded(callback)

Register state callback.

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

publish_desired_state(desired_state)

Publish desired state update to AWS IoT shadow.

Parameters:

desired_state (Dict[str, Any])

Return type:

Future

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

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

clear_intentional_disconnect_flag()

Clear the intentional disconnect flag after reconnection.

Return type:

None

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

is_connected()

Check if connected to broker.

Return type:

bool

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

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

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

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: Optional[str] = None

error: Optional[Exception] = 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.