Add new connectivity checker

custom_components/jibo/binary_sensor.py

@@ -0,0 +1,63 @@
+import asyncio
+import logging
+from datetime import timedelta
+
+from homeassistant.components.binary_sensor import (
+    BinarySensorDeviceClass,
+    BinarySensorEntity,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+SCAN_INTERVAL = timedelta(seconds=30)
+_CONNECT_TIMEOUT = 5.0
+_JIBO_PORT = 8089
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    data = hass.data[DOMAIN][entry.entry_id]
+    async_add_entities(
+        [JiboConnectivitySensor(entry, data["jibo_ip"], data["name"])],
+        update_before_add=True,
+    )
+
+
+class JiboConnectivitySensor(BinarySensorEntity):
+    _attr_device_class = BinarySensorDeviceClass.CONNECTIVITY
+    _attr_should_poll = True
+
+    def __init__(self, entry: ConfigEntry, ip: str, name: str) -> None:
+        self._ip = ip
+        self._attr_unique_id = f"{entry.entry_id}_connectivity"
+        self._attr_name = f"{name} Online"
+        self._attr_is_on = False
+        self._attr_device_info = {
+            "identifiers": {(DOMAIN, entry.entry_id)},
+            "name": name,
+            "manufacturer": "Jibo Inc.",
+            "model": "Jibo",
+        }
+
+    async def async_update(self) -> None:
+        try:
+            _, writer = await asyncio.wait_for(
+                asyncio.open_connection(self._ip, _JIBO_PORT),
+                timeout=_CONNECT_TIMEOUT,
+            )
+            writer.close()
+            try:
+                await writer.wait_closed()
+            except Exception:
+                pass
+            self._attr_is_on = True
+        except (asyncio.TimeoutError, OSError):
+            self._attr_is_on = False