indevolt-api 1.6.1

Python API client for Indevolt devices

Indevolt API

Python client library for communicating with Indevolt devices (home battery systems).

Features

• Async/await support using aiohttp
• Fully typed with type hints
• Simple and intuitive API
• Comprehensive error handling

Installation

pip install indevolt-api

Quick Start

import asyncio
import aiohttp
from indevolt_api import (
    IndevoltAPI,
    IndevoltConfig,
    IndevoltEnergyMode,
    IndevoltSystem,
    SET_REALTIME_ACTION,
    IndevoltRealtimeAction,
)

async def main():
    async with aiohttp.ClientSession() as session:
        api = IndevoltAPI(host="192.168.1.100", port=8080, session=session)

        # Get device configuration
        config = await api.get_config()
        print(f"Device config: {config}")

        # Fetch data using StrEnum members — response keys are the same strings
        data = await api.fetch_data([IndevoltConfig.READ_ENERGY_MODE, IndevoltSystem.INPUT_POWER])
        print(f"Energy mode: {data[IndevoltConfig.READ_ENERGY_MODE]}")
        print(f"Input power: {data[IndevoltSystem.INPUT_POWER]}")

        # Write a single value
        await api.set_data(IndevoltConfig.WRITE_DISCHARGE_LIMIT, 50)

        # Write a real-time charge command
        await api.set_data(SET_REALTIME_ACTION, [IndevoltRealtimeAction.CHARGE, 700, 80])

asyncio.run(main())

Device Discovery

The library supports automatic discovery of Indevolt devices on your local network using UDP broadcast.

Quick Discovery Example

import asyncio
import aiohttp
from indevolt_api import async_discover, IndevoltAPI

async def main():
    # Discover devices on the network (overrides default 3-second timeout)
    devices = await async_discover(timeout=3.0)

    if not devices:
        print("No devices found")
        return

    print(f"Found {len(devices)} device(s):")
    for device in devices:
        print(f"  - {device.host}:{device.port} (name: {device.name})")

    # Connect to the first discovered device
    async with aiohttp.ClientSession() as session:
        api = IndevoltAPI.from_discovered_device(devices[0], session)
        config = await api.get_config()
        print(f"Device config: {config}")

asyncio.run(main())

Discovery Details

The discovery mechanism:

1. Sends AT command AT+IGDEVICEIP via UDP broadcast to 255.255.255.255:8099
2. Indevolt devices on the same network respond to local port 10000 with their IP
3. Returns a list of DiscoveredDevice objects with device information

Note: Ensure your device and computer are on the same local network and that UDP port 10000 is available.

API Reference

IndevoltAPI

__init__(host: str, port: int, session: aiohttp.ClientSession, timeout: float = 10.0)

Initialize the API client.

Parameters:

• host (str): Device hostname or IP address
• port (int): Device port number (typically 80 or 8080)
• session (aiohttp.ClientSession): An aiohttp client session
• timeout (float): Request timeout in seconds (default: 10.0)

Example:

# Default 10-second timeout (recommended for local devices)
api = IndevoltAPI(host="192.168.1.100", port=8080, session=session)

# Custom timeout
api = IndevoltAPI(host="192.168.1.100", port=8080, session=session, timeout=15.0)

classmethod from_discovered_device(device: DiscoveredDevice, session: aiohttp.ClientSession, timeout: float = 10.0)

Create an API client from a discovered device.

Parameters:

• device (DiscoveredDevice): A device object returned by async_discover()
• session (aiohttp.ClientSession): An aiohttp client session
• timeout (float): Request timeout in seconds (default: 10.0)

Returns:

• IndevoltAPI instance configured for the discovered device

Example:

devices = await async_discover()
if devices:
    api = IndevoltAPI.from_discovered_device(devices[0], session)

async fetch_data(t: str | list[str]) -> dict[str, Any]

Fetch data from the device.

Parameters:

• t: A StrEnum member, a raw string key, or a list of either

Returns:

• Dictionary whose keys are strings matching the requested cJson points. StrEnum members can be used directly to index the result.

Example:

from indevolt_api import IndevoltSystem, IndevoltGrid, IndevoltBattery

# Single point
data = await api.fetch_data(IndevoltBattery.SOC)
print(data[IndevoltBattery.SOC])

