Skip to content

Valve

valve

valve domain implementation.

SPEC module-attribute 

SPEC: DomainSpec[Valve] = register_domain(DomainSpec(name='valve', entity_cls=Valve))

The DomainSpec registered with the shared DomainRegistry.

Valve

Bases: Entity

A Home Assistant valve entity (water shutoff, gas, irrigation).

Mirrors the typed Cover shape because valves share the same open/close/position model, but is exposed as a distinct domain to preserve the safety-critical semantics of valves (water mains, gas lines) and to match Home Assistant's separate valve domain introduced in 2023.9.

Uses intent-specific action names (open, close, stop, set_position) rather than raw HA service names. The set_position and stop actions degrade safely on valves that only advertise binary open/close support: they log a debug message and return without raising, so user code targeting a heterogeneous fleet of valves does not break. Callers that need to know whether the action will actually be dispatched can pre-check with supports_set_position or supports_stop.

Source code in src/haclient/domains/valve.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
class Valve(Entity):
    """A Home Assistant valve entity (water shutoff, gas, irrigation).

    Mirrors the typed `Cover` shape because valves share the same
    open/close/position model, but is exposed as a distinct domain to
    preserve the safety-critical semantics of valves (water mains, gas
    lines) and to match Home Assistant's separate ``valve`` domain
    introduced in 2023.9.

    Uses intent-specific action names (``open``, ``close``, ``stop``,
    ``set_position``) rather than raw HA service names. The
    ``set_position`` and ``stop`` actions degrade safely on valves that
    only advertise binary open/close support: they log a debug message
    and return without raising, so user code targeting a heterogeneous
    fleet of valves does not break. Callers that need to know whether
    the action will actually be dispatched can pre-check with
    `supports_set_position` or `supports_stop`.
    """

    domain = "valve"

    # -- Listener decorators ------------------------------------------

    def on_open(self, func: ValueChangeHandler) -> ValueChangeHandler:
        """Register a listener for when the valve opens.

        Parameters
        ----------
        func : callable
            Sync or async callable invoked with ``(old_state, new_state)``
            on every transition into the ``open`` state.

        Returns
        -------
        callable
            The same *func*, returned for decorator use.
        """
        return self._register_state_transition_listener("open", func)

    def on_close(self, func: ValueChangeHandler) -> ValueChangeHandler:
        """Register a listener for when the valve closes.

        Parameters
        ----------
        func : callable
            Sync or async callable invoked with ``(old_state, new_state)``
            on every transition into the ``closed`` state.

        Returns
        -------
        callable
            The same *func*, returned for decorator use.
        """
        return self._register_state_transition_listener("closed", func)

    def on_position_change(self, func: ValueChangeHandler) -> ValueChangeHandler:
        """Register a listener for position changes.

        Parameters
        ----------
        func : callable
            Callable invoked with ``(old_value, new_value)`` whenever
            the ``current_position`` attribute (0-100) changes.

        Returns
        -------
        callable
            The same *func*, returned for decorator use.
        """
        return self._register_attr_listener("current_position", func)

    # -- State properties ---------------------------------------------

    @property
    def is_open(self) -> bool:
        """Whether the valve is currently open."""
        return self.state == "open"

    @property
    def is_closed(self) -> bool:
        """Whether the valve is currently closed."""
        return self.state == "closed"

    @property
    def is_opening(self) -> bool:
        """Whether the valve is currently in the process of opening."""
        return self.state == "opening"

    @property
    def is_closing(self) -> bool:
        """Whether the valve is currently in the process of closing."""
        return self.state == "closing"

    @property
    def current_position(self) -> int | None:
        """Current position (0--100) or ``None`` if unsupported."""
        value = self.attributes.get("current_position")
        return int(value) if isinstance(value, (int, float)) else None

    def _has_feature(self, mask: int) -> bool:
        """Return ``True`` when ``supported_features`` advertises *mask*.

        Parameters
        ----------
        mask : int
            ``ValveEntityFeature`` bit to test for.

        Returns
        -------
        bool
            ``True`` if the entity advertises *mask* in its
            ``supported_features`` bitmask, otherwise ``False``.
        """
        features = self.attributes.get("supported_features")
        if not isinstance(features, int):
            return False
        return bool(features & mask)

    @property
    def supports_set_position(self) -> bool:
        """Whether the valve advertises ``SET_POSITION`` support."""
        return self._has_feature(_FEATURE_SET_POSITION)

    @property
    def supports_stop(self) -> bool:
        """Whether the valve advertises ``STOP`` support."""
        return self._has_feature(_FEATURE_STOP)

    # -- Actions ------------------------------------------------------

    async def open(self) -> None:
        """Open the valve fully.

        Invokes the ``valve.open_valve`` Home Assistant service.

        Raises
        ------
        CommandError
            If Home Assistant rejects the service call.
        HTTPError
            If the REST call returns a non-2xx response.
        TimeoutError
            If the call exceeds the configured request timeout.
        ConnectionClosedError
            If the WebSocket disconnects mid-call.
        """
        await self._call_service("open_valve")

    async def close(self) -> None:
        """Close the valve fully.

        Invokes the ``valve.close_valve`` Home Assistant service.

        Raises
        ------
        CommandError
            If Home Assistant rejects the service call.
        HTTPError
            If the REST call returns a non-2xx response.
        TimeoutError
            If the call exceeds the configured request timeout.
        ConnectionClosedError
            If the WebSocket disconnects mid-call.
        """
        await self._call_service("close_valve")

    async def stop(self) -> None:
        """Stop movement of the valve, if supported.

        Degrades safely: if the valve does not advertise the ``STOP``
        feature, this method logs a debug message and returns without
        raising. Callers can pre-check with `supports_stop`.
        """
        if not self.supports_stop:
            _LOGGER.debug(
                "stop() unsupported for %s; skipping (no ValveEntityFeature.STOP)",
                self.entity_id,
            )
            return
        await self._call_service("stop_valve")

    async def set_position(self, position: int) -> None:
        """Set the valve position, if supported.

        Parameters
        ----------
        position : int
            Target position in the range 0--100 (``0`` = fully closed,
            ``100`` = fully open).

        Raises
        ------
        ValueError
            If *position* is outside the 0--100 range.

        Notes
        -----
        Degrades safely: if the valve does not advertise the
        ``SET_POSITION`` feature (e.g. a binary water shutoff), this
        method logs a debug message and returns without raising.
        Callers can pre-check with `supports_set_position`.
        Range validation always runs, even when the feature is absent,
        so callers receive an immediate error for clearly invalid input.
        """
        value = validate_range(position, 0, 100, "position")
        if not self.supports_set_position:
            _LOGGER.debug(
                "set_position() unsupported for %s; skipping (no ValveEntityFeature.SET_POSITION)",
                self.entity_id,
            )
            return
        await self._call_service("set_valve_position", {"position": value})

    async def toggle(self) -> None:
        """Toggle open/close state.

        Invokes the ``valve.toggle`` Home Assistant service.

        Raises
        ------
        CommandError
            If Home Assistant rejects the service call.
        HTTPError
            If the REST call returns a non-2xx response.
        TimeoutError
            If the call exceeds the configured request timeout.
        ConnectionClosedError
            If the WebSocket disconnects mid-call.
        """
        await self._call_service("toggle")

