### Install project dependencies Source: https://github.com/kohlerlibs/aiokem/blob/main/CONTRIBUTING.md Install all necessary project dependencies using uv. This command synchronizes the environment with the project's requirements. ```shell $ uv sync ``` -------------------------------- ### Install aiokem Package Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/installation.md Use this command to install the aiokem package from PyPI. Ensure you have pip or an equivalent package installer available. ```bash pip install aiokem ``` -------------------------------- ### Install project dependencies Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/contributing.md Install all necessary project dependencies using uv. This command ensures your local environment is set up correctly for development. ```shell uv sync ``` -------------------------------- ### Complete Kohler Generator Monitoring Example Source: https://context7.com/kohlerlibs/aiokem/llms.txt A full example demonstrating continuous monitoring of Kohler generators. It includes setting timeouts, retry policies, refresh token callbacks, loading stored tokens, and handling generator data and alerts. ```python import asyncio import json import logging from pathlib import Path from aiohttp import ClientSession from aiokem import AioKem, AioKemError logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) TOKEN_FILE = Path("kem_tokens.json") async def save_token(refresh_token: str | None) -> None: if refresh_token: TOKEN_FILE.write_text(json.dumps({"refresh_token": refresh_token})) async def monitor_generator(email: str, password: str): async with ClientSession() as session: kem = AioKem(session=session) kem.set_timeout(30) kem.set_retry_policy(retry_count=3, retry_delays=[1, 2, 5]) kem.set_refresh_token_callback(save_token) # Load stored refresh token refresh_token = None if TOKEN_FILE.exists(): refresh_token = json.loads(TOKEN_FILE.read_text()).get("refresh_token") try: await kem.authenticate(email, password, refresh_token) homes = await kem.get_homes() while True: for home in homes: for device in home['devices']: generator_id = device['id'] data = await kem.get_generator_data(generator_id) logger.info( f"{device['displayName']}: " f"Status={data['engineState']}, " f"Utility={data['utilityVoltageV']}V, " f"Load={data['generatorLoadW']}W" ) # Check for alerts if data['device']['alertCount'] > 0: alerts = await kem.get_alerts(generator_id) for alert in alerts: logger.warning(f"ALERT: {alert['name']} - {alert['event']}") await asyncio.sleep(60) except AioKemError as e: logger.error(f"Error: {e}") finally: await kem.close() if __name__ == "__main__": asyncio.run(monitor_generator("user@example.com", "password")) ``` -------------------------------- ### Install pre-commit hooks Source: https://github.com/kohlerlibs/aiokem/blob/main/CONTRIBUTING.md Install pre-commit hooks to automatically run linters and formatters on code changes before each commit. This ensures consistent code style. ```shell $ pre-commit install ``` -------------------------------- ### Install pre-commit hooks Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/contributing.md Install pre-commit hooks to automatically run checks on each commit. This helps maintain code quality throughout the development process. ```shell pre-commit install ``` -------------------------------- ### GET /homes Source: https://context7.com/kohlerlibs/aiokem/llms.txt Retrieves the list of homes associated with the authenticated user, including generator devices at each location. ```APIDOC ## GET /homes ### Description Retrieves the list of homes associated with the authenticated user, including generator devices at each location. Each home contains device information such as serial numbers, connection status, alert counts, and maintenance schedules. ### Method GET ### Response #### Success Response (200) - **name** (string) - Home name - **weatherCondition** (string) - Current weather condition - **weatherTempF** (number) - Current temperature in Fahrenheit - **address** (object) - Address details including city and state - **devices** (array) - List of generator devices associated with the home ``` -------------------------------- ### GET /generator/{id} Source: https://context7.com/kohlerlibs/aiokem/llms.txt Fetches detailed real-time data for a specific generator including voltage readings, engine state, temperature sensors, load information, exercise schedule, and load shedding parameters. ```APIDOC ## GET /generator/{id} ### Description Fetches detailed real-time data for a specific generator including voltage readings, engine state, temperature sensors, load information, exercise schedule, and load shedding parameters. ### Method GET ### Parameters #### Path Parameters - **id** (string) - Required - The unique identifier of the generator ### Response #### Success Response (200) - **utilityVoltageV** (number) - Utility voltage in Volts - **generatorVoltageAvgV** (number) - Average generator voltage in Volts - **generatorLoadW** (number) - Generator load in Watts - **generatorLoadPercent** (number) - Generator load percentage - **powerSource** (string) - Current power source - **switchState** (string) - Current switch state - **engineState** (string) - Current engine state - **engineSpeedRpm** (number) - Engine speed in RPM - **engineFrequencyHz** (number) - Engine frequency in Hz - **batteryVoltageV** (number) - Battery voltage in Volts - **controllerTempF** (number) - Controller temperature in Fahrenheit - **lubeOilTempF** (number) - Lube oil temperature in Fahrenheit - **totalRuntimeHours** (number) - Total runtime in hours - **runtimeSinceLastMaintenanceHours** (number) - Hours since last maintenance - **fuelType** (string) - Fuel type - **exercise** (object) - Exercise schedule details - **loadShed** (object) - Load shedding parameters ``` -------------------------------- ### GET /homeowner Source: https://context7.com/kohlerlibs/aiokem/llms.txt Retrieves the authenticated homeowner's profile information including email, name, device serial numbers, and account preferences. ```APIDOC ## GET /homeowner ### Description Retrieves the authenticated homeowner's profile information including email, name, device serial numbers, and account preferences. ### Method GET ### Response #### Success Response (200) - **id** (integer) - Homeowner ID - **email** (string) - User email address - **firstName** (string) - User first name - **lastName** (string) - User last name - **hasAcceptedTC** (boolean) - Terms and conditions acceptance status - **deviceSerialNumbers** (array) - List of device serial numbers - **hasApprovedDevices** (boolean) - Status of approved devices - **hasAppAccount** (boolean) - Status of app account - **hasAcceptedMarketingOutreach** (boolean) - Marketing outreach preference #### Response Example { "id": 12345, "email": "john.doe@gmail.com", "firstName": "John", "lastName": "Doe", "hasAcceptedTC": true, "deviceSerialNumbers": ["SGV123456"], "hasApprovedDevices": true, "hasAppAccount": true, "hasAcceptedMarketingOutreach": true } ``` -------------------------------- ### Get JWT Access Token Subject Source: https://context7.com/kohlerlibs/aiokem/llms.txt Retrieves the subject claim from the JWT access token. Useful for identifying the authenticated user for caching or session management. Requires prior authentication. ```python import asyncio from aiohttp import ClientSession from aiokem import AioKem, AuthenticationError async def main(): async with ClientSession() as session: kem = AioKem(session=session) await kem.authenticate("user@example.com", "password") try: user_id = kem.get_token_subject() print(f"User ID (JWT subject): {user_id}") except AuthenticationError: print("Not authenticated") asyncio.run(main()) ``` -------------------------------- ### Initialize AioKem Class Source: https://context7.com/kohlerlibs/aiokem/llms.txt Initialize the AioKem client with an aiohttp session, optional timezone, and custom retry or timeout configurations. ```python import asyncio from datetime import timezone from zoneinfo import ZoneInfo from aiohttp import ClientSession from aiokem import AioKem async def main(): async with ClientSession() as session: # Initialize with default UTC timezone kem = AioKem(session=session) # Or with a specific timezone for local timestamps eastern = ZoneInfo("America/New_York") kem = AioKem(session=session, home_timezone=eastern) # Configure timeout (default is 20 seconds) kem.set_timeout(30) # Configure retry policy: 2 retries with 1s and 2s delays kem.set_retry_policy(retry_count=2, retry_delays=[1, 2]) asyncio.run(main()) ``` -------------------------------- ### Run all pre-commit hooks Source: https://github.com/kohlerlibs/aiokem/blob/main/CONTRIBUTING.md Execute all pre-commit hooks manually to check code quality and style across the entire project. This is useful for a one-off check before committing. ```shell $ pre-commit run -a ``` -------------------------------- ### Run all pre-commit hooks Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/contributing.md Execute all pre-commit hooks to ensure code quality and style consistency. This is a one-off command to check all hooks. ```shell pre-commit run -a ``` -------------------------------- ### Clone the aiokem repository Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/contributing.md Clone your forked repository locally to begin development. Replace 'your_name_here' with your GitHub username. ```shell git clone git@github.com:your_name_here/aiokem.git ``` -------------------------------- ### Clone the aiokem repository Source: https://github.com/kohlerlibs/aiokem/blob/main/CONTRIBUTING.md Clone your forked repository locally to begin development. Ensure you replace 'your_name_here' with your GitHub username. ```shell $ git clone git@github.com:your_name_here/aiokem.git ``` -------------------------------- ### Import aiokem Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/usage.md Initial import statement required to access the library functionality. ```python import aiokem ``` -------------------------------- ### List Homes and Devices with aiokem Source: https://context7.com/kohlerlibs/aiokem/llms.txt Retrieves all homes associated with the account and iterates through their connected generator devices. Useful for identifying device IDs for further queries. ```python import asyncio from aiohttp import ClientSession from aiokem import AioKem async def main(): async with ClientSession() as session: kem = AioKem(session=session) await kem.authenticate("user@example.com", "password") homes = await kem.get_homes() for home in homes: print(f"Home: {home['name']}") print(f"Weather: {home['weatherCondition']}, {home['weatherTempF']}F") print(f"Address: {home['address']['city']}, {home['address']['state']}") for device in home['devices']: print(f" Generator: {device['displayName']}") print(f" Serial: {device['serialNumber']}") print(f" Model: {device['modelDisplayName']}") print(f" Status: {device['status']}") print(f" Connected: {device['isConnected']}") print(f" Alert Count: {device['alertCount']}") print(f" Runtime Hours: {device['totalRuntimeHours']}") print(f" Next Maintenance: {device['nextMaintenanceTimestamp']}") # Store generator ID for detailed data retrieval generator_id = device['id'] asyncio.run(main()) ``` -------------------------------- ### Run a subset of tests Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/contributing.md Execute a specific subset of tests within the project. This is useful for quickly testing changes in a particular area. ```shell pytest tests ``` -------------------------------- ### Commit and push changes Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/contributing.md Stage all changes, commit them with a descriptive message following conventional commits, and push the branch to GitHub. The commit message format is validated by commitlint. ```shell git add . git commit -m "feat(something): your detailed description of your changes" git push origin name-of-your-bugfix-or-feature ``` -------------------------------- ### Commit and push changes Source: https://github.com/kohlerlibs/aiokem/blob/main/CONTRIBUTING.md Stage all changes, commit them with a conventional commit message, and push the branch to your GitHub repository. The commit message format is validated by commitlint. ```shell $ git add . $ git commit -m "feat(something): your detailed description of your changes" $ git push origin name-of-your-bugfix-or-feature ``` -------------------------------- ### Run a subset of tests Source: https://github.com/kohlerlibs/aiokem/blob/main/CONTRIBUTING.md Execute a specific subset of tests within the project's test suite. This is useful for quickly testing changes in a particular area without running all tests. ```shell $ pytest tests ``` -------------------------------- ### Retrieve Homeowner Profile with aiokem Source: https://context7.com/kohlerlibs/aiokem/llms.txt Fetches the authenticated user's profile, including contact details and device serial numbers. Requires an active authenticated session. ```python import asyncio from aiohttp import ClientSession from aiokem import AioKem async def main(): async with ClientSession() as session: kem = AioKem(session=session) await kem.authenticate("user@example.com", "password") homeowner = await kem.get_homeowner() print(f"Name: {homeowner['firstName']} {homeowner['lastName']}") print(f"Email: {homeowner['email']}") print(f"Device Serial Numbers: {homeowner['deviceSerialNumbers']}") print(f"Has App Account: {homeowner['hasAppAccount']}") # Example response: # { # "id": 12345, # "email": "john.doe@gmail.com", # "firstName": "John", # "lastName": "Doe", # "hasAcceptedTC": true, # "deviceSerialNumbers": ["SGV123456"], # "hasApprovedDevices": true, # "hasAppAccount": true, # "hasAcceptedMarketingOutreach": true # } asyncio.run(main()) ``` -------------------------------- ### Create a pull request Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/contributing.md Create a pull request on GitHub to merge your changes. The --fill flag attempts to pre-populate the PR details from your commit. ```shell gh pr create --fill ``` -------------------------------- ### Handle aiokem Exceptions Source: https://context7.com/kohlerlibs/aiokem/llms.txt Demonstrates comprehensive exception handling for various error conditions in aiokem, including authentication, communication, and server errors. It's recommended to catch specific exceptions before the general AioKemError. ```python import asyncio from aiohttp import ClientSession from aiokem import ( AioKem, AioKemError, AuthenticationError, AuthenticationCredentialsError, CommunicationError, ServerError, ) async def main(): async with ClientSession() as session: kem = AioKem(session=session) kem.set_retry_policy(retry_count=2, retry_delays=[1, 2]) try: await kem.authenticate("user@example.com", "password") data = await kem.get_generator_data(12345) except AuthenticationCredentialsError as e: # Invalid username or password print(f"Invalid credentials: {e}") except AuthenticationError as e: # General authentication failure (token expired, unauthorized) print(f"Authentication error: {e}") except CommunicationError as e: # Network issues, timeouts, connection errors print(f"Communication error: {e}") except ServerError as e: # Server-side errors (500, missing data in response) print(f"Server error: {e}") except AioKemError as e: # Catch-all for any aiokem exception print(f"AioKem error: {e}") finally: await kem.close() asyncio.run(main()) ``` -------------------------------- ### AioKem Class Methods Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/aiokem.md This section details the methods available within the AioKem class for interacting with the KEM API. ```APIDOC ## POST /api/authenticate ### Description Logs into the KEM server using email and password, optionally with a refresh token. ### Method POST ### Endpoint /api/authenticate ### Parameters #### Request Body - **email** (str) - Required - The user's email address. - **password** (str) - Required - The user's password. - **refresh_token** (str | None) - Optional - A refresh token for authentication. ### Response #### Success Response (200) Returns None upon successful authentication. #### Response Example null ``` ```APIDOC ## POST /api/authenticate_with_refresh_token ### Description Logs into the KEM server using a provided refresh token. ### Method POST ### Endpoint /api/authenticate_with_refresh_token ### Parameters #### Request Body - **refresh_token** (str) - Required - The refresh token for authentication. ### Response #### Success Response (200) Returns None upon successful authentication. #### Response Example null ``` ```APIDOC ## GET /api/check_and_refresh_token ### Description Checks if the current authentication token is expired and refreshes it if necessary. ### Method GET ### Endpoint /api/check_and_refresh_token ### Response #### Success Response (200) Returns None if the token is valid or successfully refreshed. #### Response Example null ``` ```APIDOC ## POST /api/close ### Description Closes the API session. ### Method POST ### Endpoint /api/close ### Response #### Success Response (200) Returns None upon successful session closure. #### Response Example null ``` ```APIDOC ## GET /api/generators/{generator_id}/alerts ### Description Retrieves a list of alerts for a specific generator. ### Method GET ### Endpoint /api/generators/{generator_id}/alerts ### Parameters #### Path Parameters - **generator_id** (int) - Required - The ID of the generator. ### Response #### Success Response (200) - **alerts** (list[dict[str, Any]]) - A list of alert dictionaries. #### Response Example { "alerts": [ { "alert_id": 1, "message": "High temperature detected", "timestamp": "2023-10-27T10:00:00Z" } ] } ``` ```APIDOC ## GET /api/generators/{generator_id}/events ### Description Retrieves a list of events for a specific generator. ### Method GET ### Endpoint /api/generators/{generator_id}/events ### Parameters #### Path Parameters - **generator_id** (int) - Required - The ID of the generator. ### Response #### Success Response (200) - **events** (list[dict[str, Any]]) - A list of event dictionaries. #### Response Example { "events": [ { "event_id": 101, "type": "startup", "timestamp": "2023-10-27T09:00:00Z" } ] } ``` ```APIDOC ## GET /api/generators/{generator_id}/data ### Description Retrieves detailed data for a specific generator. ### Method GET ### Endpoint /api/generators/{generator_id}/data ### Parameters #### Path Parameters - **generator_id** (int) - Required - The ID of the generator. ### Response #### Success Response (200) - **generator_data** (dict[str, Any]) - A dictionary containing the generator's data. #### Response Example { "generator_id": 1, "status": "running", "load": 50.5, "voltage": 240.0 } ``` ```APIDOC ## GET /api/homeowner ### Description Retrieves information about the homeowner. ### Method GET ### Endpoint /api/homeowner ### Response #### Success Response (200) - **homeowner_info** (dict[str, Any]) - A dictionary containing homeowner details. #### Response Example { "name": "John Doe", "email": "john.doe@example.com", "phone": "123-456-7890" } ``` ```APIDOC ## GET /api/homes ### Description Retrieves a list of homes associated with the account. ### Method GET ### Endpoint /api/homes ### Response #### Success Response (200) - **homes** (list[dict[str, Any]]) - A list of home dictionaries. #### Response Example [ { "home_id": 1, "address": "123 Main St" }, { "home_id": 2, "address": "456 Oak Ave" } ] ``` ```APIDOC ## GET /api/generators/{generator_id}/maintenance_notes ### Description Retrieves a list of maintenance notes for a specific generator. ### Method GET ### Endpoint /api/generators/{generator_id}/maintenance_notes ### Parameters #### Path Parameters - **generator_id** (int) - Required - The ID of the generator. ### Response #### Success Response (200) - **maintenance_notes** (list[dict[str, Any]]) - A list of maintenance note dictionaries. #### Response Example { "maintenance_notes": [ { "note_id": 5, "date": "2023-10-20", "notes": "Performed oil change." } ] } ``` ```APIDOC ## GET /api/notifications ### Description Retrieves a list of user notifications. ### Method GET ### Endpoint /api/notifications ### Response #### Success Response (200) - **notifications** (list[dict[str, Any]]) - A list of notification dictionaries. #### Response Example [ { "notification_id": 201, "message": "Generator maintenance due soon.", "timestamp": "2023-10-26T15:30:00Z" } ] ``` ```APIDOC ## GET /api/token_subject ### Description Retrieves the subject of the current JWT token, used as a unique user identifier. ### Method GET ### Endpoint /api/token_subject ### Response #### Success Response (200) - **token_subject** (str | None) - The subject of the JWT token, or None if not available. #### Response Example "user-12345abc" ``` ```APIDOC ## POST /api/on_refresh_token_update ### Description Executes the registered callback function when the refresh token is updated. ### Method POST ### Endpoint /api/on_refresh_token_update ### Parameters #### Request Body - **refresh_token** (str | None) - Required - The new refresh token, or None if it's cleared. ### Response #### Success Response (200) Returns None upon successful execution of the callback. #### Response Example null ``` ```APIDOC ## POST /api/set_refresh_token_callback ### Description Sets a callback function to be executed when the refresh token is updated. ### Method POST ### Endpoint /api/set_refresh_token_callback ### Parameters #### Request Body - **callback** (Callable[[str | None], Awaitable[None]]) - Required - The callback function. It accepts one argument: the new refresh token (str or None). ### Response #### Success Response (200) Returns None upon successful registration of the callback. #### Response Example null ``` ```APIDOC ## POST /api/set_retry_policy ### Description Configures the retry policy for API requests. ### Method POST ### Endpoint /api/set_retry_policy ### Parameters #### Request Body - **retry_count** (int) - Required - The number of retries. Zero means no retries. - **retry_delays** (list[int]) - Required - A list of integers representing the delay in seconds between retries. ### Response #### Success Response (200) Returns None upon successful configuration of the retry policy. #### Response Example null ``` ```APIDOC ## POST /api/set_timeout ### Description Sets the timeout duration for API requests. ### Method POST ### Endpoint /api/set_timeout ### Parameters #### Request Body - **timeout** (int) - Required - The timeout duration in seconds. ### Response #### Success Response (200) Returns None upon successful setting of the timeout. #### Response Example null ``` -------------------------------- ### Run project tests Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/contributing.md Execute the project's test suite using pytest. This command verifies that your changes have not introduced regressions. ```shell uv run pytest ``` -------------------------------- ### Create a new branch for development Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/contributing.md Create a new branch for your bug fix or feature. This isolates your changes and makes them easier to manage. ```shell git checkout -b name-of-your-bugfix-or-feature ``` -------------------------------- ### Home and User Information Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/aiokem.md Methods for retrieving homeowner details and the list of associated homes. ```APIDOC ## GET /homes ### Description Get the list of homes. ### Method GET ### Response #### Success Response (200) - **homes** (list) - List of homes. ## GET /homeowner ### Description Get homeowner information. ### Method GET ### Response #### Success Response (200) - **homeowner** (dict) - Homeowner information. ``` -------------------------------- ### Register Refresh Token Callback Source: https://context7.com/kohlerlibs/aiokem/llms.txt Register a callback to persist refresh tokens to disk, enabling seamless re-authentication across sessions. ```python import asyncio import json from pathlib import Path from aiohttp import ClientSession from aiokem import AioKem TOKEN_FILE = Path("tokens.json") async def save_refresh_token(refresh_token: str | None) -> None: """Callback to persist refresh token to disk.""" if refresh_token: TOKEN_FILE.write_text(json.dumps({"refresh_token": refresh_token})) print(f"Refresh token saved") async def main(): async with ClientSession() as session: kem = AioKem(session=session) # Register callback for token updates kem.set_refresh_token_callback(save_refresh_token) # Load stored refresh token if available refresh_token = None if TOKEN_FILE.exists(): data = json.loads(TOKEN_FILE.read_text()) refresh_token = data.get("refresh_token") # Authenticate - will use refresh token if valid, fallback to password await kem.authenticate( email="user@example.com", password="your_password", refresh_token=refresh_token ) # Callback will be invoked with new refresh token asyncio.run(main()) ``` -------------------------------- ### Create a pull request Source: https://github.com/kohlerlibs/aiokem/blob/main/CONTRIBUTING.md Create a pull request on GitHub to merge your changes into the main branch. The '--fill' flag attempts to automatically populate the PR title and body from the commit. ```shell $ gh pr create --fill ``` -------------------------------- ### Run tests with pytest Source: https://github.com/kohlerlibs/aiokem/blob/main/CONTRIBUTING.md Execute the project's test suite using pytest to ensure your changes are working correctly and haven't introduced regressions. This command runs all tests defined in the project. ```shell $ uv run pytest ``` -------------------------------- ### Create a new branch for development Source: https://github.com/kohlerlibs/aiokem/blob/main/CONTRIBUTING.md Create a new branch for your bug fix or feature development. This helps in organizing changes and managing contributions. ```shell $ git checkout -b name-of-your-bugfix-or-feature ``` -------------------------------- ### Fetch Detailed Generator Data with aiokem Source: https://context7.com/kohlerlibs/aiokem/llms.txt Retrieves real-time telemetry, engine state, and maintenance data for a specific generator. Requires a valid generator ID obtained from the homes list. ```python import asyncio from aiohttp import ClientSession from aiokem import AioKem async def main(): async with ClientSession() as session: kem = AioKem(session=session) await kem.authenticate("user@example.com", "password") # Get generator ID from homes list first homes = await kem.get_homes() generator_id = homes[0]['devices'][0]['id'] data = await kem.get_generator_data(generator_id) # Power and voltage information print(f"Utility Voltage: {data['utilityVoltageV']}V") print(f"Generator Voltage: {data['generatorVoltageAvgV']}V") print(f"Generator Load: {data['generatorLoadW']}W ({data['generatorLoadPercent']}%)") print(f"Power Source: {data['powerSource']}") print(f"Switch State: {data['switchState']}") # Engine state print(f"Engine State: {data['engineState']}") print(f"Engine Speed: {data['engineSpeedRpm']} RPM") print(f"Engine Frequency: {data['engineFrequencyHz']} Hz") # Temperatures print(f"Battery Voltage: {data['batteryVoltageV']}V") print(f"Controller Temp: {data['controllerTempF']}F") print(f"Lube Oil Temp: {data['lubeOilTempF']}F") # Runtime and maintenance print(f"Total Runtime: {data['totalRuntimeHours']} hours") print(f"Hours Since Maintenance: {data['runtimeSinceLastMaintenanceHours']}") print(f"Fuel Type: {data['fuelType']}") # Exercise schedule exercise = data['exercise'] print(f"Exercise Frequency: {exercise['frequency']}") print(f"Next Exercise: {exercise['nextStartTimestamp']}") print(f"Exercise Duration: {exercise['durationMinutes']} minutes") # Load shedding parameters if data['loadShed']['isConnected']: print("Load Shed Parameters:") for param in data['loadShed']['parameters']: print(f" {param['displayName']}: {param['value']}") asyncio.run(main()) ``` -------------------------------- ### Authenticate with KEM API Source: https://context7.com/kohlerlibs/aiokem/llms.txt Authenticate using email and password, with support for providing a refresh token for faster re-authentication. ```python import asyncio from aiohttp import ClientSession from aiokem import AioKem, AuthenticationCredentialsError, AuthenticationError async def main(): async with ClientSession() as session: kem = AioKem(session=session) try: # Basic authentication with email and password await kem.authenticate( email="user@example.com", password="your_password" ) print("Authentication successful!") # Or with a stored refresh token for faster re-authentication await kem.authenticate( email="user@example.com", password="your_password", refresh_token="stored_refresh_token_from_previous_session" ) except AuthenticationCredentialsError as e: print(f"Invalid credentials: {e}") except AuthenticationError as e: print(f"Authentication failed: {e}") asyncio.run(main()) ``` -------------------------------- ### Helper Functions Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/aiokem.md Utility functions provided by the aiokem.helpers module. ```APIDOC ## POST /api/helpers/convert_number_abs ### Description Converts a numerical value associated with a specific key in a dictionary to its absolute value. ### Method POST ### Endpoint /api/helpers/convert_number_abs ### Parameters #### Request Body - **response** (dict[str, Any]) - Required - The dictionary containing the number. - **key** (str) - Required - The key associated with the number to convert. ### Response #### Success Response (200) Returns None. The conversion happens in-place on the provided dictionary. #### Response Example null ``` ```APIDOC ## POST /api/helpers/convert_timestamp ### Description Converts a timestamp in a dictionary to the specified timezone, assuming it lacks timezone information. ### Method POST ### Endpoint /api/helpers/convert_timestamp ### Parameters #### Request Body - **response** (dict[str, Any]) - Required - The dictionary containing the timestamp. - **key** (str) - Required - The key associated with the timestamp to convert. - **tz** (tzinfo) - Required - The target timezone to convert the timestamp to. ### Response #### Success Response (200) Returns None. The conversion happens in-place on the provided dictionary. #### Response Example null ``` ```APIDOC ## POST /api/helpers/reverse_mac_address ### Description Reverses the byte order of a MAC address string. ### Method POST ### Endpoint /api/helpers/reverse_mac_address ### Parameters #### Request Body - **mac** (str) - Required - The MAC address string to reverse. ### Response #### Success Response (200) - **reversed_mac** (str) - The MAC address with its bytes reversed. #### Response Example { "reversed_mac": "01:02:03:04:05:06" } ``` -------------------------------- ### Authentication and Session Management Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/aiokem.md Methods for managing authentication tokens, session timeouts, and retry policies. ```APIDOC ## POST /auth/refresh ### Description Login to the server using a refresh token. ### Method POST ### Request Body - **refresh_token** (str) - Required - The refresh token string. ## POST /session/timeout ### Description Set the timeout for the session. ### Method POST ### Request Body - **timeout** (int) - Required - Timeout in seconds. ## POST /session/retry_policy ### Description Set the retry policy for the session. ### Method POST ### Request Body - **retry_count** (int) - Required - Number of retries. - **retry_delays** (list) - Required - Delay between retries in seconds. ``` -------------------------------- ### Retrieve generator maintenance notes with aiokem Source: https://context7.com/kohlerlibs/aiokem/llms.txt Fetches maintenance records logged by technicians for a specific generator. ```python import asyncio from aiohttp import ClientSession from aiokem import AioKem async def main(): async with ClientSession() as session: kem = AioKem(session=session) await kem.authenticate("user@example.com", "password") homes = await kem.get_homes() generator_id = homes[0]['devices'][0]['id'] notes = await kem.get_maintenance_notes(generator_id) if notes: for note in notes: print(f"Maintenance Note: {note}") else: print("No maintenance notes recorded") asyncio.run(main()) ``` -------------------------------- ### Retrieve generator alerts with aiokem Source: https://context7.com/kohlerlibs/aiokem/llms.txt Fetches active warnings or faults for a specific generator. Requires checking the alertCount from generator data before retrieval. ```python import asyncio from aiohttp import ClientSession from aiokem import AioKem async def main(): async with ClientSession() as session: kem = AioKem(session=session) await kem.authenticate("user@example.com", "password") homes = await kem.get_homes() generator_id = homes[0]['devices'][0]['id'] # Check if there are alerts data = await kem.get_generator_data(generator_id) if data['device']['alertCount'] > 0: alerts = await kem.get_alerts(generator_id) for alert in alerts: print(f"Alert: {alert['name']}") print(f" Type: {alert['type']}") print(f" Level: {alert['level']}") print(f" Event: {alert['event']}") print(f" FMI: {alert['fmi']}") print(f" Time: {alert['timestamp']}") print(f" Critical: {alert['isCriticalFault']}") print(f" Can Remote Reset: {alert['canRemoteResetFault']}") # Example alert response: # { # "id": -1658684129, # "name": "Not Connected", # "level": "None", # "event": "Not Connected", # "fmi": "CriticallyHigh", # "type": "notice", # "timestamp": "2025-06-01T23:14:36+00:00", # "isCriticalFault": false, # "canRemoteResetFault": null # } asyncio.run(main()) ``` -------------------------------- ### Retrieve account notifications with aiokem Source: https://context7.com/kohlerlibs/aiokem/llms.txt Fetches account-wide notifications across all devices, including warnings and status updates. ```python import asyncio from aiohttp import ClientSession from aiokem import AioKem async def main(): async with ClientSession() as session: kem = AioKem(session=session) await kem.authenticate("user@example.com", "password") notifications = await kem.get_notifications() print(f"You have {len(notifications)} notifications:") for notification in notifications: print(f"[{notification['type']}] {notification['displayName']}") print(f" Message: {notification['message']}") print(f" Time: {notification['timestamp']}") print(f" Device ID: {notification['deviceId']}") print() # Example notification types: # - EngineStartedNotice # - EngineStoppedNotice # - ExerciseStartedNotice # - ExerciseEndedNotice # - Warning asyncio.run(main()) ``` -------------------------------- ### Retrieve generator event history with aiokem Source: https://context7.com/kohlerlibs/aiokem/llms.txt Fetches historical events such as engine status changes and exercises for a specific generator. ```python import asyncio from aiohttp import ClientSession from aiokem import AioKem async def main(): async with ClientSession() as session: kem = AioKem(session=session) await kem.authenticate("user@example.com", "password") homes = await kem.get_homes() generator_id = homes[0]['devices'][0]['id'] events = await kem.get_events(generator_id) print(f"Found {len(events)} events:") for event in events[:10]: # Show first 10 events print(f"Event: {event['name']}") print(f" Level: {event['level']}") print(f" Type: {event['type']}") print(f" Time: {event['timestamp']}") if event['parameter']: print(f" Parameter: {event['parameter']}") print() # Example events: # - "Engine Started Status" (type: runningNotice) # - "Engine Stopped Status" (type: notice) # - "Diagnostic Exercise Active Status" (type: notice) # - "Ats Exercise Ended Status" (type: notice) # - "Generator Started And Ready To Provide Power Status" (type: notice) asyncio.run(main()) ``` -------------------------------- ### Custom Exceptions Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/aiokem.md Defines the custom exceptions used within the aiokem package for error handling. ```APIDOC ## Exception: aiokem.AioKemError ### Description Base exception for all errors raised by the aiokem package. ### Type Exception ``` ```APIDOC ## Exception: aiokem.AuthenticationCredentialsError ### Description Raised when authentication credentials are invalid or missing. ### Bases aiokem.AuthenticationError ### Type Exception ``` ```APIDOC ## Exception: aiokem.AuthenticationError ### Description Raised for general authentication-related errors. ### Bases aiokem.AioKemError ### Type Exception ``` ```APIDOC ## Exception: aiokem.CommunicationError ### Description Raised when there is an issue communicating with the KEM API server. ### Bases aiokem.AioKemError ### Type Exception ``` ```APIDOC ## Exception: aiokem.ServerError ### Description Raised when the KEM API server returns an error response. ### Bases aiokem.AioKemError ### Type Exception ``` -------------------------------- ### Generator Data Retrieval Source: https://github.com/kohlerlibs/aiokem/blob/main/docs/aiokem.md Methods for retrieving information related to generators, including data, alerts, events, and maintenance notes. ```APIDOC ## GET /generator/{generator_id}/data ### Description Get generator data for a specific generator. ### Method GET ### Parameters #### Path Parameters - **generator_id** (int) - Required - The unique identifier for the generator. ### Response #### Success Response (200) - **data** (dict) - Generator information. ## GET /generator/{generator_id}/alerts ### Description Get list of alerts for a generator. ### Method GET ### Parameters #### Path Parameters - **generator_id** (int) - Required - The unique identifier for the generator. ### Response #### Success Response (200) - **alerts** (list) - List of alerts. ## GET /generator/{generator_id}/events ### Description Get list of events for a generator. ### Method GET ### Parameters #### Path Parameters - **generator_id** (int) - Required - The unique identifier for the generator. ### Response #### Success Response (200) - **events** (list) - List of events. ## GET /generator/{generator_id}/maintenance_notes ### Description Get list of maintenance_notes for a generator. ### Method GET ### Parameters #### Path Parameters - **generator_id** (int) - Required - The unique identifier for the generator. ### Response #### Success Response (200) - **maintenance_notes** (list) - List of maintenance notes. ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.