API Reference

Everything below is auto-generated from the source docstrings. The public surface is curated via __all__ in proconip.__init__, so anything importable as from proconip import … shows up here.

Top-level package

Async Python library for interacting with the ProCon.IP pool controller.

BadCredentialsException

Bases: ProconipApiException

Raised when the controller rejects the request with HTTP 401 or 403.

Almost always means the username or password in the ConfigObject is wrong.

Source code in src/proconip/api.py

class BadCredentialsException(ProconipApiException):
    """Raised when the controller rejects the request with HTTP 401 or 403.

    Almost always means the username or password in the `ConfigObject` is wrong.
    """

BadStatusCodeException

Bases: ProconipApiException

Raised on any unexpected HTTP error status (4xx or 5xx) other than 401/403.

The original aiohttp.ClientResponseError is preserved as the cause and can be inspected via __cause__ if the status code or response details are needed.

Source code in src/proconip/api.py

class BadStatusCodeException(ProconipApiException):
    """Raised on any unexpected HTTP error status (4xx or 5xx) other than 401/403.

    The original `aiohttp.ClientResponseError` is preserved as the cause and
    can be inspected via `__cause__` if the status code or response details
    are needed.
    """

DigitalInputControl

Convenience wrapper that binds a session and config for triggering inputs.

Construct once with your aiohttp.ClientSession and ConfigObject, then pulse digital inputs by zero-based id (0–3) without repeating those arguments each time.

The constructor also binds a default timeout for every call made via this wrapper. Each method then accepts an optional per-call timeout that overrides the bound default when supplied.

Example

dic = DigitalInputControl(session, config)
await dic.async_trigger(0)  # pulse the first digital input

Source code in src/proconip/api.py

class DigitalInputControl:
    """Convenience wrapper that binds a session and config for triggering inputs.

    Construct once with your `aiohttp.ClientSession` and `ConfigObject`, then
    pulse digital inputs by zero-based id (0–3) without repeating those
    arguments each time.

    The constructor also binds a default `timeout` for every call made via this
    wrapper. Each method then accepts an optional per-call `timeout` that
    overrides the bound default when supplied.

    Example:
        ```python
        dic = DigitalInputControl(session, config)
        await dic.async_trigger(0)  # pulse the first digital input
        ```
    """

    def __init__(
        self,
        client_session: ClientSession,
        config: ConfigObject,
        timeout: float = 10.0,
    ):
        """Bind the session, config, and default per-request timeout.

        Args:
            client_session: An open `aiohttp.ClientSession`.
            config: Controller configuration.
            timeout: Default per-request timeout in seconds, used when a method
                is called without its own ``timeout`` argument.
        """
        self.client_session = client_session
        self.config = config
        self.timeout = timeout

    async def async_trigger(
        self,
        digital_input_id: int,
        timeout: float | None = None,
        hold_seconds: float = DIGITAL_INPUT_PULSE_SECONDS,
    ) -> str:
        """Trigger the digital input identified by ``digital_input_id``.

        Args:
            digital_input_id: Zero-based digital input index (0–3).
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.
            hold_seconds: Seconds to hold the input HIGH between press and
                release. Defaults to the web UI's ~600ms.

        See `async_trigger_digital_input` (the free function) for the full
        description of behavior and raised exceptions.
        """
        return await async_trigger_digital_input(
            client_session=self.client_session,
            config=self.config,
            digital_input_id=digital_input_id,
            timeout=self.timeout if timeout is None else timeout,
            hold_seconds=hold_seconds,
        )

async_trigger async

async_trigger(digital_input_id: int, timeout: float | None = None, hold_seconds: float = DIGITAL_INPUT_PULSE_SECONDS) -> str

Trigger the digital input identified by digital_input_id.

PARAMETER	DESCRIPTION
digital_input_id	Zero-based digital input index (0–3). TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None
hold_seconds	Seconds to hold the input HIGH between press and release. Defaults to the web UI's ~600ms. TYPE: float DEFAULT: DIGITAL_INPUT_PULSE_SECONDS

See async_trigger_digital_input (the free function) for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_trigger(
    self,
    digital_input_id: int,
    timeout: float | None = None,
    hold_seconds: float = DIGITAL_INPUT_PULSE_SECONDS,
) -> str:
    """Trigger the digital input identified by ``digital_input_id``.

    Args:
        digital_input_id: Zero-based digital input index (0–3).
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.
        hold_seconds: Seconds to hold the input HIGH between press and
            release. Defaults to the web UI's ~600ms.

    See `async_trigger_digital_input` (the free function) for the full
    description of behavior and raised exceptions.
    """
    return await async_trigger_digital_input(
        client_session=self.client_session,
        config=self.config,
        digital_input_id=digital_input_id,
        timeout=self.timeout if timeout is None else timeout,
        hold_seconds=hold_seconds,
    )

DmxControl

Convenience wrapper that binds a session and config for DMX I/O.

Construct once with your aiohttp.ClientSession and ConfigObject, then read or write DMX channels without repeating those arguments each time.

The constructor also binds a default timeout for every call made via this wrapper. Each method then accepts an optional per-call timeout that overrides the bound default when supplied.

Example

dc = DmxControl(session, config)
dmx = await dc.async_get_dmx()
for ch in dmx:
    dmx.set(ch.index, (ch.value + 64) % 256)
await dc.async_set(dmx)

Source code in src/proconip/api.py

