Repository: zigpy/zigpy-xbee (language: python)

Issue:
[REQUEST] Energy detection scan with zigpy-xbee for XBee Zigbee based adapters
Requesting an "energy detection scan" feature in zigpy-xbee if possible matching what @puddly has recently added to zigpy-znp

Suggest implement similar channel energy scanning via some energy scanner tool but for XBee based firmware and hardware.

https://github.com/zha-ng/zigpy-znp

https://github.com/zha-ng/zigpy-znp/blob/dev/README.md

# Energy scan
Perform an energy scan to find a quiet Zigbee channel:

```console
$ python -m zigpy_znp.tools.energy_scan /dev/cu.usbmodem14101
Channel energy (mean of 1 / 5):
------------------------------------------------
 + Lower energy is better
 + Active Zigbee networks on a channel may still cause congestion
 + Using 26 in the USA may have lower TX power due to FCC regulations
 + Zigbee channels 15, 20, 25 fall between WiFi channels 1, 6, 11
 + Some Zigbee devices only join networks on channels 15, 20, and 25
------------------------------------------------
 - 11    61.57%  #############################################################
 - 12    60.78%  ############################################################
 - 13    12.16%  ############
 - 14    58.43%  ##########################################################
 - 15    57.65%  #########################################################
 - 16    29.80%  #############################
 - 17    38.82%  ######################################
 - 18    47.06%  ###############################################
 - 19    36.86%  ####################################
 - 20    10.98%  ##########
 - 21    16.47%  ################
 - 22    33.73%  #################################
 - 23    30.59%  ##############################
 - 24    20.39%  ####################
 - 25     5.88%  #####
 - 26*   20.39%  ####################
```


Discussion / hints:

Fix patch:
diff --git a/zigpy_xbee/api.py b/zigpy_xbee/api.py
index ea4bb4e..f4f6ffc 100644
--- a/zigpy_xbee/api.py
+++ b/zigpy_xbee/api.py
@@ -216,6 +216,8 @@ class ModemStatus(t.uint8_t, t.UndefinedEnum):
     "D7": t.uint8_t,  # 0 - 7 (an Enum)
     "P3": t.uint8_t,  # 0 - 5 (an Enum)
     "P4": t.uint8_t,  # 0 - 5 (an Enum)
+    # MAC diagnostics commands
+    "ED": t.Bytes,  # 16-byte value
     # I/O commands
     "IR": t.uint16_t,
     "IC": t.uint16_t,
diff --git a/zigpy_xbee/zigbee/application.py b/zigpy_xbee/zigbee/application.py
index 9463955..8cad835 100644
--- a/zigpy_xbee/zigbee/application.py
+++ b/zigpy_xbee/zigbee/application.py
@@ -2,6 +2,8 @@
 
 import asyncio
 import logging
+import math
+import statistics
 import time
 from typing import Any
 
@@ -181,9 +183,38 @@ async def energy_scan(
         self, channels: zigpy.types.Channels, duration_exp: int, count: int
     ) -> dict[int, float]:
         """Runs an energy detection scan and returns the per-channel scan results."""
+        all_results = {}
 
-        LOGGER.warning("Coordinator does not support energy scanning")
-        return {c: 0 for c in channels}
+        for _ in range(count):
+            results = await self._api._at_command("ED", duration_exp)
+            results = {
+                channel: -int(rssi) for channel, rssi in zip(range(11, 27), results)
+            }
+
+            for channel, rssi in results.items():
+                all_results.setdefault(channel, []).append(rssi)
+
+        def logistic(x: float, *, L: float = 1, x_0: float = 0, k: float = 1) -> float:
+            """Logistic function."""
+            return L / (1 + math.exp(-k * (x - x_0)))
+
+        def map_rssi_to_energy(rssi: int) -> float:
+            """Remaps RSSI (in dBm) to Energy (0-255)."""
+            RSSI_MAX = -5
+            RSSI_MIN = -92
+            return logistic(
+                x=rssi,
+                L=255,
+                x_0=RSSI_MIN + 0.45 * (RSSI_MAX - RSSI_MIN),
+                k=0.13,
+            )
+
+        energy = {
+            channel: map_rssi_to_energy(statistics.mean(all_rssi))
+            for channel, all_rssi in all_results.items()
+        }
+
+        return {channel: energy.get(channel, 0) for channel in channels}
 
     async def force_remove(self, dev):
         """Forcibly remove device from NCP."""


Test patch:
diff --git a/tests/test_application.py b/tests/test_application.py
index 63de630..6cf4d76 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -597,3 +597,32 @@ async def test_move_network_to_channel(app):
 
     assert len(app._api._queued_at.mock_calls) == 1
     app._api._queued_at.assert_any_call("SC", 1 << (26 - 11))
+
+
+async def test_energy_scan(app):
+    rssi = b"\x0A\x0F\x14\x19\x1E\x23\x28\x2D\x32\x37\x3C\x41\x46\x4B\x50\x55"
+    app._api._at_command = mock.AsyncMock(spec=XBee._at_command, return_value=rssi)
+    time_s = 3
+    count = 3
+    energy = await app.energy_scan(
+        channels=[x for x in range(11, 27)], duration_exp=time_s, count=count
+    )
+    assert app._api._at_command.mock_calls == [mock.call("ED", time_s)] * count
+    assert {k: round(v, 3) for k, v in energy.items()} == {
+        11: 254.032,
+        12: 253.153,
+        13: 251.486,
+        14: 248.352,
+        15: 242.562,
+        16: 232.193,
+        17: 214.619,
+        18: 187.443,
+        19: 150.853,
+        20: 109.797,
+        21: 72.172,
+        22: 43.571,
+        23: 24.769,
+        24: 13.56,
+        25: 7.264,
+        26: 3.844,
+    }