Skip to content

API Reference

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

Top-level package

Async Python library for the Pooldigital Violet pool controller.

VioletClient

Async client for one Violet controller.

PARAMETER DESCRIPTION
session

A reusable aiohttp.ClientSession. The client does not own the session; the caller is responsible for closing it.

TYPE: ClientSession

host

Hostname or IP of the controller (e.g. "violet.local" or "violet.local:8080"). Must be a bare hostname/IP with an optional :port suffix — userinfo, paths, queries, and other URL components are rejected to prevent URL smuggling.

TYPE: str

username

Optional. Required for write endpoints and /getConfig.

TYPE: str | None DEFAULT: None

password

Optional. Required if username is set.

TYPE: str | None DEFAULT: None

timeout

Default per-request timeout in seconds.

TYPE: float DEFAULT: 10.0

scheme

"http" (default) or "https".

TYPE: str DEFAULT: 'http'

RAISES DESCRIPTION
ValueError

if host or scheme is malformed, or if exactly one of username / password is provided.

Source code in src/myviolet/client.py
class VioletClient:
    """Async client for one Violet controller.

    Args:
        session: A reusable `aiohttp.ClientSession`. The client does not own
            the session; the caller is responsible for closing it.
        host: Hostname or IP of the controller (e.g. ``"violet.local"`` or
            ``"violet.local:8080"``). Must be a bare hostname/IP with an
            optional ``:port`` suffix — userinfo, paths, queries, and other
            URL components are rejected to prevent URL smuggling.
        username: Optional. Required for write endpoints and `/getConfig`.
        password: Optional. Required if `username` is set.
        timeout: Default per-request timeout in seconds.
        scheme: ``"http"`` (default) or ``"https"``.

    Raises:
        ValueError: if `host` or `scheme` is malformed, or if exactly one of
            `username` / `password` is provided.
    """

    def __init__(
        self,
        session: aiohttp.ClientSession,
        host: str,
        *,
        username: str | None = None,
        password: str | None = None,
        timeout: float = 10.0,
        scheme: str = "http",
    ) -> None:
        scheme = _validate_scheme(scheme)
        hostname, port = _validate_host(host)
        if (username is None) != (password is None):
            raise ValueError("username and password must be provided together, or both omitted")
        auth: aiohttp.BasicAuth | None = None
        if username is not None and password is not None:
            auth = SafeBasicAuth(username, password)
        base_url = URL.build(scheme=scheme, host=hostname, port=port)
        self._transport = VioletTransport(session, base_url, auth=auth, default_timeout=timeout)

    async def __aenter__(self) -> VioletClient:
        """No-op. The aiohttp session lifecycle is the caller's responsibility.

        Implemented so callers can use ``async with VioletClient(...) as client``
        as a stylistic convention. If you add per-request retry pools or
        connection caches in the future, this is where setup/teardown belongs.
        """
        return self

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

    # ---- namespaces -------------------------------------------------------

    @cached_property
    def readings(self) -> _ReadingsNamespace:
        return _ReadingsNamespace(self._transport)

    @cached_property
    def control(self) -> _ControlNamespace:
        return _ControlNamespace(self._transport)

    @cached_property
    def targets(self) -> _TargetsNamespace:
        return _TargetsNamespace(self._transport)

    @cached_property
    def config(self) -> _ConfigNamespace:
        return _ConfigNamespace(self._transport)

    @cached_property
    def dosing_parameters(self) -> _DosingParametersNamespace:
        """`/setDosingParameters` writes. Distinct from `client.control.dosing`,
        which dispatches per-channel ON/OFF/AUTO commands."""
        return _DosingParametersNamespace(self._transport)

    @cached_property
    def history(self) -> _HistoryNamespace:
        return _HistoryNamespace(self._transport)

    @cached_property
    def calibration(self) -> _CalibrationNamespace:
        return _CalibrationNamespace(self._transport)

dosing_parameters cached property

dosing_parameters: _DosingParametersNamespace

/setDosingParameters writes. Distinct from client.control.dosing, which dispatches per-channel ON/OFF/AUTO commands.

CoverState

Bases: StrEnum

Position / motion state of the pool cover.

Source code in src/myviolet/enums.py
class CoverState(StrEnum):
    """Position / motion state of the pool cover."""

    OPEN = "OPEN"
    CLOSED = "CLOSED"
    OPENING = "OPENING"
    CLOSING = "CLOSING"
    STOPPED = "STOPPED"

DmxSceneState

Bases: IntEnum

State of a DMX scene. Subset of OutputState (no priority rules).

Source code in src/myviolet/enums.py
class DmxSceneState(IntEnum):
    """State of a DMX scene. Subset of `OutputState` (no priority rules)."""

    AUTO_OFF = 0
    AUTO_ON = 1
    MANUAL_ON = 4
    MANUAL_OFF = 6

DosingType

Bases: IntEnum

Configuration of a chlorine dosing controller.

Source code in src/myviolet/enums.py
class DosingType(IntEnum):
    """Configuration of a chlorine dosing controller."""

    ORP_ONLY = 0
    ORP_AND_CL = 1

OnewireState

Bases: StrEnum

Fault state of a 1-wire temperature sensor.

Note: DATA_MISSMATCH (sic) is spelled with two s letters to match the wire value emitted verbatim by the controller firmware.

Source code in src/myviolet/enums.py
class OnewireState(StrEnum):
    """Fault state of a 1-wire temperature sensor.

    Note: ``DATA_MISSMATCH`` (sic) is spelled with two ``s`` letters to match
    the wire value emitted verbatim by the controller firmware.
    """

    OK = "OK"
    CRC_FAULT = "CRC_FAULT"
    DATA_MISSMATCH = "DATA_MISSMATCH"
    NOT_CONNECTED = "NOT_CONNECTED"
    NO_SENSOR_CONFIGURED = "NO_SENSOR_CONFIGURED"

OutputState

Bases: IntEnum

State of any binary output controlled by the Violet controller.

These seven values are emitted by every relay-style output: pump, heater, solar, light, refill, eco, backwash, backwash rinse, the five dosing channels, and both extension relay buses.

Source code in src/myviolet/enums.py
class OutputState(IntEnum):
    """State of any binary output controlled by the Violet controller.

    These seven values are emitted by every relay-style output: pump, heater,
    solar, light, refill, eco, backwash, backwash rinse, the five dosing
    channels, and both extension relay buses.
    """

    AUTO_OFF = 0
    AUTO_ON = 1
    AUTO_PRIO_OFF = 2
    AUTO_PRIO_ON = 3
    MANUAL_ON = 4
    EMERGENCY_OFF = 5
    MANUAL_OFF = 6

    @property
    def is_on(self) -> bool:
        """Whether the output is currently energised, regardless of cause."""
        return self in (OutputState.AUTO_ON, OutputState.AUTO_PRIO_ON, OutputState.MANUAL_ON)

    @property
    def is_manual(self) -> bool:
        """Whether the output is in a user-forced (manual) state."""
        return self in (OutputState.MANUAL_ON, OutputState.MANUAL_OFF)

    @property
    def is_emergency(self) -> bool:
        """Whether an emergency or priority rule is currently in effect."""
        return self in (
            OutputState.AUTO_PRIO_OFF,
            OutputState.AUTO_PRIO_ON,
            OutputState.EMERGENCY_OFF,
        )