class DmxControl:
    """Convenience wrapper that binds a session and config for DMX I/O.

    Construct once with your `aiohttp.ClientSession` and `ConfigObject`, then
    read or write DMX channels without repeating those arguments each time.

    The constructor also binds a default `timeout` for every call made via
    this wrapper. Each method then accepts an optional per-call `timeout`
    that overrides the bound default when supplied.

    Example:
        ```python
        dc = DmxControl(session, config)
        dmx = await dc.async_get_dmx()
        for ch in dmx:
            dmx.set(ch.index, (ch.value + 64) % 256)
        await dc.async_set(dmx)
        ```
    """

    def __init__(
        self,
        client_session: ClientSession,
        config: ConfigObject,
        timeout: float = 10.0,
    ):
        """Bind the session, config, and default per-request timeout.

        Args:
            client_session: An open `aiohttp.ClientSession`.
            config: Controller configuration.
            timeout: Default per-request timeout in seconds, used when a
                method is called without its own ``timeout`` argument.
        """
        self.client_session = client_session
        self.config = config
        self.timeout = timeout

    async def async_get_raw_dmx(self, timeout: float | None = None) -> str:
        """Fetch the raw `/GetDmx.csv` body using the bound session and config.

        Args:
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        See `async_get_raw_dmx` (the free function) for the full description
        of behavior and raised exceptions.
        """
        return await async_get_raw_dmx(
            self.client_session,
            self.config,
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_get_dmx(self, timeout: float | None = None) -> GetDmxData:
        """Fetch and parse the current DMX state into a `GetDmxData` instance.

        Args:
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        See `async_get_dmx` (the free function) for the full description of
        behavior and raised exceptions.
        """
        return await async_get_dmx(
            self.client_session,
            self.config,
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_set(self, data: GetDmxData, timeout: float | None = None) -> str:
        """Push the given DMX state back to the controller.

        Args:
            data: The `GetDmxData` to write.
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        See `async_set_dmx` (the free function) for the full description of
        behavior and raised exceptions.
        """
        return await async_set_dmx(
            client_session=self.client_session,
            config=self.config,
            dmx_states=data,
            timeout=self.timeout if timeout is None else timeout,
        )

async_get_raw_dmx async

async_get_raw_dmx(timeout: float | None = None) -> str

Fetch the raw /GetDmx.csv body using the bound session and config.

PARAMETER	DESCRIPTION
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

See async_get_raw_dmx (the free function) for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_get_raw_dmx(self, timeout: float | None = None) -> str:
    """Fetch the raw `/GetDmx.csv` body using the bound session and config.

    Args:
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    See `async_get_raw_dmx` (the free function) for the full description
    of behavior and raised exceptions.
    """
    return await async_get_raw_dmx(
        self.client_session,
        self.config,
        timeout=self.timeout if timeout is None else timeout,
    )

async_get_dmx async

async_get_dmx(timeout: float | None = None) -> GetDmxData

Fetch and parse the current DMX state into a GetDmxData instance.

PARAMETER	DESCRIPTION
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

See async_get_dmx (the free function) for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_get_dmx(self, timeout: float | None = None) -> GetDmxData:
    """Fetch and parse the current DMX state into a `GetDmxData` instance.

    Args:
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    See `async_get_dmx` (the free function) for the full description of
    behavior and raised exceptions.
    """
    return await async_get_dmx(
        self.client_session,
        self.config,
        timeout=self.timeout if timeout is None else timeout,
    )

async_set async

async_set(data: GetDmxData, timeout: float | None = None) -> str

Push the given DMX state back to the controller.

PARAMETER	DESCRIPTION
data	The GetDmxData to write. TYPE: GetDmxData
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

See async_set_dmx (the free function) for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_set(self, data: GetDmxData, timeout: float | None = None) -> str:
    """Push the given DMX state back to the controller.

    Args:
        data: The `GetDmxData` to write.
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    See `async_set_dmx` (the free function) for the full description of
    behavior and raised exceptions.
    """
    return await async_set_dmx(
        client_session=self.client_session,
        config=self.config,
        dmx_states=data,
        timeout=self.timeout if timeout is None else timeout,
    )

DosageControl

Convenience wrapper for manual dosage commands.

Construct once with your aiohttp.ClientSession and ConfigObject, then trigger dosing per chemical without specifying the DosageTarget enum each time.

The constructor also binds a default timeout for every call made via this wrapper. Each method then accepts an optional per-call timeout that overrides the bound default when supplied.

Example

dc = DosageControl(session, config)
await dc.async_chlorine_dosage(60)   # 60 seconds of chlorine

Source code in src/proconip/api.py

class DosageControl:
    """Convenience wrapper for manual dosage commands.

    Construct once with your `aiohttp.ClientSession` and `ConfigObject`, then
    trigger dosing per chemical without specifying the `DosageTarget` enum
    each time.

    The constructor also binds a default `timeout` for every call made via
    this wrapper. Each method then accepts an optional per-call `timeout`
    that overrides the bound default when supplied.

    Example:
        ```python
        dc = DosageControl(session, config)
        await dc.async_chlorine_dosage(60)   # 60 seconds of chlorine
        ```
    """

    def __init__(
        self,
        client_session: ClientSession,
        config: ConfigObject,
        timeout: float = 10.0,
    ):
        """Bind the session, config, and default per-request timeout.

        Args:
            client_session: An open `aiohttp.ClientSession`.
            config: Controller configuration.
            timeout: Default per-request timeout in seconds, used when a
                method is called without its own ``timeout`` argument.
        """
        self.client_session = client_session
        self.config = config
        self.timeout = timeout

    async def async_chlorine_dosage(
        self, dosage_duration: int, timeout: float | None = None
    ) -> str:
        """Run the chlorine dosage pump for ``dosage_duration`` seconds.

        Args:
            dosage_duration: Run time in seconds.
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        Subject to the same controller-side safety interlocks as manual
        dosage from the web UI. See `async_start_dosage` for full behavior
        and raised exceptions.
        """
        return await async_start_dosage(
            client_session=self.client_session,
            config=self.config,
            dosage_target=DosageTarget.CHLORINE,
            dosage_duration=dosage_duration,
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_ph_minus_dosage(
        self, dosage_duration: int, timeout: float | None = None
    ) -> str:
        """Run the pH- dosage pump for ``dosage_duration`` seconds.

        Args:
            dosage_duration: Run time in seconds.
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        Subject to the same controller-side safety interlocks as manual
        dosage from the web UI. See `async_start_dosage` for full behavior
        and raised exceptions.
        """
        return await async_start_dosage(
            client_session=self.client_session,
            config=self.config,
            dosage_target=DosageTarget.PH_MINUS,
            dosage_duration=dosage_duration,
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_ph_plus_dosage(self, dosage_duration: int, timeout: float | None = None) -> str:
        """Run the pH+ dosage pump for ``dosage_duration`` seconds.

        Args:
            dosage_duration: Run time in seconds.
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        Subject to the same controller-side safety interlocks as manual
        dosage from the web UI. See `async_start_dosage` for full behavior
        and raised exceptions.
        """
        return await async_start_dosage(
            client_session=self.client_session,
            config=self.config,
            dosage_target=DosageTarget.PH_PLUS,
            dosage_duration=dosage_duration,
            timeout=self.timeout if timeout is None else timeout,
        )

async_chlorine_dosage async

async_chlorine_dosage(dosage_duration: int, timeout: float | None = None) -> str

Run the chlorine dosage pump for dosage_duration seconds.

PARAMETER	DESCRIPTION
dosage_duration	Run time in seconds. TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

Subject to the same controller-side safety interlocks as manual dosage from the web UI. See async_start_dosage for full behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_chlorine_dosage(
    self, dosage_duration: int, timeout: float | None = None
) -> str:
    """Run the chlorine dosage pump for ``dosage_duration`` seconds.

    Args:
        dosage_duration: Run time in seconds.
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    Subject to the same controller-side safety interlocks as manual
    dosage from the web UI. See `async_start_dosage` for full behavior
    and raised exceptions.
    """
    return await async_start_dosage(
        client_session=self.client_session,
        config=self.config,
        dosage_target=DosageTarget.CHLORINE,
        dosage_duration=dosage_duration,
        timeout=self.timeout if timeout is None else timeout,
    )

async_ph_minus_dosage async

async_ph_minus_dosage(dosage_duration: int, timeout: float | None = None) -> str

Run the pH- dosage pump for dosage_duration seconds.

PARAMETER	DESCRIPTION
dosage_duration	Run time in seconds. TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

Subject to the same controller-side safety interlocks as manual dosage from the web UI. See async_start_dosage for full behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_ph_minus_dosage(
    self, dosage_duration: int, timeout: float | None = None
) -> str:
    """Run the pH- dosage pump for ``dosage_duration`` seconds.

    Args:
        dosage_duration: Run time in seconds.
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    Subject to the same controller-side safety interlocks as manual
    dosage from the web UI. See `async_start_dosage` for full behavior
    and raised exceptions.
    """
    return await async_start_dosage(
        client_session=self.client_session,
        config=self.config,
        dosage_target=DosageTarget.PH_MINUS,
        dosage_duration=dosage_duration,
        timeout=self.timeout if timeout is None else timeout,
    )

async_ph_plus_dosage async

async_ph_plus_dosage(dosage_duration: int, timeout: float | None = None) -> str

Run the pH+ dosage pump for dosage_duration seconds.

PARAMETER	DESCRIPTION
dosage_duration	Run time in seconds. TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

Subject to the same controller-side safety interlocks as manual dosage from the web UI. See async_start_dosage for full behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_ph_plus_dosage(self, dosage_duration: int, timeout: float | None = None) -> str:
    """Run the pH+ dosage pump for ``dosage_duration`` seconds.

    Args:
        dosage_duration: Run time in seconds.
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    Subject to the same controller-side safety interlocks as manual
    dosage from the web UI. See `async_start_dosage` for full behavior
    and raised exceptions.
    """
    return await async_start_dosage(
        client_session=self.client_session,
        config=self.config,
        dosage_target=DosageTarget.PH_PLUS,
        dosage_duration=dosage_duration,
        timeout=self.timeout if timeout is None else timeout,
    )

GetState

Convenience wrapper that binds a session and config for state reads.

Construct once with your aiohttp.ClientSession and ConfigObject, then call async_get_state() (parsed) or async_get_raw_state() (CSV) without repeating those arguments each time.

The constructor also binds a default timeout for every call made via this wrapper. Each method then accepts an optional per-call timeout that overrides the bound default when supplied.

Example

async with aiohttp.ClientSession() as session:
    api = GetState(session, config, timeout=15.0)
    state = await api.async_get_state()              # uses 15.0 s
    quick = await api.async_get_state(timeout=2.0)   # overrides to 2.0 s
    print(state.ph_electrode.display_value)

Source code in src/proconip/api.py

class GetState:
    """Convenience wrapper that binds a session and config for state reads.

    Construct once with your `aiohttp.ClientSession` and `ConfigObject`, then
    call `async_get_state()` (parsed) or `async_get_raw_state()` (CSV) without
    repeating those arguments each time.

    The constructor also binds a default `timeout` for every call made via
    this wrapper. Each method then accepts an optional per-call `timeout`
    that overrides the bound default when supplied.

    Example:
        ```python
        async with aiohttp.ClientSession() as session:
            api = GetState(session, config, timeout=15.0)
            state = await api.async_get_state()              # uses 15.0 s
            quick = await api.async_get_state(timeout=2.0)   # overrides to 2.0 s
            print(state.ph_electrode.display_value)
        ```
    """

    def __init__(
        self,
        client_session: ClientSession,
        config: ConfigObject,
        timeout: float = 10.0,
    ):
        """Bind the session, config, and default per-request timeout.

        Args:
            client_session: An open `aiohttp.ClientSession`.
            config: Controller configuration.
            timeout: Default per-request timeout in seconds, used when a
                method is called without its own ``timeout`` argument.
        """
        self.client_session = client_session
        self.config = config
        self.timeout = timeout

    async def async_get_raw_state(self, timeout: float | None = None) -> str:
        """Fetch the raw `/GetState.csv` body using the bound session and config.

        Args:
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        See `async_get_raw_state` (the free function) for the full description
        of behavior and raised exceptions.
        """
        return await async_get_raw_state(
            self.client_session,
            self.config,
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_get_state(self, timeout: float | None = None) -> GetStateData:
        """Fetch and parse the controller state into a `GetStateData` instance.

        Args:
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        See `async_get_state` (the free function) for the full description of
        behavior and raised exceptions.
        """
        return await async_get_state(
            self.client_session,
            self.config,
            timeout=self.timeout if timeout is None else timeout,
        )

async_get_raw_state async

async_get_raw_state(timeout: float | None = None) -> str

Fetch the raw /GetState.csv body using the bound session and config.

PARAMETER	DESCRIPTION
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

See async_get_raw_state (the free function) for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_get_raw_state(self, timeout: float | None = None) -> str:
    """Fetch the raw `/GetState.csv` body using the bound session and config.

    Args:
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    See `async_get_raw_state` (the free function) for the full description
    of behavior and raised exceptions.
    """
    return await async_get_raw_state(
        self.client_session,
        self.config,
        timeout=self.timeout if timeout is None else timeout,
    )

async_get_state async

async_get_state(timeout: float | None = None) -> GetStateData

Fetch and parse the controller state into a GetStateData instance.

PARAMETER	DESCRIPTION
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

See async_get_state (the free function) for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_get_state(self, timeout: float | None = None) -> GetStateData:
    """Fetch and parse the controller state into a `GetStateData` instance.

    Args:
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    See `async_get_state` (the free function) for the full description of
    behavior and raised exceptions.
    """
    return await async_get_state(
        self.client_session,
        self.config,
        timeout=self.timeout if timeout is None else timeout,
    )

ProconipApiException

Bases: Exception

Base exception for any failed API call.

Catch this if you want to handle all controller-side and network failures uniformly. The more specific subclasses below let you distinguish auth, HTTP, and timeout problems when that's useful.

Source code in src/proconip/api.py

class ProconipApiException(Exception):
    """Base exception for any failed API call.

    Catch this if you want to handle all controller-side and network failures
    uniformly. The more specific subclasses below let you distinguish auth,
    HTTP, and timeout problems when that's useful.
    """

RelaySwitch

Convenience wrapper that binds a session and config for relay control.

Construct once with your aiohttp.ClientSession and ConfigObject, then switch relays by aggregated relay ID (an integer) instead of constructing Relay instances by hand.

Aggregated relay IDs run from 0 to 7 for the eight built-in relays and 8 to 15 for the eight optional external relays.

The constructor also binds a default timeout for every call made via this wrapper. Each method then accepts an optional per-call timeout that overrides the bound default when supplied.

Example

rs = RelaySwitch(session, config)
state = await GetState(session, config).async_get_state()
await rs.async_switch_on(state, relay_id=2)

Source code in src/proconip/api.py

class RelaySwitch:
    """Convenience wrapper that binds a session and config for relay control.

    Construct once with your `aiohttp.ClientSession` and `ConfigObject`, then
    switch relays by **aggregated relay ID** (an integer) instead of
    constructing `Relay` instances by hand.

    Aggregated relay IDs run from 0 to 7 for the eight built-in relays and 8
    to 15 for the eight optional external relays.

    The constructor also binds a default `timeout` for every call made via
    this wrapper. Each method then accepts an optional per-call `timeout`
    that overrides the bound default when supplied.

    Example:
        ```python
        rs = RelaySwitch(session, config)
        state = await GetState(session, config).async_get_state()
        await rs.async_switch_on(state, relay_id=2)
        ```
    """

    def __init__(
        self,
        client_session: ClientSession,
        config: ConfigObject,
        timeout: float = 10.0,
    ):
        """Bind the session, config, and default per-request timeout.

        Args:
            client_session: An open `aiohttp.ClientSession`.
            config: Controller configuration.
            timeout: Default per-request timeout in seconds, used when a
                method is called without its own ``timeout`` argument.
        """
        self.client_session = client_session
        self.config = config
        self.timeout = timeout

    async def async_switch_on(
        self,
        current_state: GetStateData,
        relay_id: int,
        timeout: float | None = None,
    ) -> str:
        """Switch the relay identified by ``relay_id`` to manual ON.

        Args:
            current_state: A recent `GetStateData` snapshot.
            relay_id: Aggregated relay ID (0–15).
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        Resolves the relay ID against ``current_state`` and delegates to the
        free function `async_switch_on`. See it for the full description of
        behavior and raised exceptions, including `BadRelayException` for
        dosage relays.
        """
        return await async_switch_on(
            client_session=self.client_session,
            config=self.config,
            current_state=current_state,
            relay=current_state.get_relay(relay_id),
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_switch_off(
        self,
        current_state: GetStateData,
        relay_id: int,
        timeout: float | None = None,
    ) -> str:
        """Switch the relay identified by ``relay_id`` to manual OFF.

        Args:
            current_state: A recent `GetStateData` snapshot.
            relay_id: Aggregated relay ID (0–15).
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        Resolves the relay ID against ``current_state`` and delegates to the
        free function `async_switch_off`. See it for the full description of
        behavior and raised exceptions.
        """
        return await async_switch_off(
            client_session=self.client_session,
            config=self.config,
            current_state=current_state,
            relay=current_state.get_relay(relay_id),
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_set_auto_mode(
        self,
        current_state: GetStateData,
        relay_id: int,
        timeout: float | None = None,
    ) -> str:
        """Hand the relay identified by ``relay_id`` back to AUTO mode.

        Args:
            current_state: A recent `GetStateData` snapshot.
            relay_id: Aggregated relay ID (0–15).
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        Resolves the relay ID against ``current_state`` and delegates to the
        free function `async_set_auto_mode`. See it for the full description
        of behavior and raised exceptions.
        """
        return await async_set_auto_mode(
            client_session=self.client_session,
            config=self.config,
            current_state=current_state,
            relay=current_state.get_relay(relay_id),
            timeout=self.timeout if timeout is None else timeout,
        )

async_switch_on async

async_switch_on(current_state: GetStateData, relay_id: int, timeout: float | None = None) -> str

Switch the relay identified by relay_id to manual ON.

PARAMETER	DESCRIPTION
current_state	A recent GetStateData snapshot. TYPE: GetStateData
relay_id	Aggregated relay ID (0–15). TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

Resolves the relay ID against current_state and delegates to the free function async_switch_on. See it for the full description of behavior and raised exceptions, including BadRelayException for dosage relays.

Source code in src/proconip/api.py

async def async_switch_on(
    self,
    current_state: GetStateData,
    relay_id: int,
    timeout: float | None = None,
) -> str:
    """Switch the relay identified by ``relay_id`` to manual ON.

    Args:
        current_state: A recent `GetStateData` snapshot.
        relay_id: Aggregated relay ID (0–15).
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    Resolves the relay ID against ``current_state`` and delegates to the
    free function `async_switch_on`. See it for the full description of
    behavior and raised exceptions, including `BadRelayException` for
    dosage relays.
    """
    return await async_switch_on(
        client_session=self.client_session,
        config=self.config,
        current_state=current_state,
        relay=current_state.get_relay(relay_id),
        timeout=self.timeout if timeout is None else timeout,
    )

async_switch_off async

async_switch_off(current_state: GetStateData, relay_id: int, timeout: float | None = None) -> str

Switch the relay identified by relay_id to manual OFF.

PARAMETER	DESCRIPTION
current_state	A recent GetStateData snapshot. TYPE: GetStateData
relay_id	Aggregated relay ID (0–15). TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

Resolves the relay ID against current_state and delegates to the free function async_switch_off. See it for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_switch_off(
    self,
    current_state: GetStateData,
    relay_id: int,
    timeout: float | None = None,
) -> str:
    """Switch the relay identified by ``relay_id`` to manual OFF.

    Args:
        current_state: A recent `GetStateData` snapshot.
        relay_id: Aggregated relay ID (0–15).
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    Resolves the relay ID against ``current_state`` and delegates to the
    free function `async_switch_off`. See it for the full description of
    behavior and raised exceptions.
    """
    return await async_switch_off(
        client_session=self.client_session,
        config=self.config,
        current_state=current_state,
        relay=current_state.get_relay(relay_id),
        timeout=self.timeout if timeout is None else timeout,
    )

async_set_auto_mode async

async_set_auto_mode(current_state: GetStateData, relay_id: int, timeout: float | None = None) -> str

Hand the relay identified by relay_id back to AUTO mode.

PARAMETER	DESCRIPTION
current_state	A recent GetStateData snapshot. TYPE: GetStateData
relay_id	Aggregated relay ID (0–15). TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

Resolves the relay ID against current_state and delegates to the free function async_set_auto_mode. See it for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_set_auto_mode(
    self,
    current_state: GetStateData,
    relay_id: int,
    timeout: float | None = None,
) -> str:
    """Hand the relay identified by ``relay_id`` back to AUTO mode.

    Args:
        current_state: A recent `GetStateData` snapshot.
        relay_id: Aggregated relay ID (0–15).
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    Resolves the relay ID against ``current_state`` and delegates to the
    free function `async_set_auto_mode`. See it for the full description
    of behavior and raised exceptions.
    """
    return await async_set_auto_mode(
        client_session=self.client_session,
        config=self.config,
        current_state=current_state,
        relay=current_state.get_relay(relay_id),
        timeout=self.timeout if timeout is None else timeout,
    )

TimeoutException

Bases: ProconipApiException

Raised when a request does not complete within the configured timeout.

This covers both network-level stalls (connection or socket I/O) and slow response bodies, since the timeout context wraps the entire exchange.

Source code in src/proconip/api.py

class TimeoutException(ProconipApiException):
    """Raised when a request does not complete within the configured timeout.

    This covers both network-level stalls (connection or socket I/O) and slow
    response bodies, since the timeout context wraps the entire exchange.
    """

BadRelayException

Bases: Exception

Raised when a relay argument doesn't make sense in context.

The two main cases are: switching a dosage relay on directly (rejected by proconip.api.async_switch_on), and passing a non-relay DataObject to GetStateData.is_dosage_relay.

Source code in src/proconip/definitions.py

class BadRelayException(Exception):
    """Raised when a relay argument doesn't make sense in context.

    The two main cases are: switching a dosage relay on directly (rejected
    by `proconip.api.async_switch_on`), and passing a non-relay `DataObject`
    to `GetStateData.is_dosage_relay`.
    """

ConfigObject

Base URL and credentials for talking to a single ProCon.IP controller.

Instances are plain holders — no network connection is opened until they are passed into one of the API helpers in proconip.api.

Source code in src/proconip/definitions.py

class ConfigObject:
    """Base URL and credentials for talking to a single ProCon.IP controller.

    Instances are plain holders — no network connection is opened until they
    are passed into one of the API helpers in `proconip.api`.
    """

    def __init__(
        self,
        base_url: str,
        username: str,
        password: str,
    ):
        """Build a config from explicit values.

        Args:
            base_url: Root URL of the controller, e.g. ``http://192.168.1.50``.
                The library appends the API paths itself; do not include them
                here. Plain HTTP is normal — these controllers are LAN-only.
            username: HTTP Basic auth username (controller default: ``admin``).
            password: HTTP Basic auth password (controller default: ``admin``).
        """
        self.base_url = base_url
        self.username = username
        self.password = password

    @staticmethod
    def from_dict(data: dict[str, str]) -> "ConfigObject":
        """Build a `ConfigObject` from a serialized dictionary.

        Useful for restoring a config that was previously written out via
        `to_dict` (e.g. into a Home Assistant config entry).

        Args:
            data: A dict containing the keys ``base_url``, ``username``, and
                ``password``. All three are required.

        Returns:
            A new `ConfigObject` populated from the dict.

        Raises:
            ValueError: If any of the required keys is missing. The exception
                message names the missing key.
        """
        if "base_url" not in data:
            raise ValueError("base_url is required")
        if "username" not in data:
            raise ValueError("username is required")
        if "password" not in data:
            raise ValueError("password is required")
        return ConfigObject(data["base_url"], data["username"], data["password"])

    def to_dict(self) -> dict[str, str]:
        """Return a plain dict copy of the config, suitable for serialization.

        The password is stored in the clear — encrypt the dict yourself if it
        will be persisted somewhere readable.
        """
        return {
            "base_url": self.base_url,
            "username": self.username,
            "password": self.password,
        }

from_dict staticmethod

from_dict(data: dict[str, str]) -> ConfigObject

Build a ConfigObject from a serialized dictionary.

Useful for restoring a config that was previously written out via to_dict (e.g. into a Home Assistant config entry).

PARAMETER	DESCRIPTION
data	A dict containing the keys base_url, username, and password. All three are required. TYPE: dict[str, str]

RETURNS	DESCRIPTION
ConfigObject	A new ConfigObject populated from the dict.

RAISES	DESCRIPTION
ValueError	If any of the required keys is missing. The exception message names the missing key.

Source code in src/proconip/definitions.py

@staticmethod
def from_dict(data: dict[str, str]) -> "ConfigObject":
    """Build a `ConfigObject` from a serialized dictionary.

    Useful for restoring a config that was previously written out via
    `to_dict` (e.g. into a Home Assistant config entry).

    Args:
        data: A dict containing the keys ``base_url``, ``username``, and
            ``password``. All three are required.

    Returns:
        A new `ConfigObject` populated from the dict.

    Raises:
        ValueError: If any of the required keys is missing. The exception
            message names the missing key.
    """
    if "base_url" not in data:
        raise ValueError("base_url is required")
    if "username" not in data:
        raise ValueError("username is required")
    if "password" not in data:
        raise ValueError("password is required")
    return ConfigObject(data["base_url"], data["username"], data["password"])

to_dict

to_dict() -> dict[str, str]

Return a plain dict copy of the config, suitable for serialization.

The password is stored in the clear — encrypt the dict yourself if it will be persisted somewhere readable.

Source code in src/proconip/definitions.py

def to_dict(self) -> dict[str, str]:
    """Return a plain dict copy of the config, suitable for serialization.

    The password is stored in the clear — encrypt the dict yourself if it
    will be persisted somewhere readable.
    """
    return {
        "base_url": self.base_url,
        "username": self.username,
        "password": self.password,
    }

DataObject

A single sensor, relay, canister, or consumption channel from /GetState.csv.

Each DataObject represents one column of the CSV response, combining the name, unit, offset, gain, and raw value rows that the controller sends. The column index alone determines which category the object falls into (analog, relay, temperature, …) — see the constructor for the exact ranges.

The actual physical reading is computed once at construction via offset + gain * raw_value and exposed as value. A pre-formatted display_value string is also produced; for relay columns it is one of "Auto (off)", "Auto (on)", "Off", or "On".

Source code in src/proconip/definitions.py

class DataObject:
    """A single sensor, relay, canister, or consumption channel from `/GetState.csv`.

    Each `DataObject` represents one column of the CSV response, combining the
    name, unit, offset, gain, and raw value rows that the controller sends. The
    column index alone determines which category the object falls into (analog,
    relay, temperature, …) — see the constructor for the exact ranges.

    The actual physical reading is computed once at construction via
    ``offset + gain * raw_value`` and exposed as `value`. A pre-formatted
    `display_value` string is also produced; for relay columns it is one of
    "Auto (off)", "Auto (on)", "Off", or "On".
    """

    _column: int
    _category: str
    _category_id: int
    _name: str
    _unit: str
    _offset: float
    _gain: float
    _raw_value: float
    _value: float
    _display_value: str

    def __init__(
        self,
        column: int,
        name: str,
        unit: str,
        offset: float,
        gain: float,
        value: float,
    ):
        """Build a `DataObject` from one column's worth of CSV data.

        Args:
            column: Zero-based column index in the CSV. Determines the category:
                ``0`` → time, ``1–5`` → analog, ``6–7`` → electrode, ``8–15`` →
                temperature, ``16–23`` → relay, ``24–27`` → digital input,
                ``28–35`` → external relay, ``36–38`` → canister, ``39–41`` →
                consumption. Anything outside this range falls into a sentinel
                "uncategorized" bucket.
            name: Sensor name as reported by the controller (e.g. ``"Redox"``).
            unit: Unit string (``"mV"``, ``"°C"``, ``"%"``, ``"--"``, …).
            offset: Calibration offset applied to ``value``.
            gain: Calibration gain applied to ``value``.
            value: Raw sensor value before calibration. Stored verbatim as
                `raw_value`; the physical `value` is computed as
                ``offset + gain * raw_value``.
        """
        self._column = column
        self._name = name
        self._unit = unit
        self._offset = offset
        self._gain = gain
        self._raw_value = value
        self._value = self._offset + (self._gain * self._raw_value)

        if column == 0:
            self._category = CATEGORY_TIME
            self._category_id = 0
            self._display_value = f"{int(self._value / 256):02d}:{int(self._value) % 256:02d}"
        elif 1 <= column <= 5:
            self._category = CATEGORY_ANALOG
            self._category_id = column - 1
            self._display_value = f"{self._value:.2f} {self._unit}"
        elif 6 <= column <= 7:
            self._category = CATEGORY_ELECTRODE
            self._category_id = column - 6
            self._display_value = f"{self._value:.2f} {self._unit}"
        elif 8 <= column <= 15:
            self._category = CATEGORY_TEMPERATURE
            self._category_id = column - 8
            self._display_value = f"{self._value:.2f} °{self._unit}"
        elif 16 <= column <= 23:
            self._category = CATEGORY_RELAY
            self._category_id = column - 16
            self._display_value = self._relay_state()
        elif 24 <= column <= 27:
            self._category = CATEGORY_DIGITAL_INPUT
            self._category_id = column - 24
            self._display_value = f"{self._value}"
        elif 28 <= column <= 35:
            self._category = CATEGORY_EXTERNAL_RELAY
            self._category_id = column - 28
            self._display_value = self._relay_state()
        elif 36 <= column <= 38:
            self._category = CATEGORY_CANISTER
            self._category_id = column - 36
            self._display_value = f"{self._value:.2f} {self._unit}"
        elif 39 <= column <= 41:
            self._category = CATEGORY_CONSUMPTION
            self._category_id = column - 39
            self._display_value = f"{self._value:.2f} {self._unit}"
        else:
            self._category = ""
            self._category_id = -1
            self._display_value = f"{self._value}"

    def __str__(self) -> str:
        """Return a short ``"name (unit): value"`` representation."""
        return f"{self._name} ({self._unit}): {self._value}"

    def _relay_state(self) -> str:
        """Render the current relay value as one of the four state strings.

        Raises:
            ValueError: If `self._value` is not one of the four valid relay
                states (0, 1, 2, 3). Indicates a malformed CSV payload.
        """
        if self._value == 0:
            return "Auto (off)"
        if self._value == 1:
            return "Auto (on)"
        if self._value == 2:
            return "Off"
        if self._value == 3:
            return "On"
        raise ValueError(f"Unexpected relay value {self._value}")

    @property
    def name(self) -> str:
        """Sensor name as reported by the controller."""
        return self._name

    @property
    def unit(self) -> str:
        """Unit string (e.g. ``"mV"``, ``"°C"``, ``"%"``)."""
        return self._unit

    @property
    def offset(self) -> float:
        """Calibration offset applied when computing `value` from `raw_value`."""
        return self._offset

    @property
    def gain(self) -> float:
        """Calibration gain applied when computing `value` from `raw_value`."""
        return self._gain

    @property
    def raw_value(self) -> float:
        """Raw value as received from the controller, before calibration."""
        return self._raw_value

    @property
    def value(self) -> float:
        """Physical value: ``offset + gain * raw_value``."""
        return self._value

    @property
    def display_value(self) -> str:
        """Pre-formatted human-readable string for display.

        For sensors this is ``value`` rendered with its unit and two decimal
        places. For relay columns it is one of "Auto (off)", "Auto (on)",
        "Off", or "On". For column 0 (the system time field) it is "HH:MM".
        """
        return self._display_value

    @property
    def column(self) -> int:
        """Zero-based column index this object came from in the raw CSV."""
        return self._column

    @property
    def category(self) -> str:
        """One of the `CATEGORY_*` constants identifying the entity type."""
        return self._category

    @property
    def category_id(self) -> int:
        """Zero-based index of this object within its category.

        For example, the third temperature sensor has ``category_id == 2``.
        Use `Relay.relay_id` instead if you need the aggregated relay ID
        across both the internal and external relay banks.
        """
        return self._category_id

name property

name: str

Sensor name as reported by the controller.

unit property

unit: str

Unit string (e.g. "mV", "°C", "%").

offset property

offset: float

Calibration offset applied when computing value from raw_value.

gain property

gain: float

Calibration gain applied when computing value from raw_value.

raw_value property

raw_value: float

Raw value as received from the controller, before calibration.

value property

value: float

Physical value: offset + gain * raw_value.

display_value property

display_value: str

Pre-formatted human-readable string for display.

For sensors this is value rendered with its unit and two decimal places. For relay columns it is one of "Auto (off)", "Auto (on)", "Off", or "On". For column 0 (the system time field) it is "HH:MM".

column property

column: int

Zero-based column index this object came from in the raw CSV.

category property

category: str

One of the CATEGORY_* constants identifying the entity type.

category_id property

category_id: int

Zero-based index of this object within its category.

For example, the third temperature sensor has category_id == 2. Use Relay.relay_id instead if you need the aggregated relay ID across both the internal and external relay banks.

DigitalInput

Bases: DataObject

Typed view of a single digital input (CSV columns 24–27).

Construct one by passing the DataObject you got from GetStateData.digital_input_objects; the column, name, calibration, and raw value are copied across so the physical value is computed exactly once. GetStateData.digital_inputs() is a shorthand that does this for you.

The wrapper exists so digital inputs can be triggered (not just read): get_bit_mask yields the bit used in the controller's WEBIO IO field.

Source code in src/proconip/definitions.py

class DigitalInput(DataObject):
    """Typed view of a single digital input (CSV columns 24–27).

    Construct one by passing the `DataObject` you got from
    `GetStateData.digital_input_objects`; the column, name, calibration, and
    raw value are copied across so the physical value is computed exactly once.
    ``GetStateData.digital_inputs()`` is a shorthand that does this for you.

    The wrapper exists so digital inputs can be *triggered* (not just read):
    `get_bit_mask` yields the bit used in the controller's WEBIO ``IO`` field.
    """

    def __init__(self, data_object: DataObject):
        """Wrap an existing digital-input `DataObject`.

        Args:
            data_object: A `DataObject` produced by parsing a `/GetState.csv`
                response. It should be in the digital_input category, but no
                check is enforced — passing another object yields a
                `DigitalInput` whose `get_bit_mask` is meaningless.
        """
        super().__init__(
            data_object.column,
            data_object.name,
            data_object.unit,
            data_object.offset,
            data_object.gain,
            data_object.raw_value,  # pass raw value so offset+gain are applied exactly once
        )

    def get_bit_mask(self) -> int:
        """Bit for this input in the WEBIO ``IO`` field (``1 << category_id``).

        The four digital inputs occupy bits 0–3, so this returns 1, 2, 4, or 8.
        To *trigger* an input, pass its ``category_id`` to
        `async_trigger_digital_input` (which derives the same mask itself); use
        this method when assembling an ``IO`` payload for a manual
        `async_post_usrcfg_cgi` write.
        """
        return 1 << self._category_id

get_bit_mask

get_bit_mask() -> int

Bit for this input in the WEBIO IO field (1 << category_id).

The four digital inputs occupy bits 0–3, so this returns 1, 2, 4, or 8. To trigger an input, pass its category_id to async_trigger_digital_input (which derives the same mask itself); use this method when assembling an IO payload for a manual async_post_usrcfg_cgi write.

Source code in src/proconip/definitions.py

def get_bit_mask(self) -> int:
    """Bit for this input in the WEBIO ``IO`` field (``1 << category_id``).

    The four digital inputs occupy bits 0–3, so this returns 1, 2, 4, or 8.
    To *trigger* an input, pass its ``category_id`` to
    `async_trigger_digital_input` (which derives the same mask itself); use
    this method when assembling an ``IO`` payload for a manual
    `async_post_usrcfg_cgi` write.
    """
    return 1 << self._category_id

DmxChannelData

A single DMX channel's index, name, and current value.

ATTRIBUTE	DESCRIPTION
value	Current channel intensity, expected in the range [0, 255]. The constructor stores the value verbatim; clamping happens in GetDmxData.set. TYPE: int

Source code in src/proconip/definitions.py

class DmxChannelData:
    """A single DMX channel's index, name, and current value.

    Attributes:
        value: Current channel intensity, expected in the range [0, 255].
            The constructor stores the value verbatim; clamping happens in
            `GetDmxData.set`.
    """

    value: int
    _index: int
    _name: str

    def __init__(self, index: int, value: int):
        """Build a channel entry.

        Args:
            index: Zero-based channel index (0 = channel 1, 15 = channel 16).
            value: Initial channel intensity. The constructor does not clamp
                values — out-of-range inputs are stored verbatim. Use
                `GetDmxData.set` if you want the [0, 255] clamp.
        """
        self.value = value
        self._index = index
        self._name = f"CH{index + 1:0>2}"

    @property
    def index(self) -> int:
        """Zero-based channel index (0 = channel 1)."""
        return self._index

    @property
    def name(self) -> str:
        """Human-friendly channel name like ``"CH01"`` or ``"CH16"``."""
        return self._name

    def __str__(self) -> str:
        """Render the channel as a bare integer string for payload building."""
        return str(self.value)

index property

index: int

Zero-based channel index (0 = channel 1).

name property

name: str

Human-friendly channel name like "CH01" or "CH16".

DosageTarget

Bases: IntEnum

Identifies which dosing pump a manual dosage command should engage.

The numeric values match the controller's MAN_DOSAGE query parameter, so this enum can be used directly in URL building.

Source code in src/proconip/definitions.py

class DosageTarget(IntEnum):
    """Identifies which dosing pump a manual dosage command should engage.

    The numeric values match the controller's `MAN_DOSAGE` query parameter,
    so this enum can be used directly in URL building.
    """

    CHLORINE = 0
    PH_MINUS = 1
    PH_PLUS = 2

GetDmxData

Mutable representation of all 16 DMX channels.

Construct from a /GetDmx.csv body, then read or modify channels via indexing, iteration, get_value, or set. Pass the (possibly mutated) instance to proconip.api.async_set_dmx to write the new state back.

Source code in src/proconip/definitions.py

class GetDmxData:
    """Mutable representation of all 16 DMX channels.

    Construct from a `/GetDmx.csv` body, then read or modify channels via
    indexing, iteration, `get_value`, or `set`. Pass the (possibly mutated)
    instance to `proconip.api.async_set_dmx` to write the new state back.
    """

    _channels: list[DmxChannelData]

    def __init__(self, raw_data: str):
        """Parse a `/GetDmx.csv` body into 16 channels.

        Args:
            raw_data: The raw CSV string returned by the controller. Leading
                blank lines are tolerated; only the first non-blank line is
                parsed.

        Raises:
            InvalidPayloadException: If the payload is empty, whitespace-only,
                or does not contain exactly 16 comma-separated channel values.
            ValueError: If a channel value cannot be parsed as an integer.
        """
        self._raw_data = raw_data
        self._channels = []

        line = 0
        lines = raw_data.splitlines()
        while line < len(lines) and len(lines[line].strip()) < 1:
            line += 1

        if line >= len(lines):
            raise InvalidPayloadException("Empty or missing DMX payload")

        values = lines[line].split(",")
        if len(values) != 16:
            raise InvalidPayloadException(
                f"GetDmx.csv must contain exactly 16 channels; got {len(values)}"
            )
        for idx, value in enumerate(values):
            self._channels.append(DmxChannelData(idx, int(value)))

    def __getitem__(self, index: int) -> DmxChannelData:
        """Return the `DmxChannelData` at the given zero-based index."""
        return self._channels[index]

    def __iter__(self) -> Iterator[DmxChannelData]:
        """Iterate over all channels in index order."""
        return iter(self._channels)

    def __str__(self) -> str:
        """Return the raw CSV body the instance was parsed from."""
        return self._raw_data

    def get_value(self, index: int) -> int:
        """Return the current value of the channel at ``index``.

        Equivalent to ``self[index].value``. Provided for symmetry with
        `set`.
        """
        return self._channels[index].value

    def set(self, index: int, value: int) -> None:
        """Update the value of one channel.

        Values outside the [0, 255] range are silently clamped — the
        controller's DMX hardware only accepts 8-bit values, so callers
        rarely need anything else.

        Args:
            index: Zero-based channel index (0 for channel 1, 15 for
                channel 16).
            value: New intensity. Clamped to [0, 255].

        Raises:
            IndexError: If ``index`` is not in 0–15.
        """
        if index > 15 or index < 0:
            raise IndexError("Index must be between 0 (channel 1) and 15 (channel 16)")
        self._channels[index].value = max(0, min(255, value))

    @property
    def post_data(self) -> dict[str, str]:
        """Form payload that updates the DMX channel state via `/usrcfg.cgi`.

        The dict has five keys, all required by the controller:

        - ``TYPE``: always ``"0"``.
        - ``LEN``: always ``"16"`` (channels per write).
        - ``CH1_8``: comma-separated values for channels 1–8.
        - ``CH9_16``: comma-separated values for channels 9–16.
        - ``DMX512``: always ``"1"``.

        URL-encode and join with ``&`` to produce the actual POST body — see
        `proconip.api.async_set_dmx` for the canonical encoding.
        """
        return {
            "TYPE": "0",
            "LEN": "16",
            "CH1_8": ",".join(map(str, self._channels[:8])),
            "CH9_16": ",".join(map(str, self._channels[8:])),
            "DMX512": "1",
        }

post_data property

post_data: dict[str, str]

Form payload that updates the DMX channel state via /usrcfg.cgi.

The dict has five keys, all required by the controller:

• TYPE: always "0".
• LEN: always "16" (channels per write).
• CH1_8: comma-separated values for channels 1–8.
• CH9_16: comma-separated values for channels 9–16.
• DMX512: always "1".

URL-encode and join with & to produce the actual POST body — see proconip.api.async_set_dmx for the canonical encoding.

get_value

get_value(index: int) -> int

Return the current value of the channel at index.

Equivalent to self[index].value. Provided for symmetry with set.

Source code in src/proconip/definitions.py

def get_value(self, index: int) -> int:
    """Return the current value of the channel at ``index``.

    Equivalent to ``self[index].value``. Provided for symmetry with
    `set`.
    """
    return self._channels[index].value

set

set(index: int, value: int) -> None

Update the value of one channel.

Values outside the [0, 255] range are silently clamped — the controller's DMX hardware only accepts 8-bit values, so callers rarely need anything else.

PARAMETER	DESCRIPTION
index	Zero-based channel index (0 for channel 1, 15 for channel 16). TYPE: int
value	New intensity. Clamped to [0, 255]. TYPE: int

RAISES	DESCRIPTION
IndexError	If index is not in 0–15.

Source code in src/proconip/definitions.py

def set(self, index: int, value: int) -> None:
    """Update the value of one channel.

    Values outside the [0, 255] range are silently clamped — the
    controller's DMX hardware only accepts 8-bit values, so callers
    rarely need anything else.

    Args:
        index: Zero-based channel index (0 for channel 1, 15 for
            channel 16).
        value: New intensity. Clamped to [0, 255].

    Raises:
        IndexError: If ``index`` is not in 0–15.
    """
    if index > 15 or index < 0:
        raise IndexError("Index must be between 0 (channel 1) and 15 (channel 16)")
    self._channels[index].value = max(0, min(255, value))

GetStateData

Parsed representation of a single /GetState.csv response.

The CSV the controller returns has six lines: SYSINFO, names, units, offsets, gains, and raw values. The constructor parses all six and builds a list of DataObject instances, then groups them by category for easy lookup.

Once constructed, an instance is read-only — it represents a snapshot. Re-fetch and reconstruct the object whenever you need fresh data.

All public properties on this class are populated eagerly at construction time, so accessing them is cheap.

Source code in src/proconip/definitions.py

class GetStateData:
    """Parsed representation of a single `/GetState.csv` response.

    The CSV the controller returns has six lines: SYSINFO, names, units,
    offsets, gains, and raw values. The constructor parses all six and
    builds a list of `DataObject` instances, then groups them by category
    for easy lookup.

    Once constructed, an instance is read-only — it represents a snapshot.
    Re-fetch and reconstruct the object whenever you need fresh data.

    All public properties on this class are populated eagerly at construction
    time, so accessing them is cheap.
    """

    _time: str
    _version: str
    _cpu_time: int
    _reset_root_cause: int
    _ntp_fault_state: int
    _config_other_enable: int
    _dosage_control: int
    _ph_plus_dosage_relay_id: int
    _ph_minus_dosage_relay_id: int
    _chlorine_dosage_relay_id: int
    _data_objects: list[DataObject]
    _analog_objects: list[DataObject]
    _electrode_objects: list[DataObject]
    _temperature_objects: list[DataObject]
    _relay_objects: list[DataObject]
    _digital_input_objects: list[DataObject]
    _external_relay_objects: list[DataObject]
    _canister_objects: list[DataObject]
    _consumption_objects: list[DataObject]

    def __init__(self, raw_data: str):
        """Parse a `/GetState.csv` body into structured data.

        Args:
            raw_data: The raw multi-line CSV string returned by the
                controller. Leading blank lines are tolerated.

        Raises:
            InvalidPayloadException: If the payload is empty, has fewer than
                six non-blank lines, or has rows whose column counts do not
                line up (names / units / offsets / gains / raw values must
                all have the same number of comma-separated entries).
            ValueError: If any of the numeric rows contains a value that is
                not parseable as a float.
        """
        self._raw_data = raw_data

        line = 0
        lines = raw_data.splitlines()
        while line < len(lines) and len(lines[line].strip()) < 1:
            line += 1
        if len(lines) < line + 6:
            raise InvalidPayloadException(
                f"GetState.csv payload is incomplete: expected at least 6 non-blank lines, "
                f"got {len(lines) - line}"
            )
        self._system_info = lines[line].split(",")
        self._data_names = lines[line + 1].split(",")
        self._data_units = lines[line + 2].split(",")
        self._data_offsets = [float(v) for v in lines[line + 3].split(",")]
        self._data_gain = [float(v) for v in lines[line + 4].split(",")]
        self._data_raw_values = [float(v) for v in lines[line + 5].split(",")]

        column_count = len(self._data_names)
        row_lengths = {
            "names": column_count,
            "units": len(self._data_units),
            "offsets": len(self._data_offsets),
            "gains": len(self._data_gain),
            "raw_values": len(self._data_raw_values),
        }
        if len(set(row_lengths.values())) != 1:
            raise InvalidPayloadException(
                "GetState.csv column counts don't line up: "
                + ", ".join(f"{k}={v}" for k, v in row_lengths.items())
            )

        self._parse_system_info()
        self._parse()
        self._time = self._data_objects[0].display_value

    def __str__(self) -> str:
        """Return the original raw CSV as it was received."""
        return self._raw_data

    def _parse_system_info(self) -> None:
        """Populate the system-level attributes from the SYSINFO line."""
        self._version = self._system_info[1]
        self._cpu_time = int(self._system_info[2])
        self._reset_root_cause = int(self._system_info[3])
        self._ntp_fault_state = int(self._system_info[4])
        self._config_other_enable = int(self._system_info[5])
        self._dosage_control = int(self._system_info[6])
        self._ph_plus_dosage_relay_id = int(self._system_info[7])
        self._ph_minus_dosage_relay_id = int(self._system_info[8])
        self._chlorine_dosage_relay_id = int(self._system_info[9])

    @property
    def time(self) -> str:
        """Controller's current local time as ``"HH:MM"``."""
        return self._time

    @property
    def version(self) -> str:
        """Firmware version string reported by the controller."""
        return self._version

    @property
    def cpu_time(self) -> int:
        """Controller CPU uptime in seconds since the last reset."""
        return self._cpu_time

    @property
    def reset_root_cause(self) -> int:
        """Numeric reset-root-cause code. Decode with `RESET_ROOT_CAUSE` or
        `get_reset_root_cause_as_str`."""
        return self._reset_root_cause

    @property
    def ntp_fault_state(self) -> int:
        """Numeric NTP fault state. Decode with `NTP_FAULT_STATE` or
        `get_ntp_fault_state_as_str`. Bits 0/1/2 indicate severity (logfile,
        warning, error); bit 16 set indicates an NTP fault — no time received
        from the NTP server (clear means the clock is in sync)."""
        return self._ntp_fault_state

    @property
    def config_other_enable(self) -> int:
        """Misc configuration flags. Use the `is_*_enabled` methods to query
        individual bits (TCP/IP boost, SD card, DMX, …)."""
        return self._config_other_enable

    @property
    def dosage_control(self) -> int:
        """Dosage configuration flags. Use the `is_*_dosage_enabled` and
        `is_electrolysis_enabled` methods to query individual bits."""
        return self._dosage_control

    @property
    def ph_plus_dosage_relay_id(self) -> int:
        """Aggregated relay ID configured to act as the pH+ dosing pump."""
        return self._ph_plus_dosage_relay_id

    @property
    def ph_minus_dosage_relay_id(self) -> int:
        """Aggregated relay ID configured to act as the pH- dosing pump."""
        return self._ph_minus_dosage_relay_id

    @property
    def chlorine_dosage_relay_id(self) -> int:
        """Aggregated relay ID configured to act as the chlorine dosing pump."""
        return self._chlorine_dosage_relay_id

    def is_chlorine_dosage_enabled(self) -> bool:
        """True if chlorine dosage control is enabled in the controller config (bit 0)."""
        return self._dosage_control & 1 == 1

    def is_electrolysis_enabled(self) -> bool:
        """True if electrolysis (saltwater) chlorination is enabled (bit 4)."""
        return self._dosage_control & 16 == 16

    def is_ph_minus_dosage_enabled(self) -> bool:
        """True if pH- dosage control is enabled in the controller config (bit 8)."""
        return self._dosage_control & 256 == 256

    def is_ph_plus_dosage_enabled(self) -> bool:
        """True if pH+ dosage control is enabled in the controller config (bit 12)."""
        return self._dosage_control & 4096 == 4096

    def is_dosage_enabled(self, data_entity: DataObject) -> bool:
        """Convenience: is the dosage chemical for this canister/consumption entity enabled?

        Args:
            data_entity: A canister (column 36–38) or consumption (column 39–41)
                `DataObject`. The chemical is inferred from the column index.

        Returns:
            True if the corresponding ``is_*_dosage_enabled`` flag is set.
            False for any other column (or if the chemical is disabled).
        """
        col = data_entity.column
        if col in (36, 39):
            return self.is_chlorine_dosage_enabled()
        if col in (37, 40):
            return self.is_ph_minus_dosage_enabled()
        if col in (38, 41):
            return self.is_ph_plus_dosage_enabled()
        return False

    def get_dosage_relay(self, data_entity: DataObject) -> int | None:
        """Aggregated relay ID that handles the dosing for this canister/consumption entity.

        Args:
            data_entity: A canister (column 36–38) or consumption (column 39–41)
                `DataObject`.

        Returns:
            The aggregated relay ID (chlorine, pH-, or pH+) corresponding to
            the entity's chemical, or ``None`` if the entity is not a
            canister/consumption object.
        """
        col = data_entity.column
        if col in (36, 39):
            return self._chlorine_dosage_relay_id
        if col in (37, 40):
            return self._ph_minus_dosage_relay_id
        if col in (38, 41):
            return self._ph_plus_dosage_relay_id
        return None

    def is_dosage_relay(
        self,
        relay_object: Relay | None = None,
        data_object: DataObject | None = None,
        relay_id: int | None = None,
    ) -> bool:
        """Check whether a relay is one of the configured dosage control relays.

        Provide one of `relay_object`, `data_object`, or `relay_id`. If more
        than one is supplied, the first non-None argument in that precedence
        order wins and the others are ignored. If none are provided the
        method returns False.

        Args:
            relay_object: A `Relay` instance. Highest-precedence argument.
            data_object: A `DataObject` of category `relay` or
                `external_relay`. Considered only when `relay_object` is None.
            relay_id: An aggregated relay ID (0–15). Considered only when
                both `relay_object` and `data_object` are None.

        Returns:
            True if the resolved argument identifies a dosage relay; False
            otherwise (including when no argument is provided).

        Raises:
            BadRelayException: If ``data_object`` is the resolved argument
                but is not a relay-category `DataObject`.

        Example:
            ```python
            # Three equivalent ways to ask "is relay 5 a dosage relay?",
            # assuming the chlorine pump is configured there.
            state.is_dosage_relay(relay_id=5)
            state.is_dosage_relay(relay_object=state.get_relay(5))
            state.is_dosage_relay(data_object=state.aggregated_relay_objects[5])
            ```
        """
        dosage_control_relays = [
            self._chlorine_dosage_relay_id,
            self._ph_minus_dosage_relay_id,
            self._ph_plus_dosage_relay_id,
        ]
        if relay_object is not None:
            return relay_object.relay_id in dosage_control_relays
        if data_object is not None:
            if data_object.category not in (CATEGORY_RELAY, CATEGORY_EXTERNAL_RELAY):
                raise BadRelayException(
                    f"DataObject category '{data_object.category}' is not a relay category"
                )
            offset = (
                EXTERNAL_RELAY_ID_OFFSET if data_object.category == CATEGORY_EXTERNAL_RELAY else 0
            )
            return data_object.category_id + offset in dosage_control_relays
        if relay_id is not None:
            return relay_id in dosage_control_relays
        return False

    def get_reset_root_cause_as_str(self) -> str:
        """Decode `reset_root_cause` to its `RESET_ROOT_CAUSE` label.

        Falls back to the "n.a." label for any value not in the lookup table.
        """
        if self._reset_root_cause not in RESET_ROOT_CAUSE:
            return RESET_ROOT_CAUSE[0]
        return RESET_ROOT_CAUSE[self._reset_root_cause]

    def get_ntp_fault_state_as_str(self) -> str:
        """Decode `ntp_fault_state` to a human-readable label from `NTP_FAULT_STATE`.

        For exact matches in the lookup table (``0``, ``1``, ``2``, ``4``,
        ``65536``) the corresponding label is returned. Composite states are
        approximated by returning the highest-severity active bit (4 → 2 →
        1), since the controller's CSV has no fixed combinations beyond the
        listed ones. Falls back to "n.a." if no severity bit is set.
        """
        if self._ntp_fault_state in NTP_FAULT_STATE:
            return NTP_FAULT_STATE[self._ntp_fault_state]
        for bit in (4, 2, 1):
            if self._ntp_fault_state & bit:
                return NTP_FAULT_STATE[bit]
        return NTP_FAULT_STATE[0]

    def is_tcpip_boost_enabled(self) -> bool:
        """True if TCP/IP boost is enabled in the controller config (bit 0)."""
        return self._config_other_enable & 1 == 1

    def is_sd_card_enabled(self) -> bool:
        """True if SD card logging is enabled in the controller config (bit 1)."""
        return self._config_other_enable & 2 == 2

    def is_dmx_enabled(self) -> bool:
        """True if DMX output is enabled in the controller config (bit 2)."""
        return self._config_other_enable & 4 == 4

    def is_avatar_enabled(self) -> bool:
        """True if the avatar feature is enabled in the controller config (bit 3)."""
        return self._config_other_enable & 8 == 8

    def is_relay_extension_enabled(self) -> bool:
        """True if the external relay extension module is connected and active (bit 4).

        Affects how `determine_overall_relay_bit_state` builds the ENA mask:
        with the extension active, the mask covers all 16 bits instead of
        just the internal 8.
        """
        return self._config_other_enable & 16 == 16

    def is_high_bus_load_enabled(self) -> bool:
        """True if high bus load mode is enabled in the controller config (bit 5)."""
        return self._config_other_enable & 32 == 32

    def is_flow_sensor_enabled(self) -> bool:
        """True if the flow sensor is enabled in the controller config (bit 6)."""
        return self._config_other_enable & 64 == 64

    def is_repeated_mails_enabled(self) -> bool:
        """True if repeated email notifications are enabled (bit 7)."""
        return self._config_other_enable & 128 == 128

    def is_dmx_extension_enabled(self) -> bool:
        """True if the DMX extension module is enabled in the controller config (bit 8)."""
        return self._config_other_enable & 256 == 256

    def _parse(self) -> None:
        """Build per-column `DataObject` instances and group them by category."""
        self._data_objects = []
        for column, name in enumerate(self._data_names):
            self._data_objects.append(
                DataObject(
                    column,
                    name,
                    self._data_units[column],
                    self._data_offsets[column],
                    self._data_gain[column],
                    self._data_raw_values[column],
                )
            )

        self._analog_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_ANALOG
        ]
        self._electrode_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_ELECTRODE
        ]
        self._temperature_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_TEMPERATURE
        ]
        self._relay_objects = [obj for obj in self._data_objects if obj.category == CATEGORY_RELAY]
        self._digital_input_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_DIGITAL_INPUT
        ]
        self._external_relay_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_EXTERNAL_RELAY
        ]
        self._canister_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_CANISTER
        ]
        self._consumption_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_CONSUMPTION
        ]

    @property
    def analog_objects(self) -> list[DataObject]:
        """The five analog inputs (columns 1–5), in column order."""
        return self._analog_objects

    @property
    def electrode_objects(self) -> list[DataObject]:
        """The two electrode readings — redox at index 0, pH at index 1."""
        return self._electrode_objects

    @property
    def temperature_objects(self) -> list[DataObject]:
        """The eight temperature sensors (columns 8–15), in column order."""
        return self._temperature_objects

    @property
    def relay_objects(self) -> list[DataObject]:
        """The eight built-in relays (columns 16–23), in column order.

        These are still untyped `DataObject` instances. Use `relays()` for
        `Relay` instances with on/off and manual/auto helpers, or
        `aggregated_relay_objects` to also include the external relays.
        """
        return self._relay_objects

    def relays(self) -> list[Relay]:
        """The eight built-in relays as `Relay` instances.

        Equivalent to wrapping each entry in `relay_objects` with `Relay(...)`.
        """
        return [Relay(obj) for obj in self._relay_objects]

    @property
    def digital_input_objects(self) -> list[DataObject]:
        """The four digital inputs (columns 24–27), in column order."""
        return self._digital_input_objects

    def digital_inputs(self) -> list[DigitalInput]:
        """The four digital inputs as `DigitalInput` instances.

        Equivalent to wrapping each entry in `digital_input_objects` with
        `DigitalInput(...)`. Use these to get `get_bit_mask()` for triggering.
        """
        return [DigitalInput(obj) for obj in self._digital_input_objects]

    @property
    def external_relay_objects(self) -> list[DataObject]:
        """The eight external relays (columns 28–35), in column order.

        Will be present in the parsed data even when the relay extension is
        not enabled in the controller config — check
        `is_relay_extension_enabled` before treating them as live.
        """
        return self._external_relay_objects

    def external_relays(self) -> list[Relay]:
        """The eight external relays as `Relay` instances."""
        return [Relay(obj) for obj in self._external_relay_objects]

    @property
    def canister_objects(self) -> list[DataObject]:
        """The three canister fill-level readings (columns 36–38).

        Order: chlorine, pH-, pH+. Convenience properties `chlorine_canister`,
        `ph_minus_canister`, and `ph_plus_canister` return individual entries.
        """
        return self._canister_objects

    @property
    def consumption_objects(self) -> list[DataObject]:
        """The three dosage consumption counters (columns 39–41).

        Order: chlorine, pH-, pH+. Convenience properties
        `chlorine_consumption`, `ph_minus_consumption`, and
        `ph_plus_consumption` return individual entries.
        """
        return self._consumption_objects

    @property
    def redox_electrode(self) -> DataObject:
        """The redox electrode reading (column 6)."""
        return self._electrode_objects[0]

    @property
    def ph_electrode(self) -> DataObject:
        """The pH electrode reading (column 7)."""
        return self._electrode_objects[1]

    @property
    def chlorine_canister(self) -> DataObject:
        """Chlorine canister fill level (column 36)."""
        return self._canister_objects[0]

    @property
    def ph_minus_canister(self) -> DataObject:
        """pH- canister fill level (column 37)."""
        return self._canister_objects[1]

    @property
    def ph_plus_canister(self) -> DataObject:
        """pH+ canister fill level (column 38)."""
        return self._canister_objects[2]

    @property
    def chlorine_consumption(self) -> DataObject:
        """Cumulative chlorine consumption counter (column 39)."""
        return self._consumption_objects[0]

    @property
    def ph_minus_consumption(self) -> DataObject:
        """Cumulative pH- consumption counter (column 40)."""
        return self._consumption_objects[1]

    @property
    def ph_plus_consumption(self) -> DataObject:
        """Cumulative pH+ consumption counter (column 41)."""
        return self._consumption_objects[2]

    @property
    def aggregated_relay_objects(self) -> list[DataObject]:
        """All 16 relay `DataObject`s — internal first, then external.

        Index in this list is the aggregated relay ID used by `Relay.relay_id`,
        `get_relay`, and the `RelaySwitch` API.
        """
        return self._relay_objects + self._external_relay_objects

    @property
    def chlorine_dosage_relay(self) -> DataObject:
        """The relay configured as the chlorine dosing pump."""
        return self.aggregated_relay_objects[self._chlorine_dosage_relay_id]

    @property
    def ph_minus_dosage_relay(self) -> DataObject:
        """The relay configured as the pH- dosing pump."""
        return self.aggregated_relay_objects[self._ph_minus_dosage_relay_id]

    @property
    def ph_plus_dosage_relay(self) -> DataObject:
        """The relay configured as the pH+ dosing pump."""
        return self.aggregated_relay_objects[self._ph_plus_dosage_relay_id]

    def get_relay(self, relay_id: int) -> Relay:
        """Return the `Relay` for the given aggregated relay ID (0–15).

        Args:
            relay_id: 0–7 for internal relays, 8–15 for external relays.

        Returns:
            A new `Relay` wrapping the underlying `DataObject`.

        Raises:
            IndexError: If ``relay_id`` is outside the 0–15 range.
        """
        return Relay(self.aggregated_relay_objects[relay_id])

    def get_relays(self) -> list[Relay]:
        """All 16 relays as `Relay` instances, in aggregated-ID order."""
        return [Relay(obj) for obj in self.aggregated_relay_objects]

    def determine_overall_relay_bit_state(self) -> list[int]:
        """Build the two-element ENA bit field that represents the current relay state.

        The controller's `/usrcfg.cgi` payload uses an ``ENA=enable_mask,on_mask``
        pair to set relay state. ``enable_mask`` selects which relays are in
        manual mode (bit set = manual, bit clear = auto), and ``on_mask``
        selects the manual-on relays among them.

        Returns:
            A two-element list ``[enable_mask, on_mask]``. Both masks cover
            bits 0–7 (internal relays) by default, or bits 0–15 if the
            external relay extension is enabled (`is_relay_extension_enabled`).

            The masks reflect the *current* state, so callers can flip a
            single relay's bit and POST the result to switch only that
            relay without touching the others.
        """
        relay_list: list[Relay] = [Relay(obj) for obj in self._relay_objects]
        bit_state = [255, 0]
        if self.is_relay_extension_enabled():
            relay_list.extend(Relay(obj) for obj in self._external_relay_objects)
            bit_state[0] = 65535
        for relay in relay_list:
            relay_bit_mask = relay.get_bit_mask()
            if relay.is_auto_mode():
                bit_state[0] &= ~relay_bit_mask
            if relay.is_on():
                bit_state[1] |= relay_bit_mask
        return bit_state

