feat: display MAC address and local IP in AirSend Box device info

Added invert option detail and modified manual installation method.

README.md

@@ -67,16 +67,18 @@ devices:
 
 ### Manual installation (until the repository is available in HACS)
 
-1. Download the latest release from [GitHub releases](https://github.com/devmel/hass_airsend/releases).
+
-2. Copy the `custom_components/airsend` directory into your `custom_components` folder.
+1. In HACS, go to **Integrations** → click the three-dot menu → **Custom repositories**.
-3. Restart Home Assistant.
+2. Add the repository URL: `https://github.com/devmel/hass_airsend` and select **Integration** as the category.
-4. Go to **Settings → Integrations → Add integration** and search for `AirSend`.
+3. Search for **AirSend** in HACS and click **Install**.
+4. **Restart** Home Assistant.
+5. Go to **Settings → Integrations → Add integration** and search for `AirSend`.
 
 ### HACS (recommended)
 
 1. Ensure [HACS](https://hacs.xyz) is installed.
 2. Search for `AirSend` in HACS and install it, or use the button below.
-3. Restart Home Assistant.
+3. **Restart** Home Assistant.
 4. Go to **Settings → Integrations → Add integration** and search for `AirSend`.
 
 [![Open your Home Assistant instance and open a repository inside the Home Assistant Community Store.](https://my.home-assistant.io/badges/hacs_repository.svg)](https://my.home-assistant.io/redirect/hacs_repository/?owner=Devmel&repository=hass_airsend)
@@ -102,6 +104,7 @@ To reload devices after modifying `airsend.yaml`, use the **Reconfigure** option
 | `sensors` | Enable temperature and illuminance sensors for AirSend box (`true`/`false`) |
 | `bind` | Channel ID to bind for incoming RF messages |
 | `refresh` | Poll interval in seconds (default: 300) |
+| `invert` | Give possibility to reverse command UP/DOWN as DOWN/UP |
 
 ## Informations
 

custom_components/airsend/device.py

@@ -1,310 +1,356 @@
-"""AirSend device."""
+"""AirSend device."""
-import logging
+import logging
-import json
+import json
-import hashlib
+import hashlib
-import aiohttp
+import re
-from . import DOMAIN
+import aiohttp
-
+from . import DOMAIN
-_LOGGER = logging.getLogger(DOMAIN)
+
-
+_LOGGER = logging.getLogger(DOMAIN)
-RTYPE_LABELS = {
+
-    0: "AirSend",
+RTYPE_LABELS = {
-    1: "AirSend Sensor",
+    0: "AirSend Box",
-    4096: "AirSend Button",
+    1: "AirSend Sensor",
-    4097: "AirSend Switch",
+    4096: "AirSend Button",
-    4098: "AirSend Cover",
+    4097: "AirSend Switch",
-    4099: "AirSend Cover (position)",
+    4098: "AirSend Cover",
-    4100: "AirSend Light",
+    4099: "AirSend Cover (position)",
-}
+    4100: "AirSend Light",
-
+}
-
+
-from enum import Enum
+
-
+from enum import Enum
-
+
-class TransferResult(Enum):
+
-    """Result of an async_transfer call."""
+class TransferResult(Enum):
-    SUCCESS = "success"          # Command sent and confirmed
+    """Result of an async_transfer call."""
-    SENT = "sent"                # Command sent, no confirmation (wait=True + 500)
+    SUCCESS = "success"          # Command sent and confirmed
-    SERVER_ERROR = "server_error" # Server responded with error (command may have been sent)
+    SENT = "sent"                # Command sent, no confirmation (wait=True + 500)
-    NETWORK_ERROR = "network_error" # Addon unreachable (timeout, connection refused)
+    SERVER_ERROR = "server_error" # Server responded with error (command may have been sent)
-
+    NETWORK_ERROR = "network_error" # Addon unreachable (timeout, connection refused)
-
+
-class Device:
+
-    """Representation of a Device."""
+class Device:
-
+    """Representation of a Device."""
-    def __init__(
+
-        self,
+    def __init__(
-        name: str,
+        self,
-        options,
+        name: str,
-        serviceurl: str,
+        options,
-    ) -> None:
+        serviceurl: str,
-        """Initialize a device."""
+    ) -> None:
-        self._name = name
+        """Initialize a device."""
-        self._serviceurl = serviceurl
+        self._name = name
-        self._uid = None
+        self._serviceurl = serviceurl
-        self._rtype = None
+        self._uid = None
-        self._apikey = None
+        self._rtype = None
-        self._spurl = None
+        self._apikey = None
-        self._wait = False
+        self._spurl = None
-        self._channel = {}
+        self._wait = False
-        self._note = None
+        self._channel = {}
-        self._bind = None
+        self._note = None
-        self._refresh = 5 * 60
+        self._bind = None
-        try:
+        self._refresh = 5 * 60
-            self._uid = options["id"]
+        try:
-        except KeyError:
+            self._uid = options["id"]
-            pass
+        except KeyError:
-        try:
+            pass
-            self._rtype = options["type"]
+        try:
-        except KeyError:
+            self._rtype = options["type"]
-            pass
+        except KeyError:
-        if self._rtype == 0:
+            pass
-            self._channel = {"id": 1}
+        if self._rtype == 0:
-        try:
+            self._channel = {"id": 1}
-            self._apikey = options["apiKey"]
+        try:
-        except KeyError:
+            self._apikey = options["apiKey"]
-            pass
+        except KeyError:
-        try:
+            pass
-            self._spurl = options["spurl"]
+        try:
-        except KeyError:
+            self._spurl = options["spurl"]
-            pass
+        except KeyError:
-        try:
+            pass
-            self._wait = eval(str(options["wait"]))
+        try:
-        except KeyError:
+            self._wait = eval(str(options["wait"]))
-            pass
+        except KeyError:
-        try:
+            pass
-            self._channel = options["channel"]
+        try:
-        except KeyError:
+            self._channel = options["channel"]
-            pass
+        except KeyError:
-        try:
+            pass
-            self._note = options["note"]
+        try:
-        except KeyError:
+            self._note = options["note"]
-            pass
+        except KeyError:
-        try:
+            pass
-            self._bind = int(options["bind"])
+        try:
-        except KeyError:
+            self._bind = int(options["bind"])
-            pass
+        except KeyError:
-        try:
+            pass
-            self._refresh = int(options["refresh"])
+        try:
-        except KeyError:
+            self._refresh = int(options["refresh"])
-            pass
+        except KeyError:
-        try:
+            pass
-            self._invert = bool(options["invert"])
+        try:
-        except KeyError:
+            self._invert = bool(options["invert"])
-            self._invert = False
+        except KeyError:
-
+            self._invert = False
-    @property
+
-    def name(self) -> str:
+        # MAC and local IP — only for AirSend Box (type 0)
-        """Return the name."""
+        # MAC derived from EUI-64 link-local IPv6 in spurl: sp://TOKEN@[fe80::xxxx]?gw=0&rhost=192.168.x.x
-        return self._name
+        self._mac = None
-
+        self._local_ip = None
-    @property
+        if self._rtype == 0 and self._spurl:
-    def unique_channel_name(self) -> str:
+            try:
-        if self._uid:
+                ipv6_match = re.search(r'@\[([^\]]+)\]', self._spurl)
-            return str(self._uid)
+                if ipv6_match:
-        if self._channel:
+                    self._mac = self._ipv6_link_local_to_mac(ipv6_match.group(1))
-            result = str(self._channel['id'])
+                rhost_match = re.search(r'rhost=([^&]+)', self._spurl)
-            if result:
+                if rhost_match:
-                uniquefield = ['source', 'mac', 'seed']
+                    self._local_ip = rhost_match.group(1)
-                for field in uniquefield:
+            except Exception:
-                    if field in self._channel:
+                pass
-                        result += "_"
+
-                        result += str(self._channel[field])
+    @staticmethod
-            return result
+    def _ipv6_link_local_to_mac(ipv6_str: str) -> str:
-        return self._name
+        """Convert a fe80:: EUI-64 link-local IPv6 address to a MAC address."""
-
+        ipv6_str = ipv6_str.lower().strip()
-    @property
+        if '::' in ipv6_str:
-    def device_info(self) -> dict:
+            left, right = ipv6_str.split('::')
-        """Return device info for Home Assistant device registry."""
+            left_groups = left.split(':') if left else []
-        return {
+            right_groups = right.split(':') if right else []
-            "identifiers": {(DOMAIN, self.unique_channel_name)},
+            missing = 8 - len(left_groups) - len(right_groups)
-            "name": self._name,
+            groups = left_groups + ['0'] * missing + right_groups
-            "manufacturer": "AirSend",
+        else:
-            "model": RTYPE_LABELS.get(self._rtype, "AirSend"),
+            groups = ipv6_str.split(':')
-        }
+        groups = [g.zfill(4) for g in groups]
-
+        eui64 = ''.join(groups[4:])
-    @property
+        b = [int(eui64[i:i+2], 16) for i in range(0, 16, 2)]
-    def extra_state_attributes(self):
+        mac_bytes = [b[0] ^ 0x02, b[1], b[2], b[5], b[6], b[7]]
-        if self._channel:
+        return ':'.join(f'{x:02X}' for x in mac_bytes)
-            return {"channel": self._channel}
+
-        return None
+    @property
-
+    def name(self) -> str:
-    @property
+        """Return the name."""
-    def is_async(self) -> bool:
+        return self._name
-        """Return if asynchronous state."""
+
-        return self._wait == False
+    @property
-
+    def unique_channel_name(self) -> str:
-    @property
+        if self._uid:
-    def is_airsend(self) -> bool:
+            return str(self._uid)
-        return self._rtype == 0
+        if self._channel:
-
+            result = str(self._channel['id'])
-    @property
+            if result:
-    def is_sensor(self) -> bool:
+                uniquefield = ['source', 'mac', 'seed']
-        return self._rtype == 1
+                for field in uniquefield:
-
+                    if field in self._channel:
-    @property
+                        result += "_"
-    def is_button(self) -> bool:
+                        result += str(self._channel[field])
-        return self._rtype == 4096
+            return result
-
+        return self._name
-    @property
+
-    def is_cover(self) -> bool:
+    @property
-        return self._rtype in (4098, 4099)
+    def device_info(self) -> dict:
-
+        """Return device info for Home Assistant device registry."""
-    @property
+        from homeassistant.helpers import device_registry as dr
-    def is_cover_with_position(self) -> bool:
+
-        return self._rtype == 4099
+        connections = set()
-
+        if self._mac:
-    @property
+            connections.add((dr.CONNECTION_NETWORK_MAC, self._mac))
-    def is_switch(self) -> bool:
+
-        return self._rtype == 4097
+        info = {
-
+            "identifiers": {(DOMAIN, self.unique_channel_name)},
-    @property
+            "name": self._name,
-    def is_light(self) -> bool:
+            "manufacturer": "AirSend",
-        return self._rtype == 4100
+            "model": RTYPE_LABELS.get(self._rtype, "AirSend"),
-
+        }
-    @property
+        if connections:
-    def is_inverted(self) -> bool:
+            info["connections"] = connections
-        """Return True if open/close logic is inverted."""
+        info["configuration_url"] = "https://app.airsend.cloud/"
-        return self._invert
+        if self._local_ip:
-
+            info["hw_version"] = f"IP: {self._local_ip}"
-    @property
+        return info
-    def refresh_value(self) -> int:
+
-        """Return refresh value in seconds."""
+    @property
-        if isinstance(self._refresh, int) and self._refresh > 0:
+    def extra_state_attributes(self):
-            return self._refresh
+        if self._channel:
-        return 5 * 60
+            return {"channel": self._channel}
-
+        return None
-    async def async_bind(self) -> bool:
+
-        """Bind a channel to listen (async)."""
+    @property
-        if not (self._serviceurl and self._spurl and isinstance(self._bind, int) and self._bind > 0):
+    def is_async(self) -> bool:
-            return False
+        """Return if asynchronous state."""
-        payload = json.dumps({
+        return self._wait == False
-            "channel": {"id": self._bind},
+
-            "duration": 0,
+    @property
-            "callback": "http://127.0.0.1/"
+    def is_airsend(self) -> bool:
-        })
+        return self._rtype == 0
-        headers = {
+
-            "Authorization": "Bearer " + self._spurl,
+    @property
-            "content-type": "application/json",
+    def is_sensor(self) -> bool:
-            "User-Agent": "hass_airsend",
+        return self._rtype == 1
-        }
+
-        try:
+    @property
-            async with aiohttp.ClientSession() as session:
+    def is_button(self) -> bool:
-                async with session.post(
+        return self._rtype == 4096
-                    self._serviceurl + "airsend/bind",
+
-                    headers=headers,
+    @property
-                    data=payload,
+    def is_cover(self) -> bool:
-                    timeout=aiohttp.ClientTimeout(total=6),
+        return self._rtype in (4098, 4099)
-                ) as response:
+
-                    return response.status == 200
+    @property
-        except aiohttp.ClientError as err:
+    def is_cover_with_position(self) -> bool:
-            _LOGGER.debug("Bind error '%s': %s", self._name, err)
+        return self._rtype == 4099
-            return False
+
-
+    @property
-    async def async_transfer(self, note, entity_id=None) -> "TransferResult":
+    def is_switch(self) -> bool:
-        """Send a command (async)."""
+        return self._rtype == 4097
-        status_code = 404
+
-        ret = False
+    @property
-        wait = 'false, "callback":"http://127.0.0.1/"'
+    def is_light(self) -> bool:
-        if self._wait:
+        return self._rtype == 4100
-            wait = 'true'
+
-
+    @property
-        if self._serviceurl and self._spurl and entity_id is not None:
+    def is_inverted(self) -> bool:
-            uid = hashlib.sha256(entity_id.encode('utf-8')).hexdigest()[:12]
+        """Return True if open/close logic is inverted."""
-            jnote = json.dumps(note)
+        return self._invert
-            if (
+
-                self._note is not None
+    @property
-                and all(k in self._note for k in ("method", "type", "value"))
+    def refresh_value(self) -> int:
-                and all(k in note for k in ("method", "type", "value"))
+        """Return refresh value in seconds."""
-                and note["method"] == 1 and note["type"] == 0
+        if isinstance(self._refresh, int) and self._refresh > 0:
-                and note["value"] in ("TOGGLE", 6)
+            return self._refresh
-            ):
+        return 5 * 60
-                jnote = json.dumps(self._note)
+
-
+    async def async_bind(self) -> bool:
-            payload = (
+        """Bind a channel to listen (async)."""
-                '{"wait": ' + wait + ', "channel":'
+        if not (self._serviceurl and self._spurl and isinstance(self._bind, int) and self._bind > 0):
-                + json.dumps(self._channel)
+            return False
-                + ', "thingnotes":{"uid":"0x' + uid + '", "notes":['
+        payload = json.dumps({
-                + jnote
+            "channel": {"id": self._bind},
-                + "]}}"
+            "duration": 0,
-            )
+            "callback": "http://127.0.0.1/"
-            headers = {
+        })
-                "Authorization": "Bearer " + self._spurl,
+        headers = {
-                "content-type": "application/json",
+            "Authorization": "Bearer " + self._spurl,
-                "User-Agent": "hass_airsend",
+            "content-type": "application/json",
-            }
+            "User-Agent": "hass_airsend",
-            try:
+        }
-                async with aiohttp.ClientSession() as session:
+        try:
-                    async with session.post(
+            async with aiohttp.ClientSession() as session:
-                        self._serviceurl + "airsend/transfer",
+                async with session.post(
-                        headers=headers,
+                    self._serviceurl + "airsend/bind",
-                        data=payload,
+                    headers=headers,
-                        timeout=aiohttp.ClientTimeout(total=6),
+                    data=payload,
-                    ) as response:
+                    timeout=aiohttp.ClientTimeout(total=6),
-                        if self._wait:
+                ) as response:
-                            ret = True
+                    return response.status == 200
-                            status_code = 500
+        except aiohttp.ClientError as err:
-                            try:
+            _LOGGER.debug("Bind error '%s': %s", self._name, err)
-                                jdata = await response.json(content_type=None)
+            return False
-                                if jdata.get("type", 0x100) < 0x100:
+
-                                    status_code = response.status
+    async def async_transfer(self, note, entity_id=None) -> "TransferResult":
-                            except Exception:
+        """Send a command (async)."""
-                                pass
+        status_code = 404
-                        else:
+        ret = False
-                            ret = None
+        wait = 'false, "callback":"http://127.0.0.1/"'
-                            status_code = response.status
+        if self._wait:
-            except aiohttp.ClientError as err:
+            wait = 'true'
-                _LOGGER.debug("Transfer local error '%s': %s", self._name, err)
+
-                return TransferResult.NETWORK_ERROR
+        if self._serviceurl and self._spurl and entity_id is not None:
-
+            uid = hashlib.sha256(entity_id.encode('utf-8')).hexdigest()[:12]
-        # Fallback to cloud API if local failed
+            jnote = json.dumps(note)
-        if status_code != 200 and self._apikey:
+            if (
-            action = "command"
+                self._note is not None
-            value = 6
+                and all(k in self._note for k in ("method", "type", "value"))
-            if all(k in note for k in ("method", "type", "value")):
+                and all(k in note for k in ("method", "type", "value"))
-                if note["method"] == 1 and note["type"] == 0:
+                and note["method"] == 1 and note["type"] == 0
-                    value = {
+                and note["value"] in ("TOGGLE", 6)
-                        "OFF": 0, "ON": 1, "STOP": 3, "DOWN": 4, "UP": 5
+            ):
-                    }.get(note["value"], 6)
+                jnote = json.dumps(self._note)
-                elif note["method"] == 1 and note["type"] == 9:
+
-                    action = "level"
+            payload = (
-                    value = int(note["value"])
+                '{"wait": ' + wait + ', "channel":'
-
+                + json.dumps(self._channel)
-            cloud_url = (
+                + ', "thingnotes":{"uid":"0x' + uid + '", "notes":['
-                "https://airsend.cloud/device/"
+                + jnote
-                + str(self._uid) + "/" + action + "/" + str(value) + "/"
+                + "]}}"
-            )
+            )
-            headers = {
+            headers = {
-                "Authorization": "Bearer " + self._apikey,
+                "Authorization": "Bearer " + self._spurl,
-                "content-type": "application/json",
+                "content-type": "application/json",
-                "User-Agent": "hass_airsend",
+                "User-Agent": "hass_airsend",
-            }
+            }
-            try:
+            try:
-                async with aiohttp.ClientSession() as session:
+                async with aiohttp.ClientSession() as session:
-                    async with session.get(
+                    async with session.post(
-                        cloud_url,
+                        self._serviceurl + "airsend/transfer",
-                        headers=headers,
+                        headers=headers,
-                        timeout=aiohttp.ClientTimeout(total=10),
+                        data=payload,
-                    ) as response:
+                        timeout=aiohttp.ClientTimeout(total=6),
-                        status_code = response.status
+                    ) as response:
-                        ret = True
+                        if self._wait:
-            except aiohttp.ClientError as err:
+                            ret = True
-                _LOGGER.warning("Transfer cloud error '%s': %s — both local and cloud failed", self._name, err)
+                            status_code = 500
-                return TransferResult.NETWORK_ERROR
+                            try:
-
+                                jdata = await response.json(content_type=None)
-        if status_code == 200:
+                                if jdata.get("type", 0x100) < 0x100:
-            return TransferResult.SUCCESS
+                                    status_code = response.status
-        if status_code == 401:
+                            except Exception:
-            _LOGGER.error("Transfer '%s' : invalid locator (401) — check spurl in airsend.yaml", self.name)
+                                pass
-            return TransferResult.SERVER_ERROR
+                        else:
-        if status_code == 405:
+                            ret = None
-            _LOGGER.error("Transfer '%s' : invalid input (405) — check channel configuration", self.name)
+                            status_code = response.status
-            return TransferResult.SERVER_ERROR
+            except aiohttp.ClientError as err:
-        if status_code == 500:
+                _LOGGER.debug("Transfer local error '%s': %s", self._name, err)
-            if self._wait:
+                return TransferResult.NETWORK_ERROR
-                _LOGGER.warning("Transfer '%s' : no RF confirmation (500), command may have been sent", self.name)
+
-                return TransferResult.SENT
+        # Fallback to cloud API if local failed
-            _LOGGER.warning("Transfer '%s' : server error (500), command may have been sent", self.name)
+        if status_code != 200 and self._apikey:
-            return TransferResult.SERVER_ERROR
+            action = "command"
-        _LOGGER.warning("Transfer '%s' : unexpected status %s", self.name, status_code)
+            value = 6
-        return TransferResult.SERVER_ERROR
+            if all(k in note for k in ("method", "type", "value")):
+                if note["method"] == 1 and note["type"] == 0:
+                    value = {
+                        "OFF": 0, "ON": 1, "STOP": 3, "DOWN": 4, "UP": 5
+                    }.get(note["value"], 6)
+                elif note["method"] == 1 and note["type"] == 9:
+                    action = "level"
+                    value = int(note["value"])
+
+            cloud_url = (
+                "https://airsend.cloud/device/"
+                + str(self._uid) + "/" + action + "/" + str(value) + "/"
+            )
+            headers = {
+                "Authorization": "Bearer " + self._apikey,
+                "content-type": "application/json",
+                "User-Agent": "hass_airsend",
+            }
+            try:
+                async with aiohttp.ClientSession() as session:
+                    async with session.get(
+                        cloud_url,
+                        headers=headers,
+                        timeout=aiohttp.ClientTimeout(total=10),
+                    ) as response:
+                        status_code = response.status
+                        ret = True
+            except aiohttp.ClientError as err:
+                _LOGGER.warning("Transfer cloud error '%s': %s — both local and cloud failed", self._name, err)
+                return TransferResult.NETWORK_ERROR
+
+        if status_code == 200:
+            return TransferResult.SUCCESS
+        if status_code == 401:
+            _LOGGER.error("Transfer '%s' : invalid locator (401) — check spurl in airsend.yaml", self.name)
+            return TransferResult.SERVER_ERROR
+        if status_code == 405:
+            _LOGGER.error("Transfer '%s' : invalid input (405) — check channel configuration", self.name)
+            return TransferResult.SERVER_ERROR
+        if status_code == 500:
+            if self._wait:
+                _LOGGER.warning("Transfer '%s' : no RF confirmation (500), command may have been sent", self.name)
+                return TransferResult.SENT
+            _LOGGER.warning("Transfer '%s' : server error (500), command may have been sent", self.name)
+            return TransferResult.SERVER_ERROR
+        _LOGGER.warning("Transfer '%s' : unexpected status %s", self.name, status_code)
+        return TransferResult.SERVER_ERROR

custom_components/airsend/manifest.json

@@ -1,12 +1,12 @@
-{
+{
-  "domain": "airsend",
+  "domain": "airsend",
-  "name": "AirSend",
+  "name": "AirSend",
-  "documentation": "https://github.com/devmel/hass_airsend",
+  "documentation": "https://github.com/devmel/hass_airsend",
-  "issue_tracker": "https://github.com/devmel/hass_airsend/issues",
+  "issue_tracker": "https://github.com/devmel/hass_airsend/issues",
-  "dependencies": ["http"],
+  "dependencies": ["http"],
-  "config_flow": true,
+  "config_flow": true,
-  "codeowners": ["@devmel"],
+  "codeowners": ["@devmel"],
-  "requirements": [],
+  "requirements": [],
-  "version": "4.0.0",
+  "version": "4.1.0",
-  "iot_class": "local_push"
+  "iot_class": "local_push"
-}
+}