python-pooldose 0.5.0

Unoffical async Python client for SEKO PoolDose devices

python-pooldose

Unofficial async Python client for SEKO Pooldosing systems. SEKO is a manufacturer of various monitoring and control devices for Pools and Spas. This client uses an undocumented local HTTP API. It provides live readings for pool sensors such as temperature, pH, ORP/Redox, as well as status information and control over the dosing logic.

Features

• Async/await support for non-blocking operations
• Dynamic sensor discovery based on device model and firmware
• Dictionary-style access to instant values
• Structured data API with type-based organization
• Secure by default - WiFi passwords excluded unless explicitly requested
• Comprehensive error handling with detailed logging
• SSL/HTTPS support for secure communication

API Overview

Program Flow

1. Create PooldoseClient
   ├── Connect to Device
   │   ├── Fetch Device Info (Debug Config)
   │   ├── WiFi Station Info (optional)
   │   ├── Access Point Info (optional)
   │   └── Network Info
   └── Load Mapping JSON (based on MODEL_ID + FW_CODE)

2. Get Static Values
   └── Device information and configuration

3. Get Instant Values
   ├── Dictionary-style access: instant_values['temperature']
   ├── Get with default: instant_values.get('ph', default)
   ├── Check existence: 'sensor_name' in instant_values
   └── Structured access: instant_values_structured()

4. Set Values via Type Methods
   ├── set_number()
   ├── set_switch()
   └── set_select()

API Architecture

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│  PooldoseClient │────│ RequestHandler  │────│   HTTP Device   │
└─────────────────┘    └─────────────────┘    └─────────────────┘
         │                       │
         │                       ▼
         │              ┌─────────────────┐
         │              │ API Endpoints   │
         │              │ • get_debug     │
         │              │ • get_wifi      │
         │              │ • get_values    │
         │              │ • set_value     │
         │              └─────────────────┘
         │
         ▼
┌─────────────────┐    ┌─────────────────┐
│   MappingInfo   │────│  JSON Files     │
└─────────────────┘    └─────────────────┘
         │
         ▼
┌─────────────────┐    ┌─────────────────┐
│  InstantValues  │────│ Dictionary API  │
└─────────────────┘    └─────────────────┘
         │
         ▼
┌─────────────────┐
│ Structured API  │
│ • sensor{}      │
│ • number{}      │
│ • switch{}      │
│ • binary_sensor{}│
│ • select{}      │
└─────────────────┘

Prerequisites

1. Install and set-up the PoolDose devices according to the user manual.
   1. In particular, connect the device to your WiFi network.
   2. Identify the IP address or hostname of the device.
2. Browse to the IP address or hostname (default port: 80).
   1. Try to log in to the web interface with the default password (0000).
   2. Check availability of data in the web interface.
3. Optionally: Block the device from internet access to ensure cloudless-only operation.

SSL/HTTPS Support

The client supports SSL/HTTPS connections for secure communication with your PoolDose device. This is particularly useful when the device is configured for HTTPS or when connecting over untrusted networks.

Basic SSL Configuration

from pooldose.client import PooldoseClient

# Enable SSL with default settings (port 443, certificate verification enabled)
client = PooldoseClient("192.168.1.100", use_ssl=True)
status = await client.connect()

SSL Configuration Options

# Custom HTTPS port
client = PooldoseClient("192.168.1.100", use_ssl=True, port=8443)

# Disable SSL certificate verification (not recommended for production)
client = PooldoseClient("192.168.1.100", use_ssl=True, ssl_verify=False)

# Complete SSL configuration example
client = PooldoseClient(
    host="pool-device.local",
    timeout=30,
    use_ssl=True,
    port=8443,
    ssl_verify=True,  # Verify SSL certificates
    include_sensitive_data=False
)

SSL Security Considerations

• Certificate Verification: By default, SSL certificate verification is enabled (ssl_verify=True). This ensures secure connections but requires valid certificates.
• Self-signed Certificates: If your device uses self-signed certificates, set ssl_verify=False. Note that this reduces security.
• Port Configuration: Use the port parameter to specify custom HTTPS ports. Defaults to 443 for HTTPS and 80 for HTTP.
• Connection Timeouts: Consider increasing the timeout value for SSL connections as they may take longer to establish.

