POC: add unified devices method

src/tadoasync/api_v3.py

@@ -0,0 +1,48 @@
+"""Wrapper for Tado v3 API."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import orjson
+
+from tadoasync.models_v3 import Device, TemperatureOffset, Zone
+
+if TYPE_CHECKING:
+    from tadoasync.tadoasync import Tado
+
+
+class ApiV3:
+    """Wrapper class for the Tado v3 API."""
+
+    def __init__(self, base: Tado) -> None:
+        """Initialize the API wrapper."""
+        self._base = base
+
+    async def get_zones(self) -> list[Zone]:
+        """Get zones."""
+        response = await self._base._request(  # noqa: SLF001
+            uri=f"homes/{self._base._home_id}/zones",  # noqa: SLF001
+        )
+        obj = orjson.loads(response)
+        return [Zone.from_dict(zone) for zone in obj]
+
+    async def get_devices(self) -> list[Device]:
+        """Get devices."""
+        response = await self._base._request(  # noqa: SLF001
+            uri=f"homes/{self._base._home_id}/devices",  # noqa: SLF001
+        )
+        obj = orjson.loads(response)
+        return [Device.from_dict(device) for device in obj]
+
+    async def get_device(self, serial_no: str) -> Device:
+        """Get device."""
+        response = await self._base._request(  # noqa: SLF001
+            uri=f"homes/{self._base._home_id}/devices/{serial_no}",  # noqa: SLF001
+        )
+        return Device.from_json(response)
+
+    async def get_device_temperature_offset(self, serial_no: str) -> TemperatureOffset:
+        """Get the device temperature offset."""
+        response = await self._base._request(f"devices/{serial_no}/temperatureOffset")  # noqa: SLF001
+        return TemperatureOffset.from_json(response)

src/tadoasync/api_x.py

@@ -0,0 +1,28 @@
+"""Wrapper for Tado X API."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from tadoasync.models_x import RoomsAndDevices
+
+API_URL = "hops.tado.com"
+
+if TYPE_CHECKING:
+    from tadoasync.tadoasync import Tado
+
+
+class ApiX:  # pylint: disable=too-few-public-methods
+    """Wrapper class for the Tado X API."""
+
+    def __init__(self, base: Tado) -> None:
+        """Initialize the API wrapper."""
+        self._base = base
+
+    async def get_rooms_and_devices(self) -> RoomsAndDevices:
+        """Get rooms and devices."""
+        response = await self._base._request(  # noqa: SLF001
+            endpoint=API_URL,
+            uri=f"homes/{self._base._home_id}/roomsAndDevices",  # noqa: SLF001
+        )
+        return RoomsAndDevices.from_json(response)

src/tadoasync/const.py

@@ -83,6 +83,8 @@
     CONST_HVAC_COOL: CONST_MODE_COOL,
 }
 
+INSIDE_TEMPERATURE_MEASUREMENT = "INSIDE_TEMPERATURE_MEASUREMENT"
+
 # These modes will not allow a temp to be set
 TADO_MODES_WITH_NO_TEMP_SETTING = [CONST_MODE_AUTO, CONST_MODE_DRY, CONST_MODE_FAN]
 
@@ -99,3 +101,10 @@ class HttpMethod(Enum):
     POST = "POST"
     PUT = "PUT"
     DELETE = "DELETE"
+
+
+class TadoLine(Enum):
+    """Supported Tado product lines."""
+
+    PRE_LINE_X = "PRE_LINE_X"
+    LINE_X = "LINE_X"

src/tadoasync/models_unified.py

