API reference

intelbras_solar_local.IntelbrasSolarClient

Talks to one inverter through its WiNet-S datalogger.

Modbus TCP is the primary source; the WiNet-S WebSocket adds datalogger connectivity, the fault list and the commercial model, best-effort.

Parameters:

Name	Type	Description	Default
host	str	IP or hostname of the WiNet-S datalogger.	required
port	int	Modbus TCP port.	DEFAULT_PORT
unit	int	Modbus unit id of the inverter.	DEFAULT_UNIT
session	ClientSession | None	optional aiohttp session for the WebSocket; one is created (and closed by :meth:async_close) when omitted.	None
winet_port	int	WebSocket port of the datalogger web UI.	DEFAULT_WINET_PORT

Source code in src/intelbras_solar_local/client.py

class IntelbrasSolarClient:
    """Talks to one inverter through its WiNet-S datalogger.

    Modbus TCP is the primary source; the WiNet-S WebSocket adds datalogger
    connectivity, the fault list and the commercial model, best-effort.

    Args:
        host: IP or hostname of the WiNet-S datalogger.
        port: Modbus TCP port.
        unit: Modbus unit id of the inverter.
        session: optional ``aiohttp`` session for the WebSocket; one is created
            (and closed by :meth:`async_close`) when omitted.
        winet_port: WebSocket port of the datalogger web UI.
    """

    def __init__(
        self,
        host: str,
        port: int = DEFAULT_PORT,
        unit: int = DEFAULT_UNIT,
        session: aiohttp.ClientSession | None = None,
        winet_port: int = DEFAULT_WINET_PORT,
    ) -> None:
        self.host = host
        self.port = port
        self.unit = unit
        self.winet_port = winet_port
        self._session = session
        self._owns_session = session is None
        self.modbus = ModbusTcpClient(host, port, unit)

    async def __aenter__(self) -> IntelbrasSolarClient:
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        await self.async_close()

    async def async_close(self) -> None:
        """Close the Modbus connection and the session this client created."""
        await self.modbus.async_close()
        if self._owns_session and self._session is not None:
            await self._session.close()
            self._session = None

    @property
    def winet(self) -> WiNetClient:
        """WebSocket client for the datalogger, sharing this client's session."""
        if self._session is None:
            self._session = aiohttp.ClientSession()
        return WiNetClient(self.host, self._session, self.winet_port)

    async def async_get_inverter_data(self) -> InverterStatus:
        """Read every polled register block over Modbus.

        Raises:
            IntelbrasSolarConnectionError: datalogger unreachable.
            IntelbrasSolarProtocolError: Modbus exception reply.
        """
        inp: dict[int, int] = {}
        hold: dict[int, int] = {}
        for start, count in INPUT_BLOCKS:
            inp |= await self.modbus.async_read(FC_READ_INPUT, start, count)
        for start, count in HOLDING_BLOCKS:
            hold |= await self.modbus.async_read(FC_READ_HOLDING, start, count)
        return parse_registers(inp, hold)

    async def async_get_status(self) -> InverterStatus:
        """Modbus snapshot plus datalogger state (``datalogger`` may be None).

        A WebSocket failure never fails the call; only Modbus errors raise.
        """
        status = await self.async_get_inverter_data()
        try:
            status.datalogger = await self.winet.async_get_status()
        except IntelbrasSolarError as err:
            _LOGGER.debug("WiNet datalogger read failed: %s", err)
        return status

    async def async_set_inverter_on(self, on: bool) -> None:
        """Start (``True``) or stop (``False``) power generation."""
        await self.modbus.async_write(REG_START_STOP, START if on else STOP)

    async def async_set_power_limit_enabled(self, enabled: bool) -> None:
        """Turn the active power limitation on or off."""
        await self.modbus.async_write(
            REG_POWER_LIMIT_SWITCH, ENABLE if enabled else DISABLE
        )

    async def async_set_power_limit(self, percent: float) -> None:
        """Set the active power limit, 0–100 % of nominal power."""
        if not 0 <= percent <= 100:
            raise ValueError("percent must be between 0 and 100")
        await self.modbus.async_write(REG_POWER_LIMIT, round(percent * 10))

    async def async_set_clock(self, when: datetime) -> None:
        """Set the inverter clock (one register per field)."""
        fields = (when.year, when.month, when.day, when.hour, when.minute, when.second)
        for offset, value in enumerate(fields):
            await self.modbus.async_write(REG_CLOCK + offset, value)

