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.clock	Real-time clock management
HistoryService	client.history	Historical trend data (100 cycles)
EventLogService	client.events	Pump event history (20 entries)

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) session = Session(transport) telemetry_service = TelemetryService(transport, session)

Read once

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

Stream continuously

async for data in telemetry_service.stream(): ... 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 print(f"Converter temp: {adv.converter_temperature_c}°C")

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 print(f"Voltage: {telemetry.voltage_ac_v}V")

__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() print(f"Flow: {data.flow_m3h} m³/h") print(f"Power: {data.power_w} W")

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): ... 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 including starting, stopping, and setting control modes with automatic validation.

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 from alpha_hwr.services import ControlService from alpha_hwr.constants import ControlMode

Initialize

control = ControlService(transport, session)

Start pump

await control.start()

Set constant pressure mode

await control.set_constant_pressure(1.5) # 1.5 meters

Stop pump

await control.stop()

__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	Optional['ScheduleService']	Optional schedule service for status reading	None

disable_remote_mode() async

Disable remote control mode (return to Auto).

Disables remote control mode (Class 3 command ID 6), returning the pump to automatic operation. The pump will resume normal operation based on its internal logic and local controls.

Returns:

Type	Description
bool	True if remote mode was disabled successfully, False otherwise

Example

await control.disable_remote_mode()

Pump returns to automatic operation

Implementation Notes

• Uses Class 3 command: [0x03, 0xC1, 0x06]
• Service ID: 0xE7, Source: 0xF8

enable_remote_mode() async

Enable remote control mode.

Enables remote control mode (Class 3 command ID 7), allowing external control of the pump via BLE/API commands. When enabled, the pump accepts control commands and ignores local controls.

Returns:

Type	Description
bool	True if remote mode was enabled successfully, False otherwise

Example

await control.enable_remote_mode()

Now you can send control commands

await control.start()

Implementation Notes

• Uses Class 3 command: [0x03, 0xC1, 0x07]
• Service ID: 0xE7, Source: 0xF8

get_cycle_time_config() async

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

Returns:

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

get_mode() async

Get the current control mode and setpoint information.

Reads from Class 10 Object 86, Sub-ID 6 (overall_operation_local_request_obj). For Temperature Range Control (mode 27), reads from Object 91, Sub-ID 430.

Returns:

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

Example

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

Implementation Notes

• Standard modes: Object 86, Sub-ID 6, 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)

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

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).

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_flow_limit(value_gpm) async

Set the maximum flow limit to prevent noise and corrosion.

Parameters:

Name	Type	Description	Default
value_gpm	float	Maximum flow limit in GPM.	required

Returns:

Type	Description
bool	True if successful, False otherwise

set_mode(mode) async

Set the control mode without changing setpoint.

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_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_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 await control.set_temperature_control(35.0, 39.0, "underfloor")

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_control(min_temp, max_temp, autoadapt=True) 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	If True, enables automatic flow adjustment (1-4 gpm). If False, uses fixed flow limits.	True

Returns:

Type	Description
bool	True if successful, False otherwise

Example

await control.set_temperature_range_control(35.0, 45.0, autoadapt=True)

start(mode=None) async

Start the pump.

Parameters:

Name	Type	Description	Default
mode	Optional[int]	Optional control mode to use (defaults to current mode)	None

Returns:

Type	Description
bool	True if successful, False otherwise

stop(mode=None) async

Stop the pump.

Parameters:

Name	Type	Description	Default
mode	Optional[int]	Optional control mode (defaults to current mode)	None

Returns:

Type	Description
bool	True if successful, False otherwise

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)

Check if schedule is enabled

enabled = await service.get_state() print(f"Schedule enabled: {enabled}")

Read current schedule

entries = await service.read_entries() for entry in entries: ... 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)

Enable schedule

await service.enable()

__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) if success: ... 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() if success: ... 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() if success: ... 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

TypeScript: const apdu = Buffer.from([ 0x0A, 0x90, (subId >> 8) & 0xFF, subId & 0xFF, 0x03, 0xF8, 0x01 ]); return await sendCommand(apdu);

Rust: let apdu = vec![ 0x0A, 0x90, (sub_id >> 8) as u8, sub_id as u8, 0x03, 0xF8, 0x01 ]; send_command(&apdu).await

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() if enabled: ... 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

TypeScript: const data = await readClass10Object(84, 1); return data[7] === 0x01;

Rust: let data = read_class10_object(84, 1).await?; Ok(data[7] == 0x01)

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()

