Services API

Services provide specialized interfaces for different aspects of pump communication. Each service is accessed as a property on the client instance.

Service Overview

Service	Property	Purpose
TelemetryService	client.telemetry	Real-time sensor data and monitoring
ControlService	client.control	Pump control operations (start, stop, modes)
ScheduleService	client.schedule	Weekly schedule management (5 layers)
DeviceInfoService	client.device_info	Device identification and statistics
ConfigurationService	client.config	Backup and restore operations
TimeService	client.time	Real-time clock management
HistoryService	client.history	Historical trend data (100 cycles)
EventLogService	client.event_log	Pump event history (20 entries)
SingleEventService	client.single_events	One-off events and vacations
WriteOperationService	client.writes	The serialized, verified write path

TelemetryService

Read telemetry data from the pump, either as a single snapshot or as a continuous stream.

Service for managing pump telemetry operations.

This service provides high-level APIs for accessing telemetry data: - One-time reads (polling) - Continuous streaming (notifications) - State management

Attributes:

Name	Type	Description
_telemetry		Current basic telemetry data
_advanced_telemetry		Current advanced telemetry data
_has_motor_state_stream		Flag indicating motor state stream is active
_has_flow_stream		Flag indicating flow/pressure stream is active

Example

from alpha_hwr.core import Transport, Session from alpha_hwr.services import TelemetryService

Initialize

transport = Transport(bleak_client) # doctest: +SKIP session = Session(transport) # doctest: +SKIP telemetry_service = TelemetryService(transport, session) # doctest: +SKIP

Read once

data = await telemetry_service.read_once() # doctest: +SKIP print(f"Flow: {data.flow_m3h} m³/h") # doctest: +SKIP

Stream continuously

async for data in telemetry_service.stream(): # doctest: +SKIP ... print(f"Power: {data.power_w} W")

advanced property

Get current advanced telemetry data.

Returns advanced telemetry including converter temperature, inlet/outlet pressure, alarms/warnings, etc.

Returns:

Type	Description
AdvancedTelemetry	Current AdvancedTelemetry

Example

adv = service.advanced # doctest: +SKIP print(f"Converter temp: {adv.converter_temperature_c}°C") # doctest: +SKIP

current property

Get current telemetry data.

Returns the most recently updated telemetry state. This may be from active polling or passive notifications.

Returns:

Type	Description
TelemetryData	Current TelemetryData

Example

telemetry = service.current # doctest: +SKIP print(f"Voltage: {telemetry.voltage_ac_v}V") # doctest: +SKIP

__init__(transport, session)

Initialize telemetry service.

Parameters:

Name	Type	Description	Default
transport	Transport	Transport layer for BLE communication	required
session	Session	Session manager for state tracking	required

read_once() async

Read telemetry snapshot using Class 10 INFO commands.

Sends INFO requests to query current telemetry data from the pump. This is the correct modern approach - NOT Class 3 register polling!

The pump responds with Class 10 data object frames containing the requested telemetry values.

Returns:

Type	Description
TelemetryData	TelemetryData with current values

Example

data = await service.read_once() # doctest: +SKIP print(f"Flow: {data.flow_m3h} m³/h") # doctest: +SKIP print(f"Power: {data.power_w} W") # doctest: +SKIP

Implementation Notes

• Uses Class 10 INFO commands (OpSpec 0x00)
• Filters out passive notifications (OpSpec 0x0E)
• Parses responses using TelemetryDecoder
• Updates internal state for subsequent queries

stream(interval=0.1, poll_if_no_stream=True) async

Stream continuous telemetry updates.

This method yields telemetry data as it's updated, either from: - Passive Class 10 notifications (if pump sends them) - Active polling (if no notifications)

Parameters:

Name	Type	Description	Default
interval	float	Polling interval in seconds (default 0.1 = 10Hz)	0.1
poll_if_no_stream	bool	If True, falls back to polling if no stream detected	True

Yields:

Type	Description
AsyncIterator[TelemetryData]	TelemetryData as it's updated

Example

async for data in service.stream(interval=0.2): # doctest: +SKIP ... print(f"Flow: {data.flow_m3h} m³/h, Power: {data.power_w} W") ... if data.power_w > 100: ... break # Stop streaming

Implementation Notes

• Non-blocking: uses async iteration
• Can be cancelled by breaking from loop
• Automatically detects if pump sends notifications
• Falls back to polling if notifications stop

update_from_notification(data)

Update telemetry state from BLE notification.

This method is called by the Client's notification handler when a notification arrives. It parses the frame, decodes telemetry, and updates state.

Parameters:

Name	Type	Description	Default
data	bytes	Raw notification bytes from BLE	required

Note

Registration of this handler is managed by the Client layer during connection setup. Services should not directly interact with the transport layer per the architecture guidelines.

Example

Handler registration happens in Client.connect()

The client automatically forwards notifications to this method

Implementation Notes

• Automatically detects Class 10 telemetry frames
• Routes to appropriate decoder based on Sub-ID/Object ID
• Updates both basic and advanced telemetry
• Sets stream detection flags
• Thread-safe (can be called from notification callback)

ControlService

Control pump operations: run state, control mode and setpoints.

The verified setters (set_enabled, set_mode_verified, set_setpoint, set_temperature_range, set_cycle_times) return a WriteResult reporting what the pump actually stored. The older set_constant_* setters return bool and do not read back — see Verified Writes.

Bases: BaseService

Service for pump control operations.

This service provides high-level APIs for controlling the pump: - Start/stop operations - Mode changes (constant pressure, flow, speed, etc.) - Setpoint management - Mode validation

Attributes:

Name	Type	Description
_current_mode	ControlMode | int	Currently active control mode
_CLASS10_CONTROL_MAP	ControlMode | int	Mapping of modes to Class 10 parameters

Example

from alpha_hwr.core import Transport, Session # doctest: +SKIP from alpha_hwr.services import ControlService # doctest: +SKIP from alpha_hwr.constants import ControlMode # doctest: +SKIP

Initialize

control = ControlService(transport, session) # doctest: +SKIP

Start pump

await control.start() # doctest: +SKIP

Set constant pressure mode

await control.set_constant_pressure(1.5) # 1.5 meters # doctest: +SKIP

Stop pump

await control.stop() # doctest: +SKIP

is_cache_valid property

Whether this service knows enough about the pump to write safely.

False until :meth:sync_cache has succeeded, and again after a disconnect - a write built from a cache filled on a previous connection can carry values the pump no longer holds.

__init__(transport, session, schedule_service=None)

Initialize control service.

Parameters:

Name	Type	Description	Default
transport	Transport	Transport layer for BLE communication	required
session	Session	Session manager for state tracking	required
schedule_service	ScheduleService | None	Optional schedule service for status reading	None

attach_write_service(writes)

Give this service the write layer its verified methods submit to.

cached_setpoint(mode)

The last setpoint seen for mode, or None if never read.

get_cycle_flow() async

Flow the pump targets during cycle-mode ON periods, in m3/h.

Stored in SI m3/s, like every other flow setpoint.

get_cycle_time_config() async

Get current cycle time configuration for Mode 25 (DHW_ON_OFF_CONTROL).

Returns:

Type	Description
tuple[int, int] | None	Tuple of (on_time_minutes, off_time_minutes) if successful, None otherwise

get_mode(retries=3) async

Get the current control mode and setpoint information.

Reads Class 10 Object 86, Sub-ID 7 (overall_operation_prioritized_request_obj) - the pump's own view of its state after it has weighed remote, local and alarm influence against each other. For Temperature Range Control (mode 27) it additionally reads Object 91, Sub-ID 430.

Sub-ID 6, which this used to read, is the request object: it echoes what was last written and reports control_source = 0 indefinitely. Measured side by side on hardware, Sub 6 returned control_source = 0 while Sub 7 returned 1 (Local/Panel), which is why is_remote was never meaningful before.

Returns:

Type	Description
SetpointInfo | None	SetpointInfo with current control mode, operation mode, and setpoint value,
SetpointInfo | None	or None if read failed

Raises:

Type	Description
ConnectionError	If the BLE connection to the pump drops while waiting for the response.

Example

info = await control.get_mode() # doctest: +SKIP if info and info.control_mode == ControlMode.CONSTANT_PRESSURE: # doctest: +SKIP ... value, unit = info.get_display_value() ... print(f"Running in constant pressure mode: {value} {unit}")

Implementation Notes

• Standard modes: Object 86, Sub-ID 7, Type 303 (OperationStatusRequest)
• Temperature Range: Object 91, Sub-ID 430, Type 1012
• Response format: [00 00 XX][control_source][operation_mode][control_mode][setpoint(4 bytes float)]
• Setpoint is big-endian float at offset 3 (after 3-byte header)

get_setpoint_range(mode)

The pump's own range for a mode, if it has been read.

Returns None when it has not. Callers should fall back to the wider inherited constants rather than refusing: letting the pump clamp a value it dislikes is better than refusing one it would have taken.

get_temperature_range() async

Read the stored temperature range and AutoAdapt flag.

Returns:

Type	Description
tuple[float, float, bool] | None	(min_c, max_c, autoadapt), or None if the read failed.

invalidate_cache()

Forget everything read from the pump.

Called on disconnect. The mode is dropped along with the rest: a command issued on one connection must not be treated as confirmed by a reading taken on the next.

read_limiters() async

Read the pump's flow limiters and whether either is limiting.

A limiter that is enabled caps delivered flow regardless of the setpoint, and nothing in the setpoint range says so: the type 301 range is the factory range. So a setpoint can be accepted, read back correct, and still not be delivered. This is the only way to see that.

Returns:

Type	Description
dict[str, dict[str, float | bool]]	{"MaxFlow": {...}, "MinFlow": {...}} with enabled,
dict[str, dict[str, float | bool]]	limit_m3h, factory_min_m3h, factory_max_m3h and
dict[str, dict[str, float | bool]]	limiting for each limiter that answered.

Examples:

>>> limiters = await client.control.read_limiters()
>>> limiters["MaxFlow"]["enabled"]
False

read_setpoint_ranges() async

Read each scalar mode's setpoint range from the pump.

The pump publishes these in the type 301 factory-config objects at Object 86 sub-ids 13, 15, 17 and 39 - the same objects the Grundfos GO app's setpoint slider binds to. Each holds a 28-byte struct whose first three floats are default, minimum and maximum, in the pump's native units.

Returns:

Type	Description
dict[int, tuple[float, float]]	{ControlMode: (minimum, maximum)} in this client's units,
dict[int, tuple[float, float]]	for as many modes as could be read.

Note

The chain is deliberately sequential and stops at the first failure. All four objects answer with the same type code (00 01 2d 01), so the transport cannot tell their replies apart. Carrying on past a failure hands read N's late reply to read N+1 and shifts every remaining range by one slot - which would bound constant pressure by constant speed's 1650-3671 read as Pascals, 0.168-0.374 m, and refuse an ordinary 1.5 m setpoint for the rest of the connection.

Examples:

>>> ranges = await client.control.read_setpoint_ranges()
>>> ranges[ControlMode.CONSTANT_SPEED]
(1650.0, 3671.0)

set_autoadapt(value_m) async

Set generic AutoAdapt mode with setpoint.

AutoAdapt mode automatically analyzes and adjusts pump operation based on system demand. This is the generic AutoAdapt mode (Mode 5).

For specific heating system types, consider using: - set_autoadapt_radiator() for radiator systems (Mode 13) - set_autoadapt_underfloor() for underfloor heating (Mode 14) - set_autoadapt_combined() for combined systems (Mode 15)

Parameters:

Name	Type	Description	Default
value_m	float	Pressure setpoint in meters of water column (e.g., 1.5 for 1.5 meters)	required

Returns:

Type	Description
bool	True if successful, False otherwise

Warning

Mode 5 (AUTO_ADAPT) has limited support on ALPHA HWR. Mode switching may not work reliably. Consider using specific AutoAdapt variants (modes 13-15) instead for better compatibility.

Example

await control.set_autoadapt(1.5) # 1.5 meters # doctest: +SKIP

set_autoadapt_combined(value_m) async

DEPRECATED: Use set_temperature_control() instead.

Legacy method that incorrectly uses pressure setpoints.

set_autoadapt_radiator(value_m) async

DEPRECATED: Use set_temperature_control() instead.

Legacy method that incorrectly uses pressure setpoints.

set_autoadapt_underfloor(value_m) async

DEPRECATED: Use set_temperature_control() instead.

Legacy method that incorrectly uses pressure setpoints.

set_constant_flow(value_m3h) async

Set constant flow mode with setpoint.

Parameters:

Name	Type	Description	Default
value_m3h	float	Flow setpoint in m³/h	required

Returns:

Type	Description
bool	True if successful, False otherwise

set_constant_pressure(value_m) async

Set constant pressure mode with setpoint.

Parameters:

Name	Type	Description	Default
value_m	float	Pressure setpoint in meters of water column	required

Returns:

Type	Description
bool	True if successful, False otherwise

set_constant_speed(value_rpm) async

Set constant speed mode with setpoint.

Parameters:

Name	Type	Description	Default
value_rpm	float	Speed setpoint in RPM	required

Returns:

Type	Description
bool	True if successful, False otherwise

set_cycle_time_control(on_minutes, off_minutes) async

Set cycle time control mode (Mode 25 / DHW_ON_OFF_CONTROL).

Writes Object 91 Sub 421 as a read-modify-write: the object also carries the flow the pump targets during ON periods, which is echoed back byte for byte rather than recomputed, so setting the periods cannot disturb it.

Parameters:

Name	Type	Description	Default
on_minutes	int	Duration pump runs (1-60)	required
off_minutes	int	Duration pump is off (1-60)	required

Returns:

Type	Description
bool	True if successful, False otherwise

set_cycle_times(on_minutes, off_minutes) async

Set the cycle periods and confirm them, preserving the flow.

set_enabled(enabled) async

Start or stop the pump, and confirm the resulting run state.

The pump sends nothing after a Class 3 run command, so the result comes from reading the state back.

set_mode(mode) async

Set the control mode, and nothing else.

Neither the run state nor any mode's stored setpoint is touched: the mode change goes through the pump's dedicated mode-request object rather than the fused control object, which used to force the pump on and overwrite the target mode's setpoint with a default.

Parameters:

Name	Type	Description	Default
mode	ControlMode | int	Control mode to set	required

Returns:

Type	Description
bool	True if mode set successfully, False otherwise

set_mode_verified(mode) async

Change the control mode and confirm the pump applied it.

set_proportional_pressure(value_m) async

Set proportional pressure mode with setpoint.

Parameters:

Name	Type	Description	Default
value_m	float	Pressure setpoint in meters of water column	required

Returns:

Type	Description
bool	True if successful, False otherwise

set_setpoint(mode, value) async

Set a mode's setpoint and report what the pump stored.

Switches the pump into mode as well - the two share one write, so this cannot merely edit a stored value in the background.

A clamped result is normal and not an error: this pump stores 1650 for a request of 600 RPM and 3671 for 4400, the ends of its own limits.

set_temperature_control(on_temp_c, off_temp_c, heating_type='radiator') async

Set Temperature Control mode with on/off temperature setpoints.

This mode maintains hot water temperature with AutoAdapt flow adjustment (1-4 gpm). The pump turns on when temperature drops below on_temp and turns off when it reaches off_temp.

Parameters:

Name	Type	Description	Default
on_temp_c	float	Turn-on temperature threshold in Celsius (e.g., 35.0)	required
off_temp_c	float	Turn-off temperature threshold in Celsius (e.g., 39.0)	required
heating_type	str	System type - "radiator" (Mode 13), "underfloor" (Mode 14), or "combined" (Mode 15). Default: "radiator"	'radiator'

Returns:

Type	Description
bool	True if successful, False otherwise

Example

await control.set_temperature_control(35.0, 39.0) # Radiator system # doctest: +SKIP await control.set_temperature_control(35.0, 39.0, "underfloor") # doctest: +SKIP

Note

For ALPHA HWR pumps, all heating_type variants likely behave the same (hot water recirculation), but the mode selection is available for compatibility with the GENI protocol.

set_temperature_range(min_temp, max_temp, autoadapt=None) async

Set the temperature range and confirm it, preserving the rest.

autoadapt=None keeps the pump's current setting; the three fields share one write, so a default here would silently change it.

set_temperature_range_control(min_temp, max_temp, autoadapt=None) async

Set temperature range control mode (Mode 27) with min/max setpoints.

Parameters:

Name	Type	Description	Default
min_temp	float	Minimum temperature in Celsius	required
max_temp	float	Maximum temperature in Celsius	required
autoadapt	bool | None	If True, enables automatic flow adjustment (1-4 gpm). If False, uses fixed flow limits. If omitted (the default), the pump's current setting is preserved - the three fields share one write, so passing a default here silently turned AutoAdapt back on every time a bound was adjusted.	None

Returns:

Type	Description
bool	True if successful, False otherwise

Example

await control.set_temperature_range_control(35.0, 45.0, autoadapt=True) # doctest: +SKIP

start(mode=None) async

Start the pump.

Uses the Class 3 START command, which changes only the run state. The pump keeps its current mode and that mode's stored setpoint.

Parameters:

Name	Type	Description	Default
mode	int | None	Optional control mode to switch to first. Sent as a separate, unfused mode change rather than folded into the start command.	None

Returns:

Type	Description
bool	True if successful, False otherwise

stop(mode=None) async

Stop the pump.

Uses the Class 3 STOP command, which changes only the run state.

Parameters:

Name	Type	Description	Default
mode	int | None	Optional control mode to switch to first. Sent as a separate, unfused mode change.	None

Returns:

Type	Description
bool	True if successful, False otherwise

sync_cache() async

Read the pump's stored configuration into this service.

Several writes carry more fields than the caller sets - the temperature range writes min, max and AutoAdapt together, the cycle config writes both periods and a flow - so they have to know what the pump currently holds. Reading it once here, after authentication, is what lets those writes preserve the fields they were not asked to change instead of inventing them.

Returns:

Type	Description
bool	True if everything needed was read.

ScheduleService

Manage weekly operation schedules across 5 independent layers with full CRUD operations.

Bases: BaseService

Manages pump schedule operations.

Handles reading, writing, and validation of weekly pump schedules. The pump supports up to 5 schedule layers, with each layer containing one time interval per day of the week.

Example

service = ScheduleService(session, transport) # doctest: +SKIP

Check if schedule is enabled

enabled = await service.get_state() # doctest: +SKIP print(f"Schedule enabled: {enabled}") # doctest: +SKIP

Read current schedule

entries = await service.read_entries() # doctest: +SKIP for entry in entries: # doctest: +SKIP ... print(f"{entry.day}: {entry.begin_time}-{entry.end_time}")

Write new schedule

new_entries = [ ... ScheduleEntry(day="Monday", begin_hour=6, begin_minute=0, ... end_hour=8, end_minute=0), ... ScheduleEntry(day="Tuesday", begin_hour=6, begin_minute=0, ... end_hour=8, end_minute=0), ... ] success = await service.write_entries(new_entries, layer=0) # doctest: +SKIP

Enable schedule

await service.enable() # doctest: +SKIP

last_read_incomplete property

Whether the last :meth:read_entries missed a layer.

An unread layer is not an empty one, and the difference matters to anything that writes the result back.

__init__(session, transport)

Initialize the schedule service.

Parameters:

Name	Type	Description	Default
session	Session	Session manager for authentication state	required
transport	Transport	BLE transport layer for communication	required

clear_entry(day, layer=0) async