time property

time: str

Controller's current local time as "HH:MM".

version property

version: str

Firmware version string reported by the controller.

cpu_time property

cpu_time: int

Controller CPU uptime in seconds since the last reset.

reset_root_cause property

reset_root_cause: int

Numeric reset-root-cause code. Decode with RESET_ROOT_CAUSE or get_reset_root_cause_as_str.

ntp_fault_state property

ntp_fault_state: int

Numeric NTP fault state. Decode with NTP_FAULT_STATE or get_ntp_fault_state_as_str. Bits 0/1/2 indicate severity (logfile, warning, error); bit 16 set indicates an NTP fault — no time received from the NTP server (clear means the clock is in sync).

config_other_enable property

config_other_enable: int

Misc configuration flags. Use the is_*_enabled methods to query individual bits (TCP/IP boost, SD card, DMX, …).

dosage_control property

dosage_control: int

Dosage configuration flags. Use the is_*_dosage_enabled and is_electrolysis_enabled methods to query individual bits.

ph_plus_dosage_relay_id property

ph_plus_dosage_relay_id: int

Aggregated relay ID configured to act as the pH+ dosing pump.

ph_minus_dosage_relay_id property

ph_minus_dosage_relay_id: int

Aggregated relay ID configured to act as the pH- dosing pump.

chlorine_dosage_relay_id property

chlorine_dosage_relay_id: int

Aggregated relay ID configured to act as the chlorine dosing pump.

analog_objects property

analog_objects: list[DataObject]

The five analog inputs (columns 1–5), in column order.

electrode_objects property

electrode_objects: list[DataObject]

The two electrode readings — redox at index 0, pH at index 1.

temperature_objects property

temperature_objects: list[DataObject]

The eight temperature sensors (columns 8–15), in column order.

relay_objects property

relay_objects: list[DataObject]

The eight built-in relays (columns 16–23), in column order.

These are still untyped DataObject instances. Use relays() for Relay instances with on/off and manual/auto helpers, or aggregated_relay_objects to also include the external relays.

digital_input_objects property

digital_input_objects: list[DataObject]

The four digital inputs (columns 24–27), in column order.

external_relay_objects property

external_relay_objects: list[DataObject]

The eight external relays (columns 28–35), in column order.

Will be present in the parsed data even when the relay extension is not enabled in the controller config — check is_relay_extension_enabled before treating them as live.

canister_objects property

canister_objects: list[DataObject]

The three canister fill-level readings (columns 36–38).

Order: chlorine, pH-, pH+. Convenience properties chlorine_canister, ph_minus_canister, and ph_plus_canister return individual entries.

consumption_objects property

consumption_objects: list[DataObject]

The three dosage consumption counters (columns 39–41).

Order: chlorine, pH-, pH+. Convenience properties chlorine_consumption, ph_minus_consumption, and ph_plus_consumption return individual entries.

redox_electrode property

redox_electrode: DataObject

The redox electrode reading (column 6).

ph_electrode property

ph_electrode: DataObject

The pH electrode reading (column 7).

chlorine_canister property

chlorine_canister: DataObject

Chlorine canister fill level (column 36).

ph_minus_canister property

ph_minus_canister: DataObject

pH- canister fill level (column 37).

ph_plus_canister property

ph_plus_canister: DataObject

pH+ canister fill level (column 38).

chlorine_consumption property

chlorine_consumption: DataObject

Cumulative chlorine consumption counter (column 39).

ph_minus_consumption property

ph_minus_consumption: DataObject

Cumulative pH- consumption counter (column 40).

ph_plus_consumption property

ph_plus_consumption: DataObject

Cumulative pH+ consumption counter (column 41).

aggregated_relay_objects property

aggregated_relay_objects: list[DataObject]

All 16 relay DataObjects — internal first, then external.

Index in this list is the aggregated relay ID used by Relay.relay_id, get_relay, and the RelaySwitch API.

chlorine_dosage_relay property

chlorine_dosage_relay: DataObject

The relay configured as the chlorine dosing pump.

ph_minus_dosage_relay property

ph_minus_dosage_relay: DataObject

The relay configured as the pH- dosing pump.

ph_plus_dosage_relay property

ph_plus_dosage_relay: DataObject

The relay configured as the pH+ dosing pump.

is_chlorine_dosage_enabled

is_chlorine_dosage_enabled() -> bool

True if chlorine dosage control is enabled in the controller config (bit 0).

Source code in src/proconip/definitions.py

def is_chlorine_dosage_enabled(self) -> bool:
    """True if chlorine dosage control is enabled in the controller config (bit 0)."""
    return self._dosage_control & 1 == 1

is_electrolysis_enabled

is_electrolysis_enabled() -> bool

True if electrolysis (saltwater) chlorination is enabled (bit 4).

Source code in src/proconip/definitions.py

def is_electrolysis_enabled(self) -> bool:
    """True if electrolysis (saltwater) chlorination is enabled (bit 4)."""
    return self._dosage_control & 16 == 16

is_ph_minus_dosage_enabled

is_ph_minus_dosage_enabled() -> bool

True if pH- dosage control is enabled in the controller config (bit 8).

Source code in src/proconip/definitions.py

def is_ph_minus_dosage_enabled(self) -> bool:
    """True if pH- dosage control is enabled in the controller config (bit 8)."""
    return self._dosage_control & 256 == 256

is_ph_plus_dosage_enabled

is_ph_plus_dosage_enabled() -> bool

True if pH+ dosage control is enabled in the controller config (bit 12).

Source code in src/proconip/definitions.py

def is_ph_plus_dosage_enabled(self) -> bool:
    """True if pH+ dosage control is enabled in the controller config (bit 12)."""
    return self._dosage_control & 4096 == 4096

is_dosage_enabled

is_dosage_enabled(data_entity: DataObject) -> bool

Convenience: is the dosage chemical for this canister/consumption entity enabled?

PARAMETER	DESCRIPTION
data_entity	A canister (column 36–38) or consumption (column 39–41) DataObject. The chemical is inferred from the column index. TYPE: DataObject

RETURNS	DESCRIPTION
bool	True if the corresponding is_*_dosage_enabled flag is set.
bool	False for any other column (or if the chemical is disabled).

Source code in src/proconip/definitions.py

def is_dosage_enabled(self, data_entity: DataObject) -> bool:
    """Convenience: is the dosage chemical for this canister/consumption entity enabled?

    Args:
        data_entity: A canister (column 36–38) or consumption (column 39–41)
            `DataObject`. The chemical is inferred from the column index.

    Returns:
        True if the corresponding ``is_*_dosage_enabled`` flag is set.
        False for any other column (or if the chemical is disabled).
    """
    col = data_entity.column
    if col in (36, 39):
        return self.is_chlorine_dosage_enabled()
    if col in (37, 40):
        return self.is_ph_minus_dosage_enabled()
    if col in (38, 41):
        return self.is_ph_plus_dosage_enabled()
    return False

get_dosage_relay

get_dosage_relay(data_entity: DataObject) -> int | None

Aggregated relay ID that handles the dosing for this canister/consumption entity.

PARAMETER	DESCRIPTION
data_entity	A canister (column 36–38) or consumption (column 39–41) DataObject. TYPE: DataObject

RETURNS	DESCRIPTION
int | None	The aggregated relay ID (chlorine, pH-, or pH+) corresponding to
int | None	the entity's chemical, or None if the entity is not a
int | None	canister/consumption object.

Source code in src/proconip/definitions.py

def get_dosage_relay(self, data_entity: DataObject) -> int | None:
    """Aggregated relay ID that handles the dosing for this canister/consumption entity.

    Args:
        data_entity: A canister (column 36–38) or consumption (column 39–41)
            `DataObject`.

    Returns:
        The aggregated relay ID (chlorine, pH-, or pH+) corresponding to
        the entity's chemical, or ``None`` if the entity is not a
        canister/consumption object.
    """
    col = data_entity.column
    if col in (36, 39):
        return self._chlorine_dosage_relay_id
    if col in (37, 40):
        return self._ph_minus_dosage_relay_id
    if col in (38, 41):
        return self._ph_plus_dosage_relay_id
    return None

is_dosage_relay

is_dosage_relay(relay_object: Relay | None = None, data_object: DataObject | None = None, relay_id: int | None = None) -> bool

Check whether a relay is one of the configured dosage control relays.

Provide one of relay_object, data_object, or relay_id. If more than one is supplied, the first non-None argument in that precedence order wins and the others are ignored. If none are provided the method returns False.

PARAMETER	DESCRIPTION
relay_object	A Relay instance. Highest-precedence argument. TYPE: Relay | None DEFAULT: None
data_object	A DataObject of category relay or external_relay. Considered only when relay_object is None. TYPE: DataObject | None DEFAULT: None
relay_id	An aggregated relay ID (0–15). Considered only when both relay_object and data_object are None. TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
bool	True if the resolved argument identifies a dosage relay; False
bool	otherwise (including when no argument is provided).

RAISES	DESCRIPTION
BadRelayException	If data_object is the resolved argument but is not a relay-category DataObject.

Example

# Three equivalent ways to ask "is relay 5 a dosage relay?",
# assuming the chlorine pump is configured there.
state.is_dosage_relay(relay_id=5)
state.is_dosage_relay(relay_object=state.get_relay(5))
state.is_dosage_relay(data_object=state.aggregated_relay_objects[5])

Source code in src/proconip/definitions.py

def is_dosage_relay(
    self,
    relay_object: Relay | None = None,
    data_object: DataObject | None = None,
    relay_id: int | None = None,
) -> bool:
    """Check whether a relay is one of the configured dosage control relays.

    Provide one of `relay_object`, `data_object`, or `relay_id`. If more
    than one is supplied, the first non-None argument in that precedence
    order wins and the others are ignored. If none are provided the
    method returns False.

    Args:
        relay_object: A `Relay` instance. Highest-precedence argument.
        data_object: A `DataObject` of category `relay` or
            `external_relay`. Considered only when `relay_object` is None.
        relay_id: An aggregated relay ID (0–15). Considered only when
            both `relay_object` and `data_object` are None.

    Returns:
        True if the resolved argument identifies a dosage relay; False
        otherwise (including when no argument is provided).

    Raises:
        BadRelayException: If ``data_object`` is the resolved argument
            but is not a relay-category `DataObject`.

    Example:
        ```python
        # Three equivalent ways to ask "is relay 5 a dosage relay?",
        # assuming the chlorine pump is configured there.
        state.is_dosage_relay(relay_id=5)
        state.is_dosage_relay(relay_object=state.get_relay(5))
        state.is_dosage_relay(data_object=state.aggregated_relay_objects[5])
        ```
    """
    dosage_control_relays = [
        self._chlorine_dosage_relay_id,
        self._ph_minus_dosage_relay_id,
        self._ph_plus_dosage_relay_id,
    ]
    if relay_object is not None:
        return relay_object.relay_id in dosage_control_relays
    if data_object is not None:
        if data_object.category not in (CATEGORY_RELAY, CATEGORY_EXTERNAL_RELAY):
            raise BadRelayException(
                f"DataObject category '{data_object.category}' is not a relay category"
            )
        offset = (
            EXTERNAL_RELAY_ID_OFFSET if data_object.category == CATEGORY_EXTERNAL_RELAY else 0
        )
        return data_object.category_id + offset in dosage_control_relays
    if relay_id is not None:
        return relay_id in dosage_control_relays
    return False

get_reset_root_cause_as_str

get_reset_root_cause_as_str() -> str

Decode reset_root_cause to its RESET_ROOT_CAUSE label.

Falls back to the "n.a." label for any value not in the lookup table.

Source code in src/proconip/definitions.py

def get_reset_root_cause_as_str(self) -> str:
    """Decode `reset_root_cause` to its `RESET_ROOT_CAUSE` label.

    Falls back to the "n.a." label for any value not in the lookup table.
    """
    if self._reset_root_cause not in RESET_ROOT_CAUSE:
        return RESET_ROOT_CAUSE[0]
    return RESET_ROOT_CAUSE[self._reset_root_cause]

get_ntp_fault_state_as_str

get_ntp_fault_state_as_str() -> str

Decode ntp_fault_state to a human-readable label from NTP_FAULT_STATE.

For exact matches in the lookup table (0, 1, 2, 4, 65536) the corresponding label is returned. Composite states are approximated by returning the highest-severity active bit (4 → 2 → 1), since the controller's CSV has no fixed combinations beyond the listed ones. Falls back to "n.a." if no severity bit is set.

Source code in src/proconip/definitions.py

def get_ntp_fault_state_as_str(self) -> str:
    """Decode `ntp_fault_state` to a human-readable label from `NTP_FAULT_STATE`.

    For exact matches in the lookup table (``0``, ``1``, ``2``, ``4``,
    ``65536``) the corresponding label is returned. Composite states are
    approximated by returning the highest-severity active bit (4 → 2 →
    1), since the controller's CSV has no fixed combinations beyond the
    listed ones. Falls back to "n.a." if no severity bit is set.
    """
    if self._ntp_fault_state in NTP_FAULT_STATE:
        return NTP_FAULT_STATE[self._ntp_fault_state]
    for bit in (4, 2, 1):
        if self._ntp_fault_state & bit:
            return NTP_FAULT_STATE[bit]
    return NTP_FAULT_STATE[0]

is_tcpip_boost_enabled

is_tcpip_boost_enabled() -> bool

True if TCP/IP boost is enabled in the controller config (bit 0).

Source code in src/proconip/definitions.py

def is_tcpip_boost_enabled(self) -> bool:
    """True if TCP/IP boost is enabled in the controller config (bit 0)."""
    return self._config_other_enable & 1 == 1

is_sd_card_enabled

is_sd_card_enabled() -> bool

True if SD card logging is enabled in the controller config (bit 1).

Source code in src/proconip/definitions.py

def is_sd_card_enabled(self) -> bool:
    """True if SD card logging is enabled in the controller config (bit 1)."""
    return self._config_other_enable & 2 == 2

is_dmx_enabled

is_dmx_enabled() -> bool

True if DMX output is enabled in the controller config (bit 2).

Source code in src/proconip/definitions.py

def is_dmx_enabled(self) -> bool:
    """True if DMX output is enabled in the controller config (bit 2)."""
    return self._config_other_enable & 4 == 4

is_avatar_enabled

is_avatar_enabled() -> bool

True if the avatar feature is enabled in the controller config (bit 3).

Source code in src/proconip/definitions.py

def is_avatar_enabled(self) -> bool:
    """True if the avatar feature is enabled in the controller config (bit 3)."""
    return self._config_other_enable & 8 == 8

is_relay_extension_enabled

is_relay_extension_enabled() -> bool

True if the external relay extension module is connected and active (bit 4).

Affects how determine_overall_relay_bit_state builds the ENA mask: with the extension active, the mask covers all 16 bits instead of just the internal 8.

Source code in src/proconip/definitions.py

def is_relay_extension_enabled(self) -> bool:
    """True if the external relay extension module is connected and active (bit 4).

    Affects how `determine_overall_relay_bit_state` builds the ENA mask:
    with the extension active, the mask covers all 16 bits instead of
    just the internal 8.
    """
    return self._config_other_enable & 16 == 16

is_high_bus_load_enabled

is_high_bus_load_enabled() -> bool

True if high bus load mode is enabled in the controller config (bit 5).

Source code in src/proconip/definitions.py

def is_high_bus_load_enabled(self) -> bool:
    """True if high bus load mode is enabled in the controller config (bit 5)."""
    return self._config_other_enable & 32 == 32

is_flow_sensor_enabled

is_flow_sensor_enabled() -> bool

True if the flow sensor is enabled in the controller config (bit 6).

Source code in src/proconip/definitions.py

def is_flow_sensor_enabled(self) -> bool:
    """True if the flow sensor is enabled in the controller config (bit 6)."""
    return self._config_other_enable & 64 == 64

is_repeated_mails_enabled

is_repeated_mails_enabled() -> bool

True if repeated email notifications are enabled (bit 7).

Source code in src/proconip/definitions.py

def is_repeated_mails_enabled(self) -> bool:
    """True if repeated email notifications are enabled (bit 7)."""
    return self._config_other_enable & 128 == 128

is_dmx_extension_enabled

is_dmx_extension_enabled() -> bool

True if the DMX extension module is enabled in the controller config (bit 8).

Source code in src/proconip/definitions.py

def is_dmx_extension_enabled(self) -> bool:
    """True if the DMX extension module is enabled in the controller config (bit 8)."""
    return self._config_other_enable & 256 == 256

relays

relays() -> list[Relay]

The eight built-in relays as Relay instances.

Equivalent to wrapping each entry in relay_objects with Relay(...).

Source code in src/proconip/definitions.py

def relays(self) -> list[Relay]:
    """The eight built-in relays as `Relay` instances.

    Equivalent to wrapping each entry in `relay_objects` with `Relay(...)`.
    """
    return [Relay(obj) for obj in self._relay_objects]

digital_inputs

digital_inputs() -> list[DigitalInput]

The four digital inputs as DigitalInput instances.

Equivalent to wrapping each entry in digital_input_objects with DigitalInput(...). Use these to get get_bit_mask() for triggering.

Source code in src/proconip/definitions.py

def digital_inputs(self) -> list[DigitalInput]:
    """The four digital inputs as `DigitalInput` instances.

    Equivalent to wrapping each entry in `digital_input_objects` with
    `DigitalInput(...)`. Use these to get `get_bit_mask()` for triggering.
    """
    return [DigitalInput(obj) for obj in self._digital_input_objects]

external_relays

external_relays() -> list[Relay]

The eight external relays as Relay instances.

Source code in src/proconip/definitions.py

def external_relays(self) -> list[Relay]:
    """The eight external relays as `Relay` instances."""
    return [Relay(obj) for obj in self._external_relay_objects]

get_relay

get_relay(relay_id: int) -> Relay

Return the Relay for the given aggregated relay ID (0–15).

PARAMETER	DESCRIPTION
relay_id	0–7 for internal relays, 8–15 for external relays. TYPE: int

RETURNS	DESCRIPTION
Relay	A new Relay wrapping the underlying DataObject.

RAISES	DESCRIPTION
IndexError	If relay_id is outside the 0–15 range.

Source code in src/proconip/definitions.py

def get_relay(self, relay_id: int) -> Relay:
    """Return the `Relay` for the given aggregated relay ID (0–15).

    Args:
        relay_id: 0–7 for internal relays, 8–15 for external relays.

    Returns:
        A new `Relay` wrapping the underlying `DataObject`.

    Raises:
        IndexError: If ``relay_id`` is outside the 0–15 range.
    """
    return Relay(self.aggregated_relay_objects[relay_id])

get_relays

get_relays() -> list[Relay]

All 16 relays as Relay instances, in aggregated-ID order.

Source code in src/proconip/definitions.py

def get_relays(self) -> list[Relay]:
    """All 16 relays as `Relay` instances, in aggregated-ID order."""
    return [Relay(obj) for obj in self.aggregated_relay_objects]

determine_overall_relay_bit_state

determine_overall_relay_bit_state() -> list[int]

Build the two-element ENA bit field that represents the current relay state.

The controller's /usrcfg.cgi payload uses an ENA=enable_mask,on_mask pair to set relay state. enable_mask selects which relays are in manual mode (bit set = manual, bit clear = auto), and on_mask selects the manual-on relays among them.

RETURNS	DESCRIPTION
list[int]	A two-element list [enable_mask, on_mask]. Both masks cover
list[int]	bits 0–7 (internal relays) by default, or bits 0–15 if the
list[int]	external relay extension is enabled (is_relay_extension_enabled).
list[int]	The masks reflect the current state, so callers can flip a
list[int]	single relay's bit and POST the result to switch only that
list[int]	relay without touching the others.

Source code in src/proconip/definitions.py

def determine_overall_relay_bit_state(self) -> list[int]:
    """Build the two-element ENA bit field that represents the current relay state.

    The controller's `/usrcfg.cgi` payload uses an ``ENA=enable_mask,on_mask``
    pair to set relay state. ``enable_mask`` selects which relays are in
    manual mode (bit set = manual, bit clear = auto), and ``on_mask``
    selects the manual-on relays among them.

    Returns:
        A two-element list ``[enable_mask, on_mask]``. Both masks cover
        bits 0–7 (internal relays) by default, or bits 0–15 if the
        external relay extension is enabled (`is_relay_extension_enabled`).

        The masks reflect the *current* state, so callers can flip a
        single relay's bit and POST the result to switch only that
        relay without touching the others.
    """
    relay_list: list[Relay] = [Relay(obj) for obj in self._relay_objects]
    bit_state = [255, 0]
    if self.is_relay_extension_enabled():
        relay_list.extend(Relay(obj) for obj in self._external_relay_objects)
        bit_state[0] = 65535
    for relay in relay_list:
        relay_bit_mask = relay.get_bit_mask()
        if relay.is_auto_mode():
            bit_state[0] &= ~relay_bit_mask
        if relay.is_on():
            bit_state[1] |= relay_bit_mask
    return bit_state

