Feature/get gateway status

custom_components/ttlock/__init__.py

@@ -11,8 +11,8 @@
 from homeassistant.helpers import aiohttp_client, config_entry_oauth2_flow
 
 from .api import TTLockApi
-from .const import DOMAIN, TT_API, TT_LOCKS
-from .coordinator import LockUpdateCoordinator
+from .const import DOMAIN, TT_API, TT_GATEWAYS, TT_LOCKS
+from .coordinator import GatewaysUpdateCoordinator, LockUpdateCoordinator
 from .services import Services
 from .webhook import WebhookHandler
 
@@ -56,6 +56,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     )
     hass.data[DOMAIN][entry.entry_id][TT_LOCKS] = locks
 
+    gateway_coordinator = GatewaysUpdateCoordinator(hass, entry, client)
+    await gateway_coordinator.async_config_entry_first_refresh()
+    hass.data[DOMAIN][entry.entry_id][TT_GATEWAYS] = gateway_coordinator
+
     await WebhookHandler(hass, entry).setup()
 
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)

custom_components/ttlock/api.py

@@ -18,6 +18,7 @@
 from .models import (
     AddPasscodeConfig,
     Features,
+    Gateway,
     Lock,
     LockRecord,
     LockState,
@@ -154,6 +155,11 @@ def lock_connectable(lock) -> bool:
 
         return [lock["lockId"] for lock in res["list"] if lock_connectable(lock)]
 
+    async def get_gateways(self) -> list[Gateway]:
+        """Enumerate all gateways in the account."""
+        res = await self.get("gateway/list", pageNo=1, pageSize=1000)
+        return [Gateway.parse_obj(gateway) for gateway in res["list"]]
+
     async def get_lock(self, lock_id: int) -> Lock:
         """Get a lock by ID."""
         res = await self.get("lock/detail", lockId=lock_id)

custom_components/ttlock/binary_sensor.py

@@ -10,9 +10,17 @@
 )
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity import DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
-from .coordinator import lock_coordinators, sensor_present
+from .const import DOMAIN
+from .coordinator import (
+    GatewaysUpdateCoordinator,
+    gateway_coordinator,
+    lock_coordinators,
+    sensor_present,
+)
 from .entity import BaseLockEntity
 
 _LOGGER = logging.getLogger(__name__)
@@ -39,6 +47,11 @@ async def async_setup_entry(
         ]
     )
 
+    coordinator = gateway_coordinator(hass, entry)
+    async_add_entities(
+        GatewaySensor(coordinator, gateway_id) for gateway_id in coordinator.data
+    )
+
 
 class Sensor(BaseLockEntity, BinarySensorEntity):
     """Current sensor state."""
@@ -62,3 +75,53 @@ def _update_from_coordinator(self) -> None:
         """Fetch state from the device."""
         self._attr_name = f"{self.coordinator.data.name} Passage Mode"
         self._attr_is_on = self.coordinator.data.passage_mode_active()
