Skip to content
Back to top
View this page

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