QuiltClient reference¶

Complete reference for the quilt_hp package, including all public exports, the QuiltClient constructor, and every method with full signatures, parameters, return types, and exceptions.

Module exports¶

Everything a library consumer needs is importable from quilt_hp directly:

from quilt_hp import (
    QuiltClient,
    Environment,
    QuiltError,
    QuiltAuthError,
    QuiltConnectionError,
    QuiltNotFoundError,
)

`Environment`¶

class Environment(Enum):
    PROD = "prod"
    STAGING = "staging"
    DEV = "dev"

Value	gRPC host
`Environment.PROD`	`api.prod.quilt.cloud:443`
`Environment.STAGING`	`api.staging.quilt.cloud:443`
`Environment.DEV`	`api.dev.quilt.cloud:443`

`QuiltError`¶

class QuiltError(Exception): ...

Base class for all library exceptions.

`QuiltAuthError`¶

class QuiltAuthError(QuiltError): ...

Raised when authentication fails. Causes: invalid or expired OTP, Cognito API error, malformed token in store, no otp_callback provided when OTP is required.

`QuiltConnectionError`¶

class QuiltConnectionError(QuiltError): ...

Raised when the library cannot connect to the Quilt gRPC API.

`QuiltNotFoundError`¶

class QuiltNotFoundError(QuiltError): ...

Raised when a requested resource does not exist (gRPC NOT_FOUND).

`version`¶

__version__: str  # e.g. "0.3.0"

`QuiltClient`¶

class QuiltClient:
    def __init__(
        self,
        email: str,
        *,
        home: str | None = None,
        environment: Environment = Environment.PROD,
        snapshot_ttl_s: float = 0,
        token_store: TokenStoreLike | None = None,
        token_refresh_hooks: TokenRefreshHooks | None = None,
        token_refresh_policy: TokenRefreshPolicy | None = None,
    ) -> None: ...

The primary user-facing class. Manages authentication, the gRPC channel lifecycle, and exposes all high-level HVAC control methods.

Parameters:

Parameter	Type	Default	Description
`email`	`str`	required	Quilt account email address. Used as the Cognito username and as the token store key.
`home`	`str \\| None`	`None`	Home name filter (substring match, case-insensitive) for multi-home accounts. When omitted, the first system returned by `ListSystems` is used.
`environment`	`Environment`	`PROD`	Which Quilt API environment to connect to.
`snapshot_ttl_s`	`float`	`0`	If > 0, `get_snapshot()` results are cached for this many seconds. `0` disables caching.
`token_store`	`TokenStoreLike \\| None`	`None`	Token persistence backend. `None` means in-memory only (tokens are lost when the process exits).
`token_refresh_hooks`	`TokenRefreshHooks \\| None`	`None`	Observer for token refresh lifecycle events.
`token_refresh_policy`	`TokenRefreshPolicy \\| None`	`None`	Controls behaviour on refresh failure. Default: fall back to OTP.

Async context manager:

async def __aenter__(self) -> QuiltClient: ...
async def __aexit__(self, *_: object) -> None: ...

__aexit__ calls close() to close the gRPC channel. Prefer the async context manager, or call await close() yourself when managing lifecycle manually.

Authentication¶

`login`¶

async def login(self, otp_callback: OtpCallback | None = None) -> None

Authenticates using the three-step token resolution: cache → refresh → OTP.

If cached tokens are valid, returns immediately. If the cached access token is expired but the refresh token is valid, performs a silent REFRESH_TOKEN_AUTH. Calls otp_callback only when no valid cached or refresh token exists.

Parameters: - otp_callback: (email: str) -> str | Awaitable[str]. Required when no valid cached token exists; pass None only when tokens are guaranteed to be cached.

Raises: QuiltAuthError if authentication fails.

`refresh_token`¶

async def refresh_token(self, context: TokenRefreshContext | None = None) -> None

Silently refreshes the auth token using the refresh token. Does not attempt OTP. Called automatically by the transport interceptor on UNAUTHENTICATED; rarely needed directly.

`get_current_token`¶

def get_current_token(self) -> str

Returns the current JWT access token held by the client.

Raises: QuiltAuthError if the client is not authenticated yet.

System discovery¶

`list_systems`¶

async def list_systems(self) -> list[SystemInfo]

Lists all Quilt systems the authenticated user has access to.

Returns: List of SystemInfo objects with id, name, and timezone.