winet property

winet: WiNetClient

WebSocket client for the datalogger, sharing this client's session.

async_close async

async_close() -> None

Close the Modbus connection and the session this client created.

Source code in src/intelbras_solar_local/client.py

async def async_close(self) -> None:
    """Close the Modbus connection and the session this client created."""
    await self.modbus.async_close()
    if self._owns_session and self._session is not None:
        await self._session.close()
        self._session = None

async_get_inverter_data async

async_get_inverter_data() -> InverterStatus

Read every polled register block over Modbus.

Raises:

Type	Description
IntelbrasSolarConnectionError	datalogger unreachable.
IntelbrasSolarProtocolError	Modbus exception reply.

Source code in src/intelbras_solar_local/client.py

async def async_get_inverter_data(self) -> InverterStatus:
    """Read every polled register block over Modbus.

    Raises:
        IntelbrasSolarConnectionError: datalogger unreachable.
        IntelbrasSolarProtocolError: Modbus exception reply.
    """
    inp: dict[int, int] = {}
    hold: dict[int, int] = {}
    for start, count in INPUT_BLOCKS:
        inp |= await self.modbus.async_read(FC_READ_INPUT, start, count)
    for start, count in HOLDING_BLOCKS:
        hold |= await self.modbus.async_read(FC_READ_HOLDING, start, count)
    return parse_registers(inp, hold)

async_get_status async

async_get_status() -> InverterStatus

Modbus snapshot plus datalogger state (datalogger may be None).

A WebSocket failure never fails the call; only Modbus errors raise.

Source code in src/intelbras_solar_local/client.py

async def async_get_status(self) -> InverterStatus:
    """Modbus snapshot plus datalogger state (``datalogger`` may be None).

    A WebSocket failure never fails the call; only Modbus errors raise.
    """
    status = await self.async_get_inverter_data()
    try:
        status.datalogger = await self.winet.async_get_status()
    except IntelbrasSolarError as err:
        _LOGGER.debug("WiNet datalogger read failed: %s", err)
    return status

async_set_inverter_on async

async_set_inverter_on(on: bool) -> None

Start (True) or stop (False) power generation.

Source code in src/intelbras_solar_local/client.py

async def async_set_inverter_on(self, on: bool) -> None:
    """Start (``True``) or stop (``False``) power generation."""
    await self.modbus.async_write(REG_START_STOP, START if on else STOP)

async_set_power_limit_enabled async

async_set_power_limit_enabled(enabled: bool) -> None

Turn the active power limitation on or off.

Source code in src/intelbras_solar_local/client.py

async def async_set_power_limit_enabled(self, enabled: bool) -> None:
    """Turn the active power limitation on or off."""
    await self.modbus.async_write(
        REG_POWER_LIMIT_SWITCH, ENABLE if enabled else DISABLE
    )

async_set_power_limit async

async_set_power_limit(percent: float) -> None

Set the active power limit, 0–100 % of nominal power.

Source code in src/intelbras_solar_local/client.py

async def async_set_power_limit(self, percent: float) -> None:
    """Set the active power limit, 0–100 % of nominal power."""
    if not 0 <= percent <= 100:
        raise ValueError("percent must be between 0 and 100")
    await self.modbus.async_write(REG_POWER_LIMIT, round(percent * 10))

async_set_clock async

async_set_clock(when: datetime) -> None

Set the inverter clock (one register per field).

Source code in src/intelbras_solar_local/client.py