is_open property 

is_open: bool

Whether the valve is currently open.

is_closed property 

is_closed: bool

Whether the valve is currently closed.

is_opening property 

is_opening: bool

Whether the valve is currently in the process of opening.

is_closing property 

is_closing: bool

Whether the valve is currently in the process of closing.

current_position property 

current_position: int | None

Current position (0--100) or None if unsupported.

supports_set_position property 

supports_set_position: bool

Whether the valve advertises SET_POSITION support.

supports_stop property 

supports_stop: bool

Whether the valve advertises STOP support.

on_open 

on_open(func: ValueChangeHandler) -> ValueChangeHandler

Register a listener for when the valve opens.

Parameters:

Name Type Description Default
func callable

Sync or async callable invoked with (old_state, new_state) on every transition into the open state.

 required

Returns:

Type Description
callable

The same func, returned for decorator use.
Source code in src/haclient/domains/valve.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def on_open(self, func: ValueChangeHandler) -> ValueChangeHandler:
    """Register a listener for when the valve opens.

    Parameters
    ----------
    func : callable
        Sync or async callable invoked with ``(old_state, new_state)``
        on every transition into the ``open`` state.

    Returns
    -------
    callable
        The same *func*, returned for decorator use.
    """
    return self._register_state_transition_listener("open", func)

on_close 

on_close(func: ValueChangeHandler) -> ValueChangeHandler

Register a listener for when the valve closes.

Parameters:

Name Type Description Default
func callable

Sync or async callable invoked with (old_state, new_state) on every transition into the closed state.

 required

Returns:

Type Description
callable

The same func, returned for decorator use.
Source code in src/haclient/domains/valve.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def on_close(self, func: ValueChangeHandler) -> ValueChangeHandler:
    """Register a listener for when the valve closes.

    Parameters
    ----------
    func : callable
        Sync or async callable invoked with ``(old_state, new_state)``
        on every transition into the ``closed`` state.

    Returns
    -------
    callable
        The same *func*, returned for decorator use.
    """
    return self._register_state_transition_listener("closed", func)

on_position_change 

on_position_change(func: ValueChangeHandler) -> ValueChangeHandler

Register a listener for position changes.

Parameters:

Name Type Description Default
func callable

Callable invoked with (old_value, new_value) whenever the current_position attribute (0-100) changes.

 required

Returns:

Type Description
callable

The same func, returned for decorator use.
Source code in src/haclient/domains/valve.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def on_position_change(self, func: ValueChangeHandler) -> ValueChangeHandler:
    """Register a listener for position changes.

    Parameters
    ----------
    func : callable
        Callable invoked with ``(old_value, new_value)`` whenever
        the ``current_position`` attribute (0-100) changes.

    Returns
    -------
    callable
        The same *func*, returned for decorator use.
    """
    return self._register_attr_listener("current_position", func)