is_on property

is_on: bool

Whether the output is currently energised, regardless of cause.

is_manual property

is_manual: bool

Whether the output is in a user-forced (manual) state.

is_emergency property

is_emergency: bool

Whether an emergency or priority rule is currently in effect.

PvSurplusState

Bases: IntEnum

Trigger source of the PV surplus function.

Source code in src/myviolet/enums.py
class PvSurplusState(IntEnum):
    """Trigger source of the PV surplus function."""

    OFF = 0
    ON_BY_INPUT = 1
    ON_BY_HTTP = 2

RuleState

Bases: IntEnum

State of a digital-input switching rule.

Source code in src/myviolet/enums.py
class RuleState(IntEnum):
    """State of a digital-input switching rule."""

    INACTIVE = 0
    ACTIVE = 1
    BLOCKED_BY_RULE = 5
    BLOCKED_MANUALLY = 6

SimpleOnOff

Bases: StrEnum

Binary on/off used by overflow and bathing-AI flags.

Source code in src/myviolet/enums.py
class SimpleOnOff(StrEnum):
    """Binary on/off used by overflow and bathing-AI flags."""

    ON = "ON"
    OFF = "OFF"

YesNo

Bases: StrEnum

Binary yes/no used by various status flags.

Source code in src/myviolet/enums.py
class YesNo(StrEnum):
    """Binary yes/no used by various status flags."""

    YES = "YES"
    NO = "NO"

BadCredentialsException

Bases: VioletApiException

The controller rejected the supplied credentials (HTTP 401 or 403).

Source code in src/myviolet/exceptions.py
class BadCredentialsException(VioletApiException):
    """The controller rejected the supplied credentials (HTTP 401 or 403)."""

    def __init__(self, status_code: int) -> None:
        self.status_code = status_code
        super().__init__(f"authentication failed (HTTP {status_code})")

BadStatusCodeException

Bases: VioletApiException

The controller returned an unexpected non-auth error HTTP status.

Source code in src/myviolet/exceptions.py
class BadStatusCodeException(VioletApiException):
    """The controller returned an unexpected non-auth error HTTP status."""

    def __init__(self, status_code: int, message: str = "") -> None:
        self.status_code = status_code
        suffix = f": {message}" if message else ""
        super().__init__(f"unexpected HTTP {status_code}{suffix}")

InvalidPayloadException

Bases: VioletApiException

The controller's response was malformed, truncated, or unparseable.

Source code in src/myviolet/exceptions.py
class InvalidPayloadException(VioletApiException):
    """The controller's response was malformed, truncated, or unparseable."""

SetpointValidationError

Bases: VioletApiException, ValueError

A setpoint value was outside its documented valid range.

Subclasses both VioletApiException (for blanket library catches) and ValueError (so generic input-validation code keeps working).

Source code in src/myviolet/exceptions.py
class SetpointValidationError(VioletApiException, ValueError):
    """A setpoint value was outside its documented valid range.

    Subclasses both `VioletApiException` (for blanket library catches) and
    `ValueError` (so generic input-validation code keeps working).
    """

    def __init__(self, field: str, value: float, low: float, high: float) -> None:
        self.field = field
        self.value = value
        self.low = low
        self.high = high
        super().__init__(f"{field} setpoint {value} is outside the valid range [{low}, {high}]")

TimeoutException

Bases: VioletApiException

An HTTP request to the controller exceeded the configured timeout.

Source code in src/myviolet/exceptions.py
class TimeoutException(VioletApiException):
    """An HTTP request to the controller exceeded the configured timeout."""

UnsafeOperationException

Bases: VioletApiException

A potentially unsafe operation was attempted without acknowledgment.

Currently raised by cover open/close calls when the caller has not passed acknowledge_unsafe=True.

Source code in src/myviolet/exceptions.py
class UnsafeOperationException(VioletApiException):
    """A potentially unsafe operation was attempted without acknowledgment.

    Currently raised by cover open/close calls when the caller has not passed
    ``acknowledge_unsafe=True``.
    """

    def __init__(self, operation: str) -> None:
        super().__init__(
            f"{operation!r} is potentially unsafe; pass acknowledge_unsafe=True "
            f"to confirm the caller has implemented external safety logic"
        )

VioletApiException

Bases: Exception

Base for every error raised by the myviolet client.

Source code in src/myviolet/exceptions.py
class VioletApiException(Exception):
    """Base for every error raised by the myviolet client."""

HardwareProfile dataclass

A summary of which controller hardware is present and configured.

Source code in src/myviolet/hardware.py
@dataclass(frozen=True, slots=True)
class HardwareProfile:
    """A summary of which controller hardware is present and configured."""

    has_pump: bool
    has_heater: bool
    has_solar: bool
    has_light: bool
    has_refill: bool
    has_eco: bool
    has_backwash: bool
    has_cover: bool
    has_pv_surplus: bool
    has_dmx: bool
    onewire_active_indices: frozenset[int] = field(default_factory=frozenset)
    dosing_channel_codes: frozenset[str] = field(default_factory=frozenset)
    extension_buses: frozenset[int] = field(default_factory=frozenset)
    extension_relays_per_bus: Mapping[int, int] = _EMPTY_INT_INT_MAP
    digital_input_count: int = 0
    can_empty_input_count: int = 0
    analog_sensor_count: int = 0

    @classmethod
    def from_raw(cls, raw: dict[str, Any]) -> HardwareProfile:
        return cls(
            has_pump="PUMP" in raw,
            has_heater="HEATER" in raw,
            has_solar="SOLAR" in raw,
            has_light="LIGHT" in raw,
            has_refill="REFILL" in raw,
            has_eco="ECO" in raw,
            has_backwash="BACKWASH" in raw,
            has_cover="COVER_STATE" in raw,
            has_pv_surplus="PVSURPLUS" in raw,
            has_dmx=any(f"DMX_SCENE{i}" in raw for i in range(1, 13)),
            onewire_active_indices=_active_onewire_indices(raw),
            dosing_channel_codes=_present_dosing_codes(raw),
            extension_buses=_present_extension_buses(raw),
            extension_relays_per_bus=_extension_relay_counts(raw),
            digital_input_count=sum(1 for i in range(1, 13) if f"INPUT{i}" in raw),
            can_empty_input_count=sum(1 for i in range(1, 5) if f"INPUT_CE{i}" in raw),
            analog_sensor_count=sum(1 for i in range(1, 7) if f"ADC{i}_value" in raw),
        )

BackwashStatus dataclass

Backwash-process status flags plus the free-form raw state string.

The vendor's BACKWASH_STATE field mixes several different message types (e.g. NEXT_BW_IN 2 BW_DAY5, BW_RUNNING_SINCE 123). Rather than try to parse all variants, we expose it verbatim as raw_state.

