thermostat

eq3btsmart.thermostat

Support for eQ-3 Bluetooth Smart thermostats.

Thermostat

Representation of an eQ-3 Bluetooth Smart thermostat.

Source code in eq3btsmart/thermostat.py

class Thermostat:
    """Representation of an eQ-3 Bluetooth Smart thermostat."""

    def __init__(
        self,
        mac_address: str,
        connection_timeout: int = DEFAULT_CONNECTION_TIMEOUT,
        command_timeout: int = DEFAULT_COMMAND_TIMEOUT,
    ):
        """Initialize the thermostat.

        The thermostat will be in a disconnected state after initialization.

        Args:
            mac_address (str): The MAC address of the thermostat.
            connection_timeout (int, optional): The connection timeout in seconds. Defaults to DEFAULT_CONNECTION_TIMEOUT.
            command_timeout (int, optional): The command timeout in seconds. Defaults to DEFAULT_COMMAND_TIMEOUT.
        """
        self._mac_address = mac_address

        self._last_status: Status | None = None
        self._last_device_data: DeviceData | None = None
        self._last_schedule: Schedule | None = None

        self._callbacks: defaultdict[
            Eq3Event, list[Union[Callable[..., None], Callable[..., Awaitable[None]]]]
        ] = defaultdict(list)
        self._conn: BleakClient = BleakClient(
            mac_address,
            disconnected_callback=self._on_disconnected,
            timeout=DEFAULT_CONNECTION_TIMEOUT,
        )
        self._device_data_future: asyncio.Future[DeviceData] | None = None
        self._status_future: asyncio.Future[Status] | None = None
        self._schedule_future: asyncio.Future[Schedule] | None = None
        self._schedule_future_counter: int = 0
        self._lock = asyncio.Lock()
        self._connection_timeout = connection_timeout
        self._command_timeout = command_timeout

    @property
    def is_connected(self) -> bool:
        """Check if the thermostat is connected.

        Returns:
            bool: True if connected, False otherwise.
        """
        return self._conn.is_connected

    @property
    def device_data(self) -> DeviceData:
        """Get the last known device data, ensuring it's not None.

        Returns:
            DeviceData: The last known device data.

        Raises:
            Eq3StateException: If the device data is None. This occurs when the thermostat has not been connected yet.
        """
        if self._last_device_data is None:
            raise Eq3StateException("Device data not set")
        return self._last_device_data

    @property
    def status(self) -> Status:
        """Get the last known status, ensuring it's not None.

        Returns:
            Status: The last known status.

        Raises:
            Eq3StateException: If the status is None. This occurs when the thermostat has not been connected yet.
        """
        if self._last_status is None:
            raise Eq3StateException("Status not set")
        return self._last_status

    @property
    def presets(self) -> Presets:
        """Get the presets from the last known status.

        Presets are only available since firmware version MIN_PRESETS_FW_VERSION.

        Returns:
            Presets: The presets.

        Raises:
            Eq3StateException: If the presets are None. This occurs when the thermostat has not been connected yet or if the thermostat's firmware does not support presets.
        """
        if self.status.presets is None:
            raise Eq3StateException("Presets not set")
        return self.status.presets

    @property
    def schedule(self) -> Schedule:
        """Get the last known schedule, ensuring it's not None.

        Returns:
            Schedule: The last known schedule.

        Raises:
            Eq3StateException: If the schedule is None. This occurs when the thermostat has not been connected
        """
        if self._last_schedule is None:
            raise Eq3StateException("Schedule not set")
        return self._last_schedule

    async def async_connect(self) -> None:
        """Connect to the thermostat.

        After connecting, the device data, status, and schedule will be queried and stored.
        When the connection is established, the CONNECTED event will be triggered.

        Raises:
            Eq3StateException: If the thermostat is already connected.
            Eq3ConnectionException: If the connection fails.
            Eq3TimeoutException: If the connection times out.
            Eq3CommandException: If an error occurs while sending a command.
        """
        if self.is_connected:
            raise Eq3StateException("Already connected")

        try:
            await asyncio.wait_for(self._conn.connect(), self._connection_timeout)
            await self._conn.start_notify(
                _Eq3Characteristic.NOTIFY, self._on_message_received
            )
            (
                self._last_device_data,
                self._last_status,
                self._last_schedule,
            ) = await asyncio.gather(
                self.async_get_device_data(),
                self.async_get_status(),
                self.async_get_schedule(),
            )
        except BleakError as ex:
            raise Eq3ConnectionException("Could not connect to the device") from ex
        except TimeoutError as ex:
            raise Eq3TimeoutException("Timeout during connection") from ex

        await self._trigger_event(
            Eq3Event.CONNECTED,
            device_data=self.device_data,
            status=self.status,
            schedule=self.schedule,
        )

    async def async_disconnect(self) -> None:
        """Disconnect from the thermostat.

        Before disconnection all pending futures will be cancelled.
        When the disconnection is complete, the DISCONNECTED event will be triggered.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3ConnectionException: If the disconnection fails.
            Eq3TimeoutException: If the disconnection times out.
        """
        if not self.is_connected:
            raise Eq3StateException("Not connected")

        exception = Eq3ConnectionException("Connection closed")
        if self._device_data_future is not None and not self._device_data_future.done():
            self._device_data_future.set_exception(exception)

        if self._status_future is not None and not self._status_future.done():
            self._status_future.set_exception(exception)

        if self._schedule_future is not None and not self._schedule_future.done():
            self._schedule_future.set_exception(exception)

        try:
            await self._conn.disconnect()
        except BleakError as ex:
            raise Eq3ConnectionException("Could not disconnect from the device") from ex
        except TimeoutError as ex:
            raise Eq3TimeoutException("Timeout during disconnection") from ex

    async def async_get_device_data(self) -> DeviceData:
        """Query the latest device data.

        Returns:
            DeviceData: The device data.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3CommandException: If an error occurs while sending the command.
            Eq3TimeoutException: If the command times out.
            Eq3AlreadyAwaitingResponseException: If a device data command is already pending.
        """
        return await self._async_write_command_with_device_data_response(
            _IdGetCommand()
        )

    async def async_get_status(self) -> Status:
        """Query the latest status.

        Returns:
            Status: The status.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3CommandException: If an error occurs while sending the command.
            Eq3TimeoutException: If the command times out.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        """
        return await self._async_write_command_with_status_response(
            _InfoGetCommand(time=datetime.now())
        )

    async def async_get_schedule(self) -> Schedule:
        """Query the schedule.

        Returns:
            Schedule: The schedule.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3CommandException: If an error occurs while sending the command.
            Eq3TimeoutException: If the command times out.
            Eq3AlreadyAwaitingResponseException: If a schedule command is already pending.
        """
        return await self._async_write_commands_with_schedule_response(
            [_ScheduleGetCommand(day=week_day) for week_day in Eq3WeekDay]
        )

    async def async_set_temperature(self, temperature: float) -> Status:
        """Set the target temperature.

        Temperatures are in degrees Celsius and specified in 0.5 degree increments.
        If the temperature is EQ3_OFF_TEMP, the thermostat will be turned off.
        If the temperature is EQ3_ON_TEMP, the thermostat will be turned on.

        Args:
            temperature (float): The new target temperature in degrees Celsius.

        Returns:
            Status: The updated status.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3CommandException: If an error occurs during the command.
            Eq3TimeoutException: If the command times out.
            Eq3InvalidDataException: If the temperature is invalid.
        """
        if temperature == EQ3_OFF_TEMP:
            return await self.async_set_mode(Eq3OperationMode.OFF)

        if temperature == EQ3_ON_TEMP:
            return await self.async_set_mode(Eq3OperationMode.ON)

        return await self._async_write_command_with_status_response(
            _TemperatureSetCommand(temperature=temperature)
        )

    async def async_set_mode(self, operation_mode: Eq3OperationMode) -> Status:
        """Set the operation mode.

        Args:
            operation_mode (Eq3OperationMode): The new operation mode.

        Returns:
            Status: The updated status.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3CommandException: If an error occurs during the command.
            Eq3TimeoutException: If the command times out.
            Eq3InvalidDataException: If the operation mode is not supported.
        """
        command: _ModeSetCommand

        match operation_mode:
            case Eq3OperationMode.AUTO:
                command = _ModeSetCommand(mode=Eq3OperationMode.AUTO)
            case Eq3OperationMode.MANUAL:
                command = _ModeSetCommand(
                    mode=Eq3OperationMode.MANUAL
                    | _Eq3Temperature.encode(self.status.target_temperature)
                )
            case Eq3OperationMode.OFF:
                command = _ModeSetCommand(
                    mode=Eq3OperationMode.MANUAL | _Eq3Temperature.encode(EQ3_OFF_TEMP)
                )
            case Eq3OperationMode.ON:
                command = _ModeSetCommand(
                    mode=Eq3OperationMode.MANUAL | _Eq3Temperature.encode(EQ3_ON_TEMP)
                )
            case _:
                raise Eq3InvalidDataException(
                    f"Unsupported operation mode: {operation_mode}"
                )

        return await self._async_write_command_with_status_response(command)

    async def async_set_preset(self, preset: Eq3Preset) -> Status:
        """Activate a preset.

        Args:
            preset (Eq3Preset): The preset to activate.

        Returns:
            Status: The updated status.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3CommandException: If an error occurs during the command.
            Eq3TimeoutException: If the command times out.
        """
        command: _ComfortSetCommand | _EcoSetCommand

        match preset:
            case Eq3Preset.COMFORT:
                command = _ComfortSetCommand()
            case Eq3Preset.ECO:
                command = _EcoSetCommand()

        return await self._async_write_command_with_status_response(command)

    async def async_set_boost(self, enable: bool) -> Status:
        """Enable or disable the boost mode.

        The boost mode will set the target temperature to EQ3_ON_TEMP for 300 seconds and then revert to the previous target temperature.

        Args:
            enable (bool): True to enable the boost mode, False to disable it.

        Returns:
            Status: The updated status.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3CommandException: If an error occurs during the command.
            Eq3TimeoutException: If the command times out.
        """
        return await self._async_write_command_with_status_response(
            _BoostSetCommand(enable=enable)
        )

    async def async_set_locked(self, enable: bool) -> Status:
        """Lock or unlock the thermostat.

        When locked, the thermostat's manual controls are disabled.

        Args:
            enable (bool): True to lock the thermostat, False to unlock it.

        Returns:
            Status: The updated status.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3CommandException: If an error occurs during the command.
            Eq3TimeoutException: If the command times out.
        """
        return await self._async_write_command_with_status_response(
            _LockSetCommand(enable=enable)
        )

    async def async_set_away(
        self,
        away_until: datetime,
        temperature: float,
    ) -> Status:
        """Set the thermostat to away mode.

        The thermostat will be set to away mode until the specified date and time.
        The target temperature will be set to the specified temperature.
        Temperatures are in degrees Celsius and specified in 0.5 degree increments.

        Args:
            away_until (datetime): The date and time until the thermostat should be in away mode.
            temperature (float): The target temperature in degrees Celsius.

        Returns:
            Status: The updated status.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3CommandException: If an error occurs during the command.
            Eq3TimeoutException: If the command times out.
            Eq3InvalidDataException: If the temperature or date is invalid.
        """
        return await self._async_write_command_with_status_response(
            _AwaySetCommand(
                mode=Eq3OperationMode.AWAY | _Eq3Temperature.encode(temperature),
                away_until=away_until,
            )
        )

    async def async_set_schedule(self, schedule: Schedule) -> Schedule:
        """Set the schedule.

        The schedule allows setting different target temperatures for each day of the week and different times of the day.
        It is only applied when the thermostat is in AUTO mode.

        Args:
            schedule: The schedule to set.

        Returns:
            The updated schedule.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a schedule command is already pending.
            Eq3CommandException: If an error occurs during the command.
            Eq3TimeoutException: If the command times out.
            Eq3InvalidDataException: If any of the schedule data is invalid.
        """
        return await self._async_write_commands_with_schedule_response(
            [
                _ScheduleSetCommand(
                    day=schedule_day.week_day,
                    hours=[
                        _ScheduleHourStruct(
                            target_temp=schedule_hour.target_temperature,
                            next_change_at=schedule_hour.next_change_at,
                        )
                        for schedule_hour in schedule_day.schedule_hours
                    ],
                )
                for schedule_day in schedule.schedule_days
            ]
        )

    async def async_delete_schedule(
        self, week_days: list[Eq3WeekDay] | Eq3WeekDay | None = None
    ) -> Schedule:
        """Delete the schedule for the specified week days.

        If no week days are specified, the schedule for all week days will be deleted.

        Args:
            week_days (list[WeekDay] | WeekDay | None, optional): The week days for which the schedule should be deleted.
                Can be a single WeekDay, a list of WeekDay, or None. Defaults to None.

        Returns:
            Schedule: The updated schedule after deletion.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a schedule command is already pending.
            Eq3CommandException: If an error occurs during the command.
            Eq3TimeoutException: If the command times out.
        """
        week_days = (
            [week_days]
            if isinstance(week_days, Eq3WeekDay)
            else week_days or list(Eq3WeekDay)
        )

        return await self._async_write_commands_with_schedule_response(
            [_ScheduleSetCommand(day=week_day, hours=[]) for week_day in week_days]
        )

    async def async_configure_presets(
        self,
        comfort_temperature: float,
        eco_temperature: float,
    ) -> Status:
        """Set the thermostat's preset temperatures.

        Temperatures are in degrees Celsius and specified in 0.5 degree increments.
        The initial values are 21.0 degrees Celsius for comfort and 17.0 degrees Celsius for eco.
        The comfort temperature is indicated by the sun symbol and the eco temperature by the moon symbol.

        Args:
            comfort_temperature (float): The comfort temperature in degrees Celsius.
            eco_temperature (float): The eco temperature in degrees Celsius.

        Returns:
            Status: The updated status.

        Raises:
            Eq3StateException: If the presets are not available or the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3CommandException: If an error occurs while sending the command.
            Eq3TimeoutException: If the command times out.
            Eq3InvalidDataException: If the temperatures are invalid.
        """
        return await self._async_write_command_with_status_response(
            _ComfortEcoConfigureCommand(
                comfort_temperature=comfort_temperature,
                eco_temperature=eco_temperature,
            )
        )

    async def async_configure_comfort_temperature(
        self, comfort_temperature: float
    ) -> Status:
        """Set the thermostat's comfort temperature.

        Temperatures are in degrees Celsius and specified in 0.5 degree increments.
        The initial value is 21.0 degrees Celsius.
        The comfort temperature is indicated by the sun symbol.

        Args:
            comfort_temperature (float): The comfort temperature in degrees Celsius.

        Returns:
            Status: The updated status.

        Raises:
            Eq3StateException: If the presets are not available or the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3CommandException: If an error occurs while sending the command.
            Eq3TimeoutException: If the command times out.
            Eq3InvalidDataException: If the temperature is invalid.
        """
        return await self.async_configure_presets(
            comfort_temperature, self.presets.eco_temperature
        )

    async def async_configure_eco_temperature(self, eco_temperature: float) -> Status:
        """Set the thermostat's eco temperature.

        Temperatures are in degrees Celsius and specified in 0.5 degree increments.
        The initial value is 17.0 degrees Celsius.
        The eco temperature is indicated by the moon symbol.

        Args:
            eco_temperature (float): The eco temperature in degrees Celsius.

        Returns:
            Status: The updated status.

        Raises:
            Eq3StateException: If the presets are not available or the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3CommandException: If an error occurs while sending the command.
            Eq3TimeoutException: If the command times out.
            Eq3InvalidDataException: If the temperature is invalid.
        """
        return await self.async_configure_presets(
            self.presets.comfort_temperature, eco_temperature
        )

    async def async_configure_temperature_offset(
        self, temperature_offset: float
    ) -> Status:
        """Configure the temperature offset.

        The temperature offset is added to the measured temperature to determine the current temperature the thermostat is using internally for its calculations.
        The initial value is 0.0 degrees Celsius.
        The offset is specified in 0.5 degree increments.
        The maximum offset is EQ3_MAX_TEMP_OFFSET and the minimum offset is EQ3_MIN_TEMP_OFFSET.

        Example:
            When the thermostat is set to 20.0 degrees and the actual temperature in the room is 18.0 degrees, the offset can be set to -2.0 degrees to align the thermostat with the actual temperature.

        Args:
            temperature_offset (float): The temperature offset in degrees Celsius.

        Returns:
            Status: The updated status.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3CommandException: If an error occurs while sending the command.
            Eq3TimeoutException: If the command times out.
            Eq3InvalidDataException: If the temperature is invalid.
        """
        return await self._async_write_command_with_status_response(
            _OffsetConfigureCommand(offset=temperature_offset)
        )

    async def async_configure_window_open(
        self, temperature: float, duration: timedelta | float | int
    ) -> Status:
        """Configure the window open behaviour.

        Temperatures are in degrees Celsius and specified in 0.5 degree increments.
        Durations are specified in 5 minute increments.
        The initial values are 12.0 degrees Celsius and 15 minutes.
        If a float is provided for the duration, it will be converted to a timedelta with minutes as the unit.

        Args:
            temperature (float): The temperature at which the window open behavior should be triggered in degrees Celsius.
            duration (timedelta | float): The duration for which the window open behavior should be active.

        Returns:
            Status: The updated status.

        Raises:
            Eq3StateException: If the presets are not available or the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3CommandException: If an error occurs while sending the command.
            Eq3TimeoutException: If the command times out.
            Eq3InvalidDataException: If the temperature or duration is invalid.
        """
        if isinstance(duration, float) or isinstance(duration, int):
            duration = timedelta(minutes=duration)

        return await self._async_write_command_with_status_response(
            _WindowOpenConfigureCommand(
                window_open_temperature=temperature,
                window_open_time=duration,
            )
        )

    async def async_configure_window_open_temperature(
        self, temperature: float
    ) -> Status:
        """Configure the window open temperature.

        Temperatures are in degrees Celsius and specified in 0.5 degree increments.
        The initial value is 12.0 degrees Celsius.

        Args:
            temperature (float): The temperature at which the window open behavior should be triggered in degrees Celsius.

        Returns:
            Status: The updated status.

        Raises:
            Eq3StateException: If the presets are not available or the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3CommandException: If an error occurs while sending the command.
            Eq3TimeoutException: If the command times out.
            Eq3InvalidDataException: If the temperature is invalid.
        """
        return await self.async_configure_window_open(
            temperature, self.presets.window_open_time
        )

    async def async_configure_window_open_duration(
        self,
        duration: timedelta | float,
    ) -> Status:
        """Configure the window open duration.

        The duration is specified in 5 minute increments.
        The initial value is 15 minutes.
        If a float is provided, it will be converted to a timedelta with minutes as the unit.

        Args:
            duration (timedelta | float): The duration for which the window open behavior should be active.

        Returns:
            Status: The updated status.

        Raises:
            Eq3StateException: If the presets are not available or the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3CommandException: If an error occurs while sending the command.
            Eq3TimeoutException: If the command times out.
            Eq3InvalidDataException: If the duration is invalid.
        """
        return await self.async_configure_window_open(
            self.presets.window_open_temperature, duration
        )

    ### Internal ###

    async def __aenter__(self) -> Self:
        """Async context manager enter.

        Connects to the thermostat. After connecting, the device data, status, and schedule will be queried and stored.
        When the connection is established, the CONNECTED event will be triggered.

        Raises:
            Eq3StateException: If the thermostat is already connected.
            Eq3ConnectionException: If the connection fails.
            Eq3TimeoutException: If the connection times out.
            Eq3CommandException: If an error occurs while sending a command.
        """
        await self.async_connect()
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        """Async context manager exit.

        Disconnects from the thermostat. Before disconnection all pending futures will be cancelled.
        When the disconnection is complete, the DISCONNECTED event will be triggered.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3ConnectionException: If the disconnection fails.
            Eq3TimeoutException: If the disconnection times out.
        """
        await self.async_disconnect()

    async def _async_write_command_with_device_data_response(
        self, command: _Eq3Struct
    ) -> DeviceData:
        """Write a command to the thermostat and wait for a device data response.

        Args:
            command (_Eq3Struct): The command to write.

        Returns:
            DeviceData: The device data.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a device data command is already pending.
            Eq3TimeoutException: If the command times out.
            Eq3CommandException: If an error occurs while sending the command.
        """
        if self._device_data_future is not None:
            raise Eq3AlreadyAwaitingResponseException(
                "Already awaiting a device data command response"
            )

        self._device_data_future = asyncio.Future()

        await self._async_write_command(command)

        try:
            device_data = await asyncio.wait_for(
                self._device_data_future, self._command_timeout
            )
        except TimeoutError as ex:
            raise Eq3TimeoutException("Timeout during device data command") from ex
        finally:
            self._device_data_future = None

        return device_data

    async def _async_write_command_with_status_response(
        self, command: _Eq3Struct
    ) -> Status:
        """Write a command to the thermostat and wait for a status response.

        Args:
            command (_Eq3Struct): The command to write.

        Returns:
            Status: The status.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a status command is already pending.
            Eq3TimeoutException: If the command times out.
            Eq3CommandException: If an error occurs while sending the command.
        """
        if self._status_future is not None:
            raise Eq3AlreadyAwaitingResponseException(
                "Already awaiting a status command response"
            )

        self._status_future = asyncio.Future()

        await self._async_write_command(command)

        try:
            status = await asyncio.wait_for(self._status_future, self._command_timeout)
        except TimeoutError as ex:
            raise Eq3TimeoutException("Timeout during status command") from ex
        finally:
            self._status_future = None

        return status

    async def _async_write_commands_with_schedule_response(
        self, commands: list[_Eq3Struct]
    ) -> Schedule:
        """Write commands to the thermostat and wait for a schedule response.

        Args:
            commands (list[_Eq3Struct]): The commands to write.

        Returns:
            Schedule: The schedule.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3AlreadyAwaitingResponseException: If a schedule command is already pending.
            Eq3TimeoutException: If the command times out.
            Eq3CommandException: If an error occurs while sending the command.
        """
        if self._schedule_future is not None:
            raise Eq3AlreadyAwaitingResponseException(
                "Already awaiting a schedule command response"
            )

        self._schedule_future = asyncio.Future()
        self._schedule_future_counter = len(commands)

        for command in commands:
            await self._async_write_command(command)

        try:
            schedule = await asyncio.wait_for(
                self._schedule_future, self._command_timeout
            )
        except TimeoutError as ex:
            raise Eq3TimeoutException("Timeout during schedule command") from ex
        finally:
            self._schedule_future = None

        return schedule

    async def _async_write_command(self, command: _Eq3Struct) -> None:
        """Write a command to the thermostat.

        Args:
            command (_Eq3Struct): The command to write.

        Raises:
            Eq3StateException: If the thermostat is not connected.
            Eq3CommandException: If an error occurs while sending the command.
            Eq3TimeoutException: If the command times out.
        """
        if not self.is_connected:
            raise Eq3StateException("Not connected")

        data = command.to_bytes()

        async with self._lock:
            try:
                await asyncio.wait_for(
                    self._conn.write_gatt_char(_Eq3Characteristic.WRITE, data),
                    self._command_timeout,
                )
            except BleakError as ex:
                raise Eq3CommandException("Error during write") from ex
            except TimeoutError as ex:
                raise Eq3TimeoutException("Timeout during write") from ex

    def _on_disconnected(self, _: BleakClient) -> None:
        """Handle disconnection from the thermostat."""
        asyncio.create_task(self._trigger_event(Eq3Event.DISCONNECTED))

    async def _on_message_received(
        self, _: BleakGATTCharacteristic, data: bytearray
    ) -> None:
        """Handle received messages from the thermostat."""
        data_bytes = bytes(data)
        command = DataclassStruct(_Eq3Message).parse(data_bytes)

        match command.cmd:
            case _Eq3Command.ID_RETURN:
                device_data = DeviceData._from_bytes(data_bytes)
                return await self._on_device_data_received(device_data)
            case _Eq3Command.INFO_RETURN:
                if not command.is_status_command:
                    return

                status = Status._from_bytes(data_bytes)
                return await self._on_status_received(status)
            case _Eq3Command.SCHEDULE_RETURN:
                schedule = Schedule._from_bytes(data_bytes)
                return await self._on_schedule_received(schedule)
            case _:
                raise Eq3InternalException(f"Unknown command: {command}")

    async def _on_device_data_received(self, device_data: DeviceData) -> None:
        """Handle received device data.

        Triggers the DEVICE_DATA_RECEIVED event.

        Args:
            device_data (DeviceData): The received device data.
        """
        self._last_device_data = device_data

        if self._device_data_future is not None:
            self._device_data_future.set_result(device_data)

        await self._trigger_event(
            Eq3Event.DEVICE_DATA_RECEIVED, device_data=device_data
        )

    async def _on_status_received(self, status: Status) -> None:
        """Handle received status.

        Triggers the STATUS_RECEIVED event.

        Args:
            status (Status): The received status.
        """
        self._last_status = status

        if self._status_future is not None:
            self._status_future.set_result(status)

        await self._trigger_event(Eq3Event.STATUS_RECEIVED, status=status)

    async def _on_schedule_received(self, schedule: Schedule) -> None:
        """Handle received schedule.

        Merges the received schedule with the last known schedule and triggers the SCHEDULE_RECEIVED event.

        Args:
            schedule (Schedule): The received schedule.
        """
        if self._last_schedule is None:
            self._last_schedule = schedule

        self._last_schedule.merge(schedule)

        if self._schedule_future is not None:
            self._schedule_future_counter -= 1

            if self._schedule_future_counter == 0:
                self._schedule_future.set_result(self._last_schedule)

        await self._trigger_event(
            Eq3Event.SCHEDULE_RECEIVED, schedule=self._last_schedule
        )

    ### Callbacks ###

    @overload
    def register_callback(
        self,
        event: Literal[Eq3Event.DISCONNECTED],
        callback: Union[Callable[[], None], Callable[[], Awaitable[None]]],
    ) -> None: ...

    @overload
    def register_callback(
        self,
        event: Union[Literal[Eq3Event.CONNECTED]],
        callback: Union[
            Callable[[DeviceData, Status, Schedule], None],
            Callable[[DeviceData, Status, Schedule], Awaitable[None]],
        ],
    ) -> None: ...

    @overload
    def register_callback(
        self,
        event: Literal[Eq3Event.DEVICE_DATA_RECEIVED],
        callback: Union[
            Callable[[DeviceData], None], Callable[[DeviceData], Awaitable[None]]
        ],
    ) -> None: ...

    @overload
    def register_callback(
        self,
        event: Literal[Eq3Event.STATUS_RECEIVED],
        callback: Union[Callable[[Status], None], Callable[[Status], Awaitable[None]]],
    ) -> None: ...

    @overload
    def register_callback(
        self,
        event: Literal[Eq3Event.SCHEDULE_RECEIVED],
        callback: Union[
            Callable[[Schedule], None], Callable[[Schedule], Awaitable[None]]
        ],
    ) -> None: ...

    def register_callback(
        self,
        event: Eq3Event,
        callback: Union[Callable[..., None], Callable[..., Awaitable[None]]],
    ) -> None:
        """Register a callback for a specific event."""
        if callback in self._callbacks[event]:
            return

        self._callbacks[event].append(callback)

    @overload
    def unregister_callback(
        self,
        event: Literal[Eq3Event.DISCONNECTED],
        callback: Union[Callable[[], None], Callable[[], Awaitable[None]]],
    ) -> None: ...

    @overload
    def unregister_callback(
        self,
        event: Union[Literal[Eq3Event.CONNECTED]],
        callback: Union[
            Callable[[DeviceData, Status, Schedule], None],
            Callable[[DeviceData, Status, Schedule], Awaitable[None]],
        ],
    ) -> None: ...

    @overload
    def unregister_callback(
        self,
        event: Literal[Eq3Event.DEVICE_DATA_RECEIVED],
        callback: Union[
            Callable[[DeviceData], None], Callable[[DeviceData], Awaitable[None]]
        ],
    ) -> None: ...

    @overload
    def unregister_callback(
        self,
        event: Literal[Eq3Event.STATUS_RECEIVED],
        callback: Union[Callable[[Status], None], Callable[[Status], Awaitable[None]]],
    ) -> None: ...

    @overload
    def unregister_callback(
        self,
        event: Literal[Eq3Event.SCHEDULE_RECEIVED],
        callback: Union[
            Callable[[Schedule], None], Callable[[Schedule], Awaitable[None]]
        ],
    ) -> None: ...

    def unregister_callback(
        self,
        event: Eq3Event,
        callback: Union[Callable[..., None], Callable[..., Awaitable[None]]],
    ) -> None:
        """Unregister a callback for a specific event."""
        if callback not in self._callbacks[event]:
            return

        self._callbacks[event].remove(callback)

    @overload
    async def _trigger_event(self, event: Literal[Eq3Event.DISCONNECTED]) -> None: ...

    @overload
    async def _trigger_event(
        self,
        event: Literal[Eq3Event.CONNECTED],
        *,
        device_data: DeviceData,
        status: Status,
        schedule: Schedule,
    ) -> None: ...

    @overload
    async def _trigger_event(
        self,
        event: Literal[Eq3Event.DEVICE_DATA_RECEIVED],
        *,
        device_data: DeviceData,
    ) -> None: ...

    @overload
    async def _trigger_event(
        self,
        event: Literal[Eq3Event.STATUS_RECEIVED],
        *,
        status: Status,
    ) -> None: ...

    @overload
    async def _trigger_event(
        self,
        event: Literal[Eq3Event.SCHEDULE_RECEIVED],
        *,
        schedule: Schedule,
    ) -> None: ...

    async def _trigger_event(
        self,
        event: Eq3Event,
        *,
        device_data: DeviceData | None = None,
        status: Status | None = None,
        schedule: Schedule | None = None,
    ) -> None:
        """Call the callbacks for a specific event."""
        async_callbacks = [
            callback
            for callback in self._callbacks[event]
            if asyncio.iscoroutinefunction(callback)
        ]
        sync_callbacks = [
            callback
            for callback in self._callbacks[event]
            if not asyncio.iscoroutinefunction(callback)
        ]

        args: (
            tuple[DeviceData, Status, Schedule]
            | tuple[DeviceData]
            | tuple[Status]
            | tuple[Schedule]
            | tuple[()]
        )

        match event:
            case Eq3Event.DISCONNECTED:
                args = ()
            case Eq3Event.CONNECTED:
                if device_data is None or status is None or schedule is None:
                    raise Eq3InternalException(
                        "device_data, status, and schedule must not be None for CONNECTED event"
                    )
                args = (device_data, status, schedule)
            case Eq3Event.DEVICE_DATA_RECEIVED:
                if device_data is None:
                    raise Eq3InternalException(
                        "device_data must not be None for DEVICE_DATA_RECEIVED event"
                    )
                args = (device_data,)
            case Eq3Event.STATUS_RECEIVED:
                if status is None:
                    raise Eq3InternalException(
                        "status must not be None for STATUS_RECEIVED event"
                    )
                args = (status,)
            case Eq3Event.SCHEDULE_RECEIVED:
                if schedule is None:
                    raise Eq3InternalException(
                        "schedule must not be None for SCHEDULE_RECEIVED event"
                    )
                args = (schedule,)

        await asyncio.gather(*[callback(*args) for callback in async_callbacks])

        for callback in sync_callbacks:
            callback(*args)

