aiorinnai - Python interface for the Rinnai Control-R API

Python library for communicating with the Rinnai Control-R Water Heaters and control devices via the Rinnai Control-R cloud API.

WARNING

• This library only works if you have migrated to the Rinnai 2.0 app. This will require a firmware update to your Control-R module.
• iOS
• Android

Note: This library is community supported. Contributions and improvements are welcome!

Features

• Secure Authentication - AWS Cognito-based authentication with automatic token refresh
• Temperature Control - Set temperature in Fahrenheit or Celsius with automatic conversion
• Recirculation Control - Start/stop recirculation pump with configurable duration (1-60 minutes)
• Device Management - Turn water heater on/off, enable/disable vacation mode
• Maintenance - Trigger maintenance data retrieval
• Input Validation - Built-in validation for temperature (100-140°F) and duration ranges
• Type Safety - Full type hints with TypedDict definitions for API responses
• Async/Await - Built on aiohttp for efficient async operations
• Retry Logic - Configurable retry with exponential backoff for transient errors

Installation

Requires Python 3.11 or higher.

pip install aiorinnai

Quick Start

import asyncio
from aiorinnai import API


async def main() -> None:
    async with API() as api:
        await api.async_login("<EMAIL>", "<PASSWORD>")

        user_info = await api.user.get_info()
        device = user_info["devices"]["items"][0]

        # Set temperature and start recirculation
        await api.device.set_temperature(device, 120)
        await api.device.start_recirculation(device, duration=5)


asyncio.run(main())

Usage Examples

Basic Example

import asyncio
from aiorinnai import API


async def main() -> None:
    """Run!"""
    async with API() as api:
        # Authenticate with your Rinnai account
        await api.async_login("<EMAIL>", "<PASSWORD>")

        # Get user account information (includes all devices)
        user_info = await api.user.get_info()

        # Get the first device
        device = user_info["devices"]["items"][0]
        print(f"Device: {device['device_name']}")

        # Get detailed device information
        device_info = await api.device.get_info(device["id"])

        # Start recirculation for 5 minutes
        response = await api.device.start_recirculation(device, 5)
        if response.success:
            print("Recirculation started!")

        # Stop recirculation
        await api.device.stop_recirculation(device)

        # Set temperature (100-140°F, increments of 5)
        await api.device.set_temperature(device, 120)

        # Turn water heater off
        await api.device.turn_off(device)

        # Turn water heater on
        await api.device.turn_on(device)

        # Enable vacation mode
        await api.device.enable_vacation_mode(device)

        # Disable vacation mode
        await api.device.disable_vacation_mode(device)


asyncio.run(main())

Temperature Unit Support

Set temperature in Fahrenheit (default) or Celsius:

import asyncio
from aiorinnai import API, TemperatureUnit


async def main() -> None:
    async with API() as api:
        await api.async_login("<EMAIL>", "<PASSWORD>")
        user_info = await api.user.get_info()
        device = user_info["devices"]["items"][0]

        # Set temperature in Fahrenheit (default)
        await api.device.set_temperature(device, 120)

        # Set temperature in Celsius (auto-converts to Fahrenheit)
        await api.device.set_temperature(device, 49, TemperatureUnit.CELSIUS)

        # Temperature conversion helpers
        fahrenheit = TemperatureUnit.celsius_to_fahrenheit(49)  # Returns 120
        celsius = TemperatureUnit.fahrenheit_to_celsius(120)    # Returns 48.89


asyncio.run(main())

Handling API Responses

All device commands return an APIResponse object for consistent error handling:

import asyncio
from aiorinnai import API, APIResponse


async def main() -> None:
    async with API() as api:
        await api.async_login("<EMAIL>", "<PASSWORD>")
        user_info = await api.user.get_info()
        device = user_info["devices"]["items"][0]

        # APIResponse provides success status and data/error info
        response: APIResponse = await api.device.set_temperature(device, 120)

        if response.success:
            print("Temperature set successfully!")
            print(f"Response data: {response.data}")
        else:
            print(f"Failed: {response.error}")


asyncio.run(main())

With Connection Pooling (Recommended)

For better performance, provide your own aiohttp.ClientSession:

import asyncio
from aiohttp import ClientSession
from aiorinnai import API


async def main() -> None:
    async with ClientSession() as session:
        async with API(session=session) as api:
            await api.async_login("<EMAIL>", "<PASSWORD>")
            user_info = await api.user.get_info()
            device = user_info["devices"]["items"][0]

            response = await api.device.start_recirculation(device, 5)
            if response.success:
                print("Recirculation started!")


asyncio.run(main())

Custom Configuration

Configure timeouts, retry behavior, and more:

import asyncio
from aiorinnai import API


async def main() -> None:
    # Full configuration options
    async with API(
        timeout=60.0,           # Request timeout in seconds (default: 30)
        retry_count=5,          # Number of retry attempts (default: 3)
        retry_delay=2.0,        # Initial delay between retries (default: 1.0)
        retry_multiplier=2.0,   # Exponential backoff multiplier (default: 2.0)
        executor_timeout=30.0,  # Timeout for blocking AWS calls (default: 30)
    ) as api:
        await api.async_login("<EMAIL>", "<PASSWORD>")
        user_info = await api.user.get_info()


