Merge pull request #92 from mdeweerd/dev

Add SONOFF FW download
mdeweerd · Oct 2, 2022 · 42e676b · 42e676b
2 parents cd92ae0 + 09b02f8
commit 42e676b
Show file tree

Hide file tree

Showing 3 changed files with 90 additions and 10 deletions.
diff --git a/STATS.md b/STATS.md
@@ -1,6 +1,7 @@
 # Badges showing number of downloads per version
 
 - ![badge latest](https://img.shields.io/github/downloads/mdeweerd/zha-toolkit/latest/total.svg)
+- ![badge v0.8.16](https://img.shields.io/github/downloads/mdeweerd/zha-toolkit/v0.8.16/total.svg)
 - ![badge v0.8.15](https://img.shields.io/github/downloads/mdeweerd/zha-toolkit/v0.8.15/total.svg)
 - ![badge v0.8.14](https://img.shields.io/github/downloads/mdeweerd/zha-toolkit/v0.8.14/total.svg)
 - ![badge v0.8.13](https://img.shields.io/github/downloads/mdeweerd/zha-toolkit/v0.8.13/total.svg)

diff --git a/custom_components/zha_toolkit/ota.py b/custom_components/zha_toolkit/ota.py
@@ -18,6 +18,8 @@
     "https://raw.githubusercontent.com/Koenkk/zigbee-OTA/master/index.json"
 )
 
+SONOFF_LIST_URL = "https://zigbee-ota.sonoff.tech/releases/upgrade.json"
+
 
 @retryable(
     (DeliveryError, asyncio.CancelledError, asyncio.TimeoutError), tries=3
@@ -56,7 +58,7 @@ async def download_koenkk_ota(listener, ota_dir):
     ]:
         manfs[info["manufacturer_code"]] = True
 
-    LOGGER.debug("Get Koenkk FW list")
+    LOGGER.debug(f"Get Koenkk FW list and check for manfs {manfs.keys()!r}")
     new_fw_info = {}
     async with aiohttp.ClientSession() as req:
         async with req.get(KOENKK_LIST_URL) as rsp:
@@ -65,23 +67,98 @@ async def download_koenkk_ota(listener, ota_dir):
                 if fw_info["url"]:
                     filename = fw_info["url"].split("/")[-1]
                     # Try to get fw corresponding to device manufacturers
-                    if (
-                        fw_info["manufacturerCode"] in manfs
-                        and filename not in ota_files_on_disk
-                    ):
-                        LOGGER.debug("OTA file to download: '%s'", filename)
+                    fw_manf = fw_info["manufacturerCode"]
+
+                    if fw_manf in manfs and filename not in ota_files_on_disk:
+                        LOGGER.debug(
+                            "OTA file to download for manf %u (0x%04X): '%s'",
+                            fw_manf,
+                            fw_manf,
+                            filename,
+                        )
                         new_fw_info[filename] = fw_info
 
     for filename, fw_info in new_fw_info.items():
         async with aiohttp.ClientSession() as req:
             url = fw_info["url"]
             try:
-                LOGGER.debug("Get '%s'", url)
+                out_filename = os.path.join(ota_dir, filename)
+
+                LOGGER.info("Download '%s' to '%s'", url, out_filename)
                 async with req.get(url) as rsp:
                     data = await rsp.read()
 
+                with open(out_filename, "wb") as ota_file:
+                    LOGGER.debug("Try to write '%s'", out_filename)
+                    ota_file.write(data)
+            except Exception as e:
+                LOGGER.warning("Exception getting '%s': %s", url, e)
+
+
+async def download_sonoff_ota(listener, ota_dir):
+    # Get all FW files that were already downloaded.
+    # The files usually have the FW version in their name, making them unique.
+    ota_glob_expr = [
+        "*.ZIGBEE",
+        "*.OTA",
+        "*.sbl-ota",
+        "*.bin",
+        "*.ota",
+        "*.zigbee",
+    ]
+
+    # Dictionary to do more efficient lookups
+    LOGGER.debug("List OTA files available on file system")
+    ota_files_on_disk = {}
+    for glob_expr in ota_glob_expr:
+        for path in [
+            os.path.basename(x) for x in glob(os.path.join(ota_dir, glob_expr))
+        ]:
+            ota_files_on_disk[path] = True
+
+    # LOGGER.debug(f"OTA files on disk {ota_files_on_disk!r}")
+
+    # Get manufacturers
+    manfs = {}
+    for info in [
+        device.zha_device_info for device in listener.devices.values()
+    ]:
+        manfs[info["manufacturer_code"]] = True
+
+    LOGGER.debug(f"Get SONOFF FW list and check for manfs {manfs.keys()!r}")
+    new_fw_info = {}
+    async with aiohttp.ClientSession() as req:
+        async with req.get(SONOFF_LIST_URL) as rsp:
+            data = json.loads(await rsp.read())
+            for fw_info in data:
+                if fw_info["fw_binary_url"]:
+                    filename = fw_info["fw_binary_url"].split("/")[-1]
+                    # Try to get fw corresponding to device manufacturers
+                    fw_manf = fw_info["fw_manufacturer_id"]
+                    fw_model_id = fw_info["model_id"]
+
+                    # Note: could check against model id in the future
+                    if fw_manf in manfs and filename not in ota_files_on_disk:
+                        LOGGER.debug(
+                            "OTA file to download for manf %u (0x%04X)"
+                            " Model:'%s': '%s'",
+                            fw_manf,
+                            fw_manf,
+                            fw_model_id,
+                            filename,
+                        )
+                        new_fw_info[filename] = fw_info
+
+    for filename, fw_info in new_fw_info.items():
+        async with aiohttp.ClientSession() as req:
+            url = fw_info["fw_binary_url"]
+            try:
                 out_filename = os.path.join(ota_dir, filename)
 
+                LOGGER.info("Download '%s' to '%s'", url, out_filename)
+                async with req.get(url) as rsp:
+                    data = await rsp.read()
+
                 with open(out_filename, "wb") as ota_file:
                     LOGGER.debug("Try to write '%s'", out_filename)
                     ota_file.write(data)
@@ -114,6 +191,7 @@ async def ota_notify(
         )
 
         await download_koenkk_ota(listener, ota_dir)
+        await download_sonoff_ota(listener, ota_dir)
 
     # Update internal image database
     await ota_update_images(
@@ -153,7 +231,7 @@ async def ota_notify(
             0,  # cmd_id
             *cmd_args,
             # expect_reply = True,
-            tries=params[p.TRIES]
+            tries=params[p.TRIES],
         )
 
     LOGGER.debug("Sent image notify command to 0x%04x: %s", device.nwk, ret)

diff --git a/examples/README.md b/examples/README.md
@@ -29,10 +29,11 @@
 
 - `script_TRV_setTemperatureReporting.yaml`:\
   Script to configure a TRV to
-  report 5 minutes or when temperature changed by 0.2°C.
+  report every 5 minutes or when temperature changed by 0.2°C.
 - `script_Thermometer_setReporting.yaml`:\
   Script to configure a
-  Thermometer to report 5 minutes or when temperature changed by 0.2°C.
+  Thermometer to report every 5 minutes or when temperature changed by
+  0.2°C.
 
 ## Download firmware from different sources.