is_connected property

is_connected

Check if the thermostat is connected.

Returns:

Name	Type	Description
bool	bool	True if connected, False otherwise.

device_data property

device_data

Get the last known device data, ensuring it's not None.

Returns:

Name	Type	Description
DeviceData	DeviceData	The last known device data.

Raises:

Type	Description
Eq3StateException	If the device data is None. This occurs when the thermostat has not been connected yet.

status property

status

Get the last known status, ensuring it's not None.

Returns:

Name	Type	Description
Status	Status	The last known status.

Raises:

Type	Description
Eq3StateException	If the status is None. This occurs when the thermostat has not been connected yet.

presets property

presets

Get the presets from the last known status.

Presets are only available since firmware version MIN_PRESETS_FW_VERSION.

Returns:

Name	Type	Description
Presets	Presets	The presets.

Raises:

Type	Description
Eq3StateException	If the presets are None. This occurs when the thermostat has not been connected yet or if the thermostat's firmware does not support presets.

schedule property

schedule

Get the last known schedule, ensuring it's not None.

Returns:

Name	Type	Description
Schedule	Schedule	The last known schedule.

Raises:

Type	Description
Eq3StateException	If the schedule is None. This occurs when the thermostat has not been connected

async_connect async

async_connect()

Connect to the thermostat.

