steamloop package

Local control for thermostat devices over mTLS.

exception steamloop.AuthenticationError

Bases: SteamloopError

Authentication with the thermostat failed.

exception steamloop.CommandError

Bases: SteamloopError

A command sent to the thermostat was rejected.

class steamloop.FanMode(*values)

Bases: IntEnum

Fan operating mode.

ALWAYS_ON = 2

AUTO = 1

CIRCULATE = 3

class steamloop.HoldType(*values)

Bases: IntEnum

Temperature hold type.

UNDEFINED: No hold type set. MANUAL: Manual override by user (permanent hold). SCHEDULE: Following the programmed schedule. HOLD: Hold until next scheduled period.

HOLD = 3

MANUAL = 1

SCHEDULE = 2

UNDEFINED = 0

exception steamloop.PairingError

Bases: SteamloopError

Pairing with the thermostat failed.

exception steamloop.SteamloopConnectionError

Bases: SteamloopError

Connection to the thermostat failed.

exception steamloop.SteamloopError

Bases: Exception

Base exception for steamloop.

class steamloop.ThermostatConnection(ip: str, port: int = 7878, *, cert_set: CertSet | None = None, secret_key: str, device_type: str = 'automation', device_id: str = 'module')

Bases: object

Async connection to a thermostat over mTLS.

After calling connect() and login(), call start_background_tasks() to begin sending heartbeats. Events are dispatched automatically via the protocol’s data_received callback.

If the connection drops, it will automatically reconnect with exponential backoff. Call disconnect() to stop everything.

add_event_callback(callback: Callable[[dict[str, Any]], None]) → Callable[[], None]

Register an event callback. Returns a callable to unregister it.

async connect() → None

Establish the TLS connection to the thermostat.

If no cert_set was specified, tries each cert set in order until one succeeds. If already connected, the existing connection is closed first.

Raises:

SteamloopConnectionError: If the connection fails.

property connected: bool

Return True if the connection is active.

async disconnect() → None

Disconnect from the thermostat and stop all background tasks.

heartbeat() → None

Send a heartbeat to keep the connection alive.

async login() → LoginResponse

Authenticate with the thermostat.

After receiving the login response, waits for the initial burst of state events (zone discovery, temperatures, etc.) to arrive before returning. This ensures state is fully populated.

Returns:

LoginResponse on success.

Raises:

AuthenticationError: If authentication fails. SteamloopConnectionError: If the connection is lost.

async pair() → SetSecretKeyRequest

Pair with the thermostat.

The thermostat must be in pairing mode. Sends a login request with an empty secret key and waits for the thermostat to send a SetSecretKey request containing the new secret key.

Returns:

SetSecretKeyRequest with the new secret_key.

Raises:

PairingError: If pairing fails or times out. SteamloopConnectionError: If the connection is lost.

property secret_key: str

Return the secret key used for authentication.

send(msg: dict[str, Any]) → None

Send a message to the thermostat.

Raises:

SteamloopConnectionError: If not connected.

send_request(command: str, data: dict[str, str]) → None

Send a Request-wrapped command to the thermostat.

set_emergency_heat(enabled: bool) → None

Enable or disable emergency heat.

set_fan_mode(mode: FanMode | int) → None

Set the fan operating mode.

set_temperature_setpoint(zone_id: str, *, heat_setpoint: str | None = None, cool_setpoint: str | None = None, hold_type: HoldType = HoldType.MANUAL) → None

Set temperature setpoints for a zone.

If a setpoint isn’t provided, the current state value is used. Deadband is always taken from current state. If the resulting setpoints would violate the deadband, the opposite setpoint is automatically adjusted to maintain the minimum gap:

• If only heat_setpoint is provided, cool is raised if needed.

• If only cool_setpoint is provided, heat is lowered if needed.

• If both are provided, cool is raised to maintain the gap.

set_zone_mode(zone_id: str, mode: ZoneMode | int) → None

Set the HVAC mode for a zone.