@@ -0,0 +1,61 @@
+"""Abstract models for interaction with the Tado API, regardless of line."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from mashumaro.mixins.orjson import DataClassORJSONMixin
+
+if TYPE_CHECKING:
+    from tadoasync import models_v3, models_x
+
+
+@dataclass
+class Device(DataClassORJSONMixin):
+    """Device model."""
+
+    device_type: str
+    serial: str
+    firmware_version: str
+
+    connection_state: bool
+
+    battery_state: str | None
+
+    temperature_offset: float | None = None
+
+    child_lock_enabled: bool | None = None
+
+    @classmethod
+    def from_v3(
+        cls,
+        v3_device: models_v3.Device,
+        offset: models_v3.TemperatureOffset | None = None,
+    ) -> Device:
+        """Create a device from the v3 API."""
+        return cls(
+            device_type=v3_device.device_type,
+            serial=v3_device.serial_no,
+            firmware_version=v3_device.current_fw_version,
+            temperature_offset=offset.celsius if offset else None,
+            connection_state=v3_device.connection_state.value,
+            battery_state=v3_device.battery_state,
+            child_lock_enabled=v3_device.child_lock_enabled,
+        )
+
+    @classmethod
+    def from_x(
+        cls,
+        x_device: models_x.Device,
+    ) -> Device:
+        """Create a device from the X API."""
+        return cls(
+            device_type=x_device.type,
+            serial=x_device.serial_number,
+            firmware_version=x_device.firmware_version,
+            temperature_offset=x_device.temperature_offset,
+            connection_state=x_device.connection.state == "CONNECTED",
+            battery_state=x_device.battery_state,
+            child_lock_enabled=x_device.child_lock_enabled,
+        )

src/tadoasync/models.py → src/tadoasync/models_v3.py

@@ -1,4 +1,4 @@
-"""Models for the Tado API."""
+"""Models for the Tado v3 API."""
 
 from __future__ import annotations
 

src/tadoasync/models_x.py

@@ -0,0 +1,76 @@
+"""Models for the Tado X API."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from mashumaro import field_options
+from mashumaro.mixins.orjson import DataClassORJSONMixin
+
+
+@dataclass
+class DeviceManualControlTermination(DataClassORJSONMixin):
+    """Represents the manual control termination settings of a device."""
+
+    type: str
+    duration_in_seconds: int | None = field(
+        metadata=field_options(alias="durationInSeconds")
+    )
+
+
+@dataclass
+class Connection(DataClassORJSONMixin):
+    """Connection model represents the connection information of a device."""
+
+    state: str
+
+
+@dataclass
+class Device(DataClassORJSONMixin):
+    """Device model represents the device information."""
+
+    serial_number: str = field(metadata=field_options(alias="serialNumber"))
+    type: str
+    firmware_version: str = field(metadata=field_options(alias="firmwareVersion"))
+    connection: Connection
+    battery_state: str | None = field(
+        default=None, metadata=field_options(alias="batteryState")
+    )
+    temperature_as_measured: float | None = field(
+        default=None, metadata=field_options(alias="temperatureAsMeasured")
+    )
+    temperature_offset: float | None = field(
+        default=None, metadata=field_options(alias="temperatureOffset")
+    )
+    mounting_state: str | None = field(
+        default=None, metadata=field_options(alias="mountingState")
+    )
+    child_lock_enabled: bool | None = field(
+        default=None, metadata=field_options(alias="childLockEnabled")
+    )
+
+
+@dataclass
+class Room(DataClassORJSONMixin):
+    """Room model represents the room information of a home."""
+
+    room_id: int = field(metadata=field_options(alias="roomId"))
+    room_name: str = field(metadata=field_options(alias="roomName"))
+    device_manual_control_termination: DeviceManualControlTermination = field(
+        metadata=field_options(alias="deviceManualControlTermination")
+    )
+    devices: list[Device]
+    zone_controller_assignable: bool = field(
+        metadata=field_options(alias="zoneControllerAssignable")
+    )
+    zone_controllers: list[Any] = field(metadata=field_options(alias="zoneControllers"))
+    room_link_available: bool = field(metadata=field_options(alias="roomLinkAvailable"))
+
+
+@dataclass
+class RoomsAndDevices(DataClassORJSONMixin):
+    """RoomsAndDevices model represents the rooms and devices information of a home."""
+
+    rooms: list[Room]
+    other_devices: list[Device] = field(metadata=field_options(alias="otherDevices"))

src/tadoasync/tadoasync.py

@@ -9,7 +9,7 @@
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from importlib import metadata
-from typing import Self
+from typing import TYPE_CHECKING, Self
 from urllib.parse import urlencode
 
 import jwt
@@ -18,6 +18,8 @@
 from aiohttp.client import ClientSession
 from yarl import URL
 
+from tadoasync.api_v3 import ApiV3
+from tadoasync.api_x import ApiX
 from tadoasync.const import (
     CONST_AWAY,
     CONST_FAN_AUTO,
@@ -37,6 +39,7 @@
     TADO_MODES_TO_HVAC_ACTION,
     TYPE_AIR_CONDITIONING,
     HttpMethod,
+    TadoLine,
 )
 from tadoasync.exceptions import (
     TadoAuthenticationError,
@@ -46,7 +49,7 @@
     TadoForbiddenError,
     TadoReadingError,
 )
-from tadoasync.models import (
+from tadoasync.models_v3 import (
     Capabilities,
     Device,
     GetMe,
@@ -57,6 +60,10 @@
     Zone,
     ZoneState,
 )
+from tadoasync.unifier import get_unifier_from_generation
+
+if TYPE_CHECKING:
+    from tadoasync import models_unified as unified_models
 
 CLIENT_ID = "1bb50063-6b0c-4d11-bd99-387f4a91cc46"
 TOKEN_URL = "https://login.tado.com/oauth2/token"  # noqa: S105
@@ -109,6 +116,10 @@ def __init__(
         self._home_id: int | None = None
         self._me: GetMe | None = None
         self._auto_geofencing_supported: bool | None = None
+        self._tado_line: TadoLine | None = None
+
+        self.api_x = ApiX(self)
+        self.api_v3 = ApiV3(self)
 
         self._user_code: str | None = None
         self._device_verification_url: str | None = None
@@ -545,6 +556,15 @@ async def get_device_info(
         response = await self._request(f"devices/{serial_no}/")
         return Device.from_json(response)
 
+    async def get_unified_devices(self) -> list[unified_models.Device]:
+        """Get devices in a unified format, compatible with both Tado X and v3."""
+        unifier = get_unifier_from_generation(
+            generation=self._tado_line,
+            api_x=self.api_x,
+            api_v3=self.api_v3,
+        )
+        return await unifier.get_devices()
+
     async def set_child_lock(self, serial_no: str, *, child_lock: bool) -> None:
         """Set the child lock."""
         await self._request(
@@ -581,6 +601,11 @@ async def _request(
         url = URL.build(scheme="https", host=TADO_HOST_URL, path=TADO_API_PATH)
         if endpoint == EIQ_HOST_URL:
             url = URL.build(scheme="https", host=EIQ_HOST_URL, path=EIQ_API_PATH)
+        elif endpoint != API_URL:
+            endpoint_url = (
+                endpoint if endpoint.startswith("http") else f"https://{endpoint}"
+            )
+            url = URL(endpoint_url)
 
         if uri:
             url = url.joinpath(uri)