aiokem package

class aiokem.AioKem(session: ClientSession, home_timezone: tzinfo = datetime.timezone.utc)

Bases: object

AioKem class for interacting with Kohler Energy Management System (KEM) API.

async authenticate(email: str, password: str, refresh_token: str | None = None) → None

Login to the server.

async authenticate_with_refresh_token(refresh_token: str) → None

Login to the server using a refresh token.

async check_and_refresh_token() → None

Check if the token is expired and refresh it if necessary.

async close() → None

Close the session.

async get_alerts(generator_id: int) → list[dict[str, Any]]

Get list of alerts for a generator.

async get_events(generator_id: int) → list[dict[str, Any]]

Get list of events for a generator.

async get_generator_data(generator_id: int) → dict[str, Any]

Get generator data for a specific generator.

async get_homeowner() → dict[str, Any]

Get homeowner information.

async get_homes() → list[dict[str, Any]]

Get the list of homes.

async get_maintenance_notes(generator_id: int) → list[dict[str, Any]]

Get list of maintenance_notes for a generator.

async get_notifications() → list[dict[str, Any]]

Get list of notifications.

get_token_subject() → str | None

Returns the subject of the JWT token, used as unique id for the user.

async on_refresh_token_update(refresh_token: str | None) → None

Execute the registered callback.

set_refresh_token_callback(callback: Callable[[str | None], Awaitable[None]]) → None

Set the callback for refresh token updates.

Args:

callback (callable): Callback function to be called when the refresh token updates. The function should accept a single argument, which is the new refresh token.

set_retry_policy(retry_count: int, retry_delays: list[int]) → None

Set the retry policy for the session.

Args:

retry_count (int): Number of retries. Zero means no retries. retry_delays (list[int]): Delay between retries in seconds for each retry.

set_timeout(timeout: int) → None

Set the timeout for the session.

Args:

timeout (int): Timeout in seconds.

exception aiokem.AioKemError

Bases: Exception

Base exception for the aiokem package.

exception aiokem.AuthenticationCredentialsError(message: str = 'Authentication failed')

Bases: AuthenticationError

Exception raised for authentication-related errors.

exception aiokem.AuthenticationError(message: str = 'Authentication failed')

Bases: AioKemError

Exception raised for authentication-related errors.

exception aiokem.CommunicationError(message: str = 'Communication error')

Bases: AioKemError

Exception raised for communication-related errors.

exception aiokem.ServerError(message: str = 'Server error')

Bases: AioKemError

Exception raised for server-related errors.

Submodules

aiokem.exceptions module

Exceptions for the aiokem package.

exception aiokem.exceptions.AioKemError

Bases: Exception

Base exception for the aiokem package.

exception aiokem.exceptions.AuthenticationCredentialsError(message: str = 'Authentication failed')

Bases: AuthenticationError

Exception raised for authentication-related errors.

exception aiokem.exceptions.AuthenticationError(message: str = 'Authentication failed')

Bases: AioKemError

Exception raised for authentication-related errors.

exception aiokem.exceptions.CommunicationError(message: str = 'Communication error')

Bases: AioKemError

Exception raised for communication-related errors.

exception aiokem.exceptions.ServerError(message: str = 'Server error')

Bases: AioKemError

Exception raised for server-related errors.

aiokem.helpers module

Helper functions for the aiokem package.

aiokem.helpers.convert_number_abs(response: dict[str, Any], key: str) → None

Convert a number to its absolute value.

aiokem.helpers.convert_timestamp(response: dict[str, Any], key: str, tz: tzinfo) → None

Convert a timestamp that does not have a tz in to the specified timezone.

aiokem.helpers.reverse_mac_address(mac: str) → str

Reverse the bytes of a MAC address.

aiokem.main module

AioKem class for interacting with Kohler Energy Management System (KEM) API.

class aiokem.main.AioKem(session: ClientSession, home_timezone: tzinfo = datetime.timezone.utc)

Bases: object

AioKem class for interacting with Kohler Energy Management System (KEM) API.

async authenticate(email: str, password: str, refresh_token: str | None = None) → None

Login to the server.

async authenticate_with_refresh_token(refresh_token: str) → None

Login to the server using a refresh token.

async check_and_refresh_token() → None

Check if the token is expired and refresh it if necessary.

async close() → None

Close the session.

async get_alerts(generator_id: int) → list[dict[str, Any]]

Get list of alerts for a generator.

async get_events(generator_id: int) → list[dict[str, Any]]

Get list of events for a generator.

async get_generator_data(generator_id: int) → dict[str, Any]

Get generator data for a specific generator.

async get_homeowner() → dict[str, Any]

Get homeowner information.

async get_homes() → list[dict[str, Any]]

Get the list of homes.

async get_maintenance_notes(generator_id: int) → list[dict[str, Any]]

Get list of maintenance_notes for a generator.

async get_notifications() → list[dict[str, Any]]

Get list of notifications.

get_token_subject() → str | None

Returns the subject of the JWT token, used as unique id for the user.

async on_refresh_token_update(refresh_token: str | None) → None

Execute the registered callback.

set_refresh_token_callback(callback: Callable[[str | None], Awaitable[None]]) → None

Set the callback for refresh token updates.

Args:

callback (callable): Callback function to be called when the refresh token updates. The function should accept a single argument, which is the new refresh token.

set_retry_policy(retry_count: int, retry_delays: list[int]) → None

Set the retry policy for the session.

Args:

retry_count (int): Number of retries. Zero means no retries. retry_delays (list[int]): Delay between retries in seconds for each retry.

set_timeout(timeout: int) → None

Set the timeout for the session.

Args:

timeout (int): Timeout in seconds.

aiokem.message_logger module

Module to log messages to files.

aiokem.message_logger.log_json_message(json_message: dict[str, Any] | list[dict[str, Any]]) → None

Logs a JSON message redacting sensitive information.

aiokem.message_logger.redact_fields(log_message: Any) → Any

Removes redacted fields from messages.