async def async_set_clock(self, when: datetime) -> None:
    """Set the inverter clock (one register per field)."""
    fields = (when.year, when.month, when.day, when.hour, when.minute, when.second)
    for offset, value in enumerate(fields):
        await self.modbus.async_write(REG_CLOCK + offset, value)

intelbras_solar_local.InverterStatus dataclass

Snapshot of the inverter at a point in time.

online is False when the datalogger could not be read (typically at night, when the inverter powers down); every other field is then None.

Source code in src/intelbras_solar_local/models.py

@dataclass
class InverterStatus:
    """Snapshot of the inverter at a point in time.

    ``online`` is ``False`` when the datalogger could not be read (typically at
    night, when the inverter powers down); every other field is then ``None``.
    """

    online: bool = True
    serial: str | None = None
    device_type_code: int | None = None
    nominal_power: float | None = None  # kW
    energy_today: float | None = None  # kWh
    energy_month: float | None = None  # kWh
    energy_year: float | None = None  # kWh
    energy_total: float | None = None  # kWh
    running_hours_total: int | None = None
    running_minutes_today: int | None = None
    temperature: float | None = None  # °C
    power_ac: int | None = None  # W
    power_dc: int | None = None  # W
    power_apparent: int | None = None  # VA
    power_reactive: int | None = None  # var
    power_factor: float | None = None
    grid_voltage: float | None = None  # V
    grid_current: float | None = None  # A
    grid_frequency: float | None = None  # Hz
    bus_voltage: float | None = None  # V
    insulation_resistance: int | None = None  # kΩ
    work_state_code: int | None = None
    fault_code: int | None = None
    inverter_on: bool | None = None
    power_limit_enabled: bool | None = None
    power_limit: float | None = None  # %
    clock: datetime | None = None
    mppts: list[Mppt] = field(default_factory=list)
    datalogger: DataloggerStatus | None = None

    @property
    def model(self) -> str | None:
        """Commercial model: from the datalogger, else the device type table."""
        if self.datalogger and self.datalogger.model:
            return self.datalogger.model
        if self.device_type_code is None:
            return None
        return DEVICE_MODELS.get(
            self.device_type_code, f"Sungrow 0x{self.device_type_code:04X}"
        )

    @property
    def work_state(self) -> str | None:
        """Work state as a snake_case key (see ``WORK_STATES``)."""
        if self.work_state_code is None:
            return None
        return WORK_STATES.get(self.work_state_code, "unknown")

    @property
    def producing(self) -> bool | None:
        """True while the inverter is feeding the grid."""
        if self.work_state_code is None:
            return None
        return self.work_state_code in PRODUCING_STATES

    @property
    def has_fault(self) -> bool | None:
        """True on a Modbus fault code, a fault state or a datalogger fault."""
        if self.work_state_code is None:
            return None
        return bool(
            self.fault_code
            or self.work_state_code == FAULT_STATE
            or (self.datalogger and self.datalogger.total_faults)
        )

model property

model: str | None

Commercial model: from the datalogger, else the device type table.

work_state property

work_state: str | None

Work state as a snake_case key (see WORK_STATES).

producing property

producing: bool | None

True while the inverter is feeding the grid.

has_fault property

has_fault: bool | None

True on a Modbus fault code, a fault state or a datalogger fault.

intelbras_solar_local.DataloggerStatus dataclass

State reported by the WiNet-S WebSocket.

Source code in src/intelbras_solar_local/models.py

@dataclass
class DataloggerStatus:
    """State reported by the WiNet-S WebSocket."""

    model: str | None = None
    wifi_connected: bool | None = None
    cloud_connected: bool | None = None
    total_faults: int = 0
    total_alarms: int = 0
    faults: list[dict[str, Any]] = field(default_factory=list)

intelbras_solar_local.Mppt dataclass

One MPPT (string) input.

Source code in src/intelbras_solar_local/models.py

@dataclass
class Mppt:
    """One MPPT (string) input."""

    voltage: float
    current: float

    @property
    def power(self) -> int:
        """DC power of this input in W."""
        return round(self.voltage * self.current)

power property

power: int

DC power of this input in W.