Clear (disable) a schedule entry for a specific day.

This disables the schedule for a specific day on the specified layer, but does not affect other days or layers.

Parameters:

Name	Type	Description	Default
day	str	Day name (Monday-Sunday)	required
layer	int	Schedule layer (0-4)	0

Returns:

Type	Description
bool	True if successfully cleared, False otherwise

Raises:

Type	Description
ConnectionError	If not connected or not authenticated
ValueError	If day name or layer is invalid

Example

Clear Monday's schedule on layer 0

success = await service.clear_entry("Monday", layer=0) # doctest: +SKIP if success: # doctest: +SKIP ... print("Monday schedule cleared")

Implementation Notes

This reads the current schedule for the layer, sets the specified day's entry to disabled, and writes it back.

disable() async

Disable the internal schedule.

Deactivates the pump's built-in schedule functionality. The pump will continue operating according to its current mode, but will not automatically start/stop based on the schedule.

Returns:

Type	Description
bool	True if successfully disabled, False otherwise

Raises:

Type	Description
ConnectionError	If not connected or not authenticated

Example

success = await service.disable() # doctest: +SKIP if success: # doctest: +SKIP ... print("Schedule disabled")

Implementation Notes

Same as enable() but with value 0x00 instead of 0x01.

enable() async

Enable the internal schedule.

Activates the pump's built-in schedule functionality. When enabled, the pump will automatically start/stop according to the programmed schedule entries.

Returns:

Type	Description
bool	True if successfully enabled, False otherwise

Raises:

Type	Description
ConnectionError	If not connected or not authenticated

Example

success = await service.enable() # doctest: +SKIP if success: # doctest: +SKIP ... print("Schedule enabled")

Implementation Notes

Protocol: Class 10, OpSpec 0x90, Object 1016 - APDU: [0x0A][0x90][SubH][SubL][0x03][0xF8][0x01] - 0x0A = Class 10 - 0x90 = OpSpec for SET operation - SubH, SubL = Discovered SubID (big-endian) - 0x03F8 = Object 1016 (big-endian) - 0x01 = Enable value

get_state() async

Get the current schedule state (enabled/disabled).

Reads Object 84 SubID 1 (ClockProgramOverview) to determine if the internal schedule is currently active.

Returns:

Type	Description
bool | None	True if enabled, False if disabled, None if failed to read

Raises:

Type	Description
ConnectionError	If not connected or not authenticated

Example

enabled = await service.get_state() # doctest: +SKIP if enabled: # doctest: +SKIP ... print("Schedule is active") ... else: ... print("Schedule is disabled")

Implementation Notes

Protocol: Class 10, Object 84, SubID 1 (ClockProgramOverview) - Response format: [Header(3)][Capabilities(4)][Enabled(1)][DefaultAction(1)][BaseSetpoint(4)] - Byte 7 is the enabled flag (0x01=enabled, 0x00=disabled) - No scanning required - SubID 1 is a fixed location

read_entries(layer=None) async

Read schedule entries from the pump.

Retrieves the current weekly schedule from one or all layers. Each layer can contain up to 7 entries (one per day of the week).

Parameters:

Name	Type	Description	Default
layer	int | None	Optional specific layer (0-4) to read. If None, reads all layers.	None

Returns:

Type	Description
list[ScheduleEntry]	List of ScheduleEntry objects. Only enabled entries are returned.

Raises:

Type	Description
ConnectionError	If not connected or not authenticated

Example

Read all layers

all_entries = await service.read_entries() # doctest: +SKIP

Read specific layer

layer0 = await service.read_entries(layer=0) # doctest: +SKIP

Display entries

for entry in all_entries: # doctest: +SKIP ... print(f"Layer {entry.layer}, {entry.day}: " ... f"{entry.begin_time}-{entry.end_time}")

Implementation Notes

Protocol: Class 10, Object 84 - SubID: 1000 + layer (1000-1004 for layers 0-4) - Response format: [Header 3 bytes] + [7 days × 6 bytes] - Total: 45 bytes

Each 6-byte entry format: - Byte 0: Enabled flag (0x01=enabled, 0x00=disabled) - Byte 1: Action code (0x02=run pump) - Byte 2: Start hour (0-23) - Byte 3: Start minute (0-59) - Byte 4: End hour (0-23) - Byte 5: End minute (0-59)

Days are in order: Mon, Tue, Wed, Thu, Fri, Sat, Sun

validate_entries(entries)

Validate a list of schedule entries for conflicts and errors.

Performs comprehensive validation including: - Time range validity (not zero duration) - No overlaps within same day/layer - Valid day names - Valid layer values (0-4)

Parameters:

Name	Type	Description	Default
entries	list[ScheduleEntry] | list[dict]	List of ScheduleEntry instances or dicts	required

Returns:

Type	Description
bool	Tuple of (is_valid, list_of_error_messages)
list[str]	(True, []) if all entries are valid
tuple[bool, list[str]]	(False, ["error1", "error2", ...]) if validation fails

Example