Raises: QuiltError if the gRPC call fails.

`get_system_id`¶

async def get_system_id(self, home: str | None = None) -> str

Returns the system ID for the current home filter. Caches the result after the first call.

`system_name`¶

@property
def system_name(self) -> str | None

Name of the resolved system (populated after get_system_id() is called).

`get_snapshot`¶

async def get_snapshot(self, system_id: str | None = None) -> SystemSnapshot

Fetches the complete system state as a SystemSnapshot.

Parameters: - system_id: explicit system ID to query. When None (default), uses get_system_id(). Passing system_id bypasses the cache.

Raises: QuiltNotFoundError if the system ID is not found. QuiltError for other gRPC failures.

`invalidate_snapshot`¶

def invalidate_snapshot(self) -> None

Discards the cached snapshot. The next get_snapshot() call fetches fresh data from the server.

`close`¶

async def close(self) -> None

Closes the underlying gRPC channel and clears the client's open channel reference. Safe to call multiple times.

Space control¶

`list_spaces`¶

async def list_spaces(self) -> list[Space]

Returns all room-level spaces. Equivalent to snapshot.rooms. Fetches a snapshot internally.

`set_space`¶

async def set_space(
    self,
    space: Space | str,
    *,
    mode: HVACMode | None = None,
    heat_setpoint_c: float | None = None,
    cool_setpoint_c: float | None = None,
) -> Space

Updates a space's HVAC mode and/or temperature setpoints.

Parameters: - space: Space object (no snapshot lookup) or space ID string (snapshot fetched internally). - mode: HVACMode.STANDBY, COOL, HEAT, AUTO, or FAN. Defaults to current mode. - heat_setpoint_c: heating setpoint in °C. Defaults to current. - cool_setpoint_c: cooling setpoint in °C. Defaults to current.

Behavioural notes: - mode=STANDBY clears comfort_setting_id; the room stays off regardless of occupancy. - mode=AUTO with a gap of less than 2.5°C: cooling setpoint is raised to heat + 2.5 automatically.

Returns: Updated Space from the server response.

Raises: QuiltError if the space is not found or the RPC fails.

`set_space_settings`¶

async def set_space_settings(
    self,
    space: Space | str,
    *,
    unoccupied_timeout_s: float | None = None,
    occupied_timeout_s: float | None = None,
) -> Space

Updates occupancy automation timeouts.

Parameters: - space: Space object or space ID string. - unoccupied_timeout_s: seconds of no-presence before auto-away. - occupied_timeout_s: seconds of presence before auto-return.

Returns: Updated Space.

Indoor unit control¶

`list_indoor_units`¶

async def list_indoor_units(self) -> list[IndoorUnit]

Returns all indoor units. Fetches a snapshot internally.

`set_indoor_unit`¶

async def set_indoor_unit(
    self,
    idu: IndoorUnit | str,
    *,
    fan_speed: FanSpeed | None = None,
    louver_mode: LouverMode | None = None,
    louver_position: float | None = None,
    led_color_code: int | None = None,
    led_brightness: float | None = None,
    led_animation: int | None = None,
) -> IndoorUnit

Updates indoor unit controls.

Parameters: - idu: IndoorUnit object or IDU ID string. - fan_speed: FanSpeed.AUTO, QUIET, LOW, MEDIUM, HIGH, or BLAST. - louver_mode: LouverMode.CLOSED, SWEEP, FIXED, or AUTO. - louver_position: position 0.0–1.0 when louver_mode=FIXED. - led_color_code: RGBW packed int32. Use LightPreset constants or compute manually. - led_brightness: brightness 0.0–1.0. - led_animation: animation ID (LedAnimation enum values).

Returns: Updated IndoorUnit.

`set_indoor_unit_settings`¶

async def set_indoor_unit_settings(
    self,
    idu: IndoorUnit | str,
    *,
    fence_left_m: float | None = None,
    fence_right_m: float | None = None,
    fence_forward_m: float | None = None,
    radar_height_m: float | None = None,
    light_brightness_default: float | None = None,
) -> IndoorUnit

Updates indoor unit calibration settings.

Parameters: - fence_left_m: left boundary of presence detection zone in metres (0 = unconfigured/max range). - fence_right_m: right boundary. - fence_forward_m: forward (depth) boundary. - radar_height_m: radar sensor mounting height from floor in metres. - light_brightness_default: default LED brightness 0.0–1.0.