# Multiple points
data = await api.fetch_data([
    IndevoltSystem.INPUT_POWER,
    IndevoltSystem.OUTPUT_POWER,
    IndevoltGrid.VOLTAGE,
])
print(data[IndevoltSystem.INPUT_POWER])
print(data[IndevoltGrid.VOLTAGE])

async set_data(t: str | int, v: Any) -> bool

Write data to the device.

Parameters:

• t: cJson point identifier (e.g., "47015" or 47015)
• v: Value(s) to write (automatically converted to list of integers)

Returns:

• True on success, False otherwise

Example:

from indevolt_api import IndevoltConfig, IndevoltEnergyMode, SET_REALTIME_ACTION, IndevoltRealtimeAction

# Single value
await api.set_data(IndevoltConfig.WRITE_DISCHARGE_LIMIT, 50)

# Real-time command with multiple values
await api.set_data(SET_REALTIME_ACTION, [IndevoltRealtimeAction.CHARGE, 700, 80])

# Set energy mode
await api.set_data(IndevoltConfig.WRITE_ENERGY_MODE, IndevoltEnergyMode.SELF_CONSUMED_PRIORITIZED)

async stop() -> bool

Stop any active real-time charge or discharge action.

Returns:

• True on success, False if the command was rejected or a connection error occurred

Example:

succeeded = await api.stop()

async charge(power: int, target_soc: int) -> bool

Send a real-time charge command to the device.

Parameters:

• power (int): Charge power in watts
• target_soc (int): Target state of charge percentage

Returns:

• True on success, False if the command was rejected or a connection error occurred

Example:

succeeded = await api.charge(power=700, target_soc=80)

async discharge(power: int, target_soc: int) -> bool

Send a real-time discharge command to the device.

Parameters:

• power (int): Discharge power in watts
• target_soc (int): Target state of charge percentage

Returns:

• True on success, False if the command was rejected or a connection error occurred

Example:

succeeded = await api.discharge(power=400, target_soc=20)

async get_config() -> dict[str, Any]

Get system configuration from the device.

Returns:

• Dictionary with device system configuration

Example:

config = await api.get_config()
print(config)

check_charge_limits(power: int, target_soc: int, generation: int) -> None

Check that charge parameters do not exceed device limits. Raises an exception if any boundary is violated.

Parameters:

• power (int): Requested charge power in watts
• target_soc (int): Target state of charge percentage
• generation (int): Device hardware generation (1 or 2), available from get_config() under device.generation

Raises:

• PowerExceedsMaxError: If power exceeds the maximum for the given generation
• SocBelowMinimumError: If target_soc is below the minimum SOC (5%)

Example:

config = await api.get_config()
generation = config["device"]["generation"]

try:
    api.check_charge_limits(power=1000, target_soc=80, generation=generation)
except PowerExceedsMaxError as e:
    print(f"Power {e.power}W exceeds max {e.max_power}W for gen {e.generation}")
except SocBelowMinimumError as e:
    print(f"Target SOC {e.target_soc}% is below minimum {e.minimum_soc}%")

check_discharge_limits(power: int, target_soc: int, generation: int) -> None

Check that discharge parameters do not exceed device limits. Raises an exception if any boundary is violated.

Parameters:

• power (int): Requested discharge power in watts
• target_soc (int): Target state of charge percentage
• generation (int): Device hardware generation (1 or 2), available from get_config() under device.generation

Raises:

• PowerExceedsMaxError: If power exceeds the maximum for the given generation
• SocBelowMinimumError: If target_soc is below the minimum SOC (5%)

Example:

config = await api.get_config()
generation = config["device"]["generation"]

try:
    api.check_discharge_limits(power=600, target_soc=10, generation=generation)
except PowerExceedsMaxError as e:
    print(f"Power {e.power}W exceeds max {e.max_power}W for gen {e.generation}")
except SocBelowMinimumError as e:
    print(f"Target SOC {e.target_soc}% is below minimum {e.minimum_soc}%")

async_discover(timeout: float = 3.0) -> list[DiscoveredDevice]

Discover Indevolt devices on the local network using UDP broadcast.

Parameters:

• timeout (float): Discovery timeout in seconds (default: 3.0)

Returns:

• List of DiscoveredDevice objects representing found devices

Example:

devices = await async_discover(timeout=3.0)
for device in devices:
    print(f"Found: {device.host}:{device.port}")