start_background_tasks() → None

Start the background loop with heartbeats and auto-reconnect.

class steamloop.ThermostatState(zones: dict[str, ~steamloop.models.Zone]=<factory>, supported_modes: list[ZoneMode] = <factory>, fan_mode: FanMode = FanMode.AUTO, emergency_heat: str = '', relative_humidity: str = '', cooling_active: str = '', heating_active: str = '')

Bases: object

Aggregated state of the thermostat and all zones.

cooling_active: str = ''

emergency_heat: str = ''

fan_mode: FanMode = 1

heating_active: str = ''

relative_humidity: str = ''

supported_modes: list[ZoneMode]

zones: dict[str, Zone]

class steamloop.Zone(zone_id: str, name: str = '', mode: ZoneMode = ZoneMode.OFF, indoor_temperature: str = '', heat_setpoint: str = '', cool_setpoint: str = '', deadband: str = '', hold_type: HoldType = HoldType.UNDEFINED)

Bases: object

State of a single thermostat zone.

cool_setpoint: str = ''

deadband: str = ''

heat_setpoint: str = ''

hold_type: HoldType = 0

indoor_temperature: str = ''

mode: ZoneMode = 0

name: str = ''

zone_id: str

class steamloop.ZoneMode(*values)

Bases: IntEnum

HVAC zone operating mode.

AUTO = 1

COOL = 2

HEAT = 3

OFF = 0

async steamloop.load_pairing(ip: str, directory: Path | None = None) → dict[str, str] | None

Load saved pairing data for a thermostat IP.

Args:

ip: Thermostat IP address. directory: Directory to load from. Defaults to current directory.

Returns:

Pairing dict with secret_key, device_type, device_id, or None.

async steamloop.save_pairing(ip: str, login_info: dict[str, str], directory: Path | None = None) → None

Save pairing data for a thermostat IP.

Args:

ip: Thermostat IP address. login_info: Dict with secret_key, device_type, device_id. directory: Directory to save to. Defaults to current directory.

Submodules

steamloop.certs module

Embedded client certificates for thermostat mTLS authentication.

class steamloop.certs.CertSet(name: str, chain_data: str, root_ca_data: str | None = None)

Bases: object

A set of client certificates for mTLS authentication.

chain_data: str

name: str

root_ca_data: str | None = None

steamloop.certs.create_ssl_context(cert_set: CertSet) → SSLContext

Create an SSL context configured for thermostat mTLS.

Decodes the embedded cert data and loads it via load_cert_chain(). A brief temp file is used because ssl.SSLContext.load_cert_chain() only accepts file paths.

steamloop.cli module

Command-line interface for steamloop thermostat control.

steamloop.cli.main() → None

Entry point for the steamloop CLI.

steamloop.connection module

Async connection to a thermostat over mTLS on port 7878.

class steamloop.connection.ThermostatConnection(ip: str, port: int = 7878, *, cert_set: CertSet | None = None, secret_key: str, device_type: str = 'automation', device_id: str = 'module')

Bases: object

Async connection to a thermostat over mTLS.

After calling connect() and login(), call start_background_tasks() to begin sending heartbeats. Events are dispatched automatically via the protocol’s data_received callback.

If the connection drops, it will automatically reconnect with exponential backoff. Call disconnect() to stop everything.

add_event_callback(callback: Callable[[dict[str, Any]], None]) → Callable[[], None]

Register an event callback. Returns a callable to unregister it.

async connect() → None

Establish the TLS connection to the thermostat.

If no cert_set was specified, tries each cert set in order until one succeeds. If already connected, the existing connection is closed first.

Raises:

SteamloopConnectionError: If the connection fails.

property connected: bool

Return True if the connection is active.

async disconnect() → None

Disconnect from the thermostat and stop all background tasks.

heartbeat() → None

Send a heartbeat to keep the connection alive.

async login() → LoginResponse

Authenticate with the thermostat.