+
+
+class GatewaySensor(CoordinatorEntity[GatewaysUpdateCoordinator], BinarySensorEntity):
+    """Gateway online status."""
+
+    _attr_device_class = BinarySensorDeviceClass.CONNECTIVITY
+
+    def __init__(self, coordinator: GatewaysUpdateCoordinator, gateway_id: int) -> None:
+        """Initialize the sensor."""
+        super().__init__(coordinator)
+        self.gateway_id = gateway_id
+        self._attr_unique_id = f"{DOMAIN}-gateway-{gateway_id}"
+        self._update_from_coordinator()
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Device info for the gateway."""
+        gateway = self.coordinator.data[self.gateway_id]
+        return DeviceInfo(
+            identifiers={(DOMAIN, gateway.mac)},
+            manufacturer="TT Lock",
+            name=gateway.name,
+            model="Gateway",
+        )
+
+    @property
+    def is_on(self) -> bool:
+        """Return true if the binary sensor is on."""
+        return self.coordinator.data[self.gateway_id].is_online
+
+    @property
+    def extra_state_attributes(self):
+        """Return the state attributes."""
+        gateway = self.coordinator.data[self.gateway_id]
+        return {
+            "network_name": gateway.network_name,
+            "mac": gateway.mac,
+        }
+
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        self._update_from_coordinator()
+        super()._handle_coordinator_update()
+
+    def _update_from_coordinator(self) -> None:
+        """Fetch state from the device."""
+        if self.gateway_id not in self.coordinator.data:
+            return
+        gateway = self.coordinator.data[self.gateway_id]
+        self._attr_name = f"{gateway.name} Status"

custom_components/ttlock/const.py

@@ -3,6 +3,7 @@
 DOMAIN = "ttlock"
 TT_API = "api"
 TT_LOCKS = "locks"
+TT_GATEWAYS = "gateways"
 
 OAUTH2_TOKEN = "https://euapi.ttlock.com/oauth2/token"
 CONF_WEBHOOK_URL = "webhook_url"

custom_components/ttlock/coordinator.py

@@ -18,8 +18,15 @@
 from homeassistant.util import dt
 
 from .api import TTLockApi
-from .const import DOMAIN, SIGNAL_NEW_DATA, TT_LOCKS
-from .models import Features, PassageModeConfig, SensorState, State, WebhookEvent
+from .const import DOMAIN, SIGNAL_NEW_DATA, TT_GATEWAYS, TT_LOCKS
+from .models import (
+    Features,
+    Gateway,
+    PassageModeConfig,
+    SensorState,
+    State,
+    WebhookEvent,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -114,6 +121,13 @@ def lock_coordinators(hass: HomeAssistant, entry: ConfigEntry):
     yield from coordinators
 
 
+def gateway_coordinator(
+    hass: HomeAssistant, entry: ConfigEntry
+) -> GatewaysUpdateCoordinator:
+    """Get the gateway coordinator."""
+    return hass.data[DOMAIN][entry.entry_id][TT_GATEWAYS]
+
+
 def coordinator_for(
     hass: HomeAssistant, entity_id: str
 ) -> LockUpdateCoordinator | None:
@@ -333,3 +347,31 @@ async def set_lock_sound(self, on: bool) -> None:
         if res:
             self.data.lock_sound = on
             self.async_update_listeners()
+
+
+class GatewaysUpdateCoordinator(DataUpdateCoordinator[dict[int, Gateway]]):
+    """Class to manage fetching Gateway data."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        config_entry: ConfigEntry,
+        api: TTLockApi,
+    ) -> None:
+        """Initialize the update co-ordinator for gateways."""
+        self.api = api
+
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=f"{DOMAIN}-gateways",
+            config_entry=config_entry,
+            update_interval=timedelta(minutes=15),
+        )
+
+    async def _async_update_data(self) -> dict[int, Gateway]:
+        try:
+            gateways = await self.api.get_gateways()
+            return {gateway.id: gateway for gateway in gateways}
+        except Exception as err:
+            raise UpdateFailed(err) from err

custom_components/ttlock/models.py

@@ -98,6 +98,16 @@ class Sensor(BaseModel):
     mac: str = Field(..., alias="mac")
 
 
+class Gateway(BaseModel):
+    """Gateway details."""
+
+    id: int = Field(..., alias="gatewayId")
+    name: str = Field(..., alias="gatewayName")
+    mac: str = Field(..., alias="gatewayMac")
+    is_online: bool = Field(..., alias="isOnline")
+    network_name: str = Field(None, alias="networkName")
+
+
 class LockState(BaseModel):
     """Lock state."""
 

tests/conftest.py

@@ -181,6 +181,9 @@ async def mock_get_passage_mode(*args, **kwargs):
         async def mock_get_lock_records(*args, **kwargs):
             return mock_data.records
 
+        async def mock_get_gateways(*args, **kwargs):
+            return []
+
         monkeypatch.setattr(
             "custom_components.ttlock.api.TTLockApi.get_locks", mock_get_locks
         )
@@ -201,6 +204,10 @@ async def mock_get_lock_records(*args, **kwargs):
             "custom_components.ttlock.api.TTLockApi.get_lock_records",
             mock_get_lock_records,
         )
+        monkeypatch.setattr(
+            "custom_components.ttlock.api.TTLockApi.get_gateways",
+            mock_get_gateways,
+        )
 
     return create_mock_responses
 

tests/test_gateway.py