DiscoveredDevice

Represents a discovered Indevolt device with the following attributes:

Attributes:

• host (str): Device IP address
• port (int): Device port number (default: 8080)
• name (str | None): Device name if provided in discovery response
• metadata (dict): Additional device information from discovery response

Example:

device = devices[0]
print(f"Device at {device.host}:{device.port}")
if device.name:
    print(f"Name: {device.name}")

Exception Handling

The library provides custom exceptions for API errors and limit violations.

APIException

Raised when there's a client error during API communication (network errors, HTTP errors).

TimeOutException

Raised when an API request times out (default timeout: 10 seconds).

PowerExceedsMaxError

Raised by check_charge_limits() or check_discharge_limits() when the requested power exceeds the device maximum.

Attributes: power, max_power, generation

SocBelowMinimumError

Raised by check_charge_limits() or check_discharge_limits() when the target SOC is below the hard minimum of 5%.

Attributes: target_soc, minimum_soc

Example:

from indevolt_api import IndevoltAPI, APIException, TimeOutException

try:
    data = await api.fetch_data("7101")
except TimeOutException:
    print("Request timed out")
except APIException as e:
    print(f"API error: {e}")

Note: You can adjust the timeout when creating the API client:

# Increase timeout if needed (e.g., for slower networks)
api = IndevoltAPI(host="192.168.1.100", port=8080, session=session, timeout=10.0)

Constants and Enums

All register keys and action values are available as typed StrEnum classes, importable directly from indevolt_api. Because StrEnum members are strings, they can be passed directly to fetch_data() and set_data(), and used to index the response dictionary — no manual conversion needed.

IndevoltConfig

Register keys for configurable device settings (read and write).

from indevolt_api import IndevoltConfig

# Write registers
IndevoltConfig.WRITE_ENERGY_MODE      # "47005"
IndevoltConfig.WRITE_DISCHARGE_LIMIT  # "1142"
# ... and more

# Read registers
IndevoltConfig.READ_ENERGY_MODE       # "7101"
IndevoltConfig.READ_DISCHARGE_LIMIT   # "6105"
# ... and more

IndevoltRealtimeAction

Action values for real-time control mode, used with SET_REALTIME_ACTION.

from indevolt_api import IndevoltRealtimeAction

IndevoltRealtimeAction.STOP       # "0"
IndevoltRealtimeAction.CHARGE     # "1"
IndevoltRealtimeAction.DISCHARGE  # "2"

IndevoltEnergyMode

Energy mode values for IndevoltConfig.WRITE_ENERGY_MODE.

from indevolt_api import IndevoltEnergyMode

IndevoltEnergyMode.OUTDOOR_PORTABLE
IndevoltEnergyMode.SELF_CONSUMED_PRIORITIZED
IndevoltEnergyMode.REAL_TIME_CONTROL
IndevoltEnergyMode.CHARGE_DISCHARGE_SCHEDULE

IndevoltBattery, IndevoltSystem, IndevoltGrid, IndevoltSolar

Register key enums for reading battery, system-level, grid, and solar data points.

from indevolt_api import IndevoltBattery, IndevoltSystem, IndevoltGrid, IndevoltSolar

data = await api.fetch_data([
    IndevoltBattery.SOC,
    IndevoltBattery.POWER,
    IndevoltSystem.OUTPUT_POWER,
    IndevoltGrid.METER_POWER_GEN2,
    IndevoltSolar.DC_INPUT_POWER_1,
])

SET_REALTIME_ACTION

The register key used to send real-time charge/discharge commands to the device.

from indevolt_api import SET_REALTIME_ACTION

await api.set_data(SET_REALTIME_ACTION, [IndevoltRealtimeAction.CHARGE, 700, 80])

DEVICE_LIMITS

Dictionary of per-generation device limits used by check_charge_limits() and check_discharge_limits().

from indevolt_api import DEVICE_LIMITS

print(DEVICE_LIMITS[1])  # {'max_discharge_power': 800, 'max_charge_power': 1200, 'minimum_soc': 5}
print(DEVICE_LIMITS[2])  # {'max_discharge_power': 2400, 'max_charge_power': 2400, 'minimum_soc': 5}

Requirements

• Python 3.11+
• aiohttp >= 3.9.0

License

MIT License - see LICENSE file for details