InvalidPayloadException

Bases: Exception

Raised when a CSV response from the controller cannot be parsed.

Typically this means the response was empty, truncated, or did not have the expected number of CSV lines. Catching this lets callers distinguish protocol-level breakage from network errors.

Source code in src/proconip/definitions.py

class InvalidPayloadException(Exception):
    """Raised when a CSV response from the controller cannot be parsed.

    Typically this means the response was empty, truncated, or did not
    have the expected number of CSV lines. Catching this lets callers
    distinguish protocol-level breakage from network errors.
    """

Relay

Bases: DataObject

A DataObject with extra methods for interrogating a relay's state.

Relay state is encoded in two bits of the underlying value:

• bit 0 — output level (0 = off, 1 = on)
• bit 1 — control mode (0 = auto, 1 = manual)

The four valid combinations correspond to the four display_value strings from DataObject._relay_state.

Construct one by passing the DataObject you got from GetStateData.relay_objects (or external_relay_objects); the relay's column, name, calibration, and raw value are copied across so the physical value is computed exactly once. GetStateData.get_relay() is a shorthand that does this for you.

Source code in src/proconip/definitions.py

class Relay(DataObject):
    """A `DataObject` with extra methods for interrogating a relay's state.

    Relay state is encoded in two bits of the underlying `value`:

    - bit 0 — output level (0 = off, 1 = on)
    - bit 1 — control mode (0 = auto, 1 = manual)

    The four valid combinations correspond to the four `display_value`
    strings from `DataObject._relay_state`.

    Construct one by passing the `DataObject` you got from
    `GetStateData.relay_objects` (or `external_relay_objects`); the relay's
    column, name, calibration, and raw value are copied across so the
    physical value is computed exactly once. ``GetStateData.get_relay()`` is
    a shorthand that does this for you.
    """

    def __init__(self, data_object: DataObject):
        """Wrap an existing relay `DataObject`.

        Args:
            data_object: A `DataObject` produced by parsing a `/GetState.csv`
                response. It should be in the relay or external_relay
                category, but no check is enforced — passing a non-relay
                object yields a `Relay` whose interrogation methods will
                still run but produce meaningless results.
        """
        super().__init__(
            data_object.column,
            data_object.name,
            data_object.unit,
            data_object.offset,
            data_object.gain,
            data_object.raw_value,  # pass raw value so offset+gain are applied exactly once
        )

    def __str__(self) -> str:
        """Return ``"name: state"`` (e.g. ``"Pumpe: Auto (off)"``)."""
        return f"{self._name}: {self._display_value}"

    @property
    def relay_id(self) -> int:
        """Aggregated relay ID across internal and external banks.

        Internal relays return ``category_id`` directly (0–7); external
        relays return ``category_id + EXTERNAL_RELAY_ID_OFFSET`` (8–15). This
        ID matches the index used by `GetStateData.aggregated_relay_objects`
        and `RelaySwitch.async_switch_*`.
        """
        offset = EXTERNAL_RELAY_ID_OFFSET if self.category == CATEGORY_EXTERNAL_RELAY else 0
        return self.category_id + offset

    def is_on(self) -> bool:
        """True if bit 0 of the relay value is set (output enabled)."""
        return int(self._value) & 1 == 1

    def is_off(self) -> bool:
        """True if bit 0 of the relay value is clear (output disabled)."""
        return not self.is_on()

    def is_manual_mode(self) -> bool:
        """True if bit 1 of the relay value is set (overridden by manual ENA)."""
        return int(self._value) & 2 == 2

    def is_auto_mode(self) -> bool:
        """True if bit 1 of the relay value is clear (controller-driven)."""
        return not self.is_manual_mode()

    def get_bit_mask(self) -> int:
        """Bit mask for this relay in the 16-bit ENA / state field.

        Internal relays occupy bits 0–7, external relays 8–15. The mask
        returned here is suitable for OR-ing into the
        `determine_overall_relay_bit_state` output before sending an ENA
        update.
        """
        if self._category == CATEGORY_EXTERNAL_RELAY:
            return 1 << (self._category_id + EXTERNAL_RELAY_ID_OFFSET)
        return 1 << self._category_id

relay_id property

relay_id: int

Aggregated relay ID across internal and external banks.

Internal relays return category_id directly (0–7); external relays return category_id + EXTERNAL_RELAY_ID_OFFSET (8–15). This ID matches the index used by GetStateData.aggregated_relay_objects and RelaySwitch.async_switch_*.

is_on

is_on() -> bool

True if bit 0 of the relay value is set (output enabled).

Source code in src/proconip/definitions.py

def is_on(self) -> bool:
    """True if bit 0 of the relay value is set (output enabled)."""
    return int(self._value) & 1 == 1

is_off

is_off() -> bool

True if bit 0 of the relay value is clear (output disabled).

Source code in src/proconip/definitions.py

def is_off(self) -> bool:
    """True if bit 0 of the relay value is clear (output disabled)."""
    return not self.is_on()

is_manual_mode

is_manual_mode() -> bool

True if bit 1 of the relay value is set (overridden by manual ENA).

Source code in src/proconip/definitions.py

def is_manual_mode(self) -> bool:
    """True if bit 1 of the relay value is set (overridden by manual ENA)."""
    return int(self._value) & 2 == 2

is_auto_mode

is_auto_mode() -> bool

True if bit 1 of the relay value is clear (controller-driven).

Source code in src/proconip/definitions.py

def is_auto_mode(self) -> bool:
    """True if bit 1 of the relay value is clear (controller-driven)."""
    return not self.is_manual_mode()

get_bit_mask

get_bit_mask() -> int

Bit mask for this relay in the 16-bit ENA / state field.

Internal relays occupy bits 0–7, external relays 8–15. The mask returned here is suitable for OR-ing into the determine_overall_relay_bit_state output before sending an ENA update.

Source code in src/proconip/definitions.py

def get_bit_mask(self) -> int:
    """Bit mask for this relay in the 16-bit ENA / state field.

    Internal relays occupy bits 0–7, external relays 8–15. The mask
    returned here is suitable for OR-ing into the
    `determine_overall_relay_bit_state` output before sending an ENA
    update.
    """
    if self._category == CATEGORY_EXTERNAL_RELAY:
        return 1 << (self._category_id + EXTERNAL_RELAY_ID_OFFSET)
    return 1 << self._category_id

async_get_dmx async

async_get_dmx(client_session: ClientSession, config: ConfigObject, timeout: float = 10.0) -> GetDmxData

Fetch and parse the controller's DMX channel state.

Returns a mutable GetDmxData you can iterate over, read with [index], or modify with set(index, value). Pass it to async_set_dmx to push the changes back to the controller.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
GetDmxData	A GetDmxData containing all 16 DMX channels.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.
InvalidPayloadException	If the response is empty.

Source code in src/proconip/api.py

async def async_get_dmx(
    client_session: ClientSession,
    config: ConfigObject,
    timeout: float = 10.0,
) -> GetDmxData:
    """Fetch and parse the controller's DMX channel state.

    Returns a mutable `GetDmxData` you can iterate over, read with `[index]`,
    or modify with `set(index, value)`. Pass it to `async_set_dmx` to push
    the changes back to the controller.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        timeout: Per-request timeout in seconds.

    Returns:
        A `GetDmxData` containing all 16 DMX channels.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
        InvalidPayloadException: If the response is empty.
    """
    raw_data = await async_get_raw_dmx(
        client_session=client_session, config=config, timeout=timeout
    )
    return GetDmxData(raw_data)

async_get_raw_data async

async_get_raw_data(client_session: ClientSession, config: ConfigObject, url: URL, timeout: float = 10.0) -> str

Send an authenticated GET request and return the response body as text.

This is the low-level primitive used by the higher-level state, DMX, and dosage helpers. Most callers will want one of those instead — reach for this directly only when you need to hit a custom URL on the controller.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession owned and closed by the caller. Reuse one session across many calls for connection pooling. TYPE: ClientSession
config	Controller configuration. The username and password are sent as HTTP Basic auth credentials. TYPE: ConfigObject
url	Fully-qualified URL to GET. Build it from config.base_url plus an API_PATH_* constant if you want to stay close to the standard endpoints. TYPE: URL
timeout	Maximum seconds to wait for the entire exchange (request and response body). Defaults to 10 seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body as a string. The controller typically returns
str	CSV; the caller is responsible for parsing.

RAISES	DESCRIPTION
BadCredentialsException	If the controller responds with HTTP 401 or 403.
BadStatusCodeException	If any other 4xx or 5xx status is returned.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For DNS failures, connection resets, and other network-level errors.

Source code in src/proconip/api.py

async def async_get_raw_data(
    client_session: ClientSession,
    config: ConfigObject,
    url: URL,
    timeout: float = 10.0,
) -> str:
    """Send an authenticated GET request and return the response body as text.

    This is the low-level primitive used by the higher-level state, DMX, and
    dosage helpers. Most callers will want one of those instead — reach for
    this directly only when you need to hit a custom URL on the controller.

    Args:
        client_session: An open `aiohttp.ClientSession` owned and closed by
            the caller. Reuse one session across many calls for connection
            pooling.
        config: Controller configuration. The username and password are sent
            as HTTP Basic auth credentials.
        url: Fully-qualified URL to GET. Build it from `config.base_url` plus
            an `API_PATH_*` constant if you want to stay close to the
            standard endpoints.
        timeout: Maximum seconds to wait for the entire exchange (request and
            response body). Defaults to 10 seconds.

    Returns:
        The raw response body as a string. The controller typically returns
        CSV; the caller is responsible for parsing.

    Raises:
        BadCredentialsException: If the controller responds with HTTP 401 or 403.
        BadStatusCodeException: If any other 4xx or 5xx status is returned.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For DNS failures, connection resets, and other
            network-level errors.
    """
    auth = BasicAuth(config.username, config.password)
    try:
        async with asyncio.timeout(timeout):
            async with client_session.get(url, auth=auth) as response:
                return await _handle_response(response)
    except TimeoutError as exc:
        raise TimeoutException("API request timed out") from exc
    except (ClientError, socket.gaierror) as exc:
        raise ProconipApiException(f"API request failed ({exc})") from exc

async_get_raw_dmx async

async_get_raw_dmx(client_session: ClientSession, config: ConfigObject, timeout: float = 10.0) -> str

Fetch the raw /GetDmx.csv body — the current 16 DMX channel values.