After connecting, the device data, status, and schedule will be queried and stored. When the connection is established, the CONNECTED event will be triggered.

Raises:

Type	Description
Eq3StateException	If the thermostat is already connected.
Eq3ConnectionException	If the connection fails.
Eq3TimeoutException	If the connection times out.
Eq3CommandException	If an error occurs while sending a command.

Source code in eq3btsmart/thermostat.py

async def async_connect(self) -> None:
    """Connect to the thermostat.

    After connecting, the device data, status, and schedule will be queried and stored.
    When the connection is established, the CONNECTED event will be triggered.

    Raises:
        Eq3StateException: If the thermostat is already connected.
        Eq3ConnectionException: If the connection fails.
        Eq3TimeoutException: If the connection times out.
        Eq3CommandException: If an error occurs while sending a command.
    """
    if self.is_connected:
        raise Eq3StateException("Already connected")

    try:
        await asyncio.wait_for(self._conn.connect(), self._connection_timeout)
        await self._conn.start_notify(
            _Eq3Characteristic.NOTIFY, self._on_message_received
        )
        (
            self._last_device_data,
            self._last_status,
            self._last_schedule,
        ) = await asyncio.gather(
            self.async_get_device_data(),
            self.async_get_status(),
            self.async_get_schedule(),
        )
    except BleakError as ex:
        raise Eq3ConnectionException("Could not connect to the device") from ex
    except TimeoutError as ex:
        raise Eq3TimeoutException("Timeout during connection") from ex

    await self._trigger_event(
        Eq3Event.CONNECTED,
        device_data=self.device_data,
        status=self.status,
        schedule=self.schedule,
    )