After receiving the login response, waits for the initial burst of state events (zone discovery, temperatures, etc.) to arrive before returning. This ensures state is fully populated.

Returns:

LoginResponse on success.

Raises:

AuthenticationError: If authentication fails. SteamloopConnectionError: If the connection is lost.

async pair() → SetSecretKeyRequest

Pair with the thermostat.

The thermostat must be in pairing mode. Sends a login request with an empty secret key and waits for the thermostat to send a SetSecretKey request containing the new secret key.

Returns:

SetSecretKeyRequest with the new secret_key.

Raises:

PairingError: If pairing fails or times out. SteamloopConnectionError: If the connection is lost.

property secret_key: str

Return the secret key used for authentication.

send(msg: dict[str, Any]) → None

Send a message to the thermostat.

Raises:

SteamloopConnectionError: If not connected.

send_request(command: str, data: dict[str, str]) → None

Send a Request-wrapped command to the thermostat.

set_emergency_heat(enabled: bool) → None

Enable or disable emergency heat.

set_fan_mode(mode: FanMode | int) → None

Set the fan operating mode.

set_temperature_setpoint(zone_id: str, *, heat_setpoint: str | None = None, cool_setpoint: str | None = None, hold_type: HoldType = HoldType.MANUAL) → None

Set temperature setpoints for a zone.

If a setpoint isn’t provided, the current state value is used. Deadband is always taken from current state. If the resulting setpoints would violate the deadband, the opposite setpoint is automatically adjusted to maintain the minimum gap:

• If only heat_setpoint is provided, cool is raised if needed.

• If only cool_setpoint is provided, heat is lowered if needed.

• If both are provided, cool is raised to maintain the gap.

set_zone_mode(zone_id: str, mode: ZoneMode | int) → None

Set the HVAC mode for a zone.

start_background_tasks() → None

Start the background loop with heartbeats and auto-reconnect.

class steamloop.connection.ThermostatProtocol(connection: ThermostatConnection)

Bases: Protocol

Low-level protocol handler for thermostat communication.

Handles framing (null-byte delimited JSON) and delegates parsed messages to the owning ThermostatConnection.

close() → None

Close the transport.

connection_lost(exc: Exception | None) → None

Called when the connection is lost.

connection_made(transport: BaseTransport) → None

Called when the TLS connection is established.

data_received(data: bytes) → None

Called when data is received from the thermostat.

send(msg: dict[str, Any]) → None

Send a message to the thermostat (sync — no drain needed).

send_request(command: str, data: dict[str, str]) → None

Send a Request-wrapped command to the thermostat.

async steamloop.connection.load_pairing(ip: str, directory: Path | None = None) → dict[str, str] | None

Load saved pairing data for a thermostat IP.

Args:

ip: Thermostat IP address. directory: Directory to load from. Defaults to current directory.

Returns:

Pairing dict with secret_key, device_type, device_id, or None.

async steamloop.connection.save_pairing(ip: str, login_info: dict[str, str], directory: Path | None = None) → None

Save pairing data for a thermostat IP.

Args:

ip: Thermostat IP address. login_info: Dict with secret_key, device_type, device_id. directory: Directory to save to. Defaults to current directory.

steamloop.const module

Constants and enums for the steamloop thermostat protocol.

class steamloop.const.FanMode(*values)

Bases: IntEnum

Fan operating mode.

ALWAYS_ON = 2

AUTO = 1

CIRCULATE = 3

class steamloop.const.HoldType(*values)

Bases: IntEnum

Temperature hold type.

UNDEFINED: No hold type set. MANUAL: Manual override by user (permanent hold). SCHEDULE: Following the programmed schedule. HOLD: Hold until next scheduled period.

HOLD = 3

MANUAL = 1

SCHEDULE = 2

UNDEFINED = 0

class steamloop.const.ZoneMode(*values)

Bases: IntEnum

HVAC zone operating mode.

AUTO = 1

COOL = 2

HEAT = 3