Use this when you want to parse the CSV yourself. To get a structured GetDmxData you can read and mutate, call async_get_dmx instead.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	A single CSV line containing the 16 channel values.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_get_raw_dmx(
    client_session: ClientSession,
    config: ConfigObject,
    timeout: float = 10.0,
) -> str:
    """Fetch the raw `/GetDmx.csv` body — the current 16 DMX channel values.

    Use this when you want to parse the CSV yourself. To get a structured
    `GetDmxData` you can read and mutate, call `async_get_dmx` instead.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        timeout: Per-request timeout in seconds.

    Returns:
        A single CSV line containing the 16 channel values.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    url = URL(config.base_url).with_path(API_PATH_GET_DMX)
    return await async_get_raw_data(client_session, config, url, timeout=timeout)

async_get_raw_state async

async_get_raw_state(client_session: ClientSession, config: ConfigObject, timeout: float = 10.0) -> str

Fetch the raw /GetState.csv body from the controller.

Use this when you want to handle the CSV yourself. To get a parsed object, call async_get_state instead.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration including base URL and credentials. TYPE: ConfigObject
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw multi-line CSV body returned by the controller.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors (DNS, connection reset).

Source code in src/proconip/api.py

async def async_get_raw_state(
    client_session: ClientSession,
    config: ConfigObject,
    timeout: float = 10.0,
) -> str:
    """Fetch the raw `/GetState.csv` body from the controller.

    Use this when you want to handle the CSV yourself. To get a parsed
    object, call `async_get_state` instead.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration including base URL and credentials.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw multi-line CSV body returned by the controller.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors (DNS, connection reset).
    """
    url = URL(config.base_url).with_path(API_PATH_GET_STATE)
    return await async_get_raw_data(client_session, config, url, timeout=timeout)

async_get_state async

async_get_state(client_session: ClientSession, config: ConfigObject, timeout: float = 10.0) -> GetStateData

Fetch and parse the controller's current state.

This is the most common entry point: it performs the GET request, parses the CSV response, and returns a GetStateData containing all sensor readings, relay states, dosage configuration, and so on.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration including base URL and credentials. TYPE: ConfigObject
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
GetStateData	A GetStateData instance with all properties populated.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.
InvalidPayloadException	If the response is empty or truncated.

Source code in src/proconip/api.py

async def async_get_state(
    client_session: ClientSession,
    config: ConfigObject,
    timeout: float = 10.0,
) -> GetStateData:
    """Fetch and parse the controller's current state.

    This is the most common entry point: it performs the GET request, parses
    the CSV response, and returns a `GetStateData` containing all sensor
    readings, relay states, dosage configuration, and so on.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration including base URL and credentials.
        timeout: Per-request timeout in seconds.

    Returns:
        A `GetStateData` instance with all properties populated.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
        InvalidPayloadException: If the response is empty or truncated.
    """
    raw_data = await async_get_raw_state(client_session, config, timeout=timeout)
    return GetStateData(raw_data)

async_post_usrcfg_cgi async

async_post_usrcfg_cgi(client_session: ClientSession, config: ConfigObject, payload: str, timeout: float = 10.0) -> str

Send a form-encoded POST to /usrcfg.cgi.

This is the low-level primitive behind relay switching and DMX writes. Most callers want async_switch_on, async_switch_off, async_set_auto_mode, or async_set_dmx instead. Use this directly only when you need to send a payload the higher-level helpers don't construct.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration including base URL and credentials. TYPE: ConfigObject
payload	The pre-encoded application/x-www-form-urlencoded body. TYPE: str
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body returned by the controller.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_post_usrcfg_cgi(
    client_session: ClientSession,
    config: ConfigObject,
    payload: str,
    timeout: float = 10.0,
) -> str:
    """Send a form-encoded POST to `/usrcfg.cgi`.

    This is the low-level primitive behind relay switching and DMX writes.
    Most callers want `async_switch_on`, `async_switch_off`,
    `async_set_auto_mode`, or `async_set_dmx` instead. Use this directly only
    when you need to send a payload the higher-level helpers don't construct.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration including base URL and credentials.
        payload: The pre-encoded `application/x-www-form-urlencoded` body.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw response body returned by the controller.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    url = URL(config.base_url).with_path(API_PATH_USRCFG)
    auth = BasicAuth(config.username, config.password)
    headers = {"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}
    try:
        async with asyncio.timeout(timeout):
            async with client_session.post(
                url=url,
                headers=headers,
                data=payload,
                auth=auth,
            ) as response:
                return await _handle_response(response)
    except TimeoutError as exc:
        raise TimeoutException("API request timed out") from exc
    except (ClientError, socket.gaierror) as exc:
        raise ProconipApiException(f"API request failed ({exc})") from exc

async_set_auto_mode async

async_set_auto_mode(client_session: ClientSession, config: ConfigObject, current_state: GetStateData, relay: Relay, timeout: float = 10.0) -> str

Hand a relay back to the controller's automatic schedule.

This is the inverse of async_switch_on/async_switch_off: instead of forcing a manual state, the relay's behavior is governed by the controller's configured rules (timer, sensor thresholds, dosage logic).

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
current_state	A recent GetStateData snapshot used to compute the ENA bit field. TYPE: GetStateData
relay	The relay to put back into auto mode. TYPE: Relay
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body returned by /usrcfg.cgi.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_set_auto_mode(
    client_session: ClientSession,
    config: ConfigObject,
    current_state: GetStateData,
    relay: Relay,
    timeout: float = 10.0,
) -> str:
    """Hand a relay back to the controller's automatic schedule.

    This is the inverse of `async_switch_on`/`async_switch_off`: instead of
    forcing a manual state, the relay's behavior is governed by the
    controller's configured rules (timer, sensor thresholds, dosage logic).

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        current_state: A recent `GetStateData` snapshot used to compute the
            ENA bit field.
        relay: The relay to put back into auto mode.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw response body returned by `/usrcfg.cgi`.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    bit_state = current_state.determine_overall_relay_bit_state()
    relay_bit_mask = relay.get_bit_mask()
    bit_state[0] &= ~relay_bit_mask
    bit_state[1] &= ~relay_bit_mask
    return await async_post_usrcfg_cgi(
        client_session=client_session,
        config=config,
        payload=f"ENA={bit_state[0]},{bit_state[1]}&MANUAL=1",
        timeout=timeout,
    )

async_set_dmx async

async_set_dmx(client_session: ClientSession, config: ConfigObject, dmx_states: GetDmxData, timeout: float = 10.0) -> str

Push DMX channel values back to the controller.

The controller's API only accepts the full 16-channel state in a single write. The usual pattern is: fetch with async_get_dmx, mutate channels with dmx_states.set(index, value), then call this function to commit.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
dmx_states	The full DMX state to write. All 16 channels are sent. TYPE: GetDmxData
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body returned by /usrcfg.cgi.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_set_dmx(
    client_session: ClientSession,
    config: ConfigObject,
    dmx_states: GetDmxData,
    timeout: float = 10.0,
) -> str:
    """Push DMX channel values back to the controller.

    The controller's API only accepts the full 16-channel state in a single
    write. The usual pattern is: fetch with `async_get_dmx`, mutate channels
    with `dmx_states.set(index, value)`, then call this function to commit.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        dmx_states: The full DMX state to write. All 16 channels are sent.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw response body returned by `/usrcfg.cgi`.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    payload = "&".join(f"{k}={v}" for k, v in dmx_states.post_data.items())
    return await async_post_usrcfg_cgi(
        client_session=client_session,
        config=config,
        payload=payload,
        timeout=timeout,
    )

async_start_dosage async

async_start_dosage(client_session: ClientSession, config: ConfigObject, dosage_target: DosageTarget, dosage_duration: int, timeout: float = 10.0) -> str

Trigger a manual, time-limited dosage on the controller.

Manual dosage is the safe alternative to switching a dosage relay on directly: the controller still applies the same interlocks it uses for the web UI (canister level, dosage enabled in config, redox/pH thresholds, …). If those checks fail, the controller silently no-ops while still returning HTTP 200 — there is no "dosage refused" error in the protocol.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
dosage_target	Which dosing pump to engage (chlorine, pH-, or pH+). TYPE: DosageTarget
dosage_duration	Run time in seconds. Allowed range depends on the controller's own dosage configuration; values that exceed the configured maximum are typically clamped silently by the device. TYPE: int
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body returned by /Command.htm.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_start_dosage(
    client_session: ClientSession,
    config: ConfigObject,
    dosage_target: DosageTarget,
    dosage_duration: int,
    timeout: float = 10.0,
) -> str:
    """Trigger a manual, time-limited dosage on the controller.

    Manual dosage is the safe alternative to switching a dosage relay on
    directly: the controller still applies the same interlocks it uses for the
    web UI (canister level, dosage enabled in config, redox/pH thresholds, …).
    If those checks fail, the controller silently no-ops while still returning
    HTTP 200 — there is no "dosage refused" error in the protocol.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        dosage_target: Which dosing pump to engage (chlorine, pH-, or pH+).
        dosage_duration: Run time in **seconds**. Allowed range depends on the
            controller's own dosage configuration; values that exceed the
            configured maximum are typically clamped silently by the device.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw response body returned by `/Command.htm`.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    query = f"MAN_DOSAGE={dosage_target},{dosage_duration}"
    url = URL(config.base_url).with_path(API_PATH_COMMAND).with_query(query)
    return await async_get_raw_data(client_session, config, url, timeout=timeout)

async_switch_off async

async_switch_off(client_session: ClientSession, config: ConfigObject, current_state: GetStateData, relay: Relay, timeout: float = 10.0) -> str

Switch a relay to manual OFF.

Like async_switch_on, this puts the relay into manual mode but with the output disabled. Other relays in current_state keep their state. Unlike switching on, this is allowed for dosage relays — turning a dosage pump off is always safe.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
current_state	A recent GetStateData snapshot used to compute the ENA bit field. TYPE: GetStateData
relay	The relay to switch off. TYPE: Relay
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body returned by /usrcfg.cgi.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_switch_off(
    client_session: ClientSession,
    config: ConfigObject,
    current_state: GetStateData,
    relay: Relay,
    timeout: float = 10.0,
) -> str:
    """Switch a relay to manual OFF.

    Like `async_switch_on`, this puts the relay into manual mode but with the
    output disabled. Other relays in ``current_state`` keep their state.
    Unlike switching on, this is allowed for dosage relays — turning a dosage
    pump off is always safe.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        current_state: A recent `GetStateData` snapshot used to compute the
            ENA bit field.
        relay: The relay to switch off.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw response body returned by `/usrcfg.cgi`.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    bit_state = current_state.determine_overall_relay_bit_state()
    relay_bit_mask = relay.get_bit_mask()
    bit_state[0] |= relay_bit_mask
    bit_state[1] &= ~relay_bit_mask
    return await async_post_usrcfg_cgi(
        client_session=client_session,
        config=config,
        payload=f"ENA={bit_state[0]},{bit_state[1]}&MANUAL=1",
        timeout=timeout,
    )

async_switch_on async

async_switch_on(client_session: ClientSession, config: ConfigObject, current_state: GetStateData, relay: Relay, timeout: float = 10.0) -> str

Switch a relay to manual ON.

Manual mode overrides the controller's schedule until the relay is explicitly switched off or set back to auto. Other relays in current_state are preserved by reading their bit field, which is why a fresh GetStateData snapshot is required.

Dosage relays (chlorine, pH+, pH-) cannot be switched on this way — the controller's safety logic interlocks them. Use async_start_dosage (or DosageControl) instead for time-limited manual dosing.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
current_state	A recent GetStateData snapshot. Used to compute the ENA bit field so that other relays keep their current state. TYPE: GetStateData
relay	The relay to switch on. TYPE: Relay
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body returned by /usrcfg.cgi.

RAISES	DESCRIPTION
BadRelayException	If relay is one of the configured dosage control relays.
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_switch_on(
    client_session: ClientSession,
    config: ConfigObject,
    current_state: GetStateData,
    relay: Relay,
    timeout: float = 10.0,
) -> str:
    """Switch a relay to manual ON.

    Manual mode overrides the controller's schedule until the relay is
    explicitly switched off or set back to auto. Other relays in
    ``current_state`` are preserved by reading their bit field, which is why
    a fresh `GetStateData` snapshot is required.

    Dosage relays (chlorine, pH+, pH-) cannot be switched on this way — the
    controller's safety logic interlocks them. Use `async_start_dosage` (or
    `DosageControl`) instead for time-limited manual dosing.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        current_state: A recent `GetStateData` snapshot. Used to compute the
            ENA bit field so that other relays keep their current state.
        relay: The relay to switch on.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw response body returned by `/usrcfg.cgi`.

    Raises:
        BadRelayException: If ``relay`` is one of the configured dosage
            control relays.
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    if current_state.is_dosage_relay(relay):
        raise BadRelayException("Cannot permanently switch on a dosage relay")
    bit_state = current_state.determine_overall_relay_bit_state()
    relay_bit_mask = relay.get_bit_mask()
    bit_state[0] |= relay_bit_mask
    bit_state[1] |= relay_bit_mask
    return await async_post_usrcfg_cgi(
        client_session=client_session,
        config=config,
        payload=f"ENA={bit_state[0]},{bit_state[1]}&MANUAL=1",
        timeout=timeout,
    )

async_trigger_digital_input async

async_trigger_digital_input(client_session: ClientSession, config: ConfigObject, digital_input_id: int, timeout: float = 10.0, hold_seconds: float = DIGITAL_INPUT_PULSE_SECONDS) -> str

Trigger (momentarily pulse) a digital input via the WEBIO IO field.

The controller's web UI exposes each digital input as a push button: it sets the input's bit, waits, then clears it. This sends the same two /usrcfg.cgi writes — IO=<mask>&WEBIO=1 (press) then, after hold_seconds, IO=0&WEBIO=1 (release) — where mask = 1 << digital_input_id. The hold mirrors the web UI (~600ms) so the controller reliably registers the pulse.

Unlike relay switching, no GetStateData snapshot is needed: the IO field is written directly rather than read-modify-written.

Because the pulse is two separate writes, the release is not fully guaranteed. If the hold is cancelled (e.g. the caller's task is shut down), a best-effort release is attempted before the cancellation propagates. If the release POST itself fails (timeout / network blip), that error propagates and the input is left asserted HIGH until the next write.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration including base URL and credentials. TYPE: ConfigObject
digital_input_id	Zero-based digital input index (0–3). TYPE: int
timeout	Per-request timeout in seconds, applied to each of the two POSTs. TYPE: float DEFAULT: 10.0
hold_seconds	Seconds to hold the input HIGH between the press and release POSTs. Defaults to the web UI's ~600ms. TYPE: float DEFAULT: DIGITAL_INPUT_PULSE_SECONDS

RETURNS	DESCRIPTION
str	The raw response body returned by the release POST.

RAISES	DESCRIPTION
ValueError	If digital_input_id is not in 0–3.
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If an exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_trigger_digital_input(
    client_session: ClientSession,
    config: ConfigObject,
    digital_input_id: int,
    timeout: float = 10.0,
    hold_seconds: float = DIGITAL_INPUT_PULSE_SECONDS,
) -> str:
    """Trigger (momentarily pulse) a digital input via the WEBIO ``IO`` field.

    The controller's web UI exposes each digital input as a push button: it
    sets the input's bit, waits, then clears it. This sends the same two
    `/usrcfg.cgi` writes — ``IO=<mask>&WEBIO=1`` (press) then, after
    ``hold_seconds``, ``IO=0&WEBIO=1`` (release) — where
    ``mask = 1 << digital_input_id``. The hold mirrors the web UI (~600ms) so
    the controller reliably registers the pulse.

    Unlike relay switching, no `GetStateData` snapshot is needed: the ``IO``
    field is written directly rather than read-modify-written.

    Because the pulse is two separate writes, the release is not fully
    guaranteed. If the hold is **cancelled** (e.g. the caller's task is shut
    down), a best-effort release is attempted before the cancellation
    propagates. If the **release POST itself fails** (timeout / network blip),
    that error propagates and the input is left asserted HIGH until the next
    write.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration including base URL and credentials.
        digital_input_id: Zero-based digital input index (0–3).
        timeout: Per-request timeout in seconds, applied to each of the two
            POSTs.
        hold_seconds: Seconds to hold the input HIGH between the press and
            release POSTs. Defaults to the web UI's ~600ms.

    Returns:
        The raw response body returned by the release POST.

    Raises:
        ValueError: If ``digital_input_id`` is not in 0–3.
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If an exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    if not 0 <= digital_input_id < DIGITAL_INPUT_COUNT:
        raise ValueError(
            f"digital_input_id must be in 0..{DIGITAL_INPUT_COUNT - 1}, got {digital_input_id}"
        )
    mask = 1 << digital_input_id
    await async_post_usrcfg_cgi(
        client_session=client_session,
        config=config,
        payload=f"IO={mask}&WEBIO=1",
        timeout=timeout,
    )
    try:
        await asyncio.sleep(hold_seconds)
    except asyncio.CancelledError:
        # Best-effort release so a cancelled hold doesn't leave the input
        # asserted HIGH. Networking inside a cancellation is awkward, so this
        # is a best effort, not a guarantee.
        with contextlib.suppress(Exception):
            await async_post_usrcfg_cgi(
                client_session=client_session,
                config=config,
                payload="IO=0&WEBIO=1",
                timeout=timeout,
            )
        raise
    return await async_post_usrcfg_cgi(
        client_session=client_session,
        config=config,
        payload="IO=0&WEBIO=1",
        timeout=timeout,
    )

Async API client (proconip.api)

api

Async HTTP client for the ProCon.IP pool controller.

The controller exposes four HTTP endpoints — /GetState.csv for sensor and relay state, /GetDmx.csv for DMX channel state, /usrcfg.cgi for configuration writes (including manual relay switching and DMX updates), and /Command.htm for manual dosage commands. This module wraps all four with typed exceptions and a configurable per-request timeout.

Each operation is available in two equivalent forms:

• Free async functions like async_get_state and async_switch_on. These take an aiohttp.ClientSession and ConfigObject explicitly, which makes them composable in larger applications that already manage their own session lifecycle.
• OO wrappers like GetState, RelaySwitch, DosageControl, and DmxControl. These bind the session and config once at construction and expose ergonomic instance methods so callers don't have to repeat those arguments on every call.

The OO wrappers delegate to the free functions, so behavior is identical.

All requests use HTTP Basic auth and run inside an asyncio.timeout block that covers both the request and the response body read. Network failures, HTTP error codes, and timeouts are all mapped to ProconipApiException subclasses so callers can handle them uniformly.

ProconipApiException

Bases: Exception

Base exception for any failed API call.

Catch this if you want to handle all controller-side and network failures uniformly. The more specific subclasses below let you distinguish auth, HTTP, and timeout problems when that's useful.

Source code in src/proconip/api.py

class ProconipApiException(Exception):
    """Base exception for any failed API call.

    Catch this if you want to handle all controller-side and network failures
    uniformly. The more specific subclasses below let you distinguish auth,
    HTTP, and timeout problems when that's useful.
    """

BadCredentialsException

Bases: ProconipApiException

Raised when the controller rejects the request with HTTP 401 or 403.

Almost always means the username or password in the ConfigObject is wrong.

Source code in src/proconip/api.py

class BadCredentialsException(ProconipApiException):
    """Raised when the controller rejects the request with HTTP 401 or 403.

    Almost always means the username or password in the `ConfigObject` is wrong.
    """

BadStatusCodeException

Bases: ProconipApiException

Raised on any unexpected HTTP error status (4xx or 5xx) other than 401/403.

The original aiohttp.ClientResponseError is preserved as the cause and can be inspected via __cause__ if the status code or response details are needed.

Source code in src/proconip/api.py

class BadStatusCodeException(ProconipApiException):
    """Raised on any unexpected HTTP error status (4xx or 5xx) other than 401/403.

    The original `aiohttp.ClientResponseError` is preserved as the cause and
    can be inspected via `__cause__` if the status code or response details
    are needed.
    """

TimeoutException

Bases: ProconipApiException

Raised when a request does not complete within the configured timeout.

This covers both network-level stalls (connection or socket I/O) and slow response bodies, since the timeout context wraps the entire exchange.

Source code in src/proconip/api.py

class TimeoutException(ProconipApiException):
    """Raised when a request does not complete within the configured timeout.

    This covers both network-level stalls (connection or socket I/O) and slow
    response bodies, since the timeout context wraps the entire exchange.
    """

GetState

Convenience wrapper that binds a session and config for state reads.

Construct once with your aiohttp.ClientSession and ConfigObject, then call async_get_state() (parsed) or async_get_raw_state() (CSV) without repeating those arguments each time.

The constructor also binds a default timeout for every call made via this wrapper. Each method then accepts an optional per-call timeout that overrides the bound default when supplied.

Example

async with aiohttp.ClientSession() as session:
    api = GetState(session, config, timeout=15.0)
    state = await api.async_get_state()              # uses 15.0 s
    quick = await api.async_get_state(timeout=2.0)   # overrides to 2.0 s
    print(state.ph_electrode.display_value)

Source code in src/proconip/api.py

class GetState:
    """Convenience wrapper that binds a session and config for state reads.

    Construct once with your `aiohttp.ClientSession` and `ConfigObject`, then
    call `async_get_state()` (parsed) or `async_get_raw_state()` (CSV) without
    repeating those arguments each time.

    The constructor also binds a default `timeout` for every call made via
    this wrapper. Each method then accepts an optional per-call `timeout`
    that overrides the bound default when supplied.

    Example:
        ```python
        async with aiohttp.ClientSession() as session:
            api = GetState(session, config, timeout=15.0)
            state = await api.async_get_state()              # uses 15.0 s
            quick = await api.async_get_state(timeout=2.0)   # overrides to 2.0 s
            print(state.ph_electrode.display_value)
        ```
    """

    def __init__(
        self,
        client_session: ClientSession,
        config: ConfigObject,
        timeout: float = 10.0,
    ):
        """Bind the session, config, and default per-request timeout.

        Args:
            client_session: An open `aiohttp.ClientSession`.
            config: Controller configuration.
            timeout: Default per-request timeout in seconds, used when a
                method is called without its own ``timeout`` argument.
        """
        self.client_session = client_session
        self.config = config
        self.timeout = timeout

    async def async_get_raw_state(self, timeout: float | None = None) -> str:
        """Fetch the raw `/GetState.csv` body using the bound session and config.

        Args:
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        See `async_get_raw_state` (the free function) for the full description
        of behavior and raised exceptions.
        """
        return await async_get_raw_state(
            self.client_session,
            self.config,
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_get_state(self, timeout: float | None = None) -> GetStateData:
        """Fetch and parse the controller state into a `GetStateData` instance.

        Args:
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        See `async_get_state` (the free function) for the full description of
        behavior and raised exceptions.
        """
        return await async_get_state(
            self.client_session,
            self.config,
            timeout=self.timeout if timeout is None else timeout,
        )

async_get_raw_state async

async_get_raw_state(timeout: float | None = None) -> str

Fetch the raw /GetState.csv body using the bound session and config.

PARAMETER	DESCRIPTION
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

See async_get_raw_state (the free function) for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_get_raw_state(self, timeout: float | None = None) -> str:
    """Fetch the raw `/GetState.csv` body using the bound session and config.

    Args:
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    See `async_get_raw_state` (the free function) for the full description
    of behavior and raised exceptions.
    """
    return await async_get_raw_state(
        self.client_session,
        self.config,
        timeout=self.timeout if timeout is None else timeout,
    )

async_get_state async

async_get_state(timeout: float | None = None) -> GetStateData

Fetch and parse the controller state into a GetStateData instance.

PARAMETER	DESCRIPTION
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

See async_get_state (the free function) for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_get_state(self, timeout: float | None = None) -> GetStateData:
    """Fetch and parse the controller state into a `GetStateData` instance.

    Args:
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    See `async_get_state` (the free function) for the full description of
    behavior and raised exceptions.
    """
    return await async_get_state(
        self.client_session,
        self.config,
        timeout=self.timeout if timeout is None else timeout,
    )

RelaySwitch

Convenience wrapper that binds a session and config for relay control.

Construct once with your aiohttp.ClientSession and ConfigObject, then switch relays by aggregated relay ID (an integer) instead of constructing Relay instances by hand.

Aggregated relay IDs run from 0 to 7 for the eight built-in relays and 8 to 15 for the eight optional external relays.

The constructor also binds a default timeout for every call made via this wrapper. Each method then accepts an optional per-call timeout that overrides the bound default when supplied.

Example

rs = RelaySwitch(session, config)
state = await GetState(session, config).async_get_state()
await rs.async_switch_on(state, relay_id=2)

Source code in src/proconip/api.py

class RelaySwitch:
    """Convenience wrapper that binds a session and config for relay control.

    Construct once with your `aiohttp.ClientSession` and `ConfigObject`, then
    switch relays by **aggregated relay ID** (an integer) instead of
    constructing `Relay` instances by hand.

    Aggregated relay IDs run from 0 to 7 for the eight built-in relays and 8
    to 15 for the eight optional external relays.

    The constructor also binds a default `timeout` for every call made via
    this wrapper. Each method then accepts an optional per-call `timeout`
    that overrides the bound default when supplied.

    Example:
        ```python
        rs = RelaySwitch(session, config)
        state = await GetState(session, config).async_get_state()
        await rs.async_switch_on(state, relay_id=2)
        ```
    """

    def __init__(
        self,
        client_session: ClientSession,
        config: ConfigObject,
        timeout: float = 10.0,
    ):
        """Bind the session, config, and default per-request timeout.

        Args:
            client_session: An open `aiohttp.ClientSession`.
            config: Controller configuration.
            timeout: Default per-request timeout in seconds, used when a
                method is called without its own ``timeout`` argument.
        """
        self.client_session = client_session
        self.config = config
        self.timeout = timeout

    async def async_switch_on(
        self,
        current_state: GetStateData,
        relay_id: int,
        timeout: float | None = None,
    ) -> str:
        """Switch the relay identified by ``relay_id`` to manual ON.

        Args:
            current_state: A recent `GetStateData` snapshot.
            relay_id: Aggregated relay ID (0–15).
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        Resolves the relay ID against ``current_state`` and delegates to the
        free function `async_switch_on`. See it for the full description of
        behavior and raised exceptions, including `BadRelayException` for
        dosage relays.
        """
        return await async_switch_on(
            client_session=self.client_session,
            config=self.config,
            current_state=current_state,
            relay=current_state.get_relay(relay_id),
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_switch_off(
        self,
        current_state: GetStateData,
        relay_id: int,
        timeout: float | None = None,
    ) -> str:
        """Switch the relay identified by ``relay_id`` to manual OFF.

        Args:
            current_state: A recent `GetStateData` snapshot.
            relay_id: Aggregated relay ID (0–15).
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        Resolves the relay ID against ``current_state`` and delegates to the
        free function `async_switch_off`. See it for the full description of
        behavior and raised exceptions.
        """
        return await async_switch_off(
            client_session=self.client_session,
            config=self.config,
            current_state=current_state,
            relay=current_state.get_relay(relay_id),
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_set_auto_mode(
        self,
        current_state: GetStateData,
        relay_id: int,
        timeout: float | None = None,
    ) -> str:
        """Hand the relay identified by ``relay_id`` back to AUTO mode.

        Args:
            current_state: A recent `GetStateData` snapshot.
            relay_id: Aggregated relay ID (0–15).
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        Resolves the relay ID against ``current_state`` and delegates to the
        free function `async_set_auto_mode`. See it for the full description
        of behavior and raised exceptions.
        """
        return await async_set_auto_mode(
            client_session=self.client_session,
            config=self.config,
            current_state=current_state,
            relay=current_state.get_relay(relay_id),
            timeout=self.timeout if timeout is None else timeout,
        )

async_switch_on async

async_switch_on(current_state: GetStateData, relay_id: int, timeout: float | None = None) -> str

Switch the relay identified by relay_id to manual ON.

PARAMETER	DESCRIPTION
current_state	A recent GetStateData snapshot. TYPE: GetStateData
relay_id	Aggregated relay ID (0–15). TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

Resolves the relay ID against current_state and delegates to the free function async_switch_on. See it for the full description of behavior and raised exceptions, including BadRelayException for dosage relays.

Source code in src/proconip/api.py

async def async_switch_on(
    self,
    current_state: GetStateData,
    relay_id: int,
    timeout: float | None = None,
) -> str:
    """Switch the relay identified by ``relay_id`` to manual ON.

    Args:
        current_state: A recent `GetStateData` snapshot.
        relay_id: Aggregated relay ID (0–15).
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    Resolves the relay ID against ``current_state`` and delegates to the
    free function `async_switch_on`. See it for the full description of
    behavior and raised exceptions, including `BadRelayException` for
    dosage relays.
    """
    return await async_switch_on(
        client_session=self.client_session,
        config=self.config,
        current_state=current_state,
        relay=current_state.get_relay(relay_id),
        timeout=self.timeout if timeout is None else timeout,
    )

async_switch_off async

async_switch_off(current_state: GetStateData, relay_id: int, timeout: float | None = None) -> str

Switch the relay identified by relay_id to manual OFF.

PARAMETER	DESCRIPTION
current_state	A recent GetStateData snapshot. TYPE: GetStateData
relay_id	Aggregated relay ID (0–15). TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

Resolves the relay ID against current_state and delegates to the free function async_switch_off. See it for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_switch_off(
    self,
    current_state: GetStateData,
    relay_id: int,
    timeout: float | None = None,
) -> str:
    """Switch the relay identified by ``relay_id`` to manual OFF.

    Args:
        current_state: A recent `GetStateData` snapshot.
        relay_id: Aggregated relay ID (0–15).
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    Resolves the relay ID against ``current_state`` and delegates to the
    free function `async_switch_off`. See it for the full description of
    behavior and raised exceptions.
    """
    return await async_switch_off(
        client_session=self.client_session,
        config=self.config,
        current_state=current_state,
        relay=current_state.get_relay(relay_id),
        timeout=self.timeout if timeout is None else timeout,
    )

async_set_auto_mode async

async_set_auto_mode(current_state: GetStateData, relay_id: int, timeout: float | None = None) -> str

Hand the relay identified by relay_id back to AUTO mode.

PARAMETER	DESCRIPTION
current_state	A recent GetStateData snapshot. TYPE: GetStateData
relay_id	Aggregated relay ID (0–15). TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

Resolves the relay ID against current_state and delegates to the free function async_set_auto_mode. See it for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_set_auto_mode(
    self,
    current_state: GetStateData,
    relay_id: int,
    timeout: float | None = None,
) -> str:
    """Hand the relay identified by ``relay_id`` back to AUTO mode.

    Args:
        current_state: A recent `GetStateData` snapshot.
        relay_id: Aggregated relay ID (0–15).
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    Resolves the relay ID against ``current_state`` and delegates to the
    free function `async_set_auto_mode`. See it for the full description
    of behavior and raised exceptions.
    """
    return await async_set_auto_mode(
        client_session=self.client_session,
        config=self.config,
        current_state=current_state,
        relay=current_state.get_relay(relay_id),
        timeout=self.timeout if timeout is None else timeout,
    )

DosageControl

Convenience wrapper for manual dosage commands.

Construct once with your aiohttp.ClientSession and ConfigObject, then trigger dosing per chemical without specifying the DosageTarget enum each time.

The constructor also binds a default timeout for every call made via this wrapper. Each method then accepts an optional per-call timeout that overrides the bound default when supplied.

Example

dc = DosageControl(session, config)
await dc.async_chlorine_dosage(60)   # 60 seconds of chlorine

Source code in src/proconip/api.py

class DosageControl:
    """Convenience wrapper for manual dosage commands.

    Construct once with your `aiohttp.ClientSession` and `ConfigObject`, then
    trigger dosing per chemical without specifying the `DosageTarget` enum
    each time.

    The constructor also binds a default `timeout` for every call made via
    this wrapper. Each method then accepts an optional per-call `timeout`
    that overrides the bound default when supplied.

    Example:
        ```python
        dc = DosageControl(session, config)
        await dc.async_chlorine_dosage(60)   # 60 seconds of chlorine
        ```
    """

    def __init__(
        self,
        client_session: ClientSession,
        config: ConfigObject,
        timeout: float = 10.0,
    ):
        """Bind the session, config, and default per-request timeout.

        Args:
            client_session: An open `aiohttp.ClientSession`.
            config: Controller configuration.
            timeout: Default per-request timeout in seconds, used when a
                method is called without its own ``timeout`` argument.
        """
        self.client_session = client_session
        self.config = config
        self.timeout = timeout

    async def async_chlorine_dosage(
        self, dosage_duration: int, timeout: float | None = None
    ) -> str:
        """Run the chlorine dosage pump for ``dosage_duration`` seconds.

        Args:
            dosage_duration: Run time in seconds.
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        Subject to the same controller-side safety interlocks as manual
        dosage from the web UI. See `async_start_dosage` for full behavior
        and raised exceptions.
        """
        return await async_start_dosage(
            client_session=self.client_session,
            config=self.config,
            dosage_target=DosageTarget.CHLORINE,
            dosage_duration=dosage_duration,
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_ph_minus_dosage(
        self, dosage_duration: int, timeout: float | None = None
    ) -> str:
        """Run the pH- dosage pump for ``dosage_duration`` seconds.

        Args:
            dosage_duration: Run time in seconds.
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        Subject to the same controller-side safety interlocks as manual
        dosage from the web UI. See `async_start_dosage` for full behavior
        and raised exceptions.
        """
        return await async_start_dosage(
            client_session=self.client_session,
            config=self.config,
            dosage_target=DosageTarget.PH_MINUS,
            dosage_duration=dosage_duration,
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_ph_plus_dosage(self, dosage_duration: int, timeout: float | None = None) -> str:
        """Run the pH+ dosage pump for ``dosage_duration`` seconds.

        Args:
            dosage_duration: Run time in seconds.
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        Subject to the same controller-side safety interlocks as manual
        dosage from the web UI. See `async_start_dosage` for full behavior
        and raised exceptions.
        """
        return await async_start_dosage(
            client_session=self.client_session,
            config=self.config,
            dosage_target=DosageTarget.PH_PLUS,
            dosage_duration=dosage_duration,
            timeout=self.timeout if timeout is None else timeout,
        )

async_chlorine_dosage async

async_chlorine_dosage(dosage_duration: int, timeout: float | None = None) -> str

Run the chlorine dosage pump for dosage_duration seconds.

PARAMETER	DESCRIPTION
dosage_duration	Run time in seconds. TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

Subject to the same controller-side safety interlocks as manual dosage from the web UI. See async_start_dosage for full behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_chlorine_dosage(
    self, dosage_duration: int, timeout: float | None = None
) -> str:
    """Run the chlorine dosage pump for ``dosage_duration`` seconds.

    Args:
        dosage_duration: Run time in seconds.
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    Subject to the same controller-side safety interlocks as manual
    dosage from the web UI. See `async_start_dosage` for full behavior
    and raised exceptions.
    """
    return await async_start_dosage(
        client_session=self.client_session,
        config=self.config,
        dosage_target=DosageTarget.CHLORINE,
        dosage_duration=dosage_duration,
        timeout=self.timeout if timeout is None else timeout,
    )

async_ph_minus_dosage async

async_ph_minus_dosage(dosage_duration: int, timeout: float | None = None) -> str

Run the pH- dosage pump for dosage_duration seconds.

PARAMETER	DESCRIPTION
dosage_duration	Run time in seconds. TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

Subject to the same controller-side safety interlocks as manual dosage from the web UI. See async_start_dosage for full behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_ph_minus_dosage(
    self, dosage_duration: int, timeout: float | None = None
) -> str:
    """Run the pH- dosage pump for ``dosage_duration`` seconds.

    Args:
        dosage_duration: Run time in seconds.
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    Subject to the same controller-side safety interlocks as manual
    dosage from the web UI. See `async_start_dosage` for full behavior
    and raised exceptions.
    """
    return await async_start_dosage(
        client_session=self.client_session,
        config=self.config,
        dosage_target=DosageTarget.PH_MINUS,
        dosage_duration=dosage_duration,
        timeout=self.timeout if timeout is None else timeout,
    )

async_ph_plus_dosage async

async_ph_plus_dosage(dosage_duration: int, timeout: float | None = None) -> str

Run the pH+ dosage pump for dosage_duration seconds.

PARAMETER	DESCRIPTION
dosage_duration	Run time in seconds. TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

Subject to the same controller-side safety interlocks as manual dosage from the web UI. See async_start_dosage for full behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_ph_plus_dosage(self, dosage_duration: int, timeout: float | None = None) -> str:
    """Run the pH+ dosage pump for ``dosage_duration`` seconds.

    Args:
        dosage_duration: Run time in seconds.
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    Subject to the same controller-side safety interlocks as manual
    dosage from the web UI. See `async_start_dosage` for full behavior
    and raised exceptions.
    """
    return await async_start_dosage(
        client_session=self.client_session,
        config=self.config,
        dosage_target=DosageTarget.PH_PLUS,
        dosage_duration=dosage_duration,
        timeout=self.timeout if timeout is None else timeout,
    )

DmxControl

Convenience wrapper that binds a session and config for DMX I/O.

Construct once with your aiohttp.ClientSession and ConfigObject, then read or write DMX channels without repeating those arguments each time.

The constructor also binds a default timeout for every call made via this wrapper. Each method then accepts an optional per-call timeout that overrides the bound default when supplied.

Example

dc = DmxControl(session, config)
dmx = await dc.async_get_dmx()
for ch in dmx:
    dmx.set(ch.index, (ch.value + 64) % 256)
await dc.async_set(dmx)

Source code in src/proconip/api.py

class DmxControl:
    """Convenience wrapper that binds a session and config for DMX I/O.

    Construct once with your `aiohttp.ClientSession` and `ConfigObject`, then
    read or write DMX channels without repeating those arguments each time.

    The constructor also binds a default `timeout` for every call made via
    this wrapper. Each method then accepts an optional per-call `timeout`
    that overrides the bound default when supplied.

    Example:
        ```python
        dc = DmxControl(session, config)
        dmx = await dc.async_get_dmx()
        for ch in dmx:
            dmx.set(ch.index, (ch.value + 64) % 256)
        await dc.async_set(dmx)
        ```
    """

    def __init__(
        self,
        client_session: ClientSession,
        config: ConfigObject,
        timeout: float = 10.0,
    ):
        """Bind the session, config, and default per-request timeout.

        Args:
            client_session: An open `aiohttp.ClientSession`.
            config: Controller configuration.
            timeout: Default per-request timeout in seconds, used when a
                method is called without its own ``timeout`` argument.
        """
        self.client_session = client_session
        self.config = config
        self.timeout = timeout

    async def async_get_raw_dmx(self, timeout: float | None = None) -> str:
        """Fetch the raw `/GetDmx.csv` body using the bound session and config.

        Args:
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        See `async_get_raw_dmx` (the free function) for the full description
        of behavior and raised exceptions.
        """
        return await async_get_raw_dmx(
            self.client_session,
            self.config,
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_get_dmx(self, timeout: float | None = None) -> GetDmxData:
        """Fetch and parse the current DMX state into a `GetDmxData` instance.

        Args:
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        See `async_get_dmx` (the free function) for the full description of
        behavior and raised exceptions.
        """
        return await async_get_dmx(
            self.client_session,
            self.config,
            timeout=self.timeout if timeout is None else timeout,
        )

    async def async_set(self, data: GetDmxData, timeout: float | None = None) -> str:
        """Push the given DMX state back to the controller.

        Args:
            data: The `GetDmxData` to write.
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.

        See `async_set_dmx` (the free function) for the full description of
        behavior and raised exceptions.
        """
        return await async_set_dmx(
            client_session=self.client_session,
            config=self.config,
            dmx_states=data,
            timeout=self.timeout if timeout is None else timeout,
        )

async_get_raw_dmx async

async_get_raw_dmx(timeout: float | None = None) -> str

Fetch the raw /GetDmx.csv body using the bound session and config.

PARAMETER	DESCRIPTION
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

See async_get_raw_dmx (the free function) for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_get_raw_dmx(self, timeout: float | None = None) -> str:
    """Fetch the raw `/GetDmx.csv` body using the bound session and config.

    Args:
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    See `async_get_raw_dmx` (the free function) for the full description
    of behavior and raised exceptions.
    """
    return await async_get_raw_dmx(
        self.client_session,
        self.config,
        timeout=self.timeout if timeout is None else timeout,
    )

async_get_dmx async

async_get_dmx(timeout: float | None = None) -> GetDmxData

Fetch and parse the current DMX state into a GetDmxData instance.

PARAMETER	DESCRIPTION
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

See async_get_dmx (the free function) for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_get_dmx(self, timeout: float | None = None) -> GetDmxData:
    """Fetch and parse the current DMX state into a `GetDmxData` instance.

    Args:
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    See `async_get_dmx` (the free function) for the full description of
    behavior and raised exceptions.
    """
    return await async_get_dmx(
        self.client_session,
        self.config,
        timeout=self.timeout if timeout is None else timeout,
    )

async_set async

async_set(data: GetDmxData, timeout: float | None = None) -> str

Push the given DMX state back to the controller.

PARAMETER	DESCRIPTION
data	The GetDmxData to write. TYPE: GetDmxData
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None

See async_set_dmx (the free function) for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_set(self, data: GetDmxData, timeout: float | None = None) -> str:
    """Push the given DMX state back to the controller.

    Args:
        data: The `GetDmxData` to write.
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.

    See `async_set_dmx` (the free function) for the full description of
    behavior and raised exceptions.
    """
    return await async_set_dmx(
        client_session=self.client_session,
        config=self.config,
        dmx_states=data,
        timeout=self.timeout if timeout is None else timeout,
    )

DigitalInputControl

Convenience wrapper that binds a session and config for triggering inputs.

Construct once with your aiohttp.ClientSession and ConfigObject, then pulse digital inputs by zero-based id (0–3) without repeating those arguments each time.

The constructor also binds a default timeout for every call made via this wrapper. Each method then accepts an optional per-call timeout that overrides the bound default when supplied.

Example

dic = DigitalInputControl(session, config)
await dic.async_trigger(0)  # pulse the first digital input

Source code in src/proconip/api.py

class DigitalInputControl:
    """Convenience wrapper that binds a session and config for triggering inputs.

    Construct once with your `aiohttp.ClientSession` and `ConfigObject`, then
    pulse digital inputs by zero-based id (0–3) without repeating those
    arguments each time.

    The constructor also binds a default `timeout` for every call made via this
    wrapper. Each method then accepts an optional per-call `timeout` that
    overrides the bound default when supplied.

    Example:
        ```python
        dic = DigitalInputControl(session, config)
        await dic.async_trigger(0)  # pulse the first digital input
        ```
    """

    def __init__(
        self,
        client_session: ClientSession,
        config: ConfigObject,
        timeout: float = 10.0,
    ):
        """Bind the session, config, and default per-request timeout.

        Args:
            client_session: An open `aiohttp.ClientSession`.
            config: Controller configuration.
            timeout: Default per-request timeout in seconds, used when a method
                is called without its own ``timeout`` argument.
        """
        self.client_session = client_session
        self.config = config
        self.timeout = timeout

    async def async_trigger(
        self,
        digital_input_id: int,
        timeout: float | None = None,
        hold_seconds: float = DIGITAL_INPUT_PULSE_SECONDS,
    ) -> str:
        """Trigger the digital input identified by ``digital_input_id``.

        Args:
            digital_input_id: Zero-based digital input index (0–3).
            timeout: Override for this call only. If ``None``, the timeout
                bound in `__init__` is used.
            hold_seconds: Seconds to hold the input HIGH between press and
                release. Defaults to the web UI's ~600ms.

        See `async_trigger_digital_input` (the free function) for the full
        description of behavior and raised exceptions.
        """
        return await async_trigger_digital_input(
            client_session=self.client_session,
            config=self.config,
            digital_input_id=digital_input_id,
            timeout=self.timeout if timeout is None else timeout,
            hold_seconds=hold_seconds,
        )

async_trigger async

async_trigger(digital_input_id: int, timeout: float | None = None, hold_seconds: float = DIGITAL_INPUT_PULSE_SECONDS) -> str

Trigger the digital input identified by digital_input_id.

PARAMETER	DESCRIPTION
digital_input_id	Zero-based digital input index (0–3). TYPE: int
timeout	Override for this call only. If None, the timeout bound in __init__ is used. TYPE: float | None DEFAULT: None
hold_seconds	Seconds to hold the input HIGH between press and release. Defaults to the web UI's ~600ms. TYPE: float DEFAULT: DIGITAL_INPUT_PULSE_SECONDS

See async_trigger_digital_input (the free function) for the full description of behavior and raised exceptions.

Source code in src/proconip/api.py

async def async_trigger(
    self,
    digital_input_id: int,
    timeout: float | None = None,
    hold_seconds: float = DIGITAL_INPUT_PULSE_SECONDS,
) -> str:
    """Trigger the digital input identified by ``digital_input_id``.

    Args:
        digital_input_id: Zero-based digital input index (0–3).
        timeout: Override for this call only. If ``None``, the timeout
            bound in `__init__` is used.
        hold_seconds: Seconds to hold the input HIGH between press and
            release. Defaults to the web UI's ~600ms.

    See `async_trigger_digital_input` (the free function) for the full
    description of behavior and raised exceptions.
    """
    return await async_trigger_digital_input(
        client_session=self.client_session,
        config=self.config,
        digital_input_id=digital_input_id,
        timeout=self.timeout if timeout is None else timeout,
        hold_seconds=hold_seconds,
    )

async_get_raw_data async

async_get_raw_data(client_session: ClientSession, config: ConfigObject, url: URL, timeout: float = 10.0) -> str

Send an authenticated GET request and return the response body as text.

This is the low-level primitive used by the higher-level state, DMX, and dosage helpers. Most callers will want one of those instead — reach for this directly only when you need to hit a custom URL on the controller.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession owned and closed by the caller. Reuse one session across many calls for connection pooling. TYPE: ClientSession
config	Controller configuration. The username and password are sent as HTTP Basic auth credentials. TYPE: ConfigObject
url	Fully-qualified URL to GET. Build it from config.base_url plus an API_PATH_* constant if you want to stay close to the standard endpoints. TYPE: URL
timeout	Maximum seconds to wait for the entire exchange (request and response body). Defaults to 10 seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body as a string. The controller typically returns
str	CSV; the caller is responsible for parsing.

RAISES	DESCRIPTION
BadCredentialsException	If the controller responds with HTTP 401 or 403.
BadStatusCodeException	If any other 4xx or 5xx status is returned.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For DNS failures, connection resets, and other network-level errors.

Source code in src/proconip/api.py

async def async_get_raw_data(
    client_session: ClientSession,
    config: ConfigObject,
    url: URL,
    timeout: float = 10.0,
) -> str:
    """Send an authenticated GET request and return the response body as text.

    This is the low-level primitive used by the higher-level state, DMX, and
    dosage helpers. Most callers will want one of those instead — reach for
    this directly only when you need to hit a custom URL on the controller.

    Args:
        client_session: An open `aiohttp.ClientSession` owned and closed by
            the caller. Reuse one session across many calls for connection
            pooling.
        config: Controller configuration. The username and password are sent
            as HTTP Basic auth credentials.
        url: Fully-qualified URL to GET. Build it from `config.base_url` plus
            an `API_PATH_*` constant if you want to stay close to the
            standard endpoints.
        timeout: Maximum seconds to wait for the entire exchange (request and
            response body). Defaults to 10 seconds.

    Returns:
        The raw response body as a string. The controller typically returns
        CSV; the caller is responsible for parsing.

    Raises:
        BadCredentialsException: If the controller responds with HTTP 401 or 403.
        BadStatusCodeException: If any other 4xx or 5xx status is returned.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For DNS failures, connection resets, and other
            network-level errors.
    """
    auth = BasicAuth(config.username, config.password)
    try:
        async with asyncio.timeout(timeout):
            async with client_session.get(url, auth=auth) as response:
                return await _handle_response(response)
    except TimeoutError as exc:
        raise TimeoutException("API request timed out") from exc
    except (ClientError, socket.gaierror) as exc:
        raise ProconipApiException(f"API request failed ({exc})") from exc

async_get_raw_state async

async_get_raw_state(client_session: ClientSession, config: ConfigObject, timeout: float = 10.0) -> str

Fetch the raw /GetState.csv body from the controller.

Use this when you want to handle the CSV yourself. To get a parsed object, call async_get_state instead.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration including base URL and credentials. TYPE: ConfigObject
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw multi-line CSV body returned by the controller.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors (DNS, connection reset).

Source code in src/proconip/api.py

async def async_get_raw_state(
    client_session: ClientSession,
    config: ConfigObject,
    timeout: float = 10.0,
) -> str:
    """Fetch the raw `/GetState.csv` body from the controller.

    Use this when you want to handle the CSV yourself. To get a parsed
    object, call `async_get_state` instead.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration including base URL and credentials.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw multi-line CSV body returned by the controller.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors (DNS, connection reset).
    """
    url = URL(config.base_url).with_path(API_PATH_GET_STATE)
    return await async_get_raw_data(client_session, config, url, timeout=timeout)

async_get_state async

async_get_state(client_session: ClientSession, config: ConfigObject, timeout: float = 10.0) -> GetStateData

Fetch and parse the controller's current state.

This is the most common entry point: it performs the GET request, parses the CSV response, and returns a GetStateData containing all sensor readings, relay states, dosage configuration, and so on.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration including base URL and credentials. TYPE: ConfigObject
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
GetStateData	A GetStateData instance with all properties populated.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.
InvalidPayloadException	If the response is empty or truncated.

Source code in src/proconip/api.py

async def async_get_state(
    client_session: ClientSession,
    config: ConfigObject,
    timeout: float = 10.0,
) -> GetStateData:
    """Fetch and parse the controller's current state.

    This is the most common entry point: it performs the GET request, parses
    the CSV response, and returns a `GetStateData` containing all sensor
    readings, relay states, dosage configuration, and so on.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration including base URL and credentials.
        timeout: Per-request timeout in seconds.

    Returns:
        A `GetStateData` instance with all properties populated.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
        InvalidPayloadException: If the response is empty or truncated.
    """
    raw_data = await async_get_raw_state(client_session, config, timeout=timeout)
    return GetStateData(raw_data)

async_post_usrcfg_cgi async

async_post_usrcfg_cgi(client_session: ClientSession, config: ConfigObject, payload: str, timeout: float = 10.0) -> str

Send a form-encoded POST to /usrcfg.cgi.

This is the low-level primitive behind relay switching and DMX writes. Most callers want async_switch_on, async_switch_off, async_set_auto_mode, or async_set_dmx instead. Use this directly only when you need to send a payload the higher-level helpers don't construct.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration including base URL and credentials. TYPE: ConfigObject
payload	The pre-encoded application/x-www-form-urlencoded body. TYPE: str
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body returned by the controller.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_post_usrcfg_cgi(
    client_session: ClientSession,
    config: ConfigObject,
    payload: str,
    timeout: float = 10.0,
) -> str:
    """Send a form-encoded POST to `/usrcfg.cgi`.

    This is the low-level primitive behind relay switching and DMX writes.
    Most callers want `async_switch_on`, `async_switch_off`,
    `async_set_auto_mode`, or `async_set_dmx` instead. Use this directly only
    when you need to send a payload the higher-level helpers don't construct.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration including base URL and credentials.
        payload: The pre-encoded `application/x-www-form-urlencoded` body.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw response body returned by the controller.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    url = URL(config.base_url).with_path(API_PATH_USRCFG)
    auth = BasicAuth(config.username, config.password)
    headers = {"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}
    try:
        async with asyncio.timeout(timeout):
            async with client_session.post(
                url=url,
                headers=headers,
                data=payload,
                auth=auth,
            ) as response:
                return await _handle_response(response)
    except TimeoutError as exc:
        raise TimeoutException("API request timed out") from exc
    except (ClientError, socket.gaierror) as exc:
        raise ProconipApiException(f"API request failed ({exc})") from exc

async_switch_on async

async_switch_on(client_session: ClientSession, config: ConfigObject, current_state: GetStateData, relay: Relay, timeout: float = 10.0) -> str

Switch a relay to manual ON.

Manual mode overrides the controller's schedule until the relay is explicitly switched off or set back to auto. Other relays in current_state are preserved by reading their bit field, which is why a fresh GetStateData snapshot is required.

Dosage relays (chlorine, pH+, pH-) cannot be switched on this way — the controller's safety logic interlocks them. Use async_start_dosage (or DosageControl) instead for time-limited manual dosing.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
current_state	A recent GetStateData snapshot. Used to compute the ENA bit field so that other relays keep their current state. TYPE: GetStateData
relay	The relay to switch on. TYPE: Relay
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body returned by /usrcfg.cgi.

RAISES	DESCRIPTION
BadRelayException	If relay is one of the configured dosage control relays.
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_switch_on(
    client_session: ClientSession,
    config: ConfigObject,
    current_state: GetStateData,
    relay: Relay,
    timeout: float = 10.0,
) -> str:
    """Switch a relay to manual ON.

    Manual mode overrides the controller's schedule until the relay is
    explicitly switched off or set back to auto. Other relays in
    ``current_state`` are preserved by reading their bit field, which is why
    a fresh `GetStateData` snapshot is required.

    Dosage relays (chlorine, pH+, pH-) cannot be switched on this way — the
    controller's safety logic interlocks them. Use `async_start_dosage` (or
    `DosageControl`) instead for time-limited manual dosing.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        current_state: A recent `GetStateData` snapshot. Used to compute the
            ENA bit field so that other relays keep their current state.
        relay: The relay to switch on.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw response body returned by `/usrcfg.cgi`.

    Raises:
        BadRelayException: If ``relay`` is one of the configured dosage
            control relays.
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    if current_state.is_dosage_relay(relay):
        raise BadRelayException("Cannot permanently switch on a dosage relay")
    bit_state = current_state.determine_overall_relay_bit_state()
    relay_bit_mask = relay.get_bit_mask()
    bit_state[0] |= relay_bit_mask
    bit_state[1] |= relay_bit_mask
    return await async_post_usrcfg_cgi(
        client_session=client_session,
        config=config,
        payload=f"ENA={bit_state[0]},{bit_state[1]}&MANUAL=1",
        timeout=timeout,
    )

async_switch_off async

async_switch_off(client_session: ClientSession, config: ConfigObject, current_state: GetStateData, relay: Relay, timeout: float = 10.0) -> str

Switch a relay to manual OFF.

Like async_switch_on, this puts the relay into manual mode but with the output disabled. Other relays in current_state keep their state. Unlike switching on, this is allowed for dosage relays — turning a dosage pump off is always safe.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
current_state	A recent GetStateData snapshot used to compute the ENA bit field. TYPE: GetStateData
relay	The relay to switch off. TYPE: Relay
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body returned by /usrcfg.cgi.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_switch_off(
    client_session: ClientSession,
    config: ConfigObject,
    current_state: GetStateData,
    relay: Relay,
    timeout: float = 10.0,
) -> str:
    """Switch a relay to manual OFF.

    Like `async_switch_on`, this puts the relay into manual mode but with the
    output disabled. Other relays in ``current_state`` keep their state.
    Unlike switching on, this is allowed for dosage relays — turning a dosage
    pump off is always safe.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        current_state: A recent `GetStateData` snapshot used to compute the
            ENA bit field.
        relay: The relay to switch off.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw response body returned by `/usrcfg.cgi`.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    bit_state = current_state.determine_overall_relay_bit_state()
    relay_bit_mask = relay.get_bit_mask()
    bit_state[0] |= relay_bit_mask
    bit_state[1] &= ~relay_bit_mask
    return await async_post_usrcfg_cgi(
        client_session=client_session,
        config=config,
        payload=f"ENA={bit_state[0]},{bit_state[1]}&MANUAL=1",
        timeout=timeout,
    )

async_set_auto_mode async

async_set_auto_mode(client_session: ClientSession, config: ConfigObject, current_state: GetStateData, relay: Relay, timeout: float = 10.0) -> str

Hand a relay back to the controller's automatic schedule.

This is the inverse of async_switch_on/async_switch_off: instead of forcing a manual state, the relay's behavior is governed by the controller's configured rules (timer, sensor thresholds, dosage logic).

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
current_state	A recent GetStateData snapshot used to compute the ENA bit field. TYPE: GetStateData
relay	The relay to put back into auto mode. TYPE: Relay
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body returned by /usrcfg.cgi.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_set_auto_mode(
    client_session: ClientSession,
    config: ConfigObject,
    current_state: GetStateData,
    relay: Relay,
    timeout: float = 10.0,
) -> str:
    """Hand a relay back to the controller's automatic schedule.

    This is the inverse of `async_switch_on`/`async_switch_off`: instead of
    forcing a manual state, the relay's behavior is governed by the
    controller's configured rules (timer, sensor thresholds, dosage logic).

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        current_state: A recent `GetStateData` snapshot used to compute the
            ENA bit field.
        relay: The relay to put back into auto mode.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw response body returned by `/usrcfg.cgi`.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    bit_state = current_state.determine_overall_relay_bit_state()
    relay_bit_mask = relay.get_bit_mask()
    bit_state[0] &= ~relay_bit_mask
    bit_state[1] &= ~relay_bit_mask
    return await async_post_usrcfg_cgi(
        client_session=client_session,
        config=config,
        payload=f"ENA={bit_state[0]},{bit_state[1]}&MANUAL=1",
        timeout=timeout,
    )

async_start_dosage async

async_start_dosage(client_session: ClientSession, config: ConfigObject, dosage_target: DosageTarget, dosage_duration: int, timeout: float = 10.0) -> str

Trigger a manual, time-limited dosage on the controller.

Manual dosage is the safe alternative to switching a dosage relay on directly: the controller still applies the same interlocks it uses for the web UI (canister level, dosage enabled in config, redox/pH thresholds, …). If those checks fail, the controller silently no-ops while still returning HTTP 200 — there is no "dosage refused" error in the protocol.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
dosage_target	Which dosing pump to engage (chlorine, pH-, or pH+). TYPE: DosageTarget
dosage_duration	Run time in seconds. Allowed range depends on the controller's own dosage configuration; values that exceed the configured maximum are typically clamped silently by the device. TYPE: int
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body returned by /Command.htm.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_start_dosage(
    client_session: ClientSession,
    config: ConfigObject,
    dosage_target: DosageTarget,
    dosage_duration: int,
    timeout: float = 10.0,
) -> str:
    """Trigger a manual, time-limited dosage on the controller.

    Manual dosage is the safe alternative to switching a dosage relay on
    directly: the controller still applies the same interlocks it uses for the
    web UI (canister level, dosage enabled in config, redox/pH thresholds, …).
    If those checks fail, the controller silently no-ops while still returning
    HTTP 200 — there is no "dosage refused" error in the protocol.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        dosage_target: Which dosing pump to engage (chlorine, pH-, or pH+).
        dosage_duration: Run time in **seconds**. Allowed range depends on the
            controller's own dosage configuration; values that exceed the
            configured maximum are typically clamped silently by the device.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw response body returned by `/Command.htm`.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    query = f"MAN_DOSAGE={dosage_target},{dosage_duration}"
    url = URL(config.base_url).with_path(API_PATH_COMMAND).with_query(query)
    return await async_get_raw_data(client_session, config, url, timeout=timeout)

async_get_raw_dmx async

async_get_raw_dmx(client_session: ClientSession, config: ConfigObject, timeout: float = 10.0) -> str

Fetch the raw /GetDmx.csv body — the current 16 DMX channel values.

Use this when you want to parse the CSV yourself. To get a structured GetDmxData you can read and mutate, call async_get_dmx instead.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	A single CSV line containing the 16 channel values.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_get_raw_dmx(
    client_session: ClientSession,
    config: ConfigObject,
    timeout: float = 10.0,
) -> str:
    """Fetch the raw `/GetDmx.csv` body — the current 16 DMX channel values.

    Use this when you want to parse the CSV yourself. To get a structured
    `GetDmxData` you can read and mutate, call `async_get_dmx` instead.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        timeout: Per-request timeout in seconds.

    Returns:
        A single CSV line containing the 16 channel values.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    url = URL(config.base_url).with_path(API_PATH_GET_DMX)
    return await async_get_raw_data(client_session, config, url, timeout=timeout)

async_get_dmx async

async_get_dmx(client_session: ClientSession, config: ConfigObject, timeout: float = 10.0) -> GetDmxData

Fetch and parse the controller's DMX channel state.

Returns a mutable GetDmxData you can iterate over, read with [index], or modify with set(index, value). Pass it to async_set_dmx to push the changes back to the controller.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
GetDmxData	A GetDmxData containing all 16 DMX channels.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.
InvalidPayloadException	If the response is empty.

Source code in src/proconip/api.py

async def async_get_dmx(
    client_session: ClientSession,
    config: ConfigObject,
    timeout: float = 10.0,
) -> GetDmxData:
    """Fetch and parse the controller's DMX channel state.

    Returns a mutable `GetDmxData` you can iterate over, read with `[index]`,
    or modify with `set(index, value)`. Pass it to `async_set_dmx` to push
    the changes back to the controller.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        timeout: Per-request timeout in seconds.

    Returns:
        A `GetDmxData` containing all 16 DMX channels.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
        InvalidPayloadException: If the response is empty.
    """
    raw_data = await async_get_raw_dmx(
        client_session=client_session, config=config, timeout=timeout
    )
    return GetDmxData(raw_data)

async_set_dmx async

async_set_dmx(client_session: ClientSession, config: ConfigObject, dmx_states: GetDmxData, timeout: float = 10.0) -> str

Push DMX channel values back to the controller.

The controller's API only accepts the full 16-channel state in a single write. The usual pattern is: fetch with async_get_dmx, mutate channels with dmx_states.set(index, value), then call this function to commit.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration. TYPE: ConfigObject
dmx_states	The full DMX state to write. All 16 channels are sent. TYPE: GetDmxData
timeout	Per-request timeout in seconds. TYPE: float DEFAULT: 10.0

RETURNS	DESCRIPTION
str	The raw response body returned by /usrcfg.cgi.

RAISES	DESCRIPTION
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If the exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_set_dmx(
    client_session: ClientSession,
    config: ConfigObject,
    dmx_states: GetDmxData,
    timeout: float = 10.0,
) -> str:
    """Push DMX channel values back to the controller.

    The controller's API only accepts the full 16-channel state in a single
    write. The usual pattern is: fetch with `async_get_dmx`, mutate channels
    with `dmx_states.set(index, value)`, then call this function to commit.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration.
        dmx_states: The full DMX state to write. All 16 channels are sent.
        timeout: Per-request timeout in seconds.

    Returns:
        The raw response body returned by `/usrcfg.cgi`.

    Raises:
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If the exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    payload = "&".join(f"{k}={v}" for k, v in dmx_states.post_data.items())
    return await async_post_usrcfg_cgi(
        client_session=client_session,
        config=config,
        payload=payload,
        timeout=timeout,
    )

async_trigger_digital_input async

async_trigger_digital_input(client_session: ClientSession, config: ConfigObject, digital_input_id: int, timeout: float = 10.0, hold_seconds: float = DIGITAL_INPUT_PULSE_SECONDS) -> str

Trigger (momentarily pulse) a digital input via the WEBIO IO field.

The controller's web UI exposes each digital input as a push button: it sets the input's bit, waits, then clears it. This sends the same two /usrcfg.cgi writes — IO=<mask>&WEBIO=1 (press) then, after hold_seconds, IO=0&WEBIO=1 (release) — where mask = 1 << digital_input_id. The hold mirrors the web UI (~600ms) so the controller reliably registers the pulse.

Unlike relay switching, no GetStateData snapshot is needed: the IO field is written directly rather than read-modify-written.

Because the pulse is two separate writes, the release is not fully guaranteed. If the hold is cancelled (e.g. the caller's task is shut down), a best-effort release is attempted before the cancellation propagates. If the release POST itself fails (timeout / network blip), that error propagates and the input is left asserted HIGH until the next write.

PARAMETER	DESCRIPTION
client_session	An open aiohttp.ClientSession. TYPE: ClientSession
config	Controller configuration including base URL and credentials. TYPE: ConfigObject
digital_input_id	Zero-based digital input index (0–3). TYPE: int
timeout	Per-request timeout in seconds, applied to each of the two POSTs. TYPE: float DEFAULT: 10.0
hold_seconds	Seconds to hold the input HIGH between the press and release POSTs. Defaults to the web UI's ~600ms. TYPE: float DEFAULT: DIGITAL_INPUT_PULSE_SECONDS

RETURNS	DESCRIPTION
str	The raw response body returned by the release POST.

RAISES	DESCRIPTION
ValueError	If digital_input_id is not in 0–3.
BadCredentialsException	On HTTP 401 or 403.
BadStatusCodeException	On any other 4xx or 5xx response.
TimeoutException	If an exchange exceeds timeout seconds.
ProconipApiException	For network-level errors.

Source code in src/proconip/api.py

async def async_trigger_digital_input(
    client_session: ClientSession,
    config: ConfigObject,
    digital_input_id: int,
    timeout: float = 10.0,
    hold_seconds: float = DIGITAL_INPUT_PULSE_SECONDS,
) -> str:
    """Trigger (momentarily pulse) a digital input via the WEBIO ``IO`` field.

    The controller's web UI exposes each digital input as a push button: it
    sets the input's bit, waits, then clears it. This sends the same two
    `/usrcfg.cgi` writes — ``IO=<mask>&WEBIO=1`` (press) then, after
    ``hold_seconds``, ``IO=0&WEBIO=1`` (release) — where
    ``mask = 1 << digital_input_id``. The hold mirrors the web UI (~600ms) so
    the controller reliably registers the pulse.

    Unlike relay switching, no `GetStateData` snapshot is needed: the ``IO``
    field is written directly rather than read-modify-written.

    Because the pulse is two separate writes, the release is not fully
    guaranteed. If the hold is **cancelled** (e.g. the caller's task is shut
    down), a best-effort release is attempted before the cancellation
    propagates. If the **release POST itself fails** (timeout / network blip),
    that error propagates and the input is left asserted HIGH until the next
    write.

    Args:
        client_session: An open `aiohttp.ClientSession`.
        config: Controller configuration including base URL and credentials.
        digital_input_id: Zero-based digital input index (0–3).
        timeout: Per-request timeout in seconds, applied to each of the two
            POSTs.
        hold_seconds: Seconds to hold the input HIGH between the press and
            release POSTs. Defaults to the web UI's ~600ms.

    Returns:
        The raw response body returned by the release POST.

    Raises:
        ValueError: If ``digital_input_id`` is not in 0–3.
        BadCredentialsException: On HTTP 401 or 403.
        BadStatusCodeException: On any other 4xx or 5xx response.
        TimeoutException: If an exchange exceeds ``timeout`` seconds.
        ProconipApiException: For network-level errors.
    """
    if not 0 <= digital_input_id < DIGITAL_INPUT_COUNT:
        raise ValueError(
            f"digital_input_id must be in 0..{DIGITAL_INPUT_COUNT - 1}, got {digital_input_id}"
        )
    mask = 1 << digital_input_id
    await async_post_usrcfg_cgi(
        client_session=client_session,
        config=config,
        payload=f"IO={mask}&WEBIO=1",
        timeout=timeout,
    )
    try:
        await asyncio.sleep(hold_seconds)
    except asyncio.CancelledError:
        # Best-effort release so a cancelled hold doesn't leave the input
        # asserted HIGH. Networking inside a cancellation is awkward, so this
        # is a best effort, not a guarantee.
        with contextlib.suppress(Exception):
            await async_post_usrcfg_cgi(
                client_session=client_session,
                config=config,
                payload="IO=0&WEBIO=1",
                timeout=timeout,
            )
        raise
    return await async_post_usrcfg_cgi(
        client_session=client_session,
        config=config,
        payload="IO=0&WEBIO=1",
        timeout=timeout,
    )

Data structures (proconip.definitions)

definitions

Data structures for the ProCon.IP CSV and form-encoded HTTP APIs.

This module contains everything that's needed to describe the controller's data shape without making any network calls — typed wrappers around the CSV responses, helpers for the bit-field flags they encode, and a few small exceptions that get raised when something is malformed.

Path constants (API_PATH_*) and category strings (CATEGORY_*) are exposed because the rest of the library and downstream callers (notably a Home Assistant integration) build on them.

The classes you will most often work with are:

• ConfigObject — base URL plus credentials.
• GetStateData — parsed /GetState.csv response. Exposes individual sensors, relays, dosage flags, and a few derived helpers.
• Relay — convenience wrapper around a relay DataObject with on/off and manual/auto interrogation methods.
• GetDmxData / DmxChannelData — parsed and mutable representation of the 16 DMX channels.

DosageTarget

Bases: IntEnum

Identifies which dosing pump a manual dosage command should engage.

The numeric values match the controller's MAN_DOSAGE query parameter, so this enum can be used directly in URL building.

Source code in src/proconip/definitions.py

class DosageTarget(IntEnum):
    """Identifies which dosing pump a manual dosage command should engage.

    The numeric values match the controller's `MAN_DOSAGE` query parameter,
    so this enum can be used directly in URL building.
    """

    CHLORINE = 0
    PH_MINUS = 1
    PH_PLUS = 2

ConfigObject

Base URL and credentials for talking to a single ProCon.IP controller.

Instances are plain holders — no network connection is opened until they are passed into one of the API helpers in proconip.api.

Source code in src/proconip/definitions.py

class ConfigObject:
    """Base URL and credentials for talking to a single ProCon.IP controller.

    Instances are plain holders — no network connection is opened until they
    are passed into one of the API helpers in `proconip.api`.
    """

    def __init__(
        self,
        base_url: str,
        username: str,
        password: str,
    ):
        """Build a config from explicit values.

        Args:
            base_url: Root URL of the controller, e.g. ``http://192.168.1.50``.
                The library appends the API paths itself; do not include them
                here. Plain HTTP is normal — these controllers are LAN-only.
            username: HTTP Basic auth username (controller default: ``admin``).
            password: HTTP Basic auth password (controller default: ``admin``).
        """
        self.base_url = base_url
        self.username = username
        self.password = password

    @staticmethod
    def from_dict(data: dict[str, str]) -> "ConfigObject":
        """Build a `ConfigObject` from a serialized dictionary.

        Useful for restoring a config that was previously written out via
        `to_dict` (e.g. into a Home Assistant config entry).

        Args:
            data: A dict containing the keys ``base_url``, ``username``, and
                ``password``. All three are required.

        Returns:
            A new `ConfigObject` populated from the dict.

        Raises:
            ValueError: If any of the required keys is missing. The exception
                message names the missing key.
        """
        if "base_url" not in data:
            raise ValueError("base_url is required")
        if "username" not in data:
            raise ValueError("username is required")
        if "password" not in data:
            raise ValueError("password is required")
        return ConfigObject(data["base_url"], data["username"], data["password"])

    def to_dict(self) -> dict[str, str]:
        """Return a plain dict copy of the config, suitable for serialization.

        The password is stored in the clear — encrypt the dict yourself if it
        will be persisted somewhere readable.
        """
        return {
            "base_url": self.base_url,
            "username": self.username,
            "password": self.password,
        }

from_dict staticmethod

from_dict(data: dict[str, str]) -> ConfigObject

Build a ConfigObject from a serialized dictionary.

Useful for restoring a config that was previously written out via to_dict (e.g. into a Home Assistant config entry).

PARAMETER	DESCRIPTION
data	A dict containing the keys base_url, username, and password. All three are required. TYPE: dict[str, str]

RETURNS	DESCRIPTION
ConfigObject	A new ConfigObject populated from the dict.

RAISES	DESCRIPTION
ValueError	If any of the required keys is missing. The exception message names the missing key.

Source code in src/proconip/definitions.py

@staticmethod
def from_dict(data: dict[str, str]) -> "ConfigObject":
    """Build a `ConfigObject` from a serialized dictionary.

    Useful for restoring a config that was previously written out via
    `to_dict` (e.g. into a Home Assistant config entry).

    Args:
        data: A dict containing the keys ``base_url``, ``username``, and
            ``password``. All three are required.

    Returns:
        A new `ConfigObject` populated from the dict.

    Raises:
        ValueError: If any of the required keys is missing. The exception
            message names the missing key.
    """
    if "base_url" not in data:
        raise ValueError("base_url is required")
    if "username" not in data:
        raise ValueError("username is required")
    if "password" not in data:
        raise ValueError("password is required")
    return ConfigObject(data["base_url"], data["username"], data["password"])

to_dict

to_dict() -> dict[str, str]

Return a plain dict copy of the config, suitable for serialization.

The password is stored in the clear — encrypt the dict yourself if it will be persisted somewhere readable.

Source code in src/proconip/definitions.py

def to_dict(self) -> dict[str, str]:
    """Return a plain dict copy of the config, suitable for serialization.

    The password is stored in the clear — encrypt the dict yourself if it
    will be persisted somewhere readable.
    """
    return {
        "base_url": self.base_url,
        "username": self.username,
        "password": self.password,
    }

DataObject

A single sensor, relay, canister, or consumption channel from /GetState.csv.

Each DataObject represents one column of the CSV response, combining the name, unit, offset, gain, and raw value rows that the controller sends. The column index alone determines which category the object falls into (analog, relay, temperature, …) — see the constructor for the exact ranges.

The actual physical reading is computed once at construction via offset + gain * raw_value and exposed as value. A pre-formatted display_value string is also produced; for relay columns it is one of "Auto (off)", "Auto (on)", "Off", or "On".

Source code in src/proconip/definitions.py

class DataObject:
    """A single sensor, relay, canister, or consumption channel from `/GetState.csv`.

    Each `DataObject` represents one column of the CSV response, combining the
    name, unit, offset, gain, and raw value rows that the controller sends. The
    column index alone determines which category the object falls into (analog,
    relay, temperature, …) — see the constructor for the exact ranges.

    The actual physical reading is computed once at construction via
    ``offset + gain * raw_value`` and exposed as `value`. A pre-formatted
    `display_value` string is also produced; for relay columns it is one of
    "Auto (off)", "Auto (on)", "Off", or "On".
    """

    _column: int
    _category: str
    _category_id: int
    _name: str
    _unit: str
    _offset: float
    _gain: float
    _raw_value: float
    _value: float
    _display_value: str

    def __init__(
        self,
        column: int,
        name: str,
        unit: str,
        offset: float,
        gain: float,
        value: float,
    ):
        """Build a `DataObject` from one column's worth of CSV data.

        Args:
            column: Zero-based column index in the CSV. Determines the category:
                ``0`` → time, ``1–5`` → analog, ``6–7`` → electrode, ``8–15`` →
                temperature, ``16–23`` → relay, ``24–27`` → digital input,
                ``28–35`` → external relay, ``36–38`` → canister, ``39–41`` →
                consumption. Anything outside this range falls into a sentinel
                "uncategorized" bucket.
            name: Sensor name as reported by the controller (e.g. ``"Redox"``).
            unit: Unit string (``"mV"``, ``"°C"``, ``"%"``, ``"--"``, …).
            offset: Calibration offset applied to ``value``.
            gain: Calibration gain applied to ``value``.
            value: Raw sensor value before calibration. Stored verbatim as
                `raw_value`; the physical `value` is computed as
                ``offset + gain * raw_value``.
        """
        self._column = column
        self._name = name
        self._unit = unit
        self._offset = offset
        self._gain = gain
        self._raw_value = value
        self._value = self._offset + (self._gain * self._raw_value)

        if column == 0:
            self._category = CATEGORY_TIME
            self._category_id = 0
            self._display_value = f"{int(self._value / 256):02d}:{int(self._value) % 256:02d}"
        elif 1 <= column <= 5:
            self._category = CATEGORY_ANALOG
            self._category_id = column - 1
            self._display_value = f"{self._value:.2f} {self._unit}"
        elif 6 <= column <= 7:
            self._category = CATEGORY_ELECTRODE
            self._category_id = column - 6
            self._display_value = f"{self._value:.2f} {self._unit}"
        elif 8 <= column <= 15:
            self._category = CATEGORY_TEMPERATURE
            self._category_id = column - 8
            self._display_value = f"{self._value:.2f} °{self._unit}"
        elif 16 <= column <= 23:
            self._category = CATEGORY_RELAY
            self._category_id = column - 16
            self._display_value = self._relay_state()
        elif 24 <= column <= 27:
            self._category = CATEGORY_DIGITAL_INPUT
            self._category_id = column - 24
            self._display_value = f"{self._value}"
        elif 28 <= column <= 35:
            self._category = CATEGORY_EXTERNAL_RELAY
            self._category_id = column - 28
            self._display_value = self._relay_state()
        elif 36 <= column <= 38:
            self._category = CATEGORY_CANISTER
            self._category_id = column - 36
            self._display_value = f"{self._value:.2f} {self._unit}"
        elif 39 <= column <= 41:
            self._category = CATEGORY_CONSUMPTION
            self._category_id = column - 39
            self._display_value = f"{self._value:.2f} {self._unit}"
        else:
            self._category = ""
            self._category_id = -1
            self._display_value = f"{self._value}"

    def __str__(self) -> str:
        """Return a short ``"name (unit): value"`` representation."""
        return f"{self._name} ({self._unit}): {self._value}"

    def _relay_state(self) -> str:
        """Render the current relay value as one of the four state strings.

        Raises:
            ValueError: If `self._value` is not one of the four valid relay
                states (0, 1, 2, 3). Indicates a malformed CSV payload.
        """
        if self._value == 0:
            return "Auto (off)"
        if self._value == 1:
            return "Auto (on)"
        if self._value == 2:
            return "Off"
        if self._value == 3:
            return "On"
        raise ValueError(f"Unexpected relay value {self._value}")

    @property
    def name(self) -> str:
        """Sensor name as reported by the controller."""
        return self._name

    @property
    def unit(self) -> str:
        """Unit string (e.g. ``"mV"``, ``"°C"``, ``"%"``)."""
        return self._unit

    @property
    def offset(self) -> float:
        """Calibration offset applied when computing `value` from `raw_value`."""
        return self._offset

    @property
    def gain(self) -> float:
        """Calibration gain applied when computing `value` from `raw_value`."""
        return self._gain

    @property
    def raw_value(self) -> float:
        """Raw value as received from the controller, before calibration."""
        return self._raw_value

    @property
    def value(self) -> float:
        """Physical value: ``offset + gain * raw_value``."""
        return self._value

    @property
    def display_value(self) -> str:
        """Pre-formatted human-readable string for display.

        For sensors this is ``value`` rendered with its unit and two decimal
        places. For relay columns it is one of "Auto (off)", "Auto (on)",
        "Off", or "On". For column 0 (the system time field) it is "HH:MM".
        """
        return self._display_value

    @property
    def column(self) -> int:
        """Zero-based column index this object came from in the raw CSV."""
        return self._column

    @property
    def category(self) -> str:
        """One of the `CATEGORY_*` constants identifying the entity type."""
        return self._category

    @property
    def category_id(self) -> int:
        """Zero-based index of this object within its category.

        For example, the third temperature sensor has ``category_id == 2``.
        Use `Relay.relay_id` instead if you need the aggregated relay ID
        across both the internal and external relay banks.
        """
        return self._category_id

name property

name: str

Sensor name as reported by the controller.

unit property

unit: str

Unit string (e.g. "mV", "°C", "%").

offset property

offset: float

Calibration offset applied when computing value from raw_value.

gain property

gain: float

Calibration gain applied when computing value from raw_value.

raw_value property

raw_value: float

Raw value as received from the controller, before calibration.

value property

value: float

Physical value: offset + gain * raw_value.

display_value property

display_value: str

Pre-formatted human-readable string for display.

For sensors this is value rendered with its unit and two decimal places. For relay columns it is one of "Auto (off)", "Auto (on)", "Off", or "On". For column 0 (the system time field) it is "HH:MM".

column property

column: int

Zero-based column index this object came from in the raw CSV.

category property

category: str

One of the CATEGORY_* constants identifying the entity type.

category_id property

category_id: int

Zero-based index of this object within its category.

For example, the third temperature sensor has category_id == 2. Use Relay.relay_id instead if you need the aggregated relay ID across both the internal and external relay banks.

Relay

Bases: DataObject

A DataObject with extra methods for interrogating a relay's state.

Relay state is encoded in two bits of the underlying value:

• bit 0 — output level (0 = off, 1 = on)
• bit 1 — control mode (0 = auto, 1 = manual)

The four valid combinations correspond to the four display_value strings from DataObject._relay_state.

Construct one by passing the DataObject you got from GetStateData.relay_objects (or external_relay_objects); the relay's column, name, calibration, and raw value are copied across so the physical value is computed exactly once. GetStateData.get_relay() is a shorthand that does this for you.

Source code in src/proconip/definitions.py

class Relay(DataObject):
    """A `DataObject` with extra methods for interrogating a relay's state.

    Relay state is encoded in two bits of the underlying `value`:

    - bit 0 — output level (0 = off, 1 = on)
    - bit 1 — control mode (0 = auto, 1 = manual)

    The four valid combinations correspond to the four `display_value`
    strings from `DataObject._relay_state`.

    Construct one by passing the `DataObject` you got from
    `GetStateData.relay_objects` (or `external_relay_objects`); the relay's
    column, name, calibration, and raw value are copied across so the
    physical value is computed exactly once. ``GetStateData.get_relay()`` is
    a shorthand that does this for you.
    """

    def __init__(self, data_object: DataObject):
        """Wrap an existing relay `DataObject`.

        Args:
            data_object: A `DataObject` produced by parsing a `/GetState.csv`
                response. It should be in the relay or external_relay
                category, but no check is enforced — passing a non-relay
                object yields a `Relay` whose interrogation methods will
                still run but produce meaningless results.
        """
        super().__init__(
            data_object.column,
            data_object.name,
            data_object.unit,
            data_object.offset,
            data_object.gain,
            data_object.raw_value,  # pass raw value so offset+gain are applied exactly once
        )

    def __str__(self) -> str:
        """Return ``"name: state"`` (e.g. ``"Pumpe: Auto (off)"``)."""
        return f"{self._name}: {self._display_value}"

    @property
    def relay_id(self) -> int:
        """Aggregated relay ID across internal and external banks.

        Internal relays return ``category_id`` directly (0–7); external
        relays return ``category_id + EXTERNAL_RELAY_ID_OFFSET`` (8–15). This
        ID matches the index used by `GetStateData.aggregated_relay_objects`
        and `RelaySwitch.async_switch_*`.
        """
        offset = EXTERNAL_RELAY_ID_OFFSET if self.category == CATEGORY_EXTERNAL_RELAY else 0
        return self.category_id + offset

    def is_on(self) -> bool:
        """True if bit 0 of the relay value is set (output enabled)."""
        return int(self._value) & 1 == 1

    def is_off(self) -> bool:
        """True if bit 0 of the relay value is clear (output disabled)."""
        return not self.is_on()

    def is_manual_mode(self) -> bool:
        """True if bit 1 of the relay value is set (overridden by manual ENA)."""
        return int(self._value) & 2 == 2

    def is_auto_mode(self) -> bool:
        """True if bit 1 of the relay value is clear (controller-driven)."""
        return not self.is_manual_mode()

    def get_bit_mask(self) -> int:
        """Bit mask for this relay in the 16-bit ENA / state field.

        Internal relays occupy bits 0–7, external relays 8–15. The mask
        returned here is suitable for OR-ing into the
        `determine_overall_relay_bit_state` output before sending an ENA
        update.
        """
        if self._category == CATEGORY_EXTERNAL_RELAY:
            return 1 << (self._category_id + EXTERNAL_RELAY_ID_OFFSET)
        return 1 << self._category_id

relay_id property

relay_id: int

Aggregated relay ID across internal and external banks.

Internal relays return category_id directly (0–7); external relays return category_id + EXTERNAL_RELAY_ID_OFFSET (8–15). This ID matches the index used by GetStateData.aggregated_relay_objects and RelaySwitch.async_switch_*.

is_on

is_on() -> bool

True if bit 0 of the relay value is set (output enabled).

Source code in src/proconip/definitions.py

def is_on(self) -> bool:
    """True if bit 0 of the relay value is set (output enabled)."""
    return int(self._value) & 1 == 1

is_off

is_off() -> bool

True if bit 0 of the relay value is clear (output disabled).

Source code in src/proconip/definitions.py

def is_off(self) -> bool:
    """True if bit 0 of the relay value is clear (output disabled)."""
    return not self.is_on()

is_manual_mode

is_manual_mode() -> bool

True if bit 1 of the relay value is set (overridden by manual ENA).

Source code in src/proconip/definitions.py

def is_manual_mode(self) -> bool:
    """True if bit 1 of the relay value is set (overridden by manual ENA)."""
    return int(self._value) & 2 == 2

is_auto_mode

is_auto_mode() -> bool

True if bit 1 of the relay value is clear (controller-driven).

Source code in src/proconip/definitions.py

def is_auto_mode(self) -> bool:
    """True if bit 1 of the relay value is clear (controller-driven)."""
    return not self.is_manual_mode()

get_bit_mask

get_bit_mask() -> int

Bit mask for this relay in the 16-bit ENA / state field.

Internal relays occupy bits 0–7, external relays 8–15. The mask returned here is suitable for OR-ing into the determine_overall_relay_bit_state output before sending an ENA update.

Source code in src/proconip/definitions.py

def get_bit_mask(self) -> int:
    """Bit mask for this relay in the 16-bit ENA / state field.

    Internal relays occupy bits 0–7, external relays 8–15. The mask
    returned here is suitable for OR-ing into the
    `determine_overall_relay_bit_state` output before sending an ENA
    update.
    """
    if self._category == CATEGORY_EXTERNAL_RELAY:
        return 1 << (self._category_id + EXTERNAL_RELAY_ID_OFFSET)
    return 1 << self._category_id

DigitalInput

Bases: DataObject

Typed view of a single digital input (CSV columns 24–27).

Construct one by passing the DataObject you got from GetStateData.digital_input_objects; the column, name, calibration, and raw value are copied across so the physical value is computed exactly once. GetStateData.digital_inputs() is a shorthand that does this for you.

The wrapper exists so digital inputs can be triggered (not just read): get_bit_mask yields the bit used in the controller's WEBIO IO field.

Source code in src/proconip/definitions.py

class DigitalInput(DataObject):
    """Typed view of a single digital input (CSV columns 24–27).

    Construct one by passing the `DataObject` you got from
    `GetStateData.digital_input_objects`; the column, name, calibration, and
    raw value are copied across so the physical value is computed exactly once.
    ``GetStateData.digital_inputs()`` is a shorthand that does this for you.

    The wrapper exists so digital inputs can be *triggered* (not just read):
    `get_bit_mask` yields the bit used in the controller's WEBIO ``IO`` field.
    """

    def __init__(self, data_object: DataObject):
        """Wrap an existing digital-input `DataObject`.

        Args:
            data_object: A `DataObject` produced by parsing a `/GetState.csv`
                response. It should be in the digital_input category, but no
                check is enforced — passing another object yields a
                `DigitalInput` whose `get_bit_mask` is meaningless.
        """
        super().__init__(
            data_object.column,
            data_object.name,
            data_object.unit,
            data_object.offset,
            data_object.gain,
            data_object.raw_value,  # pass raw value so offset+gain are applied exactly once
        )

    def get_bit_mask(self) -> int:
        """Bit for this input in the WEBIO ``IO`` field (``1 << category_id``).

        The four digital inputs occupy bits 0–3, so this returns 1, 2, 4, or 8.
        To *trigger* an input, pass its ``category_id`` to
        `async_trigger_digital_input` (which derives the same mask itself); use
        this method when assembling an ``IO`` payload for a manual
        `async_post_usrcfg_cgi` write.
        """
        return 1 << self._category_id

get_bit_mask

get_bit_mask() -> int

Bit for this input in the WEBIO IO field (1 << category_id).

The four digital inputs occupy bits 0–3, so this returns 1, 2, 4, or 8. To trigger an input, pass its category_id to async_trigger_digital_input (which derives the same mask itself); use this method when assembling an IO payload for a manual async_post_usrcfg_cgi write.

Source code in src/proconip/definitions.py

def get_bit_mask(self) -> int:
    """Bit for this input in the WEBIO ``IO`` field (``1 << category_id``).

    The four digital inputs occupy bits 0–3, so this returns 1, 2, 4, or 8.
    To *trigger* an input, pass its ``category_id`` to
    `async_trigger_digital_input` (which derives the same mask itself); use
    this method when assembling an ``IO`` payload for a manual
    `async_post_usrcfg_cgi` write.
    """
    return 1 << self._category_id

GetStateData

Parsed representation of a single /GetState.csv response.

The CSV the controller returns has six lines: SYSINFO, names, units, offsets, gains, and raw values. The constructor parses all six and builds a list of DataObject instances, then groups them by category for easy lookup.

Once constructed, an instance is read-only — it represents a snapshot. Re-fetch and reconstruct the object whenever you need fresh data.

All public properties on this class are populated eagerly at construction time, so accessing them is cheap.

Source code in src/proconip/definitions.py

class GetStateData:
    """Parsed representation of a single `/GetState.csv` response.

    The CSV the controller returns has six lines: SYSINFO, names, units,
    offsets, gains, and raw values. The constructor parses all six and
    builds a list of `DataObject` instances, then groups them by category
    for easy lookup.

    Once constructed, an instance is read-only — it represents a snapshot.
    Re-fetch and reconstruct the object whenever you need fresh data.

    All public properties on this class are populated eagerly at construction
    time, so accessing them is cheap.
    """

    _time: str
    _version: str
    _cpu_time: int
    _reset_root_cause: int
    _ntp_fault_state: int
    _config_other_enable: int
    _dosage_control: int
    _ph_plus_dosage_relay_id: int
    _ph_minus_dosage_relay_id: int
    _chlorine_dosage_relay_id: int
    _data_objects: list[DataObject]
    _analog_objects: list[DataObject]
    _electrode_objects: list[DataObject]
    _temperature_objects: list[DataObject]
    _relay_objects: list[DataObject]
    _digital_input_objects: list[DataObject]
    _external_relay_objects: list[DataObject]
    _canister_objects: list[DataObject]
    _consumption_objects: list[DataObject]

    def __init__(self, raw_data: str):
        """Parse a `/GetState.csv` body into structured data.

        Args:
            raw_data: The raw multi-line CSV string returned by the
                controller. Leading blank lines are tolerated.

        Raises:
            InvalidPayloadException: If the payload is empty, has fewer than
                six non-blank lines, or has rows whose column counts do not
                line up (names / units / offsets / gains / raw values must
                all have the same number of comma-separated entries).
            ValueError: If any of the numeric rows contains a value that is
                not parseable as a float.
        """
        self._raw_data = raw_data

        line = 0
        lines = raw_data.splitlines()
        while line < len(lines) and len(lines[line].strip()) < 1:
            line += 1
        if len(lines) < line + 6:
            raise InvalidPayloadException(
                f"GetState.csv payload is incomplete: expected at least 6 non-blank lines, "
                f"got {len(lines) - line}"
            )
        self._system_info = lines[line].split(",")
        self._data_names = lines[line + 1].split(",")
        self._data_units = lines[line + 2].split(",")
        self._data_offsets = [float(v) for v in lines[line + 3].split(",")]
        self._data_gain = [float(v) for v in lines[line + 4].split(",")]
        self._data_raw_values = [float(v) for v in lines[line + 5].split(",")]

        column_count = len(self._data_names)
        row_lengths = {
            "names": column_count,
            "units": len(self._data_units),
            "offsets": len(self._data_offsets),
            "gains": len(self._data_gain),
            "raw_values": len(self._data_raw_values),
        }
        if len(set(row_lengths.values())) != 1:
            raise InvalidPayloadException(
                "GetState.csv column counts don't line up: "
                + ", ".join(f"{k}={v}" for k, v in row_lengths.items())
            )

        self._parse_system_info()
        self._parse()
        self._time = self._data_objects[0].display_value

    def __str__(self) -> str:
        """Return the original raw CSV as it was received."""
        return self._raw_data

    def _parse_system_info(self) -> None:
        """Populate the system-level attributes from the SYSINFO line."""
        self._version = self._system_info[1]
        self._cpu_time = int(self._system_info[2])
        self._reset_root_cause = int(self._system_info[3])
        self._ntp_fault_state = int(self._system_info[4])
        self._config_other_enable = int(self._system_info[5])
        self._dosage_control = int(self._system_info[6])
        self._ph_plus_dosage_relay_id = int(self._system_info[7])
        self._ph_minus_dosage_relay_id = int(self._system_info[8])
        self._chlorine_dosage_relay_id = int(self._system_info[9])

    @property
    def time(self) -> str:
        """Controller's current local time as ``"HH:MM"``."""
        return self._time

    @property
    def version(self) -> str:
        """Firmware version string reported by the controller."""
        return self._version

    @property
    def cpu_time(self) -> int:
        """Controller CPU uptime in seconds since the last reset."""
        return self._cpu_time

    @property
    def reset_root_cause(self) -> int:
        """Numeric reset-root-cause code. Decode with `RESET_ROOT_CAUSE` or
        `get_reset_root_cause_as_str`."""
        return self._reset_root_cause

    @property
    def ntp_fault_state(self) -> int:
        """Numeric NTP fault state. Decode with `NTP_FAULT_STATE` or
        `get_ntp_fault_state_as_str`. Bits 0/1/2 indicate severity (logfile,
        warning, error); bit 16 set indicates an NTP fault — no time received
        from the NTP server (clear means the clock is in sync)."""
        return self._ntp_fault_state

    @property
    def config_other_enable(self) -> int:
        """Misc configuration flags. Use the `is_*_enabled` methods to query
        individual bits (TCP/IP boost, SD card, DMX, …)."""
        return self._config_other_enable

    @property
    def dosage_control(self) -> int:
        """Dosage configuration flags. Use the `is_*_dosage_enabled` and
        `is_electrolysis_enabled` methods to query individual bits."""
        return self._dosage_control

    @property
    def ph_plus_dosage_relay_id(self) -> int:
        """Aggregated relay ID configured to act as the pH+ dosing pump."""
        return self._ph_plus_dosage_relay_id

    @property
    def ph_minus_dosage_relay_id(self) -> int:
        """Aggregated relay ID configured to act as the pH- dosing pump."""
        return self._ph_minus_dosage_relay_id

    @property
    def chlorine_dosage_relay_id(self) -> int:
        """Aggregated relay ID configured to act as the chlorine dosing pump."""
        return self._chlorine_dosage_relay_id

    def is_chlorine_dosage_enabled(self) -> bool:
        """True if chlorine dosage control is enabled in the controller config (bit 0)."""
        return self._dosage_control & 1 == 1

    def is_electrolysis_enabled(self) -> bool:
        """True if electrolysis (saltwater) chlorination is enabled (bit 4)."""
        return self._dosage_control & 16 == 16

    def is_ph_minus_dosage_enabled(self) -> bool:
        """True if pH- dosage control is enabled in the controller config (bit 8)."""
        return self._dosage_control & 256 == 256

    def is_ph_plus_dosage_enabled(self) -> bool:
        """True if pH+ dosage control is enabled in the controller config (bit 12)."""
        return self._dosage_control & 4096 == 4096

    def is_dosage_enabled(self, data_entity: DataObject) -> bool:
        """Convenience: is the dosage chemical for this canister/consumption entity enabled?

        Args:
            data_entity: A canister (column 36–38) or consumption (column 39–41)
                `DataObject`. The chemical is inferred from the column index.

        Returns:
            True if the corresponding ``is_*_dosage_enabled`` flag is set.
            False for any other column (or if the chemical is disabled).
        """
        col = data_entity.column
        if col in (36, 39):
            return self.is_chlorine_dosage_enabled()
        if col in (37, 40):
            return self.is_ph_minus_dosage_enabled()
        if col in (38, 41):
            return self.is_ph_plus_dosage_enabled()
        return False

    def get_dosage_relay(self, data_entity: DataObject) -> int | None:
        """Aggregated relay ID that handles the dosing for this canister/consumption entity.

        Args:
            data_entity: A canister (column 36–38) or consumption (column 39–41)
                `DataObject`.

        Returns:
            The aggregated relay ID (chlorine, pH-, or pH+) corresponding to
            the entity's chemical, or ``None`` if the entity is not a
            canister/consumption object.
        """
        col = data_entity.column
        if col in (36, 39):
            return self._chlorine_dosage_relay_id
        if col in (37, 40):
            return self._ph_minus_dosage_relay_id
        if col in (38, 41):
            return self._ph_plus_dosage_relay_id
        return None

    def is_dosage_relay(
        self,
        relay_object: Relay | None = None,
        data_object: DataObject | None = None,
        relay_id: int | None = None,
    ) -> bool:
        """Check whether a relay is one of the configured dosage control relays.

        Provide one of `relay_object`, `data_object`, or `relay_id`. If more
        than one is supplied, the first non-None argument in that precedence
        order wins and the others are ignored. If none are provided the
        method returns False.

        Args:
            relay_object: A `Relay` instance. Highest-precedence argument.
            data_object: A `DataObject` of category `relay` or
                `external_relay`. Considered only when `relay_object` is None.
            relay_id: An aggregated relay ID (0–15). Considered only when
                both `relay_object` and `data_object` are None.

        Returns:
            True if the resolved argument identifies a dosage relay; False
            otherwise (including when no argument is provided).

        Raises:
            BadRelayException: If ``data_object`` is the resolved argument
                but is not a relay-category `DataObject`.

        Example:
            ```python
            # Three equivalent ways to ask "is relay 5 a dosage relay?",
            # assuming the chlorine pump is configured there.
            state.is_dosage_relay(relay_id=5)
            state.is_dosage_relay(relay_object=state.get_relay(5))
            state.is_dosage_relay(data_object=state.aggregated_relay_objects[5])
            ```
        """
        dosage_control_relays = [
            self._chlorine_dosage_relay_id,
            self._ph_minus_dosage_relay_id,
            self._ph_plus_dosage_relay_id,
        ]
        if relay_object is not None:
            return relay_object.relay_id in dosage_control_relays
        if data_object is not None:
            if data_object.category not in (CATEGORY_RELAY, CATEGORY_EXTERNAL_RELAY):
                raise BadRelayException(
                    f"DataObject category '{data_object.category}' is not a relay category"
                )
            offset = (
                EXTERNAL_RELAY_ID_OFFSET if data_object.category == CATEGORY_EXTERNAL_RELAY else 0
            )
            return data_object.category_id + offset in dosage_control_relays
        if relay_id is not None:
            return relay_id in dosage_control_relays
        return False

    def get_reset_root_cause_as_str(self) -> str:
        """Decode `reset_root_cause` to its `RESET_ROOT_CAUSE` label.

        Falls back to the "n.a." label for any value not in the lookup table.
        """
        if self._reset_root_cause not in RESET_ROOT_CAUSE:
            return RESET_ROOT_CAUSE[0]
        return RESET_ROOT_CAUSE[self._reset_root_cause]

    def get_ntp_fault_state_as_str(self) -> str:
        """Decode `ntp_fault_state` to a human-readable label from `NTP_FAULT_STATE`.

        For exact matches in the lookup table (``0``, ``1``, ``2``, ``4``,
        ``65536``) the corresponding label is returned. Composite states are
        approximated by returning the highest-severity active bit (4 → 2 →
        1), since the controller's CSV has no fixed combinations beyond the
        listed ones. Falls back to "n.a." if no severity bit is set.
        """
        if self._ntp_fault_state in NTP_FAULT_STATE:
            return NTP_FAULT_STATE[self._ntp_fault_state]
        for bit in (4, 2, 1):
            if self._ntp_fault_state & bit:
                return NTP_FAULT_STATE[bit]
        return NTP_FAULT_STATE[0]

    def is_tcpip_boost_enabled(self) -> bool:
        """True if TCP/IP boost is enabled in the controller config (bit 0)."""
        return self._config_other_enable & 1 == 1

    def is_sd_card_enabled(self) -> bool:
        """True if SD card logging is enabled in the controller config (bit 1)."""
        return self._config_other_enable & 2 == 2

    def is_dmx_enabled(self) -> bool:
        """True if DMX output is enabled in the controller config (bit 2)."""
        return self._config_other_enable & 4 == 4

    def is_avatar_enabled(self) -> bool:
        """True if the avatar feature is enabled in the controller config (bit 3)."""
        return self._config_other_enable & 8 == 8

    def is_relay_extension_enabled(self) -> bool:
        """True if the external relay extension module is connected and active (bit 4).

        Affects how `determine_overall_relay_bit_state` builds the ENA mask:
        with the extension active, the mask covers all 16 bits instead of
        just the internal 8.
        """
        return self._config_other_enable & 16 == 16

    def is_high_bus_load_enabled(self) -> bool:
        """True if high bus load mode is enabled in the controller config (bit 5)."""
        return self._config_other_enable & 32 == 32

    def is_flow_sensor_enabled(self) -> bool:
        """True if the flow sensor is enabled in the controller config (bit 6)."""
        return self._config_other_enable & 64 == 64

    def is_repeated_mails_enabled(self) -> bool:
        """True if repeated email notifications are enabled (bit 7)."""
        return self._config_other_enable & 128 == 128

    def is_dmx_extension_enabled(self) -> bool:
        """True if the DMX extension module is enabled in the controller config (bit 8)."""
        return self._config_other_enable & 256 == 256

    def _parse(self) -> None:
        """Build per-column `DataObject` instances and group them by category."""
        self._data_objects = []
        for column, name in enumerate(self._data_names):
            self._data_objects.append(
                DataObject(
                    column,
                    name,
                    self._data_units[column],
                    self._data_offsets[column],
                    self._data_gain[column],
                    self._data_raw_values[column],
                )
            )

        self._analog_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_ANALOG
        ]
        self._electrode_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_ELECTRODE
        ]
        self._temperature_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_TEMPERATURE
        ]
        self._relay_objects = [obj for obj in self._data_objects if obj.category == CATEGORY_RELAY]
        self._digital_input_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_DIGITAL_INPUT
        ]
        self._external_relay_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_EXTERNAL_RELAY
        ]
        self._canister_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_CANISTER
        ]
        self._consumption_objects = [
            obj for obj in self._data_objects if obj.category == CATEGORY_CONSUMPTION
        ]

    @property
    def analog_objects(self) -> list[DataObject]:
        """The five analog inputs (columns 1–5), in column order."""
        return self._analog_objects

    @property
    def electrode_objects(self) -> list[DataObject]:
        """The two electrode readings — redox at index 0, pH at index 1."""
        return self._electrode_objects

    @property
    def temperature_objects(self) -> list[DataObject]:
        """The eight temperature sensors (columns 8–15), in column order."""
        return self._temperature_objects

    @property
    def relay_objects(self) -> list[DataObject]:
        """The eight built-in relays (columns 16–23), in column order.

        These are still untyped `DataObject` instances. Use `relays()` for
        `Relay` instances with on/off and manual/auto helpers, or
        `aggregated_relay_objects` to also include the external relays.
        """
        return self._relay_objects

    def relays(self) -> list[Relay]:
        """The eight built-in relays as `Relay` instances.

        Equivalent to wrapping each entry in `relay_objects` with `Relay(...)`.
        """
        return [Relay(obj) for obj in self._relay_objects]

    @property
    def digital_input_objects(self) -> list[DataObject]:
        """The four digital inputs (columns 24–27), in column order."""
        return self._digital_input_objects

    def digital_inputs(self) -> list[DigitalInput]:
        """The four digital inputs as `DigitalInput` instances.

        Equivalent to wrapping each entry in `digital_input_objects` with
        `DigitalInput(...)`. Use these to get `get_bit_mask()` for triggering.
        """
        return [DigitalInput(obj) for obj in self._digital_input_objects]

    @property
    def external_relay_objects(self) -> list[DataObject]:
        """The eight external relays (columns 28–35), in column order.

        Will be present in the parsed data even when the relay extension is
        not enabled in the controller config — check
        `is_relay_extension_enabled` before treating them as live.
        """
        return self._external_relay_objects

    def external_relays(self) -> list[Relay]:
        """The eight external relays as `Relay` instances."""
        return [Relay(obj) for obj in self._external_relay_objects]

    @property
    def canister_objects(self) -> list[DataObject]:
        """The three canister fill-level readings (columns 36–38).

        Order: chlorine, pH-, pH+. Convenience properties `chlorine_canister`,
        `ph_minus_canister`, and `ph_plus_canister` return individual entries.
        """
        return self._canister_objects

    @property
    def consumption_objects(self) -> list[DataObject]:
        """The three dosage consumption counters (columns 39–41).

        Order: chlorine, pH-, pH+. Convenience properties
        `chlorine_consumption`, `ph_minus_consumption`, and
        `ph_plus_consumption` return individual entries.
        """
        return self._consumption_objects

    @property
    def redox_electrode(self) -> DataObject:
        """The redox electrode reading (column 6)."""
        return self._electrode_objects[0]

    @property
    def ph_electrode(self) -> DataObject:
        """The pH electrode reading (column 7)."""
        return self._electrode_objects[1]

    @property
    def chlorine_canister(self) -> DataObject:
        """Chlorine canister fill level (column 36)."""
        return self._canister_objects[0]

    @property
    def ph_minus_canister(self) -> DataObject:
        """pH- canister fill level (column 37)."""
        return self._canister_objects[1]

    @property
    def ph_plus_canister(self) -> DataObject:
        """pH+ canister fill level (column 38)."""
        return self._canister_objects[2]

    @property
    def chlorine_consumption(self) -> DataObject:
        """Cumulative chlorine consumption counter (column 39)."""
        return self._consumption_objects[0]

    @property
    def ph_minus_consumption(self) -> DataObject:
        """Cumulative pH- consumption counter (column 40)."""
        return self._consumption_objects[1]

    @property
    def ph_plus_consumption(self) -> DataObject:
        """Cumulative pH+ consumption counter (column 41)."""
        return self._consumption_objects[2]

    @property
    def aggregated_relay_objects(self) -> list[DataObject]:
        """All 16 relay `DataObject`s — internal first, then external.

        Index in this list is the aggregated relay ID used by `Relay.relay_id`,
        `get_relay`, and the `RelaySwitch` API.
        """
        return self._relay_objects + self._external_relay_objects

    @property
    def chlorine_dosage_relay(self) -> DataObject:
        """The relay configured as the chlorine dosing pump."""
        return self.aggregated_relay_objects[self._chlorine_dosage_relay_id]

    @property
    def ph_minus_dosage_relay(self) -> DataObject:
        """The relay configured as the pH- dosing pump."""
        return self.aggregated_relay_objects[self._ph_minus_dosage_relay_id]

    @property
    def ph_plus_dosage_relay(self) -> DataObject:
        """The relay configured as the pH+ dosing pump."""
        return self.aggregated_relay_objects[self._ph_plus_dosage_relay_id]

    def get_relay(self, relay_id: int) -> Relay:
        """Return the `Relay` for the given aggregated relay ID (0–15).

        Args:
            relay_id: 0–7 for internal relays, 8–15 for external relays.

        Returns:
            A new `Relay` wrapping the underlying `DataObject`.

        Raises:
            IndexError: If ``relay_id`` is outside the 0–15 range.
        """
        return Relay(self.aggregated_relay_objects[relay_id])

    def get_relays(self) -> list[Relay]:
        """All 16 relays as `Relay` instances, in aggregated-ID order."""
        return [Relay(obj) for obj in self.aggregated_relay_objects]

    def determine_overall_relay_bit_state(self) -> list[int]:
        """Build the two-element ENA bit field that represents the current relay state.

        The controller's `/usrcfg.cgi` payload uses an ``ENA=enable_mask,on_mask``
        pair to set relay state. ``enable_mask`` selects which relays are in
        manual mode (bit set = manual, bit clear = auto), and ``on_mask``
        selects the manual-on relays among them.

        Returns:
            A two-element list ``[enable_mask, on_mask]``. Both masks cover
            bits 0–7 (internal relays) by default, or bits 0–15 if the
            external relay extension is enabled (`is_relay_extension_enabled`).

            The masks reflect the *current* state, so callers can flip a
            single relay's bit and POST the result to switch only that
            relay without touching the others.
        """
        relay_list: list[Relay] = [Relay(obj) for obj in self._relay_objects]
        bit_state = [255, 0]
        if self.is_relay_extension_enabled():
            relay_list.extend(Relay(obj) for obj in self._external_relay_objects)
            bit_state[0] = 65535
        for relay in relay_list:
            relay_bit_mask = relay.get_bit_mask()
            if relay.is_auto_mode():
                bit_state[0] &= ~relay_bit_mask
            if relay.is_on():
                bit_state[1] |= relay_bit_mask
        return bit_state