intelbras_solar_local.parse_registers

parse_registers(inp: dict[int, int], hold: dict[int, int]) -> InverterStatus

Build an :class:InverterStatus from raw input and holding registers.

Sungrow stores 32-bit values low word first.

Source code in src/intelbras_solar_local/models.py

def parse_registers(inp: dict[int, int], hold: dict[int, int]) -> InverterStatus:
    """Build an :class:`InverterStatus` from raw input and holding registers.

    Sungrow stores 32-bit values low word first.
    """

    def u32(a: int) -> int:
        return inp[a] | inp[a + 1] << 16

    def s16(v: int) -> int:
        return v - 0x10000 if v & 0x8000 else v

    def s32(a: int) -> int:
        v = u32(a)
        return v - 0x100000000 if v & 0x80000000 else v

    serial = b"".join(struct.pack(">H", inp[a]) for a in range(4989, 4999))
    try:
        clock = datetime(*(hold[a] for a in range(4999, 5005)))
    except ValueError:
        clock = None
    return InverterStatus(
        serial=serial.decode("ascii", "ignore").strip("\x00 "),
        device_type_code=inp[4999],
        nominal_power=inp[5000] / 10,
        energy_today=inp[5002] / 10,
        energy_month=u32(5127) / 10,
        energy_year=u32(5225) / 10,
        energy_total=u32(5143) / 10,
        running_hours_total=u32(5005),
        running_minutes_today=inp[5112],
        temperature=s16(inp[5007]) / 10,
        power_apparent=u32(5008),
        power_dc=u32(5016),
        power_ac=u32(5030),
        power_reactive=s32(5032),
        power_factor=s16(inp[5034]) / 1000,
        grid_voltage=inp[5018] / 10,
        grid_current=inp[5021] / 10,
        grid_frequency=inp[5147] / 100,
        bus_voltage=inp[5146] / 10,
        insulation_resistance=inp[5070],
        work_state_code=inp[5037],
        fault_code=inp[5044],
        inverter_on=hold[REG_START_STOP] != STOP,
        power_limit_enabled=hold[REG_POWER_LIMIT_SWITCH] == ENABLE,
        power_limit=hold[REG_POWER_LIMIT] / 10,
        clock=clock,
        # Two MPPTs on the IONS-5K M5; models with a third use 5014/5015.
        mppts=[Mppt(inp[a] / 10, inp[a + 1] / 10) for a in (5010, 5012)],
    )

intelbras_solar_local.parse_datalogger

parse_datalogger(state: dict[str, Any], faults: dict[str, Any], devices: dict[str, Any]) -> DataloggerStatus

Build a :class:DataloggerStatus from WiNet-S result_data payloads.

Source code in src/intelbras_solar_local/models.py

def parse_datalogger(
    state: dict[str, Any], faults: dict[str, Any], devices: dict[str, Any]
) -> DataloggerStatus:
    """Build a :class:`DataloggerStatus` from WiNet-S ``result_data`` payloads."""
    device = (devices.get("list") or [{}])[0]
    return DataloggerStatus(
        model=device.get("dev_model"),
        wifi_connected=state.get("wifi_conn_sts") == "1",
        cloud_connected=state.get("cloud_conn_sts") == "1",
        total_faults=int(state.get("total_fault", 0) or 0),
        total_alarms=int(state.get("total_alarm", 0) or 0),
        faults=list(faults.get("list") or []),
    )

intelbras_solar_local.ModbusTcpClient

One persistent Modbus TCP connection, reopened on demand after errors.

Source code in src/intelbras_solar_local/modbus.py