Read specific layer

layer0 = await service.read_entries(layer=0)

Display entries

for entry in all_entries: ... 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

TypeScript: const data = await readClass10Object(84, 1000 + layer); const entries = []; const payload = data.slice(3); // Skip header for (let day = 0; day < 7; day++) { const offset = day * 6; const entry = payload.slice(offset, offset + 6); if (entry[0] === 0x01) { // If enabled entries.push(parseScheduleEntry(entry, day)); } }

Rust: let data = read_class10_object(84, 1000 + layer).await?; let payload = &data[3..]; // Skip header let mut entries = Vec::new(); for day in 0..7 { let offset = day * 6; let entry = &payload[offset..offset+6]; if entry[0] == 0x01 { entries.push(parse_schedule_entry(entry, day)?); } }

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) print(is_valid) # False print(errors) ['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

TypeScript: function validateSchedule(entries: ScheduleEntry[]): [boolean, string[]] { const errors: string[] = [];

// Validate time ranges
for (const entry of entries) {
  if (entry.getDuration() === 0) {
    errors.push(`Invalid time range: ${entry.day} ${entry.beginTime}`);
  }
}

// Check overlaps
const enabled = entries.filter(e => e.enabled);
for (let i = 0; i < enabled.length; i++) {
  for (let j = i + 1; j < enabled.length; j++) {
    if (enabled[i].overlapsWith(enabled[j])) {
      errors.push(`Overlap: ${enabled[i].day} ...`);
    }
  }
}

return [errors.length === 0, errors];

}

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) if success: ... 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

TypeScript: const payload = Buffer.alloc(42); // Initialize with zeros for (const entry of entries) { const dayIdx = getDayIndex(entry.day); const offset = dayIdx * 6; payload[offset + 0] = entry.enabled ? 0x01 : 0x00; payload[offset + 1] = entry.action; payload[offset + 2] = entry.begin_hour; payload[offset + 3] = entry.begin_minute; payload[offset + 4] = entry.end_hour; payload[offset + 5] = entry.end_minute; }

const apdu = Buffer.concat([ Buffer.from([0x0A, 0xB3, 84, subH, subL, 0x00]), Buffer.from([0xDE, 0x01, 0x00, 0x00, 0x2A]), payload ]);

Rust: let mut payload = vec![0u8; 42]; for entry in entries { let day_idx = get_day_index(&entry.day); let offset = day_idx * 6; payload[offset] = if entry.enabled { 0x01 } else { 0x00 }; payload[offset + 1] = entry.action; payload[offset + 2] = entry.begin_hour; payload[offset + 3] = entry.begin_minute; payload[offset + 4] = entry.end_hour; payload[offset + 5] = entry.end_minute; }

let mut apdu = vec![ 0x0A, 0xB3, 84, sub_h, sub_l, 0x00, 0xDE, 0x01, 0x00, 0x00, 0x2A ]; apdu.extend_from_slice(&payload);

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

Initialize

device_info = DeviceInfoService(transport, session)

Read basic info (no connection needed)

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

Read detailed info (requires connection)

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

__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
Optional[AlarmInfo]	AlarmInfo with active alarm/warning codes, or None if read failed

Example

alarms = await device_info.read_alarms() if alarms.active_alarms: ... print(f"Active alarms: {alarms.active_alarms}") if alarms.active_warnings: ... 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
Optional[DeviceInfo]	DeviceInfo with product_family, product_type, product_version,
Optional[DeviceInfo]	or None if scan failed

Example

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

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
Optional[DeviceInfo]	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() print(f"Serial: {info.serial_number}") print(f"SW Version: {info.software_version}") print(f"HW Version: {info.hardware_version}")

Implementation Notes

• Uses Class 7 ReadString command (0x07, 0x01)
• String IDs: 9=serial, 50=sw_ver, 52=hw_ver, 58=ble_ver
• Response format: [Frame Header][String Data...][CRC]
• 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
Optional[DeviceInfo]	DeviceInfo with all available fields, or None if read failed

Example

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

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
Optional[Statistics]	Statistics object with available data, or None if read failed

Example

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

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( ... device_info_service, ... control_service, ... schedule_service ... )

Backup configuration

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

Restore configuration

success = await config_service.restore( ... "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") if success: ... 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 } ] } }

TypeScript: interface BackupData { version: string; timestamp: string; device: DeviceInfo; control_mode: ControlModeData; schedule: ScheduleData; }