time property

time: str

Controller's current local time as "HH:MM".

version property

version: str

Firmware version string reported by the controller.

cpu_time property

cpu_time: int

Controller CPU uptime in seconds since the last reset.

reset_root_cause property

reset_root_cause: int

Numeric reset-root-cause code. Decode with RESET_ROOT_CAUSE or get_reset_root_cause_as_str.

ntp_fault_state property

ntp_fault_state: int

Numeric NTP fault state. Decode with NTP_FAULT_STATE or get_ntp_fault_state_as_str. Bits 0/1/2 indicate severity (logfile, warning, error); bit 16 set indicates an NTP fault — no time received from the NTP server (clear means the clock is in sync).

config_other_enable property

config_other_enable: int

Misc configuration flags. Use the is_*_enabled methods to query individual bits (TCP/IP boost, SD card, DMX, …).

dosage_control property

dosage_control: int

Dosage configuration flags. Use the is_*_dosage_enabled and is_electrolysis_enabled methods to query individual bits.

ph_plus_dosage_relay_id property

ph_plus_dosage_relay_id: int

Aggregated relay ID configured to act as the pH+ dosing pump.

ph_minus_dosage_relay_id property

ph_minus_dosage_relay_id: int

Aggregated relay ID configured to act as the pH- dosing pump.