async_disconnect async

async_disconnect()

Disconnect from the thermostat.

Before disconnection all pending futures will be cancelled. When the disconnection is complete, the DISCONNECTED event will be triggered.

Raises:

Type	Description
Eq3StateException	If the thermostat is not connected.
Eq3ConnectionException	If the disconnection fails.
Eq3TimeoutException	If the disconnection times out.

Source code in eq3btsmart/thermostat.py

async def async_disconnect(self) -> None:
    """Disconnect from the thermostat.

    Before disconnection all pending futures will be cancelled.
    When the disconnection is complete, the DISCONNECTED event will be triggered.

    Raises:
        Eq3StateException: If the thermostat is not connected.
        Eq3ConnectionException: If the disconnection fails.
        Eq3TimeoutException: If the disconnection times out.
    """
    if not self.is_connected:
        raise Eq3StateException("Not connected")

    exception = Eq3ConnectionException("Connection closed")
    if self._device_data_future is not None and not self._device_data_future.done():
        self._device_data_future.set_exception(exception)

    if self._status_future is not None and not self._status_future.done():
        self._status_future.set_exception(exception)

    if self._schedule_future is not None and not self._schedule_future.done():
        self._schedule_future.set_exception(exception)

    try:
        await self._conn.disconnect()
    except BleakError as ex:
        raise Eq3ConnectionException("Could not disconnect from the device") from ex
    except TimeoutError as ex:
        raise Eq3TimeoutException("Timeout during disconnection") from ex