class ModbusTcpClient:
    """One persistent Modbus TCP connection, reopened on demand after errors."""

    def __init__(
        self,
        host: str,
        port: int = DEFAULT_PORT,
        unit: int = DEFAULT_UNIT,
        timeout: float = MODBUS_TIMEOUT,
    ) -> None:
        self.host = host
        self.port = port
        self.unit = unit
        self.timeout = timeout
        self._reader: asyncio.StreamReader | None = None
        self._writer: asyncio.StreamWriter | None = None
        # Random start: the WiNet-S can leak replies across concurrent clients,
        # and two clients both counting from 1 would accept each other's frames.
        self._tid = random.randrange(0x10000)
        # The WiNet-S serves one request at a time; serialize every call.
        self._lock = asyncio.Lock()

    async def async_close(self) -> None:
        """Close the TCP connection (it reopens on the next request)."""
        if self._writer is not None:
            self._writer.close()
        self._reader = self._writer = None

    async def async_read(
        self, function: int, address: int, count: int
    ) -> dict[int, int]:
        """Read ``count`` registers starting at ``address``.

        Args:
            function: ``3`` (holding) or ``4`` (input).
            address: 0-based start register.
            count: number of registers (max 125).

        Returns:
            Mapping of register address to its unsigned 16-bit value.
        """
        body = await self._request(struct.pack(">BHH", function, address, count))
        if body[0] != function or len(body) != 2 + count * 2 or body[1] != count * 2:
            # A reply to some other request: the stream is out of sync.
            await self.async_close()
            raise IntelbrasSolarProtocolError(
                f"Modbus reply does not match function {function} x {count} registers"
            )
        values = struct.unpack(f">{count}H", body[2:])
        return {address + i: v for i, v in enumerate(values)}

    async def async_write(self, address: int, value: int) -> None:
        """Write one holding register (FC6)."""
        await self._request(struct.pack(">BHH", FC_WRITE_SINGLE, address, value))

    async def _request(self, pdu: bytes) -> bytes:
        async with self._lock:
            try:
                body = await asyncio.wait_for(self._roundtrip(pdu), self.timeout)
            except (OSError, TimeoutError, asyncio.IncompleteReadError) as err:
                await self.async_close()
                raise IntelbrasSolarConnectionError(
                    f"Modbus {self.host}:{self.port}: {err!r}"
                ) from err
            except _FramingError as err:
                await self.async_close()
                raise IntelbrasSolarProtocolError(
                    f"Modbus {self.host}:{self.port}: {err}"
                ) from err
        if body[0] & 0x80:
            raise IntelbrasSolarProtocolError(
                f"Modbus exception {body[1]} on function {body[0] & 0x7F}"
            )
        return body

    async def _roundtrip(self, pdu: bytes) -> bytes:
        if self._writer is None:
            self._reader, self._writer = await asyncio.open_connection(
                self.host, self.port
            )
        assert self._reader is not None
        self._tid = (self._tid + 1) & 0xFFFF
        self._writer.write(_MBAP.pack(self._tid, 0, len(pdu) + 1, self.unit) + pdu)
        await self._writer.drain()
        while True:
            tid, _, length, _ = _MBAP.unpack(await self._reader.readexactly(7))
            if length < 3:
                raise _FramingError(f"bad MBAP length {length}")
            body = await self._reader.readexactly(length - 1)
            # Drop late replies to a request that already timed out.
            if tid == self._tid:
                return body

async_close async

async_close() -> None

Close the TCP connection (it reopens on the next request).

Source code in src/intelbras_solar_local/modbus.py

async def async_close(self) -> None:
    """Close the TCP connection (it reopens on the next request)."""
    if self._writer is not None:
        self._writer.close()
    self._reader = self._writer = None

async_read async

async_read(function: int, address: int, count: int) -> dict[int, int]

Read count registers starting at address.

Parameters:

Name	Type	Description	Default
function	int	3 (holding) or 4 (input).	required
address	int	0-based start register.	required
count	int	number of registers (max 125).	required

Returns:

Type	Description
dict[int, int]	Mapping of register address to its unsigned 16-bit value.

Source code in src/intelbras_solar_local/modbus.py