Source code in src/myviolet/models/system_states.py
@dataclass(frozen=True, slots=True)
class BackwashStatus:
    """Backwash-process status flags plus the free-form raw state string.

    The vendor's `BACKWASH_STATE` field mixes several different message types
    (e.g. `NEXT_BW_IN 2 BW_DAY5`, `BW_RUNNING_SINCE 123`). Rather than try to
    parse all variants, we expose it verbatim as `raw_state`.
    """

    delay_running: bool
    delay_started_at: datetime | None
    last_auto_run: datetime | None
    last_manual_run: datetime | None
    omni_moving: YesNo | None
    omni_state: str | None
    step: int
    raw_state: str

    @classmethod
    def from_raw(cls, raw: dict[str, Any]) -> BackwashStatus | None:
        if "BACKWASH_STATE" not in raw:
            return None
        return cls(
            delay_running=_parse_yesno(raw.get("BACKWASH_DELAY_RUNNING", "NO")) is YesNo.YES,
            delay_started_at=parse_epoch_seconds(raw.get("BACKWASH_DELAY_TIMESTAMP")),
            last_auto_run=parse_epoch_seconds(raw.get("BACKWASH_LAST_AUTO_RUN")),
            last_manual_run=parse_epoch_seconds(raw.get("BACKWASH_LAST_MANUAL_RUN")),
            omni_moving=_parse_yesno(raw.get("BACKWASH_OMNI_MOVING")),
            omni_state=raw.get("BACKWASH_OMNI_STATE"),
            step=int(raw.get("BACKWASH_STEP", 0)),
            raw_state=str(raw["BACKWASH_STATE"]),
        )

BathingAi dataclass

Bathing-AI overflow-vessel surveillance state.

The vendor's spec documents surveillance_state as YES/NO; the live controller emits a third value (TRIGGERED) too, so we keep it as a plain string instead of a strict enum.

Source code in src/myviolet/models/system_states.py
@dataclass(frozen=True, slots=True)
class BathingAi:
    """Bathing-AI overflow-vessel surveillance state.

    The vendor's spec documents `surveillance_state` as YES/NO; the live
    controller emits a third value (`TRIGGERED`) too, so we keep it as a
    plain string instead of a strict enum.
    """

    surveillance_state: str
    start_level: float
    last_level: float
    pump_state: SimpleOnOff | None
    pump_started_at: datetime | None

    @classmethod
    def from_raw(cls, raw: dict[str, Any]) -> BathingAi | None:
        if "BATHING_AI_SURVEILLANCE_STATE" not in raw:
            return None
        return cls(
            surveillance_state=str(raw["BATHING_AI_SURVEILLANCE_STATE"]),
            start_level=float(raw.get("BATHING_AI_START_LEVEL", 0.0)),
            last_level=float(raw.get("BATHING_AI_LAST_LEVEL", 0.0)),
            pump_state=_parse_simple_onoff(raw.get("BATHING_AI_PUMP_STATE", "OFF")),
            pump_started_at=parse_epoch_seconds(raw.get("BATHING_AI_PUMP_TIMESTAMP")),
        )

Cover dataclass

Pool cover state (read-only; control is via client.control.cover_*).

Source code in src/myviolet/models/cover.py
@dataclass(frozen=True, slots=True)
class Cover:
    """Pool cover state (read-only; control is via `client.control.cover_*`)."""

    state: CoverState

    @classmethod
    def from_raw(cls, raw: dict[str, Any]) -> Cover | None:
        """Build from `COVER_STATE`. Unknown firmware values return ``None``."""
        state_raw = raw.get("COVER_STATE")
        if state_raw is None:
            return None
        try:
            return cls(state=CoverState(str(state_raw)))
        except ValueError:
            return None

from_raw classmethod

from_raw(raw: dict[str, Any]) -> Cover | None

Build from COVER_STATE. Unknown firmware values return None.

Source code in src/myviolet/models/cover.py
@classmethod
def from_raw(cls, raw: dict[str, Any]) -> Cover | None:
    """Build from `COVER_STATE`. Unknown firmware values return ``None``."""
    state_raw = raw.get("COVER_STATE")
    if state_raw is None:
        return None
    try:
        return cls(state=CoverState(str(state_raw)))
    except ValueError:
        return None

DmxScene dataclass

One DMX scene (12 documented). The subset enum forbids priority states.

Source code in src/myviolet/models/dmx.py
@dataclass(frozen=True, slots=True)
class DmxScene:
    """One DMX scene (12 documented). The subset enum forbids priority states."""

    index: int
    state: DmxSceneState

DosingChannel dataclass

One dosing channel. Optional DOSAGE-keyword fields default to None.

Source code in src/myviolet/models/dosing.py
@dataclass(frozen=True, slots=True)
class DosingChannel:
    """One dosing channel. Optional DOSAGE-keyword fields default to `None`."""

    code: str
    channel_number: int
    state: OutputState
    last_on: datetime | None
    last_off: datetime | None
    runtime: timedelta
    daily_amount_ml: int | None = None
    total_can_amount_ml: int | None = None
    last_can_reset: datetime | None = None
    enabled: bool | None = None
    type: DosingType | None = None
    state_blocks: tuple[str, ...] = ()

ExtensionRelay dataclass

One relay on a relay-extension board. Bus 1 is documented; bus 2 is observed.

Source code in src/myviolet/models/extension.py
@dataclass(frozen=True, slots=True)
class ExtensionRelay:
    """One relay on a relay-extension board. Bus 1 is documented; bus 2 is observed."""

    bus: int
    index: int
    state: OutputState
    last_on: datetime | None
    last_off: datetime | None
    runtime: timedelta

Heater dataclass

Heater output, including the optional postrun-timer remaining time.

Source code in src/myviolet/models/outputs.py
@dataclass(frozen=True, slots=True)
class Heater:
    """Heater output, including the optional postrun-timer remaining time."""

    state: OutputState
    last_on: datetime | None
    last_off: datetime | None
    runtime: timedelta
    postrun_remaining: timedelta | None = None

    @classmethod
    def from_raw(cls, raw: dict[str, Any]) -> Heater | None:
        base = OutputBase.from_raw(raw, "HEATER")
        if base is None:
            return None
        return cls(
            state=base.state,
            last_on=base.last_on,
            last_off=base.last_off,
            runtime=base.runtime,
            postrun_remaining=parse_optional_seconds(raw.get("HEATER_POSTRUN_TIME")),
        )

MeasuredValue dataclass

A sensor reading plus its daily min/max envelope.

Many sensor categories (water chemistry, 1-wire temperatures, analog inputs) expose <KEY>_value, <KEY>_value_min, <KEY>_value_max triplets. MeasuredValue is the typed wrapper.