async_get_device_data async

async_get_device_data()

Query the latest device data.

Returns:

Name	Type	Description
DeviceData	DeviceData	The device data.

Raises:

Type	Description
Eq3StateException	If the thermostat is not connected.
Eq3CommandException	If an error occurs while sending the command.
Eq3TimeoutException	If the command times out.
Eq3AlreadyAwaitingResponseException	If a device data command is already pending.

Source code in eq3btsmart/thermostat.py

async def async_get_device_data(self) -> DeviceData:
    """Query the latest device data.

    Returns:
        DeviceData: The device data.

    Raises:
        Eq3StateException: If the thermostat is not connected.
        Eq3CommandException: If an error occurs while sending the command.
        Eq3TimeoutException: If the command times out.
        Eq3AlreadyAwaitingResponseException: If a device data command is already pending.
    """
    return await self._async_write_command_with_device_data_response(
        _IdGetCommand()
    )

async_get_status async

async_get_status()

Query the latest status.

Returns:

Name	Type	Description
Status	Status	The status.

Raises:

Type	Description
Eq3StateException	If the thermostat is not connected.
Eq3CommandException	If an error occurs while sending the command.
Eq3TimeoutException	If the command times out.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.

Source code in eq3btsmart/thermostat.py

async def async_get_status(self) -> Status:
    """Query the latest status.

    Returns:
        Status: The status.

    Raises:
        Eq3StateException: If the thermostat is not connected.
        Eq3CommandException: If an error occurs while sending the command.
        Eq3TimeoutException: If the command times out.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
    """
    return await self._async_write_command_with_status_response(
        _InfoGetCommand(time=datetime.now())
    )

async_get_schedule async

async_get_schedule()

Query the schedule.

Returns:

Name	Type	Description
Schedule	Schedule	The schedule.

Raises:

Type	Description
Eq3StateException	If the thermostat is not connected.
Eq3CommandException	If an error occurs while sending the command.
Eq3TimeoutException	If the command times out.
Eq3AlreadyAwaitingResponseException	If a schedule command is already pending.

Source code in eq3btsmart/thermostat.py

async def async_get_schedule(self) -> Schedule:
    """Query the schedule.

    Returns:
        Schedule: The schedule.

    Raises:
        Eq3StateException: If the thermostat is not connected.
        Eq3CommandException: If an error occurs while sending the command.
        Eq3TimeoutException: If the command times out.
        Eq3AlreadyAwaitingResponseException: If a schedule command is already pending.
    """
    return await self._async_write_commands_with_schedule_response(
        [_ScheduleGetCommand(day=week_day) for week_day in Eq3WeekDay]
    )

async_set_temperature async

async_set_temperature(temperature)

Set the target temperature.

Temperatures are in degrees Celsius and specified in 0.5 degree increments. If the temperature is EQ3_OFF_TEMP, the thermostat will be turned off. If the temperature is EQ3_ON_TEMP, the thermostat will be turned on.

Parameters:

Name	Type	Description	Default
temperature	float	The new target temperature in degrees Celsius.	required

Returns:

Name	Type	Description
Status	Status	The updated status.

Raises:

Type	Description
Eq3StateException	If the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.
Eq3CommandException	If an error occurs during the command.
Eq3TimeoutException	If the command times out.
Eq3InvalidDataException	If the temperature is invalid.

Source code in eq3btsmart/thermostat.py

async def async_set_temperature(self, temperature: float) -> Status:
    """Set the target temperature.

    Temperatures are in degrees Celsius and specified in 0.5 degree increments.
    If the temperature is EQ3_OFF_TEMP, the thermostat will be turned off.
    If the temperature is EQ3_ON_TEMP, the thermostat will be turned on.

    Args:
        temperature (float): The new target temperature in degrees Celsius.

    Returns:
        Status: The updated status.

    Raises:
        Eq3StateException: If the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        Eq3CommandException: If an error occurs during the command.
        Eq3TimeoutException: If the command times out.
        Eq3InvalidDataException: If the temperature is invalid.
    """
    if temperature == EQ3_OFF_TEMP:
        return await self.async_set_mode(Eq3OperationMode.OFF)

    if temperature == EQ3_ON_TEMP:
        return await self.async_set_mode(Eq3OperationMode.ON)

    return await self._async_write_command_with_status_response(
        _TemperatureSetCommand(temperature=temperature)
    )

async_set_mode async

async_set_mode(operation_mode)

Set the operation mode.

Parameters:

Name	Type	Description	Default
operation_mode	Eq3OperationMode	The new operation mode.	required

Returns:

Name	Type	Description
Status	Status	The updated status.

Raises:

Type	Description
Eq3StateException	If the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.
Eq3CommandException	If an error occurs during the command.
Eq3TimeoutException	If the command times out.
Eq3InvalidDataException	If the operation mode is not supported.

Source code in eq3btsmart/thermostat.py

async def async_set_mode(self, operation_mode: Eq3OperationMode) -> Status:
    """Set the operation mode.

    Args:
        operation_mode (Eq3OperationMode): The new operation mode.

    Returns:
        Status: The updated status.

    Raises:
        Eq3StateException: If the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        Eq3CommandException: If an error occurs during the command.
        Eq3TimeoutException: If the command times out.
        Eq3InvalidDataException: If the operation mode is not supported.
    """
    command: _ModeSetCommand

    match operation_mode:
        case Eq3OperationMode.AUTO:
            command = _ModeSetCommand(mode=Eq3OperationMode.AUTO)
        case Eq3OperationMode.MANUAL:
            command = _ModeSetCommand(
                mode=Eq3OperationMode.MANUAL
                | _Eq3Temperature.encode(self.status.target_temperature)
            )
        case Eq3OperationMode.OFF:
            command = _ModeSetCommand(
                mode=Eq3OperationMode.MANUAL | _Eq3Temperature.encode(EQ3_OFF_TEMP)
            )
        case Eq3OperationMode.ON:
            command = _ModeSetCommand(
                mode=Eq3OperationMode.MANUAL | _Eq3Temperature.encode(EQ3_ON_TEMP)
            )
        case _:
            raise Eq3InvalidDataException(
                f"Unsupported operation mode: {operation_mode}"
            )

    return await self._async_write_command_with_status_response(command)

async_set_preset async

async_set_preset(preset)

Activate a preset.

Parameters:

Name	Type	Description	Default
preset	Eq3Preset	The preset to activate.	required

Returns:

Name	Type	Description
Status	Status	The updated status.

Raises:

Type	Description
Eq3StateException	If the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.
Eq3CommandException	If an error occurs during the command.
Eq3TimeoutException	If the command times out.

Source code in eq3btsmart/thermostat.py

async def async_set_preset(self, preset: Eq3Preset) -> Status:
    """Activate a preset.

    Args:
        preset (Eq3Preset): The preset to activate.

    Returns:
        Status: The updated status.

    Raises:
        Eq3StateException: If the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        Eq3CommandException: If an error occurs during the command.
        Eq3TimeoutException: If the command times out.
    """
    command: _ComfortSetCommand | _EcoSetCommand

    match preset:
        case Eq3Preset.COMFORT:
            command = _ComfortSetCommand()
        case Eq3Preset.ECO:
            command = _EcoSetCommand()

    return await self._async_write_command_with_status_response(command)

async_set_boost async

async_set_boost(enable)

Enable or disable the boost mode.

The boost mode will set the target temperature to EQ3_ON_TEMP for 300 seconds and then revert to the previous target temperature.

Parameters:

Name	Type	Description	Default
enable	bool	True to enable the boost mode, False to disable it.	required

Returns:

Name	Type	Description
Status	Status	The updated status.

Raises:

Type	Description
Eq3StateException	If the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.
Eq3CommandException	If an error occurs during the command.
Eq3TimeoutException	If the command times out.

Source code in eq3btsmart/thermostat.py

async def async_set_boost(self, enable: bool) -> Status:
    """Enable or disable the boost mode.

    The boost mode will set the target temperature to EQ3_ON_TEMP for 300 seconds and then revert to the previous target temperature.

    Args:
        enable (bool): True to enable the boost mode, False to disable it.

    Returns:
        Status: The updated status.

    Raises:
        Eq3StateException: If the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        Eq3CommandException: If an error occurs during the command.
        Eq3TimeoutException: If the command times out.
    """
    return await self._async_write_command_with_status_response(
        _BoostSetCommand(enable=enable)
    )

async_set_locked async

async_set_locked(enable)

Lock or unlock the thermostat.

When locked, the thermostat's manual controls are disabled.

Parameters:

Name	Type	Description	Default
enable	bool	True to lock the thermostat, False to unlock it.	required

Returns:

Name	Type	Description
Status	Status	The updated status.

Raises:

Type	Description
Eq3StateException	If the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.
Eq3CommandException	If an error occurs during the command.
Eq3TimeoutException	If the command times out.

Source code in eq3btsmart/thermostat.py

async def async_set_locked(self, enable: bool) -> Status:
    """Lock or unlock the thermostat.

    When locked, the thermostat's manual controls are disabled.

    Args:
        enable (bool): True to lock the thermostat, False to unlock it.

    Returns:
        Status: The updated status.

    Raises:
        Eq3StateException: If the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        Eq3CommandException: If an error occurs during the command.
        Eq3TimeoutException: If the command times out.
    """
    return await self._async_write_command_with_status_response(
        _LockSetCommand(enable=enable)
    )

async_set_away async

async_set_away(away_until, temperature)

Set the thermostat to away mode.

The thermostat will be set to away mode until the specified date and time. The target temperature will be set to the specified temperature. Temperatures are in degrees Celsius and specified in 0.5 degree increments.

Parameters:

Name	Type	Description	Default
away_until	datetime	The date and time until the thermostat should be in away mode.	required
temperature	float	The target temperature in degrees Celsius.	required

Returns:

Name	Type	Description
Status	Status	The updated status.

Raises:

Type	Description
Eq3StateException	If the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.
Eq3CommandException	If an error occurs during the command.
Eq3TimeoutException	If the command times out.
Eq3InvalidDataException	If the temperature or date is invalid.

Source code in eq3btsmart/thermostat.py

async def async_set_away(
    self,
    away_until: datetime,
    temperature: float,
) -> Status:
    """Set the thermostat to away mode.

    The thermostat will be set to away mode until the specified date and time.
    The target temperature will be set to the specified temperature.
    Temperatures are in degrees Celsius and specified in 0.5 degree increments.

    Args:
        away_until (datetime): The date and time until the thermostat should be in away mode.
        temperature (float): The target temperature in degrees Celsius.

    Returns:
        Status: The updated status.

    Raises:
        Eq3StateException: If the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        Eq3CommandException: If an error occurs during the command.
        Eq3TimeoutException: If the command times out.
        Eq3InvalidDataException: If the temperature or date is invalid.
    """
    return await self._async_write_command_with_status_response(
        _AwaySetCommand(
            mode=Eq3OperationMode.AWAY | _Eq3Temperature.encode(temperature),
            away_until=away_until,
        )
    )

async_set_schedule async

async_set_schedule(schedule)

Set the schedule.

The schedule allows setting different target temperatures for each day of the week and different times of the day. It is only applied when the thermostat is in AUTO mode.

Parameters:

Name	Type	Description	Default
schedule	Schedule	The schedule to set.	required

Returns:

Type	Description
Schedule	The updated schedule.

Raises:

Type	Description
Eq3StateException	If the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a schedule command is already pending.
Eq3CommandException	If an error occurs during the command.
Eq3TimeoutException	If the command times out.
Eq3InvalidDataException	If any of the schedule data is invalid.

Source code in eq3btsmart/thermostat.py

async def async_set_schedule(self, schedule: Schedule) -> Schedule:
    """Set the schedule.

    The schedule allows setting different target temperatures for each day of the week and different times of the day.
    It is only applied when the thermostat is in AUTO mode.

    Args:
        schedule: The schedule to set.

    Returns:
        The updated schedule.

    Raises:
        Eq3StateException: If the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a schedule command is already pending.
        Eq3CommandException: If an error occurs during the command.
        Eq3TimeoutException: If the command times out.
        Eq3InvalidDataException: If any of the schedule data is invalid.
    """
    return await self._async_write_commands_with_schedule_response(
        [
            _ScheduleSetCommand(
                day=schedule_day.week_day,
                hours=[
                    _ScheduleHourStruct(
                        target_temp=schedule_hour.target_temperature,
                        next_change_at=schedule_hour.next_change_at,
                    )
                    for schedule_hour in schedule_day.schedule_hours
                ],
            )
            for schedule_day in schedule.schedule_days
        ]
    )