Returns: Updated IndoorUnit.

Comfort settings¶

`list_comfort_settings`¶

async def list_comfort_settings(self) -> list[ComfortSetting]

Returns all comfort presets. Fetches a snapshot internally.

`update_comfort_setting`¶

async def update_comfort_setting(
    self,
    setting: ComfortSetting | str,
    *,
    name: str | None = None,
    hvac_mode: HVACMode | None = None,
    heat_setpoint_c: float | None = None,
    cool_setpoint_c: float | None = None,
    fan_speed: FanSpeed | None = None,
) -> ComfortSetting

Updates a comfort preset. Omitted fields keep their current values.

Returns: Updated ComfortSetting.

Schedules¶

`create_schedule_day`¶

async def create_schedule_day(
    self,
    space_id: str,
    name: str,
    events: list[ScheduleEvent],
) -> ScheduleDay

Creates a new schedule day program for a space.

`update_schedule_day`¶

async def update_schedule_day(
    self,
    schedule_day_id: str,
    space_id: str,
    name: str | None = None,
    events: list[ScheduleEvent] | None = None,
) -> ScheduleDay

Updates an existing schedule day's name and/or events.

`delete_schedule_day`¶

async def delete_schedule_day(self, schedule_day_id: str) -> None

Deletes a schedule day program.

`create_schedule_week`¶

async def create_schedule_week(
    self,
    space_id: str,
    days: list[ScheduleWeekDay] | None = None,
) -> ScheduleWeek

Creates a new schedule week, optionally mapping weekdays to day programs.

`update_schedule_week`¶

async def update_schedule_week(
    self,
    schedule_week_id: str,
    space_id: str,
    days: list[ScheduleWeekDay],
) -> ScheduleWeek

Updates a schedule week's day assignments.

`delete_schedule_week`¶

async def delete_schedule_week(self, schedule_week_id: str) -> None

Deletes a schedule week.

`set_schedule_execution`¶

async def set_schedule_execution(self, paused: bool) -> None

Globally pauses or resumes all schedules for the primary location. True pauses; False resumes.

Energy¶

`get_energy`¶

async def get_energy(
    self,
    start: datetime,
    end: datetime,
    system_id: str | None = None,
) -> list[SpaceEnergyMetrics]

Returns hourly energy consumption for all spaces.

Parameters: - start, end: Timezone-aware datetime objects. - system_id: explicit system ID; defaults to the primary system.

Returns: List of SpaceEnergyMetrics, each with space_id and a list of EnergyBucket objects (each with start_time, energy_kwh, status).

Streaming¶

`stream`¶

def stream(
    self,
    topics: list[str],
    *,
    max_reconnects: int = -1,
    reconnect_delay_s: float = 1.0,
) -> NotifierStream

Creates a NotifierStream. It does not start the stream; call start(), run_forever(), or use it as an async context manager.

Parameters: - topics: topic strings. Use snapshot.stream_topics() to get all topics for a system. - max_reconnects: maximum reconnect attempts per disconnect. -1 = unlimited (default). 0 = no retries. - reconnect_delay_s: initial back-off in seconds. Doubles on each attempt, capped at 60 s.

Returns: NotifierStream instance.

See The streaming protocol for the full reconnect state machine.

User¶

`get_current_user`¶

async def get_current_user(self) -> User

Returns the authenticated user's profile: quilt_user_id, first_name, last_name, email, phone_number.

`update_current_user`¶

async def update_current_user(
    self,
    *,
    first_name: str,
    last_name: str,
    phone_number: str | None = None,
) -> User

Updates the authenticated user's name and optional phone number.

`get_user_attributes`¶

async def get_user_attributes(self) -> UserAttributes

Returns user attributes including declared_user_type (HOMEOWNER or PARTNER).

`patch_user_attributes`¶

async def patch_user_attributes(
    self,
    *,
    declared_user_type: DeclaredUserType,
) -> UserAttributes

Updates user attributes.

QuiltClient reference¶

Module exports¶

Environment¶

QuiltError¶

QuiltAuthError¶

QuiltConnectionError¶

QuiltNotFoundError¶

__version__¶

QuiltClient¶

Authentication¶

login¶

refresh_token¶

get_current_token¶

System discovery¶

list_systems¶

get_system_id¶

system_name¶

get_snapshot¶

invalidate_snapshot¶

close¶

Space control¶

list_spaces¶

set_space¶

set_space_settings¶