open async 

open() -> None

Open the valve fully.

Invokes the valve.open_valve Home Assistant service.

Raises:

Type Description
CommandError

If Home Assistant rejects the service call.
HTTPError

If the REST call returns a non-2xx response.
TimeoutError

If the call exceeds the configured request timeout.
ConnectionClosedError

If the WebSocket disconnects mid-call.
Source code in src/haclient/domains/valve.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
async def open(self) -> None:
    """Open the valve fully.

    Invokes the ``valve.open_valve`` Home Assistant service.

    Raises
    ------
    CommandError
        If Home Assistant rejects the service call.
    HTTPError
        If the REST call returns a non-2xx response.
    TimeoutError
        If the call exceeds the configured request timeout.
    ConnectionClosedError
        If the WebSocket disconnects mid-call.
    """
    await self._call_service("open_valve")

close async 

close() -> None

Close the valve fully.

Invokes the valve.close_valve Home Assistant service.

Raises:

Type Description
CommandError

If Home Assistant rejects the service call.
HTTPError

If the REST call returns a non-2xx response.
TimeoutError

If the call exceeds the configured request timeout.
ConnectionClosedError

If the WebSocket disconnects mid-call.
Source code in src/haclient/domains/valve.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
async def close(self) -> None:
    """Close the valve fully.

    Invokes the ``valve.close_valve`` Home Assistant service.

    Raises
    ------
    CommandError
        If Home Assistant rejects the service call.
    HTTPError
        If the REST call returns a non-2xx response.
    TimeoutError
        If the call exceeds the configured request timeout.
    ConnectionClosedError
        If the WebSocket disconnects mid-call.
    """
    await self._call_service("close_valve")

stop async 

stop() -> None

Stop movement of the valve, if supported.

Degrades safely: if the valve does not advertise the STOP feature, this method logs a debug message and returns without raising. Callers can pre-check with supports_stop.

Source code in src/haclient/domains/valve.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
async def stop(self) -> None:
    """Stop movement of the valve, if supported.

    Degrades safely: if the valve does not advertise the ``STOP``
    feature, this method logs a debug message and returns without
    raising. Callers can pre-check with `supports_stop`.
    """
    if not self.supports_stop:
        _LOGGER.debug(
            "stop() unsupported for %s; skipping (no ValveEntityFeature.STOP)",
            self.entity_id,
        )
        return
    await self._call_service("stop_valve")

set_position async 

set_position(position: int) -> None

Set the valve position, if supported.

Parameters:

Name Type Description Default
position int

Target position in the range 0--100 (0 = fully closed, 100 = fully open).

 required

Raises:

Type Description
ValueError

If position is outside the 0--100 range.
Notes

Degrades safely: if the valve does not advertise the SET_POSITION feature (e.g. a binary water shutoff), this method logs a debug message and returns without raising. Callers can pre-check with supports_set_position. Range validation always runs, even when the feature is absent, so callers receive an immediate error for clearly invalid input.

Source code in src/haclient/domains/valve.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
async def set_position(self, position: int) -> None:
    """Set the valve position, if supported.

    Parameters
    ----------
    position : int
        Target position in the range 0--100 (``0`` = fully closed,
        ``100`` = fully open).

    Raises
    ------
    ValueError
        If *position* is outside the 0--100 range.

    Notes
    -----
    Degrades safely: if the valve does not advertise the
    ``SET_POSITION`` feature (e.g. a binary water shutoff), this
    method logs a debug message and returns without raising.
    Callers can pre-check with `supports_set_position`.
    Range validation always runs, even when the feature is absent,
    so callers receive an immediate error for clearly invalid input.
    """
    value = validate_range(position, 0, 100, "position")
    if not self.supports_set_position:
        _LOGGER.debug(
            "set_position() unsupported for %s; skipping (no ValveEntityFeature.SET_POSITION)",
            self.entity_id,
        )
        return
    await self._call_service("set_valve_position", {"position": value})

toggle async 

toggle() -> None

Toggle open/close state.

Invokes the valve.toggle Home Assistant service.

Raises:

Type Description
CommandError

If Home Assistant rejects the service call.
HTTPError

If the REST call returns a non-2xx response.
TimeoutError

If the call exceeds the configured request timeout.
ConnectionClosedError

If the WebSocket disconnects mid-call.
Source code in src/haclient/domains/valve.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
async def toggle(self) -> None:
    """Toggle open/close state.

    Invokes the ``valve.toggle`` Home Assistant service.

    Raises
    ------
    CommandError
        If Home Assistant rejects the service call.
    HTTPError
        If the REST call returns a non-2xx response.
    TimeoutError
        If the call exceeds the configured request timeout.
    ConnectionClosedError
        If the WebSocket disconnects mid-call.
    """
    await self._call_service("toggle")