Migration from HTTP to HTTPS

To migrate existing code from HTTP to HTTPS:

# Before (HTTP)
client = PooldoseClient("192.168.1.100")

# After (HTTPS with SSL verification)
client = PooldoseClient("192.168.1.100", use_ssl=True)

# After (HTTPS with custom port and no verification)
client = PooldoseClient("192.168.1.100", use_ssl=True, port=8443, ssl_verify=False)

Installation

pip install python-pooldose

Example Usage

Basic Example

import asyncio
import json
from pooldose.client import PooldoseClient
from pooldose.request_status import RequestStatus

HOST = "192.168.1.100"  # Change this to your device's host or IP address
TIMEOUT = 30

async def main() -> None:
    """Demonstrate PooldoseClient usage with dictionary-based API."""
    
    # Create client instance (excludes WiFi passwords by default)
    client = PooldoseClient(host=HOST, timeout=TIMEOUT)
    
    # Optional: Include sensitive data like WiFi passwords
    # client = PooldoseClient(host=HOST, timeout=TIMEOUT, include_sensitive_data=True)
    
    # Connect to device
    status = await client.connect()
    if status != RequestStatus.SUCCESS:
        print(f"Error connecting to device: {status}")
        return
    
    print(f"Connected to {HOST}")
    print("Device Info:", json.dumps(client.device_info, indent=2))

    # --- Get static values ---
    status, static_values = client.static_values()
    if status == RequestStatus.SUCCESS:
        print(f"Device Name: {static_values.sensor_name}")
        print(f"Serial Number: {static_values.sensor_serial_number}")
        print(f"Firmware Version: {static_values.sensor_fw_version}")

    # --- Get instant values (dictionary-style) ---
    status, instant_values = await client.instant_values()
    if status != RequestStatus.SUCCESS:
        print(f"Error getting instant values: {status}")
        return

    # Dictionary-style individual access
    if "temperature" in instant_values:
        temp = instant_values["temperature"]
        print(f"Temperature: {temp[0]} {temp[1]}")

    # Get with default
    ph_value = instant_values.get("ph", "Not available")
    print(f"pH: {ph_value}")

    # --- Get structured instant values ---
    status, structured_data = await client.instant_values_structured()
    if status != RequestStatus.SUCCESS:
        print(f"Error getting structured values: {status}")
        return

    # Access sensors
    sensors = structured_data.get("sensor", {})
    print("\nSensor Values:")
    for key, sensor_data in sensors.items():
        value = sensor_data.get("value")
        unit = sensor_data.get("unit")
        if unit:
            print(f"  {key}: {value} {unit}")
        else:
            print(f"  {key}: {value}")

    # Access numbers (setpoints)
    numbers = structured_data.get("number", {})
    print("\nSetpoints:")
    for key, number_data in numbers.items():
        value = number_data.get("value")
        unit = number_data.get("unit")
        min_val = number_data.get("min")
        max_val = number_data.get("max")
        
        if unit:
            print(f"  {key}: {value} {unit} (Range: {min_val}-{max_val})")
        else:
            print(f"  {key}: {value} (Range: {min_val}-{max_val})")

    # Access switches
    switches = structured_data.get("switch", {})
    print("\nSwitches:")
    for key, switch_data in switches.items():
        value = switch_data.get("value")
        status_text = "ON" if value else "OFF"
        print(f"  {key}: {status_text}")

    # Access binary sensors (alarms/status)
    binary_sensors = structured_data.get("binary_sensor", {})
    print("\nAlarms & Status:")
    for key, sensor_data in binary_sensors.items():
        value = sensor_data.get("value")
        status_text = "ACTIVE" if value else "OK"
        print(f"  {key}: {status_text}")

    # Access selects (configuration options)
    selects = structured_data.get("select", {})
    print("\nSettings:")
    for key, select_data in selects.items():
        value = select_data.get("value")
        print(f"  {key}: {value}")

    # --- Setting values ---
    
    # Set number values (via InstantValues)
    result = await instant_values.set_number("target_ph", 7.2)
    print(f"Set pH target to 7.2: {result}")

    # Set switch values
    result = await instant_values.set_switch("stop_dosing", True)
    print(f"Set stop dosing: {result}")

    # Set select values
    result = await instant_values.set_select("water_meter_unit", "L/h")
    print(f"Set water meter unit: {result}")