OFF = 0

steamloop.exceptions module

Exception classes for steamloop.

exception steamloop.exceptions.AuthenticationError

Bases: SteamloopError

Authentication with the thermostat failed.

exception steamloop.exceptions.CommandError

Bases: SteamloopError

A command sent to the thermostat was rejected.

exception steamloop.exceptions.PairingError

Bases: SteamloopError

Pairing with the thermostat failed.

exception steamloop.exceptions.SteamloopConnectionError

Bases: SteamloopError

Connection to the thermostat failed.

exception steamloop.exceptions.SteamloopError

Bases: Exception

Base exception for steamloop.

steamloop.models module

Data models for thermostat events, responses, and state.

class steamloop.models.CoolingStatusUpdatedEvent

Bases: TypedDict

Cooling compressor status changed.

cooling_active: str

class steamloop.models.EmergencyHeatUpdatedEvent

Bases: TypedDict

Emergency heat status changed.

emergency_heat: str

class steamloop.models.ErrorResponse

Bases: TypedDict

Error response from the thermostat.

description: str

error_type: str

class steamloop.models.FanModeUpdatedEvent

Bases: TypedDict

The fan mode changed.

fan_mode: str

class steamloop.models.HeatingStatusUpdatedEvent

Bases: TypedDict

Heating system status changed.

heating_active: str

class steamloop.models.IndoorRelativeHumidityUpdatedEvent

Bases: TypedDict

Indoor humidity changed.

relative_humidity: str

class steamloop.models.IndoorTemperatureUpdatedEvent

Bases: TypedDict

A zone’s indoor temperature changed.

indoor_temperature: str

zone_id: str

class steamloop.models.LoginResponse

Bases: TypedDict

Login response from the thermostat.

status: str

class steamloop.models.SetSecretKeyRequest

Bases: TypedDict

Secret key sent by the thermostat during pairing.

secret_key: str

class steamloop.models.SupportedZoneModesUpdatedEvent

Bases: TypedDict

The list of supported zone modes was received.

modes: str

class steamloop.models.TemperatureSetpointUpdatedEvent

Bases: _TemperatureSetpointRequired

A zone’s temperature setpoints changed.

Only zone_id is guaranteed; other fields may be absent when only a subset of setpoints changed.

cool_setpoint: str

deadband: str

heat_setpoint: str

hold_type: str

zone_id: str

class steamloop.models.ThermostatState(zones: dict[str, ~steamloop.models.Zone]=<factory>, supported_modes: list[ZoneMode] = <factory>, fan_mode: FanMode = FanMode.AUTO, emergency_heat: str = '', relative_humidity: str = '', cooling_active: str = '', heating_active: str = '')

Bases: object

Aggregated state of the thermostat and all zones.

cooling_active: str = ''

emergency_heat: str = ''

fan_mode: FanMode = 1

heating_active: str = ''

relative_humidity: str = ''

supported_modes: list[ZoneMode]

zones: dict[str, Zone]

class steamloop.models.Zone(zone_id: str, name: str = '', mode: ZoneMode = ZoneMode.OFF, indoor_temperature: str = '', heat_setpoint: str = '', cool_setpoint: str = '', deadband: str = '', hold_type: HoldType = HoldType.UNDEFINED)

Bases: object

State of a single thermostat zone.

cool_setpoint: str = ''

deadband: str = ''

heat_setpoint: str = ''

hold_type: HoldType = 0

indoor_temperature: str = ''

mode: ZoneMode = 0

name: str = ''

zone_id: str

class steamloop.models.ZoneAddedEvent

Bases: TypedDict

A new zone was discovered.

zone_id: str

class steamloop.models.ZoneModeUpdatedEvent

Bases: TypedDict

A zone’s HVAC mode changed.

zone_id: str

zone_mode: str

class steamloop.models.ZoneNameUpdatedEvent

Bases: TypedDict

A zone’s name changed.

zone_id: str

zone_name: str