async def async_read(
    self, function: int, address: int, count: int
) -> dict[int, int]:
    """Read ``count`` registers starting at ``address``.

    Args:
        function: ``3`` (holding) or ``4`` (input).
        address: 0-based start register.
        count: number of registers (max 125).

    Returns:
        Mapping of register address to its unsigned 16-bit value.
    """
    body = await self._request(struct.pack(">BHH", function, address, count))
    if body[0] != function or len(body) != 2 + count * 2 or body[1] != count * 2:
        # A reply to some other request: the stream is out of sync.
        await self.async_close()
        raise IntelbrasSolarProtocolError(
            f"Modbus reply does not match function {function} x {count} registers"
        )
    values = struct.unpack(f">{count}H", body[2:])
    return {address + i: v for i, v in enumerate(values)}

async_write async

async_write(address: int, value: int) -> None

Write one holding register (FC6).

Source code in src/intelbras_solar_local/modbus.py

async def async_write(self, address: int, value: int) -> None:
    """Write one holding register (FC6)."""
    await self._request(struct.pack(">BHH", FC_WRITE_SINGLE, address, value))

intelbras_solar_local.WiNetClient

Reads datalogger state, faults and the device list over WebSocket.

Source code in src/intelbras_solar_local/winet.py

class WiNetClient:
    """Reads datalogger state, faults and the device list over WebSocket."""

    def __init__(
        self,
        host: str,
        session: aiohttp.ClientSession,
        port: int = DEFAULT_WINET_PORT,
        timeout: float = WINET_TIMEOUT,
    ) -> None:
        self.url = f"ws://{host}:{port}{WINET_WS_PATH}"
        self._session = session
        self._timeout = timeout

    async def async_get_status(self) -> DataloggerStatus:
        """Open a session, read ``state``, ``fault`` and ``devicelist``, close."""
        try:
            async with asyncio.timeout(self._timeout):
                async with self._session.ws_connect(self.url) as ws:
                    token = (await _call(ws, "connect", token=""))["token"]
                    state = await _call(ws, "state", token=token)
                    faults = await _call(ws, "fault", token=token)
                    devices = await _call(
                        ws, "devicelist", token=token, type="0", is_check_token="0"
                    )
        except (aiohttp.ClientError, TimeoutError) as err:
            raise IntelbrasSolarConnectionError(f"WiNet {self.url}: {err!r}") from err
        return parse_datalogger(state, faults, devices)

async_get_status async

async_get_status() -> DataloggerStatus

Open a session, read state, fault and devicelist, close.

Source code in src/intelbras_solar_local/winet.py

async def async_get_status(self) -> DataloggerStatus:
    """Open a session, read ``state``, ``fault`` and ``devicelist``, close."""
    try:
        async with asyncio.timeout(self._timeout):
            async with self._session.ws_connect(self.url) as ws:
                token = (await _call(ws, "connect", token=""))["token"]
                state = await _call(ws, "state", token=token)
                faults = await _call(ws, "fault", token=token)
                devices = await _call(
                    ws, "devicelist", token=token, type="0", is_check_token="0"
                )
    except (aiohttp.ClientError, TimeoutError) as err:
        raise IntelbrasSolarConnectionError(f"WiNet {self.url}: {err!r}") from err
    return parse_datalogger(state, faults, devices)

intelbras_solar_local.IntelbrasSolarError

Bases: Exception

Base error for all Intelbras Solar client failures.

Source code in src/intelbras_solar_local/exceptions.py

class IntelbrasSolarError(Exception):
    """Base error for all Intelbras Solar client failures."""

intelbras_solar_local.IntelbrasSolarConnectionError

Bases: IntelbrasSolarError

Raised when the datalogger is unreachable or the transport fails.

Source code in src/intelbras_solar_local/exceptions.py

class IntelbrasSolarConnectionError(IntelbrasSolarError):
    """Raised when the datalogger is unreachable or the transport fails."""

intelbras_solar_local.IntelbrasSolarProtocolError

Bases: IntelbrasSolarError

Raised when the device answers with an error (Modbus exception, bad reply).

Source code in src/intelbras_solar_local/exceptions.py

class IntelbrasSolarProtocolError(IntelbrasSolarError):
    """Raised when the device answers with an error (Modbus exception, bad reply)."""