chlorine_dosage_relay_id property

chlorine_dosage_relay_id: int

Aggregated relay ID configured to act as the chlorine dosing pump.

analog_objects property

analog_objects: list[DataObject]

The five analog inputs (columns 1–5), in column order.

electrode_objects property

electrode_objects: list[DataObject]

The two electrode readings — redox at index 0, pH at index 1.

temperature_objects property

temperature_objects: list[DataObject]

The eight temperature sensors (columns 8–15), in column order.

relay_objects property

relay_objects: list[DataObject]

The eight built-in relays (columns 16–23), in column order.

These are still untyped DataObject instances. Use relays() for Relay instances with on/off and manual/auto helpers, or aggregated_relay_objects to also include the external relays.

digital_input_objects property

digital_input_objects: list[DataObject]

The four digital inputs (columns 24–27), in column order.

external_relay_objects property

external_relay_objects: list[DataObject]

The eight external relays (columns 28–35), in column order.

Will be present in the parsed data even when the relay extension is not enabled in the controller config — check is_relay_extension_enabled before treating them as live.

canister_objects property

canister_objects: list[DataObject]

The three canister fill-level readings (columns 36–38).

Order: chlorine, pH-, pH+. Convenience properties chlorine_canister, ph_minus_canister, and ph_plus_canister return individual entries.