Source code in src/myviolet/models/_common.py
@dataclass(frozen=True, slots=True)
class MeasuredValue:
    """A sensor reading plus its daily min/max envelope.

    Many sensor categories (water chemistry, 1-wire temperatures, analog
    inputs) expose ``<KEY>_value``, ``<KEY>_value_min``, ``<KEY>_value_max``
    triplets. `MeasuredValue` is the typed wrapper.
    """

    value: float
    unit: str
    min: float | None = None
    max: float | None = None

    @classmethod
    def from_raw(
        cls,
        raw: dict[str, Any],
        key_base: str,
        *,
        unit: str,
    ) -> MeasuredValue | None:
        """Assemble from `<key_base>_value`/`_value_min`/`_value_max` siblings.

        Returns ``None`` when the canonical ``<key_base>_value`` key is absent,
        which lets callers distinguish "sensor not installed" from "value zero".
        """
        value = raw.get(f"{key_base}_value")
        if value is None:
            return None
        return cls(
            value=float(value),
            unit=unit,
            min=_maybe_float(raw.get(f"{key_base}_value_min")),
            max=_maybe_float(raw.get(f"{key_base}_value_max")),
        )

from_raw classmethod

from_raw(raw: dict[str, Any], key_base: str, *, unit: str) -> MeasuredValue | None

Assemble from <key_base>_value/_value_min/_value_max siblings.

Returns None when the canonical <key_base>_value key is absent, which lets callers distinguish "sensor not installed" from "value zero".

Source code in src/myviolet/models/_common.py
@classmethod
def from_raw(
    cls,
    raw: dict[str, Any],
    key_base: str,
    *,
    unit: str,
) -> MeasuredValue | None:
    """Assemble from `<key_base>_value`/`_value_min`/`_value_max` siblings.

    Returns ``None`` when the canonical ``<key_base>_value`` key is absent,
    which lets callers distinguish "sensor not installed" from "value zero".
    """
    value = raw.get(f"{key_base}_value")
    if value is None:
        return None
    return cls(
        value=float(value),
        unit=unit,
        min=_maybe_float(raw.get(f"{key_base}_value_min")),
        max=_maybe_float(raw.get(f"{key_base}_value_max")),
    )

OneWireSensor dataclass

A single 1-wire temperature sensor (12 slots, only configured ones).

Source code in src/myviolet/models/sensors.py
@dataclass(frozen=True, slots=True)
class OneWireSensor:
    """A single 1-wire temperature sensor (12 slots, only configured ones)."""

    index: int
    state: OnewireState
    rom_code: str
    value: float
    unit: str
    value_min: float | None = None
    value_max: float | None = None

    @property
    def measured(self) -> MeasuredValue:
        """Expose the reading as a `MeasuredValue` for uniform processing."""
        return MeasuredValue(
            value=self.value,
            unit=self.unit,
            min=self.value_min,
            max=self.value_max,
        )

measured property

measured: MeasuredValue

Expose the reading as a MeasuredValue for uniform processing.

OutputBase dataclass

Shared shape for the ~30 binary outputs (pump, heater, relays, ...).

Every documented binary output (PUMP, HEATER, SOLAR, LIGHT, REFILL, ECO, BACKWASH, BACKWASHRINSE, EXT1_, EXT2_) emits the same four siblings: a state integer, two epoch-seconds timestamps, and a runtime string.

Source code in src/myviolet/models/_common.py
@dataclass(frozen=True, slots=True)
class OutputBase:
    """Shared shape for the ~30 binary outputs (pump, heater, relays, ...).

    Every documented binary output (PUMP, HEATER, SOLAR, LIGHT, REFILL, ECO,
    BACKWASH, BACKWASHRINSE, EXT1_*, EXT2_*) emits the same four siblings:
    a state integer, two epoch-seconds timestamps, and a runtime string.
    """

    state: OutputState
    last_on: datetime | None
    last_off: datetime | None
    runtime: timedelta

    @classmethod
    def from_raw(cls, raw: dict[str, Any], key_base: str) -> OutputBase | None:
        """Assemble from ``<key_base>``, ``_LAST_ON``, ``_LAST_OFF``, ``_RUNTIME``.

        Returns ``None`` when the canonical state key is absent (output not
        installed) **or** when the controller emits an `OutputState` value
        not in our documented enum — that lets the typed view layer stay
        forward-compatible with future firmware revisions instead of
        crashing the entire snapshot.
        """
        state_raw = raw.get(key_base)
        if state_raw is None:
            return None
        try:
            state = OutputState(int(state_raw))
        except (ValueError, TypeError):
            return None
        runtime_raw = raw.get(f"{key_base}_RUNTIME", "00h 00m 00s")
        try:
            runtime = parse_runtime_string(runtime_raw)
        except ValueError:
            runtime = timedelta(0)
        return cls(
            state=state,
            last_on=parse_epoch_seconds(raw.get(f"{key_base}_LAST_ON")),
            last_off=parse_epoch_seconds(raw.get(f"{key_base}_LAST_OFF")),
            runtime=runtime,
        )

from_raw classmethod

from_raw(raw: dict[str, Any], key_base: str) -> OutputBase | None

Assemble from <key_base>, _LAST_ON, _LAST_OFF, _RUNTIME.

Returns None when the canonical state key is absent (output not installed) or when the controller emits an OutputState value not in our documented enum — that lets the typed view layer stay forward-compatible with future firmware revisions instead of crashing the entire snapshot.

Source code in src/myviolet/models/_common.py
@classmethod
def from_raw(cls, raw: dict[str, Any], key_base: str) -> OutputBase | None:
    """Assemble from ``<key_base>``, ``_LAST_ON``, ``_LAST_OFF``, ``_RUNTIME``.

    Returns ``None`` when the canonical state key is absent (output not
    installed) **or** when the controller emits an `OutputState` value
    not in our documented enum — that lets the typed view layer stay
    forward-compatible with future firmware revisions instead of
    crashing the entire snapshot.
    """
    state_raw = raw.get(key_base)
    if state_raw is None:
        return None
    try:
        state = OutputState(int(state_raw))
    except (ValueError, TypeError):
        return None
    runtime_raw = raw.get(f"{key_base}_RUNTIME", "00h 00m 00s")
    try:
        runtime = parse_runtime_string(runtime_raw)
    except ValueError:
        runtime = timedelta(0)
    return cls(
        state=state,
        last_on=parse_epoch_seconds(raw.get(f"{key_base}_LAST_ON")),
        last_off=parse_epoch_seconds(raw.get(f"{key_base}_LAST_OFF")),
        runtime=runtime,
    )

OverflowState dataclass

Overflow-vessel control flags (dryrun, overfill, refill triggers).

Source code in src/myviolet/models/system_states.py
@dataclass(frozen=True, slots=True)
class OverflowState:
    """Overflow-vessel control flags (dryrun, overfill, refill triggers)."""

    dryrun: SimpleOnOff
    overfill: SimpleOnOff
    refill: SimpleOnOff

    @classmethod
    def from_raw(cls, raw: dict[str, Any]) -> OverflowState | None:
        if "OVERFLOW_DRYRUN_STATE" not in raw:
            return None
        try:
            return cls(
                dryrun=SimpleOnOff(raw["OVERFLOW_DRYRUN_STATE"]),
                overfill=SimpleOnOff(raw.get("OVERFLOW_OVERFILL_STATE", "OFF")),
                refill=SimpleOnOff(raw.get("OVERFLOW_REFILL_STATE", "OFF")),
            )
        except ValueError:
            # Forward-compat: an unknown firmware value on any flag drops the
            # whole snapshot rather than half-populating it, matching the
            # `Cover.from_raw` / `OutputBase.from_raw` pattern.
            return None