@@ -0,0 +1,124 @@
+"""Test the TTLock Gateway Binary Sensor."""
+
+import pytest
+
+from custom_components.ttlock.const import DOMAIN, TT_GATEWAYS
+from custom_components.ttlock.models import Gateway
+from homeassistant.const import STATE_OFF, STATE_ON
+from homeassistant.core import HomeAssistant
+
+GATEWAY_DETAILS = {
+    "gatewayId": 1461158,
+    "gatewayName": "Test Gateway",
+    "gatewayMac": "05:F6:1E:93:8B:2B",
+    "isOnline": 1,
+    "networkName": "Battat2",
+}
+
+
+@pytest.fixture
+def mock_gateway_response(monkeypatch):
+    """Mock the get_gateways API response."""
+
+    async def mock_get_gateways(*args, **kwargs):
+        return [Gateway.parse_obj(GATEWAY_DETAILS)]
+
+    monkeypatch.setattr(
+        "custom_components.ttlock.api.TTLockApi.get_gateways", mock_get_gateways
+    )
+
+
+async def test_gateway_sensor_setup(
+    hass: HomeAssistant, component_setup, mock_api_responses, monkeypatch
+):
+    """Test that the gateway binary sensor is set up correctly."""
+    mock_api_responses("default")
+
+    # Override the default empty gateway mock
+    async def mock_get_gateways(*args, **kwargs):
+        return [Gateway.parse_obj(GATEWAY_DETAILS)]
+
+    monkeypatch.setattr(
+        "custom_components.ttlock.api.TTLockApi.get_gateways", mock_get_gateways
+    )
+
+    await component_setup()
+
+    # Get the gateway coordinator from hass.data to verify it's there
+    entries = hass.config_entries.async_entries(DOMAIN)
+    assert len(entries) == 1
+    entry = entries[0]
+
+    # Check if coordinator is stored
+    assert TT_GATEWAYS in hass.data[DOMAIN][entry.entry_id]
+
+    # Verify entity state
+    # Entity ID should be binary_sensor.test_gateway_status (based on slugified name + status)
+    # Actually, the name logic in implementation was f"{gateway.name} Status"
+    # So it should be binary_sensor.test_gateway_status
+    entity_id = "binary_sensor.test_gateway_status"
+    state = hass.states.get(entity_id)
+
+    assert state is not None
+    assert state.state == STATE_ON
+    assert state.attributes["network_name"] == "Battat2"
+    assert state.attributes["mac"] == "05:F6:1E:93:8B:2B"
+    assert state.attributes["device_class"] == "connectivity"
+
+
+async def test_gateway_sensor_offline(
+    hass: HomeAssistant, component_setup, mock_api_responses, monkeypatch
+):
+    """Test that the gateway binary sensor reports offline."""
+    mock_api_responses("default")
+
+    offline_gateway = GATEWAY_DETAILS.copy()
+    offline_gateway["isOnline"] = 0
+
+    async def mock_get_gateways_offline(*args, **kwargs):
+        return [Gateway.parse_obj(offline_gateway)]
+
+    monkeypatch.setattr(
+        "custom_components.ttlock.api.TTLockApi.get_gateways", mock_get_gateways_offline
+    )
+
+    await component_setup()
+
+    entity_id = "binary_sensor.test_gateway_status"
+    state = hass.states.get(entity_id)
+
+    assert state is not None
+    assert state.state == STATE_OFF
+
+
+async def test_setup_with_no_gateways(
+    hass: HomeAssistant, component_setup, mock_api_responses, monkeypatch
+):
+    """Test setup when account has no gateways."""
+
+    async def mock_get_gateways_empty(*args, **kwargs):
+        return []
+
+    monkeypatch.setattr(
+        "custom_components.ttlock.api.TTLockApi.get_gateways", mock_get_gateways_empty
+    )
+
+    mock_api_responses("default")
+    await component_setup()
+
+    # Verify coordinator exists but has empty data
+    entries = hass.config_entries.async_entries(DOMAIN)
+    entry = entries[0]
+    assert TT_GATEWAYS in hass.data[DOMAIN][entry.entry_id]
+    coordinator = hass.data[DOMAIN][entry.entry_id][TT_GATEWAYS]
+    assert coordinator.data == {}
+
+    # Verify no gateway entities created
+    # We can check specific naming pattern or simply that no binary_sensor.gateway* exists
+    states = hass.states.async_all()
+    gateway_sensors = [
+        state.entity_id
+        for state in states
+        if state.entity_id.startswith("binary_sensor.") and "gateway" in state.entity_id
+    ]
+    assert len(gateway_sensors) == 0