entries = [ ... ScheduleEntry(day="Monday", begin_hour=6, begin_minute=0, ... end_hour=8, end_minute=0), ... ScheduleEntry(day="Monday", begin_hour=7, begin_minute=0, ... end_hour=9, end_minute=0), # Overlaps! ... ] is_valid, errors = service.validate_entries(entries) # doctest: +SKIP print(is_valid) # False # doctest: +SKIP print(errors) # doctest: +SKIP ['Overlap detected: Monday layer 0: 06:00-08:00 overlaps with 07:00-09:00']

Implementation Notes

Validation logic: 1. Convert all entries to ScheduleEntry objects 2. Validate each entry's time range (not zero duration) 3. Check for overlaps between enabled entries on same day/layer 4. Warn if too many entries per day/layer combination

write_entries(entries, layer=0) async

Write schedule entries to the pump.

Writes a complete weekly schedule to the specified layer. Each layer can contain up to 7 entries (one per day). Entries are validated before writing to ensure no overlaps or invalid time ranges.

Parameters:

Name	Type	Description	Default
entries	list[ScheduleEntry] | list[dict]	List of ScheduleEntry objects or dicts with schedule data	required
layer	int	Schedule layer to write to (0-4)	0

Returns:

Type	Description
bool	True if successfully written, False otherwise

Raises:

Type	Description
ConnectionError	If not connected or not authenticated
ValueError	If layer is invalid (not 0-4)

Example

entries = [ ... ScheduleEntry(day="Monday", begin_hour=6, begin_minute=0, ... end_hour=8, end_minute=0, layer=0), ... ScheduleEntry(day="Tuesday", begin_hour=6, begin_minute=0, ... end_hour=8, end_minute=0, layer=0), ... ] success = await service.write_entries(entries, layer=0) # doctest: +SKIP if success: # doctest: +SKIP ... print("Schedule written successfully")

Implementation Notes

Protocol: Class 10, Object 84, OpSpec 0xB3 - SubID: 1000 + layer - Payload: 42 bytes (7 days × 6 bytes) - APDU format:

[0x0A]              # Class 10
[0xB3]              # OpSpec 5
[84]                # Object ID
[SubH][SubL]        # SubID (big-endian)
[0x00]              # Reserved
[0xDE][0x01][0x00]  # Type 222 header
[0x00][0x2A]        # Size (42 bytes)
[42 bytes data]     # Schedule entries

DeviceInfoService

Read device identification information, firmware versions, and cumulative statistics.

Bases: BaseService

Service for reading device information and metadata.

This service provides APIs for accessing device identification, version information, and operational statistics.

Example

from alpha_hwr.services import DeviceInfoService # doctest: +SKIP

Initialize

device_info = DeviceInfoService(transport, session) # doctest: +SKIP

Read basic info (no connection needed)

info = await device_info.read_basic() # doctest: +SKIP print(f"Product: {info.product_family}/{info.product_type}") # doctest: +SKIP

Read detailed info (requires connection)

info = await device_info.read_detailed() # doctest: +SKIP print(f"Serial: {info.serial_number}") # doctest: +SKIP print(f"SW Version: {info.software_version}") # doctest: +SKIP

__init__(transport, session, address=None, cached_product_info=None)

Initialize device info service.

Parameters:

Name	Type	Description	Default
transport	Transport	Transport layer for BLE communication	required
session	Session	Session manager for state tracking	required
address	str | None	Optional BLE device address for reading advertisement data	None
cached_product_info	dict[str, int] | None	Optional cached product info from advertisement scan	None

read_alarms() async

Read current alarm state.

Reads active alarms and warnings from Class 10 Object 88: - Sub-ID 0: Active alarms - Sub-ID 11: Active warnings

Returns:

Type	Description
AlarmInfo | None	AlarmInfo with active alarm/warning codes, or None if read failed

Example

alarms = await device_info.read_alarms() # doctest: +SKIP if alarms.active_alarms: # doctest: +SKIP ... print(f"Active alarms: {alarms.active_alarms}") if alarms.active_warnings: # doctest: +SKIP ... print(f"Active warnings: {alarms.active_warnings}")

Implementation Notes

• Object 88, Sub-ID 0: Active alarms (Type 570)
• Object 88, Sub-ID 11: Active warnings (Type 570)
• Format: Array of uint16 codes
• Zero codes indicate "no alarm/warning"

read_basic(address) async

Read basic device info from BLE advertisement.

This method scans for the device and extracts product information from the BLE service data. No connection is required.

Parameters:

Name	Type	Description	Default
address	str	BLE MAC address of the device	required

Returns:

Type	Description
DeviceInfo | None	DeviceInfo with product_family, product_type, product_version,
DeviceInfo | None	or None if scan failed

Example

info = await device_info.read_basic("AA:BB:CC:DD:EE:FF") # doctest: +SKIP print(f"Product family: {info.product_family}") # doctest: +SKIP print(f"Product type: {info.product_type}") # doctest: +SKIP print(f"Product version: {info.product_version}") # doctest: +SKIP

Implementation Notes

• GENI service UUID: 0000fdd0-0000-1000-8000-00805f9b34fb
• Service data format: [??][??][??][Family][Type][Version]...
• Can be called without being connected
• Uses BleakScanner to discover devices

read_detailed() async

Read detailed device info via Class 7 string parameters.

This method reads device identification strings including: - Serial number - Software version - Hardware version - BLE version

Requires an active authenticated connection.

Returns:

Type	Description
DeviceInfo | None	DeviceInfo with all available fields, or None if read failed

Raises:

Type	Description
ConnectionError	If not connected or not authenticated

Example

info = await device_info.read_detailed() # doctest: +SKIP print(f"Serial: {info.serial_number}") # doctest: +SKIP print(f"SW Version: {info.software_version}") # doctest: +SKIP print(f"HW Version: {info.hardware_version}") # doctest: +SKIP

Implementation Notes

• Uses Class 7 ReadString command (0x07, 0x01)
• String IDs: 9=serial, 50=sw_ver, 52=hw_ver, 58=ble_ver
• Response format: [STX][LEN][DST][SRC][0x07][Count][String][CRC]
• six header bytes, then the text
• Strings are UTF-8 encoded, null-terminated

read_info() async

Read complete device information (combined basic + detailed).

This is a convenience method that reads both basic info from BLE advertisement and detailed info from Class 7 strings.

Returns:

Type	Description
DeviceInfo | None	DeviceInfo with all available fields, or None if read failed

Example

info = await device_info.read_info() # doctest: +SKIP print(f"Product: {info.product_family}/{info.product_type}") # doctest: +SKIP print(f"Serial: {info.serial_number}") # doctest: +SKIP print(f"SW Version: {info.software_version}") # doctest: +SKIP

read_statistics() async

Read device operational statistics.

Reads statistics from Class 10 Object 93, Sub-ID 1: - Total runtime hours - Start count

Energy consumption (kWh) is NOT available on ALPHA HWR.

Object 77 (cumulative energy) is not implemented. Only instantaneous power (W) is available via telemetry.

Returns:

Type	Description
Statistics | None	Statistics object with available data, or None if read failed

Example

stats = await device_info.read_statistics() # doctest: +SKIP print(f"Runtime: {stats.operating_hours} hours") # doctest: +SKIP print(f"Starts: {stats.start_count}") # doctest: +SKIP

Implementation Notes

• Object 93, Sub-ID 1 (Type 248: operation_history_pump_obj)
• Format: [starts(4)][starts_1h(2)][starts_24h(2)][operating_time(4)]...
• Response has 3-byte header [00 00 XX]
• operating_time is in seconds, convert to hours

ConfigurationService

Backup and restore complete pump configurations to JSON files.

Manages pump configuration backup and restore.

Provides JSON-based backup/restore functionality for the complete pump configuration including control mode, setpoint, and schedule.

Example

config_service = ConfigurationService( # doctest: +SKIP ... device_info_service, ... control_service, ... schedule_service ... )

Backup configuration

success = await config_service.backup("pump_backup.json") # doctest: +SKIP if success: # doctest: +SKIP ... print("Configuration backed up successfully")

Restore configuration

success = await config_service.restore( # doctest: +SKIP ... "pump_backup.json", ... restore_mode=True, ... restore_schedule=True, ... verify_device=True ... )

__init__(device_info_service, control_service, schedule_service)

Initialize the configuration service.

Parameters:

Name	Type	Description	Default
device_info_service	DeviceInfoService	Service for reading device information	required
control_service	ControlService	Service for control mode operations	required
schedule_service	ScheduleService	Service for schedule operations	required

backup(filepath) async

Backup the complete pump configuration to a JSON file.

Creates a comprehensive backup including: - Device information (serial, product name, versions) - Control mode and current setpoint - Schedule enabled status and all entries

Parameters:

Name	Type	Description	Default
filepath	str	Path to the output JSON file	required

Returns:

Type	Description
bool	True if successful, False otherwise

Raises:

Type	Description
IOError	If file cannot be written

Example

success = await service.backup("pump_backup.json") # doctest: +SKIP if success: # doctest: +SKIP ... print("Backup saved")

Implementation Notes

JSON structure: { "version": "1.0", "timestamp": "2026-01-30T12:34:56Z", "device": { "serial_number": "12345678", "product_name": "ALPHA HWR 15-60", "hardware_version": "1.0", "software_version": "2.3" }, "control_mode": { "mode_name": "CONSTANT_PRESSURE", "setpoint": 4.5, "max_setpoint": null, "setpoint_unit": "m" }, "schedule": { "enabled": true, "days": [ { "day": "Monday", "begin_hour": 6, "begin_minute": 0, "end_hour": 8, "end_minute": 0, "action": 2, "layer": 0, "enabled": true } ] } }

export_json(filepath) async

Export configuration as JSON (alias for backup).

This is provided for API consistency with import_json.

Parameters:

Name	Type	Description	Default
filepath	str	Path to the output JSON file	required

Returns:

Type	Description
bool	True if successful, False otherwise

Example

await service.export_json("config.json") # doctest: +SKIP

import_json(filepath, restore_mode=True, restore_schedule=True, verify_device=True) async

Import configuration from JSON (alias for restore).

This is provided for API consistency with export_json.

Parameters:

Name	Type	Description	Default
filepath	str	Path to the backup JSON file	required
restore_mode	bool	Restore control mode and setpoint	True
restore_schedule	bool	Restore schedule configuration	True
verify_device	bool	Verify device serial number matches	True

Returns:

Type	Description
bool	True if successful, False otherwise

Example

await service.import_json("config.json") # doctest: +SKIP

restore(filepath, restore_mode=True, restore_schedule=True, verify_device=True) async

Restore pump configuration from a JSON backup file.

Restores the pump to a previously saved configuration state. Individual restore options can be enabled/disabled.

Parameters:

Name	Type	Description	Default
filepath	str	Path to the backup JSON file	required
restore_mode	bool	Restore control mode and setpoint (default True)	True
restore_schedule	bool	Restore schedule configuration (default True)	True
verify_device	bool	Verify device serial number matches backup (default True)	True

Returns:

Type	Description
bool	True if successful, False otherwise

Raises:

Type	Description
FileNotFoundError	If backup file doesn't exist
ValueError	If backup version is unsupported
ConnectionError	If not connected to pump

Example

Restore everything

success = await service.restore("pump_backup.json") # doctest: +SKIP

Restore only schedule

success = await service.restore( # doctest: +SKIP ... "pump_backup.json", ... restore_mode=False, ... verify_device=False ... )

Implementation Notes

Restore process: 1. Read and validate backup file 2. Verify backup version compatibility 3. Optionally verify device serial number 4. Restore control mode and setpoint if requested 5. Restore schedule entries and enabled state if requested

TimeService

Read and synchronize the pump's real-time clock.

Bases: BaseService

Service for managing pump real-time clock (RTC).

This service provides APIs for reading and synchronizing the pump's internal clock. The RTC is used for schedule execution and event logging.

Example

from alpha_hwr.services import TimeService

Initialize

time_service = TimeService(transport, session) # doctest: +SKIP

Read pump time

pump_time = await time_service.get_clock() # doctest: +SKIP print(f"Pump time: {pump_time}") # doctest: +SKIP

Sync with system time

success = await time_service.set_clock() # doctest: +SKIP if success: # doctest: +SKIP ... print("Clock synchronized")

Set to specific time

from datetime import datetime dt = datetime(2026, 12, 25, 10, 0, 0) await time_service.set_clock(dt) # doctest: +SKIP

__init__(transport, session)

Initialize time service.

Parameters:

Name	Type	Description	Default
transport	Transport	Transport layer for BLE communication	required
session	Session	Session manager for state tracking	required

get_clock() async

Read the pump's internal real-time clock.

Reads from Object 94, SubID 101 (DateTimeActual, Type 322). Returns the current pump time as a datetime object.

Returns:

Type	Description
datetime | None	Current pump time as datetime, or None if read failed or clock is unset.
datetime | None	If clock is unset (year < 1970), returns epoch time (1970-01-01 00:00:00).

Raises:

Type	Description
ConnectionError	If not connected

Example

pump_time = await time_service.get_clock() # doctest: +SKIP if pump_time: # doctest: +SKIP ... if pump_time.year < 1980: ... print("Clock is unset, needs sync") ... else: ... print(f"Pump time: {pump_time.strftime('%Y-%m-%d %H:%M:%S')}")

Implementation Notes

• Uses Class 10 GET on Object 94, SubID 101
• Response format: [Status(2)][Length(1)][Year(2)][Month(1)][Day(1)][Hour(1)][Minute(1)][Second(1)]
• Status 0x0000 = valid, 0xFFFF = unset
• Year is big-endian uint16
• Invalid dates (year < 1970, month/day = 0) indicate unset clock

set_clock(dt=None) async

Synchronize the pump's internal real-time clock.

Sends a standard Class 10 SET to SubID 0x5E00 (Object 94), ObjID 0x6401 (SubID 100 = DateTimeConfig) with a Type 322 data payload.

Parameters:

Name	Type	Description	Default
dt	datetime | None	Datetime to set. If None, uses current LOCAL system time.	None

Returns:

Type	Description
bool	True if clock was successfully set, False otherwise.

Raises:

Type	Description
ConnectionError	If not connected or not authenticated.

Example

Sync with system time

await time_service.set_clock() # doctest: +SKIP

Set to specific time

from datetime import datetime dt = datetime(2026, 1, 30, 11, 35, 0) await time_service.set_clock(dt) # doctest: +SKIP

Implementation Notes

• Uses build_data_object_set(0x5E00, 0x6401, data)
• Data format (16 bytes): Type 322 header (6) + [Year(2BE)][Month][Day][Hour][Min][Sec] + padding (3)
• Type 322 header is constant: 41 02 00 00 0B 01
• Pump responds with Class 10 ACK (OpSpec 0x01)

HistoryService

Access historical trend data for flow, head, temperature, and power over the last 100 cycles.

Bases: BaseService

Service for accessing historical trend data from the pump.

This service provides methods to retrieve and parse historical measurements including flow, head, and temperature data over the last 10 and 100 cycles.

Example

from alpha_hwr.services import HistoryService

Initialize

history = HistoryService(transport, session) # doctest: +SKIP

Get all trend data

trends = await history.get_trend_data() # doctest: +SKIP if trends.flow_series: # doctest: +SKIP ... print(f"Current flow: {trends.flow_series.cycle_10_points[0].value} m³/h")

Get cycle timestamps

timestamps = await history.get_cycle_timestamps(count=10) # doctest: +SKIP print(f"Last cycle: {timestamps[0]}") # doctest: +SKIP

__init__(transport, session)

Initialize history service.

Parameters:

Name	Type	Description	Default
transport	Transport	Transport layer for BLE communication	required
session	Session	Session manager for state tracking	required

get_cycle_timestamps(count=10) async

Get timestamps of recent pump cycles.

Parameters:

Name	Type	Description	Default
count	int	Number of timestamps to retrieve (10 or 100)	10

Returns:

Type	Description
list[datetime] | None	List of datetime objects for each cycle, or None if read failed.
list[datetime] | None	Most recent cycle is first in list.

Example

timestamps = await history.get_cycle_timestamps(count=10) # doctest: +SKIP if timestamps: # doctest: +SKIP ... print(f"Last cycle: {timestamps[0]}") ... print(f"Cycle 10 ago: {timestamps[-1]}")

get_trend_data() async

Retrieve all historical trend data from the pump.

Fetches: - 10-cycle timestamp map (Obj 88, Sub 13300) - 100-cycle timestamp map (Obj 88, Sub 13301) - Flow history (Obj 53, Sub 451) - Head history (Obj 53, Sub 452) - Media temperature history (Obj 53, Sub 453)

Returns:

Type	Description
TrendDataCollection | None	TrendDataCollection with all series, or None if retrieval failed.

Example

trends = await history.get_trend_data() # doctest: +SKIP if trends and trends.flow_series: # doctest: +SKIP ... for point in trends.flow_series.cycle_10_points: ... print(f"{point.timestamp}: {point.value} m³/h")

EventLogService

Access the pump's event log containing the last 20 pump events with timestamps.

Bases: BaseService

Service for accessing pump event log.

The pump maintains a circular buffer of 20 event log entries that record historical events such as pump start/stop cycles, mode changes, and errors.

Example

from alpha_hwr.services import EventLogService

Initialize

event_log = EventLogService(transport, session) # doctest: +SKIP

Get all entries

entries = await event_log.get_all_entries() # doctest: +SKIP for entry in entries: # doctest: +SKIP ... print(f"{entry.timestamp}: Cycle {entry.cycle_counter}")

Get single entry

newest = await event_log.get_entry(0) # doctest: +SKIP oldest = await event_log.get_entry(19) # doctest: +SKIP

__init__(transport, session)

Initialize event log service.

Parameters:

Name	Type	Description	Default
transport	Transport	Transport layer for BLE communication	required
session	Session	Session manager for state tracking	required

get_all_entries() async

Read all event log entries from the pump.

Returns:

Type	Description
list[EventLogEntry]	List of EventLogEntry objects, ordered from newest (0) to
list[EventLogEntry]	oldest (19). An entry the pump will not return is skipped -
list[EventLogEntry]	that is ordinary, since a log with fewer than twenty entries
list[EventLogEntry]	reports the empty slots as unreadable.

Raises:

Type	Description
ConnectionError	The link dropped part-way through. The entries read so far are discarded rather than returned, because a partial read and a short log are indistinguishable once the list is handed back - "Retrieved 5/20" is exactly what a five-entry log looks like.

Example

entries = await event_log.get_all_entries() # doctest: +SKIP print(f"Retrieved {len(entries)} event log entries") # doctest: +SKIP for entry in entries[:5]: # Show 5 most recent # doctest: +SKIP ... print(f" {entry.timestamp}: Cycle {entry.cycle_counter}")

get_cycle_timestamps(count=10) async

Get timestamps of recent pump cycles from cycle timestamp map.

.. deprecated:: Use client.history.get_cycle_timestamps(count) instead. This method will be removed in a future release.

Parameters:

Name	Type	Description	Default
count	int	Number of timestamps to retrieve (10 or 100)	10

Returns:

Type	Description
list[datetime] | None	List of datetime objects, or None if read failed

get_entry(index) async

Read a single event log entry from the pump.

Parameters:

Name	Type	Description	Default
index	int	Entry index (0 = newest, 19 = oldest)	required

Returns:

Type	Description
EventLogEntry | None	EventLogEntry object, or None if read failed

Raises:

Type	Description
ValueError	If index is not in range 0-19
ConnectionError	If not connected

Example

Get newest entry

entry = await event_log.get_entry(0) # doctest: +SKIP if entry: # doctest: +SKIP ... print(f"Last event: {entry.timestamp}")

get_metadata() async

Read event log metadata from the pump.

The metadata (SubID 10199) contains information about the current cycle counter and available entries.

Structure (7 bytes): - Bytes 0-1: Current cycle counter (uint16 BE) - Bytes 2-3: Available entries (uint16 BE) - Bytes 4-5: Max buffer size (uint16 BE, always 20) - Byte 6: Reserved/flags (uint8, typically 0)

Returns:

Type	Description
EventLogMetadata | None	EventLogMetadata object with decoded fields, or None if read failed

Example

metadata = await event_log.get_metadata() # doctest: +SKIP if metadata: # doctest: +SKIP ... print(f"Current cycle: {metadata.current_cycle}") ... print(f"Available entries: {metadata.available_entries}")

SingleEventService

One-off scheduled windows and vacations. See Run State and Schedules — in particular the local-Unix timestamp rule, which cannot be caught by verification.

Bases: BaseService

Reads and writes the pump's one-off schedule entries.

The number of slots is taken from the schedule overview rather than assumed: the pump reports its own capacity, and it is not the 35 that the sub-id range would suggest - the unit this was written against exposes 5, and reading past them simply goes unanswered.

build_apdu(slot, begin, end, action=ACTION_RUN, enabled=True)

Build the write frame for one slot.

[0A][93][54][SubH][SubL][00][DC][01][00][00][0A][enabled][action] [begin u32 BE][end u32 BE] - the object is addressed first as a single byte here, then a 16-bit sub-id.

The head is 0x93: SET, with the 19 payload bytes that follow it. It was 0xB3 - SET with 51 - borrowed from the schedule layer write, whose 53-byte APDU really does carry 51. This frame carries 19, so it declared a length it did not have.

The pump accepts both, so nothing was visibly failing; the capture corpus is what settles it. Every one of the 29 single-event writes the Grundfos GO app makes uses 0x93, and the 8 layer writes use 0xB3. A firmware that checked the field would have refused ours with no diagnostic.

clear(slot) async

Empty one slot.

clear_vacation() async

Clear the vacation that is running, or the next one due.

This used to clear the first enabled Stop event in slot order, with no reference to the clock. A finished vacation sitting in an early slot therefore shadowed a live one later on: the call reported success, and the pump stayed off.

find_free_slot one method up has always been clocked. The asymmetry was the bug.

confirm(slot, begin, end, action) async

Read a slot back and check the pump kept what was asked for.

The ACTION byte is compared, and that is the point of this method. It is half the meaning of a single event - 0x01 holds the pump off across the window (which is what a vacation is), 0x02 runs it once - and a confirm that checked only the window and the enabled flag would settle a vacation as written while the pump was scheduled to run for a week, or the reverse.

Not used on a clear: clearing disables the slot whatever it held, so there is no requested action to compare against.

find_free_slot() async

A slot that can be written without losing anything.

Prefers a genuinely empty one, and only falls back to a slot whose event has already finished. Both are safe to take, but an expired event is still a record of something the owner scheduled, so it is not overwritten while an untouched slot exists. Without the fallback the pool would exhaust and never recover, since the pump does not clear events once they pass.

Reads every slot first. Choosing without looking is how slot 0 gets handed out over a live event, since an unread slot looks empty.

read(slot) async

Read one slot.

Returns None if the slot cannot be read - which includes slots past the pump's capacity, where it answers with a short error frame rather than data.

read_all() async

Read every slot the pump has.

All-or-nothing: a partial read is not published, because an unread slot looks free, and handing one out would overwrite a live event that was simply never seen. Slots that read back empty are a legitimate result and not a failure - emptiness says nothing about whether the read worked.

set_vacation(begin, end) async

Hold the pump off across a date range.

A vacation is a Stop single event: it overrides the weekly schedule for its window.

slot_count() async

How many single-event slots this pump has.

Read from ClockProgramOverview, because it varies by model - the unit this was written against reports 5 - and clamped to :data:SLOT_LIMIT, because a wire-supplied number should not be a loop bound with no ceiling.

Callers turn this straight into one Class 10 read per slot, so a pump reporting 255 would spend minutes walking sub-ids that cannot hold a single event. The ceiling is not a judgement about what is reasonable: the sub-id is 900 + slot and the weekly schedule's layer records begin at 1000, so slot 100 addresses layer 0 and anything past 99 is a different object however the pump counts.

See esphome-alpha-hwr#284, where the same shape has two bytes behind it rather than one and can ask for 65,535 reads.

write(slot, begin, end, action=ACTION_RUN, confirm=True) async

Write one slot and commit it.

Parameters:

Name	Type	Description	Default
slot	int	Which slot to use.	required
begin	datetime	Wall clock the window opens. Naive, and taken as local.	required
end	datetime	Wall clock it closes.	required
action	int	:data:ACTION_RUN for a one-off run, :data:ACTION_STOP for a vacation.	ACTION_RUN
confirm	bool	Read the slot back and check the pump kept it, including the ACTION byte. Pass False only when the caller is going to verify some other way.	True

Returns:

Type	Description
bool	True if the pump holds what was asked for. With confirm
bool	off, only that the frame was sent and committed.

alpha_hwr.services.single_event.SingleEvent dataclass

One scheduled one-off window.

is_vacation property

True when this event holds the pump off rather than running it.

alpha_hwr.services.single_event.to_pump_time(when)

Encode a wall clock the way the pump stores it.

Takes a naive datetime as local, which is the only reading that makes sense for a schedule, and stamps its fields as though they were UTC.

Raises:

Type	Description
ValueError	The instant is outside the uint32 range the wire field can carry.

Examples:

>>> to_pump_time(datetime(2026, 8, 20, 9, 30))
1787218200

alpha_hwr.services.single_event.from_pump_time(value)

Decode a stored timestamp back to the wall clock it denotes.

Naive by design, and the exact inverse of :func:to_pump_time. The pump stores no offset, so attaching one here would invent information - and a UTC-labelled value is worse than a naive one, because astimezone() will then shift it by the local offset and produce a time the pump never meant.

Examples:

>>> from_pump_time(1787218200)
datetime.datetime(2026, 8, 20, 9, 30)
>>> from_pump_time(to_pump_time(datetime(2026, 11, 1, 1, 30)))
datetime.datetime(2026, 11, 1, 1, 30)

WriteOperationService

The single serialized path every verified write passes through. Callers normally reach it through client.control's verified setters rather than directly. See Verified Writes.

Serializes writes, confirms them, and reports one result each.

Not constructed directly by callers - the client owns one and the service methods submit through it.

on_disconnect() async

Settle everything pending after the link drops.

Must run before the transport tears its queue down, or the callbacks these operations are waiting on are discarded and their callers wait forever.

submit(command, resource, **args) async

Queue a write and wait for its settled result.

A newer write to the same resource supersedes any still waiting to start - last write wins - but never interrupts one already on the wire, which would leave the pump half-written.

Run state

Pure logic — no I/O — describing how the run flag and the schedule flag combine, including the one combination that can never run.

alpha_hwr.services.run_state.RunState

Bases: StrEnum

The states the run flag and schedule flag can express together.

ENGAGED = 'engaged' class-attribute instance-attribute

Running, schedule off: the pump follows its control mode continuously.

Whether the motor actually spins is then a question for the mode - Temperature and Cycle-Time both idle between their own cycles.

OFF = 'off' class-attribute instance-attribute

Stopped, schedule off. The pump is idle and will stay that way.

SCHEDULED = 'scheduled' class-attribute instance-attribute

Running, schedule on: the pump runs inside its windows and idles between.

STALLED = 'stalled' class-attribute instance-attribute

Stopped, schedule on. Nothing will ever run.

The schedule is armed and the pump ignores it, so every window passes with the motor idle. Reachable by setting the two flags independently, which is why it is worth detecting rather than assuming away.

alpha_hwr.services.run_state.run_state(enabled, schedule_enabled)

Name the state the two flags put the pump in.

alpha_hwr.services.run_state.is_stalled(enabled, schedule_enabled)

True for the combination that can never run.