Pump dataclass

Main pump output plus its 4 speed sub-outputs.

Source code in src/myviolet/models/outputs.py
@dataclass(frozen=True, slots=True)
class Pump:
    """Main pump output plus its 4 speed sub-outputs."""

    state: OutputState
    last_on: datetime | None
    last_off: datetime | None
    runtime: timedelta
    speeds: Mapping[int, PumpSpeed] = field(default_factory=lambda: _EMPTY_PUMP_SPEEDS)

    @classmethod
    def from_raw(cls, raw: dict[str, Any]) -> Pump | None:
        base = OutputBase.from_raw(raw, "PUMP")
        if base is None:
            return None
        speeds: dict[int, PumpSpeed] = {}
        for rpm in range(4):
            state_raw = raw.get(f"PUMP_RPM_{rpm}")
            if state_raw is None:
                continue
            try:
                state = OutputState(int(state_raw))
            except ValueError:
                # Forward-compat: skip unknown firmware enum values rather than crash.
                continue
            speeds[rpm] = PumpSpeed(
                rpm=rpm,
                state=state,
                runtime=parse_runtime_string(raw.get(f"PUMP_RPM_{rpm}_RUNTIME", "00h 00m 00s")),
            )
        return cls(
            state=base.state,
            last_on=base.last_on,
            last_off=base.last_off,
            runtime=base.runtime,
            speeds=MappingProxyType(speeds),
        )

PumpSpeed dataclass

One of the pump's 4 speed sub-outputs (RPM 0=stop, 1-3=speeds).

The vendor documents PUMP_RPM_*_LAST_ON/OFF as always "00:00:00" and not useful, so they are surfaced as None rather than parsed.

Source code in src/myviolet/models/outputs.py
@dataclass(frozen=True, slots=True)
class PumpSpeed:
    """One of the pump's 4 speed sub-outputs (RPM 0=stop, 1-3=speeds).

    The vendor documents `PUMP_RPM_*_LAST_ON/OFF` as always `"00:00:00"` and
    not useful, so they are surfaced as `None` rather than parsed.
    """

    rpm: int
    state: OutputState
    runtime: timedelta
    last_on: datetime | None = None
    last_off: datetime | None = None

PvSurplus dataclass

PV-surplus function: off, on-by-input, or on-by-HTTP.

Source code in src/myviolet/models/system_states.py
@dataclass(frozen=True, slots=True)
class PvSurplus:
    """PV-surplus function: off, on-by-input, or on-by-HTTP."""

    state: PvSurplusState

    @classmethod
    def from_raw(cls, raw: dict[str, Any]) -> PvSurplus | None:
        value = raw.get("PVSURPLUS")
        if value is None:
            return None
        try:
            return cls(state=PvSurplusState(int(value)))
        except ValueError:
            # Forward-compat: unknown firmware enum value drops the snapshot
            # rather than crashing parsing of unrelated readings.
            return None

SystemInfo dataclass

Top-level system metadata from /getReadings.

Covers the spec's "System informations" section plus a few undocumented extras the live demo controller emits (HW_*_CARRIER, MEMORY_USED, CURRENT_TIME_UNIX).

Source code in src/myviolet/models/system.py
@dataclass(frozen=True, slots=True)
class SystemInfo:
    """Top-level system metadata from `/getReadings`.

    Covers the spec's "System informations" section plus a few undocumented
    extras the live demo controller emits (`HW_*_CARRIER`, `MEMORY_USED`,
    `CURRENT_TIME_UNIX`).
    """

    date: str
    time: str
    cpu_temp: float
    cpu_temp_carrier: float
    uptime: timedelta
    system_memory_mb: float
    sw_version: str
    sw_version_carrier: str
    hw_version_carrier: str | None = None
    hw_serial_carrier: str | None = None
    current_time: datetime | None = None

    @classmethod
    def from_raw(cls, raw: dict[str, Any]) -> SystemInfo:
        return cls(
            date=str(raw.get("date", "")),
            time=str(raw.get("time", "")),
            cpu_temp=float(raw.get("CPU_TEMP", 0.0)),
            cpu_temp_carrier=float(raw.get("CPU_TEMP_CARRIER", 0.0)),
            uptime=parse_uptime_string(raw.get("CPU_UPTIME", "0d 0h 0m")),
            system_memory_mb=float(raw.get("SYSTEM_MEMORY", 0.0)),
            sw_version=str(raw.get("SW_VERSION", "")),
            sw_version_carrier=str(raw.get("SW_VERSION_CARRIER", "")),
            hw_version_carrier=_maybe_str(raw.get("HW_VERSION_CARRIER")),
            hw_serial_carrier=_maybe_str(raw.get("HW_SERIAL_CARRIER")),
            current_time=parse_epoch_seconds(raw.get("CURRENT_TIME_UNIX")),
        )

WaterChemistry dataclass

Pool water chemistry readings (pH, ORP, free chlorine).

Source code in src/myviolet/models/chemistry.py
@dataclass(frozen=True, slots=True)
class WaterChemistry:
    """Pool water chemistry readings (pH, ORP, free chlorine)."""

    ph: MeasuredValue | None
    orp: MeasuredValue | None
    chlorine: MeasuredValue | None

    @classmethod
    def from_raw(cls, raw: dict[str, Any]) -> WaterChemistry:
        return cls(
            ph=MeasuredValue.from_raw(raw, "pH", unit="pH"),
            orp=MeasuredValue.from_raw(raw, "orp", unit="mV"),
            chlorine=MeasuredValue.from_raw(raw, "pot", unit="mg/L"),
        )

VioletReadings

Typed read-only view of one /getReadings snapshot.

Wraps the parsed JSON dict from the controller. Every typed accessor is lazy and memoized; the underlying dict remains accessible as .raw so callers can still reach undocumented or firmware-specific keys.