if __name__ == "__main__":
    asyncio.run(main())

Advanced Usage

Connection Management

from pooldose.client import PooldoseClient
from pooldose.request_status import RequestStatus

# HTTP connection (default)
client = PooldoseClient("192.168.1.100", timeout=30)
status = await client.connect()

# HTTPS connection with SSL verification
client = PooldoseClient("192.168.1.100", timeout=30, use_ssl=True)
status = await client.connect()

# HTTPS connection with custom port and disabled verification
client = PooldoseClient("192.168.1.100", use_ssl=True, port=8443, ssl_verify=False)
status = await client.connect()

# Check connection status
if client.is_connected:
    print("Client is connected")
else:
    print("Client is not connected")

Error Handling

from pooldose.client import PooldoseClient

client = PooldoseClient("192.168.1.100")
status = await client.connect()

if status == RequestStatus.SUCCESS:
    print("Connected successfully")
elif status == RequestStatus.HOST_UNREACHABLE:
    print("Could not reach device")
elif status == RequestStatus.PARAMS_FETCH_FAILED:
    print("Failed to fetch device parameters")
elif status == RequestStatus.API_VERSION_UNSUPPORTED:
    print("Unsupported API version")
else:
    print(f"Other error: {status}")

Working with Structured Data

# Get all data types at once
status, structured_data = await client.instant_values_structured()

if status == RequestStatus.SUCCESS:
    # Check what types are available
    available_types = list(structured_data.keys())
    print("Available types:", available_types)
    
    # Process each type
    for data_type, items in structured_data.items():
        print(f"\n{data_type.title()} ({len(items)} items):")
        for key, data in items.items():
            if data_type in ["sensor", "number"]:
                value = data.get("value")
                unit = data.get("unit")
                if unit:
                    print(f"  {key}: {value} {unit}")
                else:
                    print(f"  {key}: {value}")
            elif data_type in ["switch", "binary_sensor"]:
                value = data.get("value")
                print(f"  {key}: {'ON' if value else 'OFF'}")
            elif data_type == "select":
                value = data.get("value")
                print(f"  {key}: {value}")

Working with Mappings

Mapping Discovery Process:
┌─────────────────┐
│ Device Connect  │
└─────────────────┘
         │
         ▼
┌─────────────────┐
│ Get MODEL_ID    │ ──────► PDPR1H1HAW100
│ Get FW_CODE     │ ──────► 539187
└─────────────────┘
         │
         ▼
┌─────────────────┐
│ Load JSON File  │ ──────► model_PDPR1H1HAW100_FW539187.json
└─────────────────┘
         │
         ▼
┌─────────────────┐
│ Type Discovery  │
│ ┌─────────────┐ │
│ │ Sensors     │ │ ──────► temperature, ph, orp, ...
│ │ Switches    │ │ ──────► stop_dosing, pump_detection, ...
│ │ Numbers     │ │ ──────► ph_target, orp_target, ...
│ │ Selects     │ │ ──────► water_meter_unit, ...
│ │ Binary Sens │ │ ──────► alarm_ph, alarm_orp, ...
│ └─────────────┘ │
└─────────────────┘

API Reference

PooldoseClient Class

Constructor

PooldoseClient(host, timeout=30, include_sensitive_data=False, use_ssl=False, port=None, ssl_verify=True)

Parameters:

• host (str): The hostname or IP address of the device
• timeout (int): Request timeout in seconds (default: 30)
• include_sensitive_data (bool): Whether to include sensitive data like WiFi passwords (default: False)
• use_ssl (bool): Whether to use HTTPS instead of HTTP (default: False)
• port (Optional[int]): Custom port for connections. Defaults to 80 for HTTP, 443 for HTTPS (default: None)
• ssl_verify (bool): Whether to verify SSL certificates when using HTTPS (default: True)

Methods