async function backup(filepath: string): Promise { const data: BackupData = { version: "1.0", timestamp: new Date().toISOString(), device: await readDeviceInfo(), control_mode: await readControlMode(), schedule: await readSchedule() };

await fs.writeFile(filepath, JSON.stringify(data, null, 2));
return true;

}

Rust: #[derive(Serialize)] struct BackupData { version: String, timestamp: String, device: DeviceInfo, control_mode: ControlModeData, schedule: ScheduleData, }

async fn backup(filepath: &str) -> Result { let data = BackupData { version: "1.0".to_string(), timestamp: Utc::now().to_rfc3339(), device: read_device_info().await?, control_mode: read_control_mode().await?, schedule: read_schedule().await?, };

let json = serde_json::to_string_pretty(&data)?;
tokio::fs::write(filepath, json).await?;
Ok(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")

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")

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")

Restore only schedule

success = await service.restore( ... "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

TypeScript: async function restore( filepath: string, options: RestoreOptions ): Promise { // Read backup file const data = JSON.parse(await fs.readFile(filepath, 'utf8'));

// Verify version
if (data.version !== "1.0") {
  throw new Error(`Unsupported version: ${data.version}`);
}

// Verify device if requested
if (options.verifyDevice) {
  const currentDevice = await readDeviceInfo();
  if (currentDevice.serial !== data.device.serial_number) {
    throw new Error("Device serial mismatch");
  }
}

// Restore mode
if (options.restoreMode && data.control_mode) {
  await setControlMode(
    data.control_mode.mode_name,
    data.control_mode.setpoint
  );
}

// Restore schedule
if (options.restoreSchedule && data.schedule) {
  const entries = data.schedule.days.map(parseScheduleEntry);

  // Group by layer
  const layers = groupBy(entries, e => e.layer);
  for (const [layer, layerEntries] of layers) {
    await writeSchedule(layerEntries, layer);
  }

  await setScheduleEnabled(data.schedule.enabled);
}

return true;

}

Rust: async fn restore( filepath: &str, options: RestoreOptions ) -> Result { // Read backup file let json = tokio::fs::read_to_string(filepath).await?; let data: BackupData = serde_json::from_str(&json)?;

// Verify version
if data.version != "1.0" {
  return Err(Error::UnsupportedVersion(data.version));
}

// Verify device
if options.verify_device {
  let current = read_device_info().await?;
  if current.serial != data.device.serial_number {
    return Err(Error::DeviceMismatch);
  }
}

// Restore mode
if options.restore_mode && let Some(mode) = data.control_mode {
  set_control_mode(&mode.mode_name, mode.setpoint).await?;
}

// Restore schedule
if options.restore_schedule && let Some(schedule) = data.schedule {
  // Group by layer
  let mut layers: HashMap<u8, Vec<ScheduleEntry>> = HashMap::new();
  for entry in schedule.days {
    layers.entry(entry.layer).or_default().push(entry);
  }

  for (layer, entries) in layers {
    write_schedule(&entries, layer).await?;
  }

  set_schedule_enabled(schedule.enabled).await?;
}

Ok(true)

}

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)

Read pump time

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

Sync with system time

success = await time_service.set_clock() if success: ... 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)

__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() if pump_time: ... 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()

Set to specific time

from datetime import datetime dt = datetime(2026, 1, 30, 11, 35, 0) await time_service.set_clock(dt)

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)

Get all trend data

trends = await history.get_trend_data() if trends.flow_series: ... 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) print(f"Last cycle: {timestamps[0]}")

__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
Optional[list[datetime]]	List of datetime objects for each cycle, or None if read failed.
Optional[list[datetime]]	Most recent cycle is first in list.

Example

timestamps = await history.get_cycle_timestamps(count=10) if timestamps: ... 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
Optional[TrendDataCollection]	TrendDataCollection with all series, or None if retrieval failed.

Example

trends = await history.get_trend_data() if trends and trends.flow_series: ... 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)

Get all entries

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

Get single entry

newest = await event_log.get_entry(0) oldest = await event_log.get_entry(19)

__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 oldest (19).
list[EventLogEntry]	Entries that fail to read will be skipped.

Example

entries = await event_log.get_all_entries() print(f"Retrieved {len(entries)} event log entries") for entry in entries[:5]: # Show 5 most recent ... 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
Optional[EventLogEntry]	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) if entry: ... 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
Optional[EventLogMetadata]	EventLogMetadata object with decoded fields, or None if read failed

Example

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