async_delete_schedule async

async_delete_schedule(week_days=None)

Delete the schedule for the specified week days.

If no week days are specified, the schedule for all week days will be deleted.

Parameters:

Name	Type	Description	Default
week_days	list[WeekDay] | WeekDay | None	The week days for which the schedule should be deleted. Can be a single WeekDay, a list of WeekDay, or None. Defaults to None.	None

Returns:

Name	Type	Description
Schedule	Schedule	The updated schedule after deletion.

Raises:

Type	Description
Eq3StateException	If the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a schedule command is already pending.
Eq3CommandException	If an error occurs during the command.
Eq3TimeoutException	If the command times out.

Source code in eq3btsmart/thermostat.py

async def async_delete_schedule(
    self, week_days: list[Eq3WeekDay] | Eq3WeekDay | None = None
) -> Schedule:
    """Delete the schedule for the specified week days.

    If no week days are specified, the schedule for all week days will be deleted.

    Args:
        week_days (list[WeekDay] | WeekDay | None, optional): The week days for which the schedule should be deleted.
            Can be a single WeekDay, a list of WeekDay, or None. Defaults to None.

    Returns:
        Schedule: The updated schedule after deletion.

    Raises:
        Eq3StateException: If the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a schedule command is already pending.
        Eq3CommandException: If an error occurs during the command.
        Eq3TimeoutException: If the command times out.
    """
    week_days = (
        [week_days]
        if isinstance(week_days, Eq3WeekDay)
        else week_days or list(Eq3WeekDay)
    )

    return await self._async_write_commands_with_schedule_response(
        [_ScheduleSetCommand(day=week_day, hours=[]) for week_day in week_days]
    )

async_configure_presets async

async_configure_presets(comfort_temperature, eco_temperature)

Set the thermostat's preset temperatures.

Temperatures are in degrees Celsius and specified in 0.5 degree increments. The initial values are 21.0 degrees Celsius for comfort and 17.0 degrees Celsius for eco. The comfort temperature is indicated by the sun symbol and the eco temperature by the moon symbol.

Parameters:

Name	Type	Description	Default
comfort_temperature	float	The comfort temperature in degrees Celsius.	required
eco_temperature	float	The eco temperature in degrees Celsius.	required

Returns:

Name	Type	Description
Status	Status	The updated status.

Raises:

Type	Description
Eq3StateException	If the presets are not available or the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.
Eq3CommandException	If an error occurs while sending the command.
Eq3TimeoutException	If the command times out.
Eq3InvalidDataException	If the temperatures are invalid.

Source code in eq3btsmart/thermostat.py

async def async_configure_presets(
    self,
    comfort_temperature: float,
    eco_temperature: float,
) -> Status:
    """Set the thermostat's preset temperatures.

    Temperatures are in degrees Celsius and specified in 0.5 degree increments.
    The initial values are 21.0 degrees Celsius for comfort and 17.0 degrees Celsius for eco.
    The comfort temperature is indicated by the sun symbol and the eco temperature by the moon symbol.

    Args:
        comfort_temperature (float): The comfort temperature in degrees Celsius.
        eco_temperature (float): The eco temperature in degrees Celsius.

    Returns:
        Status: The updated status.

    Raises:
        Eq3StateException: If the presets are not available or the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        Eq3CommandException: If an error occurs while sending the command.
        Eq3TimeoutException: If the command times out.
        Eq3InvalidDataException: If the temperatures are invalid.
    """
    return await self._async_write_command_with_status_response(
        _ComfortEcoConfigureCommand(
            comfort_temperature=comfort_temperature,
            eco_temperature=eco_temperature,
        )
    )

async_configure_comfort_temperature async

async_configure_comfort_temperature(comfort_temperature)

Set the thermostat's comfort temperature.

Temperatures are in degrees Celsius and specified in 0.5 degree increments. The initial value is 21.0 degrees Celsius. The comfort temperature is indicated by the sun symbol.

Parameters:

Name	Type	Description	Default
comfort_temperature	float	The comfort temperature in degrees Celsius.	required

Returns:

Name	Type	Description
Status	Status	The updated status.

Raises:

Type	Description
Eq3StateException	If the presets are not available or the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.
Eq3CommandException	If an error occurs while sending the command.
Eq3TimeoutException	If the command times out.
Eq3InvalidDataException	If the temperature is invalid.

Source code in eq3btsmart/thermostat.py

async def async_configure_comfort_temperature(
    self, comfort_temperature: float
) -> Status:
    """Set the thermostat's comfort temperature.

    Temperatures are in degrees Celsius and specified in 0.5 degree increments.
    The initial value is 21.0 degrees Celsius.
    The comfort temperature is indicated by the sun symbol.

    Args:
        comfort_temperature (float): The comfort temperature in degrees Celsius.

    Returns:
        Status: The updated status.

    Raises:
        Eq3StateException: If the presets are not available or the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        Eq3CommandException: If an error occurs while sending the command.
        Eq3TimeoutException: If the command times out.
        Eq3InvalidDataException: If the temperature is invalid.
    """
    return await self.async_configure_presets(
        comfort_temperature, self.presets.eco_temperature
    )

async_configure_eco_temperature async

async_configure_eco_temperature(eco_temperature)

Set the thermostat's eco temperature.

Temperatures are in degrees Celsius and specified in 0.5 degree increments. The initial value is 17.0 degrees Celsius. The eco temperature is indicated by the moon symbol.

Parameters:

Name	Type	Description	Default
eco_temperature	float	The eco temperature in degrees Celsius.	required

Returns:

Name	Type	Description
Status	Status	The updated status.

Raises:

Type	Description
Eq3StateException	If the presets are not available or the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.
Eq3CommandException	If an error occurs while sending the command.
Eq3TimeoutException	If the command times out.
Eq3InvalidDataException	If the temperature is invalid.

Source code in eq3btsmart/thermostat.py

async def async_configure_eco_temperature(self, eco_temperature: float) -> Status:
    """Set the thermostat's eco temperature.

    Temperatures are in degrees Celsius and specified in 0.5 degree increments.
    The initial value is 17.0 degrees Celsius.
    The eco temperature is indicated by the moon symbol.

    Args:
        eco_temperature (float): The eco temperature in degrees Celsius.

    Returns:
        Status: The updated status.

    Raises:
        Eq3StateException: If the presets are not available or the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        Eq3CommandException: If an error occurs while sending the command.
        Eq3TimeoutException: If the command times out.
        Eq3InvalidDataException: If the temperature is invalid.
    """
    return await self.async_configure_presets(
        self.presets.comfort_temperature, eco_temperature
    )

async_configure_temperature_offset async

async_configure_temperature_offset(temperature_offset)

Configure the temperature offset.

The temperature offset is added to the measured temperature to determine the current temperature the thermostat is using internally for its calculations. The initial value is 0.0 degrees Celsius. The offset is specified in 0.5 degree increments. The maximum offset is EQ3_MAX_TEMP_OFFSET and the minimum offset is EQ3_MIN_TEMP_OFFSET.

Example

When the thermostat is set to 20.0 degrees and the actual temperature in the room is 18.0 degrees, the offset can be set to -2.0 degrees to align the thermostat with the actual temperature.

Parameters:

Name	Type	Description	Default
temperature_offset	float	The temperature offset in degrees Celsius.	required

Returns:

Name	Type	Description
Status	Status	The updated status.

Raises:

Type	Description
Eq3StateException	If the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.
Eq3CommandException	If an error occurs while sending the command.
Eq3TimeoutException	If the command times out.
Eq3InvalidDataException	If the temperature is invalid.

Source code in eq3btsmart/thermostat.py