Source code in src/myviolet/readings.py
class VioletReadings:
    """Typed read-only view of one `/getReadings` snapshot.

    Wraps the parsed JSON dict from the controller. Every typed accessor is
    lazy and memoized; the underlying dict remains accessible as `.raw` so
    callers can still reach undocumented or firmware-specific keys.
    """

    def __init__(self, raw: dict[str, Any]) -> None:
        # Defensive copy: the caller may still hold a reference to `raw` and
        # later mutate it; without a copy, that would silently corrupt the
        # memoized typed views on this snapshot. Shallow is enough — the
        # /getReadings payload is a flat dict of JSON scalars.
        self._raw = dict(raw)

    @property
    def raw(self) -> Mapping[str, Any]:
        """Read-only view of the original `/getReadings` JSON dict.

        Returned as a `MappingProxyType` so callers can read any key
        (including undocumented or firmware-future ones) but cannot mutate
        the underlying dict in a way that would corrupt cached typed views.
        """
        return MappingProxyType(self._raw)

    # ---- system & metadata ------------------------------------------------

    @cached_property
    def system_info(self) -> SystemInfo:
        return SystemInfo.from_raw(self._raw)

    @cached_property
    def hardware_profile(self) -> HardwareProfile:
        return HardwareProfile.from_raw(self._raw)

    # ---- sensors ----------------------------------------------------------

    @cached_property
    def analog_sensors(self) -> dict[int, AnalogSensor]:
        return collect_analog_sensors(self._raw)

    @cached_property
    def impulse_inputs(self) -> dict[int, ImpulseInput]:
        return collect_impulse_inputs(self._raw)

    @cached_property
    def onewire_sensors(self) -> dict[int, OneWireSensor]:
        return collect_onewire_sensors(self._raw)

    @cached_property
    def water_chemistry(self) -> WaterChemistry:
        return WaterChemistry.from_raw(self._raw)

    # ---- outputs ----------------------------------------------------------

    @cached_property
    def pump(self) -> Pump | None:
        return Pump.from_raw(self._raw)

    @cached_property
    def heater(self) -> Heater | None:
        return Heater.from_raw(self._raw)

    @cached_property
    def solar(self) -> Solar | None:
        return Solar.from_raw(self._raw)

    @cached_property
    def light(self) -> Light | None:
        return Light.from_raw(self._raw)

    @cached_property
    def refill(self) -> Refill | None:
        return Refill.from_raw(self._raw)

    @cached_property
    def eco(self) -> Eco | None:
        return Eco.from_raw(self._raw)

    @cached_property
    def backwash(self) -> Backwash | None:
        return Backwash.from_raw(self._raw)

    @cached_property
    def backwash_rinse(self) -> BackwashRinse | None:
        return BackwashRinse.from_raw(self._raw)

    # ---- cover, dmx, extension, dosing -----------------------------------

    @cached_property
    def cover(self) -> Cover | None:
        return Cover.from_raw(self._raw)

    @cached_property
    def dmx_scenes(self) -> dict[int, DmxScene]:
        return collect_dmx_scenes(self._raw)

    @cached_property
    def extension_relays(self) -> dict[int, dict[int, ExtensionRelay]]:
        return collect_extension_relays(self._raw)

    @cached_property
    def dosing_channels(self) -> dict[str, DosingChannel]:
        return collect_dosing_channels(self._raw)

    # ---- inputs & rules ---------------------------------------------------

    @cached_property
    def digital_inputs(self) -> dict[int, DigitalInput]:
        return collect_digital_inputs(self._raw)

    @cached_property
    def can_empty_inputs(self) -> dict[int, CanEmptyInput]:
        return collect_can_empty_inputs(self._raw)

    @cached_property
    def switching_rules(self) -> dict[int, SwitchingRule]:
        return collect_switching_rules(self._raw)

    # ---- mixed system states ---------------------------------------------

    @cached_property
    def overflow(self) -> OverflowState | None:
        return OverflowState.from_raw(self._raw)

    @cached_property
    def backwash_status(self) -> BackwashStatus | None:
        return BackwashStatus.from_raw(self._raw)

    @cached_property
    def bathing_ai(self) -> BathingAi | None:
        return BathingAi.from_raw(self._raw)

    @cached_property
    def pv_surplus(self) -> PvSurplus | None:
        return PvSurplus.from_raw(self._raw)

raw property

raw: Mapping[str, Any]

Read-only view of the original /getReadings JSON dict.

Returned as a MappingProxyType so callers can read any key (including undocumented or firmware-future ones) but cannot mutate the underlying dict in a way that would corrupt cached typed views.

Async client (myviolet.client)

client

VioletClient — the public async client for the Violet pool controller.

The client is constructed once per controller and exposes domain-grouped namespaces (readings, control, targets, config, dosing_parameters, history, calibration). Per-channel dosing ON/OFF/AUTO lives under client.control.dosing; client.dosing_parameters handles /setDosingParameters writes. All namespaces share a single VioletTransport instance, so retries / observability hooks can be added in one place.

VioletClient

Async client for one Violet controller.

PARAMETER DESCRIPTION
session

A reusable aiohttp.ClientSession. The client does not own the session; the caller is responsible for closing it.

TYPE: ClientSession

host

Hostname or IP of the controller (e.g. "violet.local" or "violet.local:8080"). Must be a bare hostname/IP with an optional :port suffix — userinfo, paths, queries, and other URL components are rejected to prevent URL smuggling.

TYPE: str

username

Optional. Required for write endpoints and /getConfig.

TYPE: str | None DEFAULT: None

password

Optional. Required if username is set.

TYPE: str | None DEFAULT: None

timeout

Default per-request timeout in seconds.

TYPE: float DEFAULT: 10.0

scheme

"http" (default) or "https".

TYPE: str DEFAULT: 'http'

RAISES DESCRIPTION
ValueError

if host or scheme is malformed, or if exactly one of username / password is provided.

Source code in src/myviolet/client.py
class VioletClient:
    """Async client for one Violet controller.

    Args:
        session: A reusable `aiohttp.ClientSession`. The client does not own
            the session; the caller is responsible for closing it.
        host: Hostname or IP of the controller (e.g. ``"violet.local"`` or
            ``"violet.local:8080"``). Must be a bare hostname/IP with an
            optional ``:port`` suffix — userinfo, paths, queries, and other
            URL components are rejected to prevent URL smuggling.
        username: Optional. Required for write endpoints and `/getConfig`.
        password: Optional. Required if `username` is set.
        timeout: Default per-request timeout in seconds.
        scheme: ``"http"`` (default) or ``"https"``.

    Raises:
        ValueError: if `host` or `scheme` is malformed, or if exactly one of
            `username` / `password` is provided.
    """

    def __init__(
        self,
        session: aiohttp.ClientSession,
        host: str,
        *,
        username: str | None = None,
        password: str | None = None,
        timeout: float = 10.0,
        scheme: str = "http",
    ) -> None:
        scheme = _validate_scheme(scheme)
        hostname, port = _validate_host(host)
        if (username is None) != (password is None):
            raise ValueError("username and password must be provided together, or both omitted")
        auth: aiohttp.BasicAuth | None = None
        if username is not None and password is not None:
            auth = SafeBasicAuth(username, password)
        base_url = URL.build(scheme=scheme, host=hostname, port=port)
        self._transport = VioletTransport(session, base_url, auth=auth, default_timeout=timeout)

    async def __aenter__(self) -> VioletClient:
        """No-op. The aiohttp session lifecycle is the caller's responsibility.

        Implemented so callers can use ``async with VioletClient(...) as client``
        as a stylistic convention. If you add per-request retry pools or
        connection caches in the future, this is where setup/teardown belongs.
        """
        return self

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

    # ---- namespaces -------------------------------------------------------

    @cached_property
    def readings(self) -> _ReadingsNamespace:
        return _ReadingsNamespace(self._transport)

    @cached_property
    def control(self) -> _ControlNamespace:
        return _ControlNamespace(self._transport)

    @cached_property
    def targets(self) -> _TargetsNamespace:
        return _TargetsNamespace(self._transport)

    @cached_property
    def config(self) -> _ConfigNamespace:
        return _ConfigNamespace(self._transport)

    @cached_property
    def dosing_parameters(self) -> _DosingParametersNamespace:
        """`/setDosingParameters` writes. Distinct from `client.control.dosing`,
        which dispatches per-channel ON/OFF/AUTO commands."""
        return _DosingParametersNamespace(self._transport)

    @cached_property
    def history(self) -> _HistoryNamespace:
        return _HistoryNamespace(self._transport)

    @cached_property
    def calibration(self) -> _CalibrationNamespace:
        return _CalibrationNamespace(self._transport)

