neopool-modbus

Async Python client for Sugar Valley NeoPool pool controllers (sold under brands VistaPool, Hidrolife, Aquascenic, Oxilife, Hayward, Brilix, Bayrol) connected via Modbus TCP.

This library is the communication layer extracted from the Home Assistant neopool integration and is suitable for any async Python project — Home Assistant integrations, scripts, dashboards, or custom automation.

Installation

pip install neopool-modbus

Requires Python 3.13+ and pymodbus>=3.10.0 (installed transitively).

Quick start

import asyncio

from neopool_modbus import NeoPoolModbusClient


async def main() -> None:
    client = NeoPoolModbusClient(
        {"host": "192.168.1.42", "port": 502, "slave_id": 1}
    )
    try:
        data = await client.async_read_all()
        # Keys are the upstream Sugar Valley register names from
        # https://github.com/arendst/Tasmota/.../xsns_83_neopool.ino,
        # values are decoded into native Python types.
        print(f"pH:          {data['MBF_MEASURE_PH']}")           # e.g. 7.42
        print(f"Temperature: {data['MBF_MEASURE_TEMPERATURE']} °C")  # e.g. 27.3
        print(f"Hydrolysis:  {data['MBF_HIDRO_CURRENT']}")        # e.g. 6.5
    finally:
        await client.close()


asyncio.run(main())

The client is lazy — it opens the TCP connection on first use and reuses it across calls; close() releases the socket and resets retry/backoff state.

Public API

from neopool_modbus import (
    NeoPoolModbusClient,
    NeoPoolError,
    NeoPoolConnectionError,
    NeoPoolModbusError,
    NeoPoolTimeoutError,
    async_probe_serial,
)
from neopool_modbus.registers import (
    DEFAULT_MODBUS_FRAMER,
    EXEC_REGISTER,
    EEPROM_SAVE_REGISTER,
    HEATING_SETPOINT_REGISTER,
    INTELLIGENT_SETPOINT_REGISTER,
    MANUAL_FILTRATION_REGISTER,
    TIMER_BLOCKS,
    is_valid_relay_gpio,
)
from neopool_modbus.decoders import (
    parse_timer_block,
    build_timer_block,
    hhmm_to_seconds,
    seconds_to_hhmm,
    get_machine_name,
    is_hydrolysis_in_percent,
    # ... see neopool_modbus.decoders for the full list
)
from neopool_modbus.status_mask import (
    decode_relay_state,
    decode_named_relay_states,
    decode_uv_lamp_state,
    decode_hidro_status_bits,
    decode_ion_status_bits,
    decode_ph_rx_cl_cd_status_bits,
)

All client methods translate underlying pymodbus exceptions into the NeoPoolError hierarchy at the library boundary, so callers never need to import pymodbus to catch errors:

Class	Raised when
NeoPoolConnectionError	TCP connect fails, returned False, or the client is in its post-failure backoff
NeoPoolTimeoutError	Connect, read, or write times out (asyncio.TimeoutError)
NeoPoolModbusError	A read returns a Modbus exception response (isError() true), or async_write_aux_relay / one of the timer write follow-ups returns isError()
NeoPoolError	Common base; catch this to handle any of the above

⚠️ NeoPoolModbusClient.async_write_register() is the exception to the table above: it returns None (rather than raising) on isError() so existing callers in the Home Assistant integration keep working. A future major release will tighten this to raise NeoPoolModbusError for consistency.

from neopool_modbus import NeoPoolError, NeoPoolModbusClient

client = NeoPoolModbusClient({"host": "192.168.1.42"})
try:
    data = await client.async_read_all()
except NeoPoolError as exc:
    # exc.__cause__ is the original pymodbus / asyncio exception, if any.
    print(f"NeoPool read failed: {exc}")

ValueError is still raised directly for programmer errors such as an out-of-range AUX relay index — those are not transport failures.

Features

• Async I/O on top of pymodbus.AsyncModbusTcpClient
• Batched register reads — one round-trip per protocol page, with notification-bit-driven cache invalidation so unchanged pages skip the read
• Exponential connection retry with bounded backoff
• Write-and-verify cycle for configuration registers
• Capability detection (hydrolysis, pH, Redox, chlorine, conductivity, ION)
• Strict type hints (py.typed), 100 % unit-test coverage

Logging

The library uses a single logger named neopool_modbus. Enable it like any other Python logger:

import logging
logging.getLogger("neopool_modbus").setLevel(logging.DEBUG)

Home Assistant users can flip the integration's "Enable debug logging" toggle in the UI; the integration's manifest.json lists neopool_modbus so the toggle covers the library too.

Based On

• Tasmota NeoPool driver — implements the NeoPool Modbus register protocol originally documented by Sugar Valley
• NeoPool Control System MODBUS Register description — a Markdown transcription of the official Modbus register documentation by Sugar Valley (see docs/modbus-registers.md)

Disclaimer

This library is provided "AS IS" and without any warranty or guarantee of any kind. The author takes no responsibility for any damage, loss, or malfunction resulting from the use or misuse of this code. Use at your own risk.

This project is not affiliated with or endorsed by Sugar Valley, Hayward, or any other pool equipment manufacturer or distributor.

License

Apache 2.0 — see LICENSE.