async def async_configure_temperature_offset(
    self, temperature_offset: float
) -> Status:
    """Configure the temperature offset.

    The temperature offset is added to the measured temperature to determine the current temperature the thermostat is using internally for its calculations.
    The initial value is 0.0 degrees Celsius.
    The offset is specified in 0.5 degree increments.
    The maximum offset is EQ3_MAX_TEMP_OFFSET and the minimum offset is EQ3_MIN_TEMP_OFFSET.

    Example:
        When the thermostat is set to 20.0 degrees and the actual temperature in the room is 18.0 degrees, the offset can be set to -2.0 degrees to align the thermostat with the actual temperature.

    Args:
        temperature_offset (float): The temperature offset in degrees Celsius.

    Returns:
        Status: The updated status.

    Raises:
        Eq3StateException: If the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        Eq3CommandException: If an error occurs while sending the command.
        Eq3TimeoutException: If the command times out.
        Eq3InvalidDataException: If the temperature is invalid.
    """
    return await self._async_write_command_with_status_response(
        _OffsetConfigureCommand(offset=temperature_offset)
    )

async_configure_window_open async

async_configure_window_open(temperature, duration)

Configure the window open behaviour.

Temperatures are in degrees Celsius and specified in 0.5 degree increments. Durations are specified in 5 minute increments. The initial values are 12.0 degrees Celsius and 15 minutes. If a float is provided for the duration, it will be converted to a timedelta with minutes as the unit.

Parameters:

Name	Type	Description	Default
temperature	float	The temperature at which the window open behavior should be triggered in degrees Celsius.	required
duration	timedelta | float	The duration for which the window open behavior should be active.	required

Returns:

Name	Type	Description
Status	Status	The updated status.

Raises:

Type	Description
Eq3StateException	If the presets are not available or the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.
Eq3CommandException	If an error occurs while sending the command.
Eq3TimeoutException	If the command times out.
Eq3InvalidDataException	If the temperature or duration is invalid.

Source code in eq3btsmart/thermostat.py

async def async_configure_window_open(
    self, temperature: float, duration: timedelta | float | int
) -> Status:
    """Configure the window open behaviour.

    Temperatures are in degrees Celsius and specified in 0.5 degree increments.
    Durations are specified in 5 minute increments.
    The initial values are 12.0 degrees Celsius and 15 minutes.
    If a float is provided for the duration, it will be converted to a timedelta with minutes as the unit.

    Args:
        temperature (float): The temperature at which the window open behavior should be triggered in degrees Celsius.
        duration (timedelta | float): The duration for which the window open behavior should be active.

    Returns:
        Status: The updated status.

    Raises:
        Eq3StateException: If the presets are not available or the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        Eq3CommandException: If an error occurs while sending the command.
        Eq3TimeoutException: If the command times out.
        Eq3InvalidDataException: If the temperature or duration is invalid.
    """
    if isinstance(duration, float) or isinstance(duration, int):
        duration = timedelta(minutes=duration)

    return await self._async_write_command_with_status_response(
        _WindowOpenConfigureCommand(
            window_open_temperature=temperature,
            window_open_time=duration,
        )
    )

async_configure_window_open_temperature async

async_configure_window_open_temperature(temperature)

Configure the window open temperature.

Temperatures are in degrees Celsius and specified in 0.5 degree increments. The initial value is 12.0 degrees Celsius.

Parameters:

Name	Type	Description	Default
temperature	float	The temperature at which the window open behavior should be triggered in degrees Celsius.	required

Returns:

Name	Type	Description
Status	Status	The updated status.

Raises:

Type	Description
Eq3StateException	If the presets are not available or the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.
Eq3CommandException	If an error occurs while sending the command.
Eq3TimeoutException	If the command times out.
Eq3InvalidDataException	If the temperature is invalid.

Source code in eq3btsmart/thermostat.py

async def async_configure_window_open_temperature(
    self, temperature: float
) -> Status:
    """Configure the window open temperature.

    Temperatures are in degrees Celsius and specified in 0.5 degree increments.
    The initial value is 12.0 degrees Celsius.

    Args:
        temperature (float): The temperature at which the window open behavior should be triggered in degrees Celsius.

    Returns:
        Status: The updated status.

    Raises:
        Eq3StateException: If the presets are not available or the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        Eq3CommandException: If an error occurs while sending the command.
        Eq3TimeoutException: If the command times out.
        Eq3InvalidDataException: If the temperature is invalid.
    """
    return await self.async_configure_window_open(
        temperature, self.presets.window_open_time
    )

async_configure_window_open_duration async

async_configure_window_open_duration(duration)

Configure the window open duration.

The duration is specified in 5 minute increments. The initial value is 15 minutes. If a float is provided, it will be converted to a timedelta with minutes as the unit.

Parameters:

Name	Type	Description	Default
duration	timedelta | float	The duration for which the window open behavior should be active.	required

Returns:

Name	Type	Description
Status	Status	The updated status.

Raises:

Type	Description
Eq3StateException	If the presets are not available or the thermostat is not connected.
Eq3AlreadyAwaitingResponseException	If a status command is already pending.
Eq3CommandException	If an error occurs while sending the command.
Eq3TimeoutException	If the command times out.
Eq3InvalidDataException	If the duration is invalid.

Source code in eq3btsmart/thermostat.py

async def async_configure_window_open_duration(
    self,
    duration: timedelta | float,
) -> Status:
    """Configure the window open duration.

    The duration is specified in 5 minute increments.
    The initial value is 15 minutes.
    If a float is provided, it will be converted to a timedelta with minutes as the unit.

    Args:
        duration (timedelta | float): The duration for which the window open behavior should be active.

    Returns:
        Status: The updated status.

    Raises:
        Eq3StateException: If the presets are not available or the thermostat is not connected.
        Eq3AlreadyAwaitingResponseException: If a status command is already pending.
        Eq3CommandException: If an error occurs while sending the command.
        Eq3TimeoutException: If the command times out.
        Eq3InvalidDataException: If the duration is invalid.
    """
    return await self.async_configure_window_open(
        self.presets.window_open_temperature, duration
    )

register_callback

register_callback(event: Literal[DISCONNECTED], callback: Union[Callable[[], None], Callable[[], Awaitable[None]]]) -> None

register_callback(event: Union[Literal[CONNECTED]], callback: Union[Callable[[DeviceData, Status, Schedule], None], Callable[[DeviceData, Status, Schedule], Awaitable[None]]]) -> None

register_callback(event: Literal[DEVICE_DATA_RECEIVED], callback: Union[Callable[[DeviceData], None], Callable[[DeviceData], Awaitable[None]]]) -> None

register_callback(event: Literal[STATUS_RECEIVED], callback: Union[Callable[[Status], None], Callable[[Status], Awaitable[None]]]) -> None

register_callback(event: Literal[SCHEDULE_RECEIVED], callback: Union[Callable[[Schedule], None], Callable[[Schedule], Awaitable[None]]]) -> None

register_callback(event, callback)

Register a callback for a specific event.

Source code in eq3btsmart/thermostat.py

def register_callback(
    self,
    event: Eq3Event,
    callback: Union[Callable[..., None], Callable[..., Awaitable[None]]],
) -> None:
    """Register a callback for a specific event."""
    if callback in self._callbacks[event]:
        return

    self._callbacks[event].append(callback)

unregister_callback

unregister_callback(event: Literal[DISCONNECTED], callback: Union[Callable[[], None], Callable[[], Awaitable[None]]]) -> None

unregister_callback(event: Union[Literal[CONNECTED]], callback: Union[Callable[[DeviceData, Status, Schedule], None], Callable[[DeviceData, Status, Schedule], Awaitable[None]]]) -> None

unregister_callback(event: Literal[DEVICE_DATA_RECEIVED], callback: Union[Callable[[DeviceData], None], Callable[[DeviceData], Awaitable[None]]]) -> None

unregister_callback(event: Literal[STATUS_RECEIVED], callback: Union[Callable[[Status], None], Callable[[Status], Awaitable[None]]]) -> None

unregister_callback(event: Literal[SCHEDULE_RECEIVED], callback: Union[Callable[[Schedule], None], Callable[[Schedule], Awaitable[None]]]) -> None

unregister_callback(event, callback)

Unregister a callback for a specific event.

Source code in eq3btsmart/thermostat.py

def unregister_callback(
    self,
    event: Eq3Event,
    callback: Union[Callable[..., None], Callable[..., Awaitable[None]]],
) -> None:
    """Unregister a callback for a specific event."""
    if callback not in self._callbacks[event]:
        return

    self._callbacks[event].remove(callback)