dosing_parameters cached property

dosing_parameters: _DosingParametersNamespace

/setDosingParameters writes. Distinct from client.control.dosing, which dispatches per-channel ON/OFF/AUTO commands.

Readings & typed views (myviolet.readings)

readings

VioletReadings — the typed facade over a raw /getReadings JSON dict.

All typed views are computed lazily and memoized via cached_property so that repeatedly accessing the same view on a snapshot is free and returns the same object identity.

VioletReadings

Typed read-only view of one /getReadings snapshot.

Wraps the parsed JSON dict from the controller. Every typed accessor is lazy and memoized; the underlying dict remains accessible as .raw so callers can still reach undocumented or firmware-specific keys.

Source code in src/myviolet/readings.py
class VioletReadings:
    """Typed read-only view of one `/getReadings` snapshot.

    Wraps the parsed JSON dict from the controller. Every typed accessor is
    lazy and memoized; the underlying dict remains accessible as `.raw` so
    callers can still reach undocumented or firmware-specific keys.
    """

    def __init__(self, raw: dict[str, Any]) -> None:
        # Defensive copy: the caller may still hold a reference to `raw` and
        # later mutate it; without a copy, that would silently corrupt the
        # memoized typed views on this snapshot. Shallow is enough — the
        # /getReadings payload is a flat dict of JSON scalars.
        self._raw = dict(raw)

    @property
    def raw(self) -> Mapping[str, Any]:
        """Read-only view of the original `/getReadings` JSON dict.

        Returned as a `MappingProxyType` so callers can read any key
        (including undocumented or firmware-future ones) but cannot mutate
        the underlying dict in a way that would corrupt cached typed views.
        """
        return MappingProxyType(self._raw)

    # ---- system & metadata ------------------------------------------------

    @cached_property
    def system_info(self) -> SystemInfo:
        return SystemInfo.from_raw(self._raw)

    @cached_property
    def hardware_profile(self) -> HardwareProfile:
        return HardwareProfile.from_raw(self._raw)

    # ---- sensors ----------------------------------------------------------

    @cached_property
    def analog_sensors(self) -> dict[int, AnalogSensor]:
        return collect_analog_sensors(self._raw)

    @cached_property
    def impulse_inputs(self) -> dict[int, ImpulseInput]:
        return collect_impulse_inputs(self._raw)

    @cached_property
    def onewire_sensors(self) -> dict[int, OneWireSensor]:
        return collect_onewire_sensors(self._raw)

    @cached_property
    def water_chemistry(self) -> WaterChemistry:
        return WaterChemistry.from_raw(self._raw)

    # ---- outputs ----------------------------------------------------------

    @cached_property
    def pump(self) -> Pump | None:
        return Pump.from_raw(self._raw)

    @cached_property
    def heater(self) -> Heater | None:
        return Heater.from_raw(self._raw)

    @cached_property
    def solar(self) -> Solar | None:
        return Solar.from_raw(self._raw)

    @cached_property
    def light(self) -> Light | None:
        return Light.from_raw(self._raw)

    @cached_property
    def refill(self) -> Refill | None:
        return Refill.from_raw(self._raw)

    @cached_property
    def eco(self) -> Eco | None:
        return Eco.from_raw(self._raw)

    @cached_property
    def backwash(self) -> Backwash | None:
        return Backwash.from_raw(self._raw)

    @cached_property
    def backwash_rinse(self) -> BackwashRinse | None:
        return BackwashRinse.from_raw(self._raw)

    # ---- cover, dmx, extension, dosing -----------------------------------

    @cached_property
    def cover(self) -> Cover | None:
        return Cover.from_raw(self._raw)

    @cached_property
    def dmx_scenes(self) -> dict[int, DmxScene]:
        return collect_dmx_scenes(self._raw)

    @cached_property
    def extension_relays(self) -> dict[int, dict[int, ExtensionRelay]]:
        return collect_extension_relays(self._raw)

    @cached_property
    def dosing_channels(self) -> dict[str, DosingChannel]:
        return collect_dosing_channels(self._raw)

    # ---- inputs & rules ---------------------------------------------------

    @cached_property
    def digital_inputs(self) -> dict[int, DigitalInput]:
        return collect_digital_inputs(self._raw)

    @cached_property
    def can_empty_inputs(self) -> dict[int, CanEmptyInput]:
        return collect_can_empty_inputs(self._raw)

    @cached_property
    def switching_rules(self) -> dict[int, SwitchingRule]:
        return collect_switching_rules(self._raw)

    # ---- mixed system states ---------------------------------------------

    @cached_property
    def overflow(self) -> OverflowState | None:
        return OverflowState.from_raw(self._raw)

    @cached_property
    def backwash_status(self) -> BackwashStatus | None:
        return BackwashStatus.from_raw(self._raw)

    @cached_property
    def bathing_ai(self) -> BathingAi | None:
        return BathingAi.from_raw(self._raw)

    @cached_property
    def pv_surplus(self) -> PvSurplus | None:
        return PvSurplus.from_raw(self._raw)

raw property

raw: Mapping[str, Any]

Read-only view of the original /getReadings JSON dict.

Returned as a MappingProxyType so callers can read any key (including undocumented or firmware-future ones) but cannot mutate the underlying dict in a way that would corrupt cached typed views.

Shared enums (myviolet.enums)

enums

Shared enums for fields returned by the Violet controller API.

Each enum mirrors a documented value table from the vendor's getReadings.xlsx specification. The OutputState enum is reused across ~30 binary outputs (pump, heater, solar, light, refill, eco, backwash, backwash rinse, all dosing channels, and all extension relays).

OutputState

Bases: IntEnum

State of any binary output controlled by the Violet controller.

These seven values are emitted by every relay-style output: pump, heater, solar, light, refill, eco, backwash, backwash rinse, the five dosing channels, and both extension relay buses.

Source code in src/myviolet/enums.py
class OutputState(IntEnum):
    """State of any binary output controlled by the Violet controller.

    These seven values are emitted by every relay-style output: pump, heater,
    solar, light, refill, eco, backwash, backwash rinse, the five dosing
    channels, and both extension relay buses.
    """

    AUTO_OFF = 0
    AUTO_ON = 1
    AUTO_PRIO_OFF = 2
    AUTO_PRIO_ON = 3
    MANUAL_ON = 4
    EMERGENCY_OFF = 5
    MANUAL_OFF = 6

    @property
    def is_on(self) -> bool:
        """Whether the output is currently energised, regardless of cause."""
        return self in (OutputState.AUTO_ON, OutputState.AUTO_PRIO_ON, OutputState.MANUAL_ON)

    @property
    def is_manual(self) -> bool:
        """Whether the output is in a user-forced (manual) state."""
        return self in (OutputState.MANUAL_ON, OutputState.MANUAL_OFF)

    @property
    def is_emergency(self) -> bool:
        """Whether an emergency or priority rule is currently in effect."""
        return self in (
            OutputState.AUTO_PRIO_OFF,
            OutputState.AUTO_PRIO_ON,
            OutputState.EMERGENCY_OFF,
        )