asyncio.run(main())

Token Persistence

For long-running applications (like Home Assistant integrations), you can persist tokens to avoid re-authenticating:

import asyncio
from aiorinnai import API


async def main() -> None:
    async with API() as api:
        # Initial login
        await api.async_login("<EMAIL>", "<PASSWORD>")

        # Save tokens for later (store securely!)
        saved_tokens = {
            "email": api.username,
            "access_token": api.access_token,
            "refresh_token": api.refresh_token,
        }
        print(f"Tokens saved: {saved_tokens}")


async def restore_session() -> None:
    """Restore a session from saved tokens."""
    # Load your saved tokens
    saved_tokens = {
        "email": "user@example.com",
        "access_token": "...",
        "refresh_token": "...",
    }

    async with API() as api:
        # Restore session without password
        await api.async_renew_access_token(
            email=saved_tokens["email"],
            access_token=saved_tokens["access_token"],
            refresh_token=saved_tokens["refresh_token"],
        )

        # Now you can use the API
        user_info = await api.user.get_info()
        print(f"Restored session for: {user_info['email']}")


asyncio.run(main())

Token Properties (read-only):

• api.id_token - JWT ID token for API authentication
• api.access_token - JWT access token for Cognito operations
• api.refresh_token - Refresh token for obtaining new tokens

Input Validation

The library validates inputs and raises ValueError for invalid values:

import asyncio
from aiorinnai import API


async def main() -> None:
    async with API() as api:
        await api.async_login("<EMAIL>", "<PASSWORD>")
        user_info = await api.user.get_info()
        device = user_info["devices"]["items"][0]

        try:
            # Temperature must be 100-140°F
            await api.device.set_temperature(device, 99)  # Raises ValueError
        except ValueError as e:
            print(f"Invalid temperature: {e}")

        try:
            # Duration must be 1-60 minutes
            await api.device.start_recirculation(device, 0)  # Raises ValueError
        except ValueError as e:
            print(f"Invalid duration: {e}")

        try:
            # Device must have 'thing_name' attribute
            await api.device.turn_on({})  # Raises ValueError
        except ValueError as e:
            print(f"Invalid device: {e}")


asyncio.run(main())

Error Handling

The library provides specific exceptions for different error conditions:

import asyncio
from aiorinnai import (
    API,
    Unauthenticated,
    UserNotFound,
    RequestError,
)


async def main() -> None:
    async with API() as api:
        try:
            await api.async_login("<EMAIL>", "<PASSWORD>")
            user_info = await api.user.get_info()

        except Unauthenticated:
            print("Invalid email or password")
        except UserNotFound:
            print("User account not found")
        except RequestError as err:
            print(f"API request failed: {err}")
        except asyncio.TimeoutError:
            print("Request timed out")


asyncio.run(main())

API Reference

API Class

Constructor:

API(
    session: ClientSession | None = None,  # Optional aiohttp session
    timeout: float = 30.0,                 # Request timeout in seconds
    retry_count: int = 3,                  # Number of retry attempts
    retry_delay: float = 1.0,              # Initial retry delay in seconds
    retry_multiplier: float = 2.0,         # Exponential backoff multiplier
    executor_timeout: float = 30.0,        # Timeout for blocking AWS calls
)

Methods:

• async_login(email, password) - Authenticate with your Rinnai account
• async_renew_access_token(email, access_token, refresh_token) - Restore session from saved tokens
• async_check_token() - Check and refresh token if needed (called automatically)
• close() - Close the API client and release resources

Properties:

• is_connected - Returns True if connected with valid authentication
• username - The authenticated user's email address
• id_token - JWT ID token (read-only, for API authentication)
• access_token - JWT access token (read-only, for Cognito operations)
• refresh_token - Refresh token (read-only, for token renewal)

Context Manager:

• The API class supports async with for automatic resource cleanup

Device Methods

All device methods return APIResponse with success, data, and error attributes.

User Methods

• get_info() - Get user account information including all devices (returns UserInfo | None)

Type Definitions

The library exports TypedDict definitions for type-safe access to API responses:

from aiorinnai import (
    APIResponse,      # Dataclass: success, data, error
    DeviceInfo,       # TypedDict for device data
    ShadowData,       # TypedDict for device shadow state
    UserInfo,         # TypedDict for user account data
    TemperatureUnit,  # Enum: FAHRENHEIT, CELSIUS
)

Exceptions

Validation Ranges

Known Issues

• Not all APIs supported

Development

Setup

# Clone the repository
git clone https://github.com/explosivo22/aiorinnai.git
cd aiorinnai

# Create virtual environment
python -m venv .venv
source .venv/bin/activate

# Install development dependencies
pip install -e ".[dev]"

# Install pre-commit hooks
pip install pre-commit
pre-commit install

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=aiorinnai --cov-report=term-missing

Code Quality

# Format code
ruff format .

# Lint code
ruff check .

# Type checking
mypy aiorinnai

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.