• async connect() → RequestStatus - Connect to device and initialize all components
• static_values() → tuple[RequestStatus, StaticValues | None] - Get static device information
• async instant_values() → tuple[RequestStatus, InstantValues | None] - Get current sensor readings and device state
• async instant_values_structured() → tuple[RequestStatus, dict[str, Any]] - Get structured data organized by type
• check_apiversion_supported() → tuple[RequestStatus, dict] - Check API version compatibility

Properties

• is_connected: bool - Check if client is connected to device
• device_info: dict - Dictionary containing device information

RequestStatus

All client methods return RequestStatus enum values:

from pooldose.request_status import RequestStatus

RequestStatus.SUCCESS                    # Operation successful
RequestStatus.HOST_UNREACHABLE           # Device not reachable
RequestStatus.PARAMS_FETCH_FAILED        # Failed to fetch device parameters
RequestStatus.API_VERSION_UNSUPPORTED    # API version not supported
RequestStatus.NO_DATA                    # No data received
RequestStatus.LAST_DATA                  # Last valid data used
RequestStatus.CLIENT_ERROR_SET           # Error setting client value
RequestStatus.UNKNOWN_ERROR              # Other error occurred

InstantValues Interface

The InstantValues class provides dictionary-style access to sensor data:

# Dictionary Interface
value = instant_values["sensor_name"]                    # Direct access
value = instant_values.get("sensor_name", default)      # Get with default
exists = "sensor_name" in instant_values                 # Check existence

# Setting values (async, with validation)
await instant_values.set_number("ph_target", 7.2)       # Set number value
await instant_values.set_switch("stop_dosing", True)    # Set switch value
await instant_values.set_select("unit", "L/h")          # Set select value

Structured Data Format

The instant_values_structured() method returns data organized by type:

{
    "sensor": {
        "temperature": {"value": 25.5, "unit": "°C"},
        "ph": {"value": 7.2, "unit": None}
    },
    "number": {
        "target_ph": {"value": 7.0, "unit": None, "min": 6.0, "max": 8.0, "step": 0.1}
    },
    "switch": {
        "stop_dosing": {"value": False}
    },
    "binary_sensor": {
        "alarm_ph": {"value": False}
    },
    "select": {
        "water_meter_unit": {"value": "L/h"}
    }
}

Data Types

• sensor: Read-only sensor values with optional units
• number: Configurable numeric values with min/max/step constraints
• switch: Boolean on/off controls
• binary_sensor: Read-only boolean status indicators
• select: Configurable selection options

Supported Devices

This client has been tested with:

• PoolDose Double/Dual WiFi (Model: PDPR1H1HAW100, FW: 539187)

Other SEKO PoolDose models may work but are untested. The client uses JSON mapping files to adapt to different device models and firmware versions (see e.g. src/pooldose/mappings/model_PDPR1H1HAW100_FW539187.json).

Note: The JSON files in the mappings directory define the device-specific data keys and their human-readable names for different PoolDose models and firmware versions.

Security

By default, the client excludes sensitive information like WiFi passwords from device info. To include sensitive data:

client = PooldoseClient(
    host="192.168.1.100", 
    include_sensitive_data=True
)
status = await client.connect()

Security Model

Data Classification:
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Public Data   │    │ Sensitive Data  │    │  Never Exposed  │
├─────────────────┤    ├─────────────────┤    ├─────────────────┤
│ • Device Name   │    │ • WiFi Password │    │ • Admin Creds   │
│ • Model ID      │    │ • AP Password   │    │ • Internal Keys │
│ • Serial Number │    │                 │    │                 │
│ • Sensor Values │    │                 │    │                 │
│ • IP Address    │    │                 │    │                 │
│ • MAC Address   │    │                 │    │                 │
└─────────────────┘    └─────────────────┘    └─────────────────┘
        │                       │                       │
        ▼                       ▼                       ▼
  Always Included      include_sensitive_data=True    Never Included

Changelog

For detailed release notes and version history, please see CHANGELOG.md.

Latest Release (0.5.0)

• BREAKING: Redesigned API with dictionary-style access (instant_values["key"])
• New structured data method instant_values_structured() grouped by type
• Removed deprecated getter methods, enhanced validation, improved error handling
• Complete code modernization and comprehensive test