is_on property

is_on: bool

Whether the output is currently energised, regardless of cause.

is_manual property

is_manual: bool

Whether the output is in a user-forced (manual) state.

is_emergency property

is_emergency: bool

Whether an emergency or priority rule is currently in effect.

DmxSceneState

Bases: IntEnum

State of a DMX scene. Subset of OutputState (no priority rules).

Source code in src/myviolet/enums.py
class DmxSceneState(IntEnum):
    """State of a DMX scene. Subset of `OutputState` (no priority rules)."""

    AUTO_OFF = 0
    AUTO_ON = 1
    MANUAL_ON = 4
    MANUAL_OFF = 6

RuleState

Bases: IntEnum

State of a digital-input switching rule.

Source code in src/myviolet/enums.py
class RuleState(IntEnum):
    """State of a digital-input switching rule."""

    INACTIVE = 0
    ACTIVE = 1
    BLOCKED_BY_RULE = 5
    BLOCKED_MANUALLY = 6

CoverState

Bases: StrEnum

Position / motion state of the pool cover.

Source code in src/myviolet/enums.py
class CoverState(StrEnum):
    """Position / motion state of the pool cover."""

    OPEN = "OPEN"
    CLOSED = "CLOSED"
    OPENING = "OPENING"
    CLOSING = "CLOSING"
    STOPPED = "STOPPED"

OnewireState

Bases: StrEnum

Fault state of a 1-wire temperature sensor.

Note: DATA_MISSMATCH (sic) is spelled with two s letters to match the wire value emitted verbatim by the controller firmware.

Source code in src/myviolet/enums.py
class OnewireState(StrEnum):
    """Fault state of a 1-wire temperature sensor.

    Note: ``DATA_MISSMATCH`` (sic) is spelled with two ``s`` letters to match
    the wire value emitted verbatim by the controller firmware.
    """

    OK = "OK"
    CRC_FAULT = "CRC_FAULT"
    DATA_MISSMATCH = "DATA_MISSMATCH"
    NOT_CONNECTED = "NOT_CONNECTED"
    NO_SENSOR_CONFIGURED = "NO_SENSOR_CONFIGURED"

SimpleOnOff

Bases: StrEnum

Binary on/off used by overflow and bathing-AI flags.

Source code in src/myviolet/enums.py
class SimpleOnOff(StrEnum):
    """Binary on/off used by overflow and bathing-AI flags."""

    ON = "ON"
    OFF = "OFF"

YesNo

Bases: StrEnum

Binary yes/no used by various status flags.

Source code in src/myviolet/enums.py
class YesNo(StrEnum):
    """Binary yes/no used by various status flags."""

    YES = "YES"
    NO = "NO"

DosingType

Bases: IntEnum

Configuration of a chlorine dosing controller.

Source code in src/myviolet/enums.py
class DosingType(IntEnum):
    """Configuration of a chlorine dosing controller."""

    ORP_ONLY = 0
    ORP_AND_CL = 1

PvSurplusState

Bases: IntEnum

Trigger source of the PV surplus function.

Source code in src/myviolet/enums.py
class PvSurplusState(IntEnum):
    """Trigger source of the PV surplus function."""

    OFF = 0
    ON_BY_INPUT = 1
    ON_BY_HTTP = 2

Exceptions (myviolet.exceptions)

exceptions

Exception hierarchy for the myviolet client.

All exceptions raised by the client (transport-level HTTP errors, payload problems, validation failures, deliberate safety guards) subclass VioletApiException so callers can catch the entire library's failure modes with a single blanket clause, or narrow to a specific subclass when needed.

VioletApiException

Bases: Exception

Base for every error raised by the myviolet client.

Source code in src/myviolet/exceptions.py
class VioletApiException(Exception):
    """Base for every error raised by the myviolet client."""

BadCredentialsException

Bases: VioletApiException

The controller rejected the supplied credentials (HTTP 401 or 403).

Source code in src/myviolet/exceptions.py
class BadCredentialsException(VioletApiException):
    """The controller rejected the supplied credentials (HTTP 401 or 403)."""

    def __init__(self, status_code: int) -> None:
        self.status_code = status_code
        super().__init__(f"authentication failed (HTTP {status_code})")

BadStatusCodeException

Bases: VioletApiException

The controller returned an unexpected non-auth error HTTP status.

Source code in src/myviolet/exceptions.py
class BadStatusCodeException(VioletApiException):
    """The controller returned an unexpected non-auth error HTTP status."""

    def __init__(self, status_code: int, message: str = "") -> None:
        self.status_code = status_code
        suffix = f": {message}" if message else ""
        super().__init__(f"unexpected HTTP {status_code}{suffix}")

TimeoutException

Bases: VioletApiException

An HTTP request to the controller exceeded the configured timeout.

Source code in src/myviolet/exceptions.py
class TimeoutException(VioletApiException):
    """An HTTP request to the controller exceeded the configured timeout."""

InvalidPayloadException

Bases: VioletApiException

The controller's response was malformed, truncated, or unparseable.

Source code in src/myviolet/exceptions.py
class InvalidPayloadException(VioletApiException):
    """The controller's response was malformed, truncated, or unparseable."""

UnsafeOperationException

Bases: VioletApiException

A potentially unsafe operation was attempted without acknowledgment.

Currently raised by cover open/close calls when the caller has not passed acknowledge_unsafe=True.

Source code in src/myviolet/exceptions.py
class UnsafeOperationException(VioletApiException):
    """A potentially unsafe operation was attempted without acknowledgment.

    Currently raised by cover open/close calls when the caller has not passed
    ``acknowledge_unsafe=True``.
    """

    def __init__(self, operation: str) -> None:
        super().__init__(
            f"{operation!r} is potentially unsafe; pass acknowledge_unsafe=True "
            f"to confirm the caller has implemented external safety logic"
        )

SetpointValidationError

Bases: VioletApiException, ValueError

A setpoint value was outside its documented valid range.

Subclasses both VioletApiException (for blanket library catches) and ValueError (so generic input-validation code keeps working).

Source code in src/myviolet/exceptions.py
class SetpointValidationError(VioletApiException, ValueError):
    """A setpoint value was outside its documented valid range.

    Subclasses both `VioletApiException` (for blanket library catches) and
    `ValueError` (so generic input-validation code keeps working).
    """

    def __init__(self, field: str, value: float, low: float, high: float) -> None:
        self.field = field
        self.value = value
        self.low = low
        self.high = high
        super().__init__(f"{field} setpoint {value} is outside the valid range [{low}, {high}]")