consumption_objects property

consumption_objects: list[DataObject]

The three dosage consumption counters (columns 39–41).

Order: chlorine, pH-, pH+. Convenience properties chlorine_consumption, ph_minus_consumption, and ph_plus_consumption return individual entries.

redox_electrode property

redox_electrode: DataObject

The redox electrode reading (column 6).

ph_electrode property

ph_electrode: DataObject

The pH electrode reading (column 7).

chlorine_canister property

chlorine_canister: DataObject

Chlorine canister fill level (column 36).

ph_minus_canister property

ph_minus_canister: DataObject

pH- canister fill level (column 37).

ph_plus_canister property

ph_plus_canister: DataObject

pH+ canister fill level (column 38).

chlorine_consumption property

chlorine_consumption: DataObject

Cumulative chlorine consumption counter (column 39).

ph_minus_consumption property

ph_minus_consumption: DataObject

Cumulative pH- consumption counter (column 40).

ph_plus_consumption property

ph_plus_consumption: DataObject

Cumulative pH+ consumption counter (column 41).

aggregated_relay_objects property

aggregated_relay_objects: list[DataObject]

All 16 relay DataObjects — internal first, then external.

Index in this list is the aggregated relay ID used by Relay.relay_id, get_relay, and the RelaySwitch API.

chlorine_dosage_relay property

chlorine_dosage_relay: DataObject

The relay configured as the chlorine dosing pump.

ph_minus_dosage_relay property

ph_minus_dosage_relay: DataObject

The relay configured as the pH- dosing pump.

ph_plus_dosage_relay property

ph_plus_dosage_relay: DataObject

The relay configured as the pH+ dosing pump.

is_chlorine_dosage_enabled

is_chlorine_dosage_enabled() -> bool

True if chlorine dosage control is enabled in the controller config (bit 0).

Source code in src/proconip/definitions.py

def is_chlorine_dosage_enabled(self) -> bool:
    """True if chlorine dosage control is enabled in the controller config (bit 0)."""
    return self._dosage_control & 1 == 1

is_electrolysis_enabled

is_electrolysis_enabled() -> bool

True if electrolysis (saltwater) chlorination is enabled (bit 4).

Source code in src/proconip/definitions.py

def is_electrolysis_enabled(self) -> bool:
    """True if electrolysis (saltwater) chlorination is enabled (bit 4)."""
    return self._dosage_control & 16 == 16

is_ph_minus_dosage_enabled

is_ph_minus_dosage_enabled() -> bool

True if pH- dosage control is enabled in the controller config (bit 8).

Source code in src/proconip/definitions.py

def is_ph_minus_dosage_enabled(self) -> bool:
    """True if pH- dosage control is enabled in the controller config (bit 8)."""
    return self._dosage_control & 256 == 256

is_ph_plus_dosage_enabled

is_ph_plus_dosage_enabled() -> bool

True if pH+ dosage control is enabled in the controller config (bit 12).

Source code in src/proconip/definitions.py

def is_ph_plus_dosage_enabled(self) -> bool:
    """True if pH+ dosage control is enabled in the controller config (bit 12)."""
    return self._dosage_control & 4096 == 4096

is_dosage_enabled

is_dosage_enabled(data_entity: DataObject) -> bool

Convenience: is the dosage chemical for this canister/consumption entity enabled?

PARAMETER	DESCRIPTION
data_entity	A canister (column 36–38) or consumption (column 39–41) DataObject. The chemical is inferred from the column index. TYPE: DataObject

RETURNS	DESCRIPTION
bool	True if the corresponding is_*_dosage_enabled flag is set.
bool	False for any other column (or if the chemical is disabled).

Source code in src/proconip/definitions.py

def is_dosage_enabled(self, data_entity: DataObject) -> bool:
    """Convenience: is the dosage chemical for this canister/consumption entity enabled?

    Args:
        data_entity: A canister (column 36–38) or consumption (column 39–41)
            `DataObject`. The chemical is inferred from the column index.

    Returns:
        True if the corresponding ``is_*_dosage_enabled`` flag is set.
        False for any other column (or if the chemical is disabled).
    """
    col = data_entity.column
    if col in (36, 39):
        return self.is_chlorine_dosage_enabled()
    if col in (37, 40):
        return self.is_ph_minus_dosage_enabled()
    if col in (38, 41):
        return self.is_ph_plus_dosage_enabled()
    return False

get_dosage_relay

get_dosage_relay(data_entity: DataObject) -> int | None

Aggregated relay ID that handles the dosing for this canister/consumption entity.

PARAMETER	DESCRIPTION
data_entity	A canister (column 36–38) or consumption (column 39–41) DataObject. TYPE: DataObject

RETURNS	DESCRIPTION
int | None	The aggregated relay ID (chlorine, pH-, or pH+) corresponding to
int | None	the entity's chemical, or None if the entity is not a
int | None	canister/consumption object.

Source code in src/proconip/definitions.py

def get_dosage_relay(self, data_entity: DataObject) -> int | None:
    """Aggregated relay ID that handles the dosing for this canister/consumption entity.

    Args:
        data_entity: A canister (column 36–38) or consumption (column 39–41)
            `DataObject`.

    Returns:
        The aggregated relay ID (chlorine, pH-, or pH+) corresponding to
        the entity's chemical, or ``None`` if the entity is not a
        canister/consumption object.
    """
    col = data_entity.column
    if col in (36, 39):
        return self._chlorine_dosage_relay_id
    if col in (37, 40):
        return self._ph_minus_dosage_relay_id
    if col in (38, 41):
        return self._ph_plus_dosage_relay_id
    return None

is_dosage_relay

is_dosage_relay(relay_object: Relay | None = None, data_object: DataObject | None = None, relay_id: int | None = None) -> bool

Check whether a relay is one of the configured dosage control relays.

Provide one of relay_object, data_object, or relay_id. If more than one is supplied, the first non-None argument in that precedence order wins and the others are ignored. If none are provided the method returns False.

PARAMETER	DESCRIPTION
relay_object	A Relay instance. Highest-precedence argument. TYPE: Relay | None DEFAULT: None
data_object	A DataObject of category relay or external_relay. Considered only when relay_object is None. TYPE: DataObject | None DEFAULT: None
relay_id	An aggregated relay ID (0–15). Considered only when both relay_object and data_object are None. TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
bool	True if the resolved argument identifies a dosage relay; False
bool	otherwise (including when no argument is provided).

RAISES	DESCRIPTION
BadRelayException	If data_object is the resolved argument but is not a relay-category DataObject.

Example

# Three equivalent ways to ask "is relay 5 a dosage relay?",
# assuming the chlorine pump is configured there.
state.is_dosage_relay(relay_id=5)
state.is_dosage_relay(relay_object=state.get_relay(5))
state.is_dosage_relay(data_object=state.aggregated_relay_objects[5])

Source code in src/proconip/definitions.py

def is_dosage_relay(
    self,
    relay_object: Relay | None = None,
    data_object: DataObject | None = None,
    relay_id: int | None = None,
) -> bool:
    """Check whether a relay is one of the configured dosage control relays.

    Provide one of `relay_object`, `data_object`, or `relay_id`. If more
    than one is supplied, the first non-None argument in that precedence
    order wins and the others are ignored. If none are provided the
    method returns False.

    Args:
        relay_object: A `Relay` instance. Highest-precedence argument.
        data_object: A `DataObject` of category `relay` or
            `external_relay`. Considered only when `relay_object` is None.
        relay_id: An aggregated relay ID (0–15). Considered only when
            both `relay_object` and `data_object` are None.

    Returns:
        True if the resolved argument identifies a dosage relay; False
        otherwise (including when no argument is provided).

    Raises:
        BadRelayException: If ``data_object`` is the resolved argument
            but is not a relay-category `DataObject`.

    Example:
        ```python
        # Three equivalent ways to ask "is relay 5 a dosage relay?",
        # assuming the chlorine pump is configured there.
        state.is_dosage_relay(relay_id=5)
        state.is_dosage_relay(relay_object=state.get_relay(5))
        state.is_dosage_relay(data_object=state.aggregated_relay_objects[5])
        ```
    """
    dosage_control_relays = [
        self._chlorine_dosage_relay_id,
        self._ph_minus_dosage_relay_id,
        self._ph_plus_dosage_relay_id,
    ]
    if relay_object is not None:
        return relay_object.relay_id in dosage_control_relays
    if data_object is not None:
        if data_object.category not in (CATEGORY_RELAY, CATEGORY_EXTERNAL_RELAY):
            raise BadRelayException(
                f"DataObject category '{data_object.category}' is not a relay category"
            )
        offset = (
            EXTERNAL_RELAY_ID_OFFSET if data_object.category == CATEGORY_EXTERNAL_RELAY else 0
        )
        return data_object.category_id + offset in dosage_control_relays
    if relay_id is not None:
        return relay_id in dosage_control_relays
    return False

get_reset_root_cause_as_str

get_reset_root_cause_as_str() -> str

Decode reset_root_cause to its RESET_ROOT_CAUSE label.

Falls back to the "n.a." label for any value not in the lookup table.

Source code in src/proconip/definitions.py

def get_reset_root_cause_as_str(self) -> str:
    """Decode `reset_root_cause` to its `RESET_ROOT_CAUSE` label.

    Falls back to the "n.a." label for any value not in the lookup table.
    """
    if self._reset_root_cause not in RESET_ROOT_CAUSE:
        return RESET_ROOT_CAUSE[0]
    return RESET_ROOT_CAUSE[self._reset_root_cause]

get_ntp_fault_state_as_str

get_ntp_fault_state_as_str() -> str

Decode ntp_fault_state to a human-readable label from NTP_FAULT_STATE.

For exact matches in the lookup table (0, 1, 2, 4, 65536) the corresponding label is returned. Composite states are approximated by returning the highest-severity active bit (4 → 2 → 1), since the controller's CSV has no fixed combinations beyond the listed ones. Falls back to "n.a." if no severity bit is set.

Source code in src/proconip/definitions.py

def get_ntp_fault_state_as_str(self) -> str:
    """Decode `ntp_fault_state` to a human-readable label from `NTP_FAULT_STATE`.

    For exact matches in the lookup table (``0``, ``1``, ``2``, ``4``,
    ``65536``) the corresponding label is returned. Composite states are
    approximated by returning the highest-severity active bit (4 → 2 →
    1), since the controller's CSV has no fixed combinations beyond the
    listed ones. Falls back to "n.a." if no severity bit is set.
    """
    if self._ntp_fault_state in NTP_FAULT_STATE:
        return NTP_FAULT_STATE[self._ntp_fault_state]
    for bit in (4, 2, 1):
        if self._ntp_fault_state & bit:
            return NTP_FAULT_STATE[bit]
    return NTP_FAULT_STATE[0]

is_tcpip_boost_enabled

is_tcpip_boost_enabled() -> bool

True if TCP/IP boost is enabled in the controller config (bit 0).

Source code in src/proconip/definitions.py

def is_tcpip_boost_enabled(self) -> bool:
    """True if TCP/IP boost is enabled in the controller config (bit 0)."""
    return self._config_other_enable & 1 == 1

is_sd_card_enabled

is_sd_card_enabled() -> bool

True if SD card logging is enabled in the controller config (bit 1).

Source code in src/proconip/definitions.py

def is_sd_card_enabled(self) -> bool:
    """True if SD card logging is enabled in the controller config (bit 1)."""
    return self._config_other_enable & 2 == 2

is_dmx_enabled

is_dmx_enabled() -> bool

True if DMX output is enabled in the controller config (bit 2).

Source code in src/proconip/definitions.py

def is_dmx_enabled(self) -> bool:
    """True if DMX output is enabled in the controller config (bit 2)."""
    return self._config_other_enable & 4 == 4

is_avatar_enabled

is_avatar_enabled() -> bool

True if the avatar feature is enabled in the controller config (bit 3).

Source code in src/proconip/definitions.py

def is_avatar_enabled(self) -> bool:
    """True if the avatar feature is enabled in the controller config (bit 3)."""
    return self._config_other_enable & 8 == 8

is_relay_extension_enabled

is_relay_extension_enabled() -> bool

True if the external relay extension module is connected and active (bit 4).

Affects how determine_overall_relay_bit_state builds the ENA mask: with the extension active, the mask covers all 16 bits instead of just the internal 8.

Source code in src/proconip/definitions.py

def is_relay_extension_enabled(self) -> bool:
    """True if the external relay extension module is connected and active (bit 4).

    Affects how `determine_overall_relay_bit_state` builds the ENA mask:
    with the extension active, the mask covers all 16 bits instead of
    just the internal 8.
    """
    return self._config_other_enable & 16 == 16

is_high_bus_load_enabled

is_high_bus_load_enabled() -> bool

True if high bus load mode is enabled in the controller config (bit 5).

Source code in src/proconip/definitions.py

def is_high_bus_load_enabled(self) -> bool:
    """True if high bus load mode is enabled in the controller config (bit 5)."""
    return self._config_other_enable & 32 == 32

is_flow_sensor_enabled

is_flow_sensor_enabled() -> bool

True if the flow sensor is enabled in the controller config (bit 6).

Source code in src/proconip/definitions.py

def is_flow_sensor_enabled(self) -> bool:
    """True if the flow sensor is enabled in the controller config (bit 6)."""
    return self._config_other_enable & 64 == 64

is_repeated_mails_enabled

is_repeated_mails_enabled() -> bool

True if repeated email notifications are enabled (bit 7).

Source code in src/proconip/definitions.py

def is_repeated_mails_enabled(self) -> bool:
    """True if repeated email notifications are enabled (bit 7)."""
    return self._config_other_enable & 128 == 128

is_dmx_extension_enabled

is_dmx_extension_enabled() -> bool

True if the DMX extension module is enabled in the controller config (bit 8).

Source code in src/proconip/definitions.py

def is_dmx_extension_enabled(self) -> bool:
    """True if the DMX extension module is enabled in the controller config (bit 8)."""
    return self._config_other_enable & 256 == 256

relays

relays() -> list[Relay]

The eight built-in relays as Relay instances.

Equivalent to wrapping each entry in relay_objects with Relay(...).

Source code in src/proconip/definitions.py

def relays(self) -> list[Relay]:
    """The eight built-in relays as `Relay` instances.

    Equivalent to wrapping each entry in `relay_objects` with `Relay(...)`.
    """
    return [Relay(obj) for obj in self._relay_objects]

digital_inputs

digital_inputs() -> list[DigitalInput]

The four digital inputs as DigitalInput instances.

Equivalent to wrapping each entry in digital_input_objects with DigitalInput(...). Use these to get get_bit_mask() for triggering.

Source code in src/proconip/definitions.py

def digital_inputs(self) -> list[DigitalInput]:
    """The four digital inputs as `DigitalInput` instances.

    Equivalent to wrapping each entry in `digital_input_objects` with
    `DigitalInput(...)`. Use these to get `get_bit_mask()` for triggering.
    """
    return [DigitalInput(obj) for obj in self._digital_input_objects]

external_relays

external_relays() -> list[Relay]

The eight external relays as Relay instances.

Source code in src/proconip/definitions.py

def external_relays(self) -> list[Relay]:
    """The eight external relays as `Relay` instances."""
    return [Relay(obj) for obj in self._external_relay_objects]

get_relay

get_relay(relay_id: int) -> Relay

Return the Relay for the given aggregated relay ID (0–15).

PARAMETER	DESCRIPTION
relay_id	0–7 for internal relays, 8–15 for external relays. TYPE: int

RETURNS	DESCRIPTION
Relay	A new Relay wrapping the underlying DataObject.

RAISES	DESCRIPTION
IndexError	If relay_id is outside the 0–15 range.

Source code in src/proconip/definitions.py

def get_relay(self, relay_id: int) -> Relay:
    """Return the `Relay` for the given aggregated relay ID (0–15).

    Args:
        relay_id: 0–7 for internal relays, 8–15 for external relays.

    Returns:
        A new `Relay` wrapping the underlying `DataObject`.

    Raises:
        IndexError: If ``relay_id`` is outside the 0–15 range.
    """
    return Relay(self.aggregated_relay_objects[relay_id])

get_relays

get_relays() -> list[Relay]

All 16 relays as Relay instances, in aggregated-ID order.

Source code in src/proconip/definitions.py

def get_relays(self) -> list[Relay]:
    """All 16 relays as `Relay` instances, in aggregated-ID order."""
    return [Relay(obj) for obj in self.aggregated_relay_objects]

determine_overall_relay_bit_state

determine_overall_relay_bit_state() -> list[int]

Build the two-element ENA bit field that represents the current relay state.

The controller's /usrcfg.cgi payload uses an ENA=enable_mask,on_mask pair to set relay state. enable_mask selects which relays are in manual mode (bit set = manual, bit clear = auto), and on_mask selects the manual-on relays among them.

RETURNS	DESCRIPTION
list[int]	A two-element list [enable_mask, on_mask]. Both masks cover
list[int]	bits 0–7 (internal relays) by default, or bits 0–15 if the
list[int]	external relay extension is enabled (is_relay_extension_enabled).
list[int]	The masks reflect the current state, so callers can flip a
list[int]	single relay's bit and POST the result to switch only that
list[int]	relay without touching the others.

Source code in src/proconip/definitions.py

def determine_overall_relay_bit_state(self) -> list[int]:
    """Build the two-element ENA bit field that represents the current relay state.

    The controller's `/usrcfg.cgi` payload uses an ``ENA=enable_mask,on_mask``
    pair to set relay state. ``enable_mask`` selects which relays are in
    manual mode (bit set = manual, bit clear = auto), and ``on_mask``
    selects the manual-on relays among them.

    Returns:
        A two-element list ``[enable_mask, on_mask]``. Both masks cover
        bits 0–7 (internal relays) by default, or bits 0–15 if the
        external relay extension is enabled (`is_relay_extension_enabled`).

        The masks reflect the *current* state, so callers can flip a
        single relay's bit and POST the result to switch only that
        relay without touching the others.
    """
    relay_list: list[Relay] = [Relay(obj) for obj in self._relay_objects]
    bit_state = [255, 0]
    if self.is_relay_extension_enabled():
        relay_list.extend(Relay(obj) for obj in self._external_relay_objects)
        bit_state[0] = 65535
    for relay in relay_list:
        relay_bit_mask = relay.get_bit_mask()
        if relay.is_auto_mode():
            bit_state[0] &= ~relay_bit_mask
        if relay.is_on():
            bit_state[1] |= relay_bit_mask
    return bit_state

DmxChannelData

A single DMX channel's index, name, and current value.

ATTRIBUTE	DESCRIPTION
value	Current channel intensity, expected in the range [0, 255]. The constructor stores the value verbatim; clamping happens in GetDmxData.set. TYPE: int

Source code in src/proconip/definitions.py

class DmxChannelData:
    """A single DMX channel's index, name, and current value.

    Attributes:
        value: Current channel intensity, expected in the range [0, 255].
            The constructor stores the value verbatim; clamping happens in
            `GetDmxData.set`.
    """

    value: int
    _index: int
    _name: str

    def __init__(self, index: int, value: int):
        """Build a channel entry.

        Args:
            index: Zero-based channel index (0 = channel 1, 15 = channel 16).
            value: Initial channel intensity. The constructor does not clamp
                values — out-of-range inputs are stored verbatim. Use
                `GetDmxData.set` if you want the [0, 255] clamp.
        """
        self.value = value
        self._index = index
        self._name = f"CH{index + 1:0>2}"

    @property
    def index(self) -> int:
        """Zero-based channel index (0 = channel 1)."""
        return self._index

    @property
    def name(self) -> str:
        """Human-friendly channel name like ``"CH01"`` or ``"CH16"``."""
        return self._name

    def __str__(self) -> str:
        """Render the channel as a bare integer string for payload building."""
        return str(self.value)

index property

index: int

Zero-based channel index (0 = channel 1).

name property

name: str

Human-friendly channel name like "CH01" or "CH16".

GetDmxData

Mutable representation of all 16 DMX channels.

Construct from a /GetDmx.csv body, then read or modify channels via indexing, iteration, get_value, or set. Pass the (possibly mutated) instance to proconip.api.async_set_dmx to write the new state back.

Source code in src/proconip/definitions.py

class GetDmxData:
    """Mutable representation of all 16 DMX channels.

    Construct from a `/GetDmx.csv` body, then read or modify channels via
    indexing, iteration, `get_value`, or `set`. Pass the (possibly mutated)
    instance to `proconip.api.async_set_dmx` to write the new state back.
    """

    _channels: list[DmxChannelData]

    def __init__(self, raw_data: str):
        """Parse a `/GetDmx.csv` body into 16 channels.

        Args:
            raw_data: The raw CSV string returned by the controller. Leading
                blank lines are tolerated; only the first non-blank line is
                parsed.

        Raises:
            InvalidPayloadException: If the payload is empty, whitespace-only,
                or does not contain exactly 16 comma-separated channel values.
            ValueError: If a channel value cannot be parsed as an integer.
        """
        self._raw_data = raw_data
        self._channels = []

        line = 0
        lines = raw_data.splitlines()
        while line < len(lines) and len(lines[line].strip()) < 1:
            line += 1

        if line >= len(lines):
            raise InvalidPayloadException("Empty or missing DMX payload")

        values = lines[line].split(",")
        if len(values) != 16:
            raise InvalidPayloadException(
                f"GetDmx.csv must contain exactly 16 channels; got {len(values)}"
            )
        for idx, value in enumerate(values):
            self._channels.append(DmxChannelData(idx, int(value)))

    def __getitem__(self, index: int) -> DmxChannelData:
        """Return the `DmxChannelData` at the given zero-based index."""
        return self._channels[index]

    def __iter__(self) -> Iterator[DmxChannelData]:
        """Iterate over all channels in index order."""
        return iter(self._channels)

    def __str__(self) -> str:
        """Return the raw CSV body the instance was parsed from."""
        return self._raw_data

    def get_value(self, index: int) -> int:
        """Return the current value of the channel at ``index``.

        Equivalent to ``self[index].value``. Provided for symmetry with
        `set`.
        """
        return self._channels[index].value

    def set(self, index: int, value: int) -> None:
        """Update the value of one channel.

        Values outside the [0, 255] range are silently clamped — the
        controller's DMX hardware only accepts 8-bit values, so callers
        rarely need anything else.

        Args:
            index: Zero-based channel index (0 for channel 1, 15 for
                channel 16).
            value: New intensity. Clamped to [0, 255].

        Raises:
            IndexError: If ``index`` is not in 0–15.
        """
        if index > 15 or index < 0:
            raise IndexError("Index must be between 0 (channel 1) and 15 (channel 16)")
        self._channels[index].value = max(0, min(255, value))

    @property
    def post_data(self) -> dict[str, str]:
        """Form payload that updates the DMX channel state via `/usrcfg.cgi`.

        The dict has five keys, all required by the controller:

        - ``TYPE``: always ``"0"``.
        - ``LEN``: always ``"16"`` (channels per write).
        - ``CH1_8``: comma-separated values for channels 1–8.
        - ``CH9_16``: comma-separated values for channels 9–16.
        - ``DMX512``: always ``"1"``.

        URL-encode and join with ``&`` to produce the actual POST body — see
        `proconip.api.async_set_dmx` for the canonical encoding.
        """
        return {
            "TYPE": "0",
            "LEN": "16",
            "CH1_8": ",".join(map(str, self._channels[:8])),
            "CH9_16": ",".join(map(str, self._channels[8:])),
            "DMX512": "1",
        }

post_data property

post_data: dict[str, str]

Form payload that updates the DMX channel state via /usrcfg.cgi.

The dict has five keys, all required by the controller:

• TYPE: always "0".
• LEN: always "16" (channels per write).
• CH1_8: comma-separated values for channels 1–8.
• CH9_16: comma-separated values for channels 9–16.
• DMX512: always "1".

URL-encode and join with & to produce the actual POST body — see proconip.api.async_set_dmx for the canonical encoding.

get_value

get_value(index: int) -> int

Return the current value of the channel at index.

Equivalent to self[index].value. Provided for symmetry with set.

Source code in src/proconip/definitions.py

def get_value(self, index: int) -> int:
    """Return the current value of the channel at ``index``.

    Equivalent to ``self[index].value``. Provided for symmetry with
    `set`.
    """
    return self._channels[index].value

set

set(index: int, value: int) -> None

Update the value of one channel.

Values outside the [0, 255] range are silently clamped — the controller's DMX hardware only accepts 8-bit values, so callers rarely need anything else.

PARAMETER	DESCRIPTION
index	Zero-based channel index (0 for channel 1, 15 for channel 16). TYPE: int
value	New intensity. Clamped to [0, 255]. TYPE: int

RAISES	DESCRIPTION
IndexError	If index is not in 0–15.

Source code in src/proconip/definitions.py

def set(self, index: int, value: int) -> None:
    """Update the value of one channel.

    Values outside the [0, 255] range are silently clamped — the
    controller's DMX hardware only accepts 8-bit values, so callers
    rarely need anything else.

    Args:
        index: Zero-based channel index (0 for channel 1, 15 for
            channel 16).
        value: New intensity. Clamped to [0, 255].

    Raises:
        IndexError: If ``index`` is not in 0–15.
    """
    if index > 15 or index < 0:
        raise IndexError("Index must be between 0 (channel 1) and 15 (channel 16)")
    self._channels[index].value = max(0, min(255, value))

BadRelayException

Bases: Exception

Raised when a relay argument doesn't make sense in context.

The two main cases are: switching a dosage relay on directly (rejected by proconip.api.async_switch_on), and passing a non-relay DataObject to GetStateData.is_dosage_relay.

Source code in src/proconip/definitions.py

class BadRelayException(Exception):
    """Raised when a relay argument doesn't make sense in context.

    The two main cases are: switching a dosage relay on directly (rejected
    by `proconip.api.async_switch_on`), and passing a non-relay `DataObject`
    to `GetStateData.is_dosage_relay`.
    """

InvalidPayloadException

Bases: Exception

Raised when a CSV response from the controller cannot be parsed.

Typically this means the response was empty, truncated, or did not have the expected number of CSV lines. Catching this lets callers distinguish protocol-level breakage from network errors.

Source code in src/proconip/definitions.py

class InvalidPayloadException(Exception):
    """Raised when a CSV response from the controller cannot be parsed.

    Typically this means the response was empty, truncated, or did not
    have the expected number of CSV lines. Catching this lets callers
    distinguish protocol-level breakage from network errors.
    """