Merge pull request #4 from natekspencer/zwavejs-device-config-db

Add zwavejs device config db lookup

.gitignore

@@ -130,3 +130,6 @@ dmypy.json
 
 # VS Code Settings / Launch files
 .vscode/
+
+# Temp directory
+.tmp/

setup.cfg

@@ -19,4 +19,7 @@ python_requires = >=3.6
 install_requires =
     aiohttp>=3.6
     certifi>=2019.9.11
-    pubnub>=5.0.1
+    pubnub>=5.0.1
+
+[options.package_data]
+vivintpy = zjs_device_config_db.json

setup.py

@@ -1,4 +1,11 @@
 """vivintpy setup script."""
+import os
+
 import setuptools
 
+file_name = "vivintpy/zjs_device_config_db.json"
+
+with open(file_name, "w") as file:
+    pass
+
 setuptools.setup()

vivintpy/devices/__init__.py

@@ -1,14 +1,13 @@
 """This package contains the various devices attached to a Vivint system."""
 from __future__ import annotations
 
-import asyncio
-import concurrent.futures
 from typing import TYPE_CHECKING, Callable, Dict, List, Optional
 
 from ..const import VivintDeviceAttribute as Attribute
 from ..entity import Entity
 from ..enums import CapabilityCategoryType, CapabilityType
 from ..vivintskyapi import VivintSkyApi
+from ..zjs_device_config_db import get_zwave_device_info
 
 if TYPE_CHECKING:
     from .alarm_panel import AlarmPanel
@@ -130,19 +129,27 @@ def get_zwave_details(self):
         if self.data.get("zpd") is None:
             return None
 
-        pool = concurrent.futures.ThreadPoolExecutor()
-        result = pool.submit(asyncio.run, self.get_zwave_details_async()).result()
-        return result
-
-    async def get_zwave_details_async(self):
-        manufacturer_id = f"{self.data.get('manid'):04x}"
-        product_id = f"{self.data.get('prid'):04x}"
-        product_type_id = f"{self.data.get('prtid'):04x}"
-        result = await self.vivintskyapi.get_zwave_details(
-            manufacturer_id, product_id, product_type_id
+        result = get_zwave_device_info(
+            self.data.get("manid"),
+            self.data.get("prtid"),
+            self.data.get("prid"),
         )
-        [self._manufacturer, self._model] = result
-        return result
+
+        self._manufacturer = result.get("manufacturer", "Unknown")
+
+        label = result.get("label")
+        description = result.get("description")
+
+        if label and description:
+            self._model = f"{description} ({label})"
+        elif label:
+            self._model = label
+        elif description:
+            self._model = description
+        else:
+            self._model = "Unknown"
+
+        return [self._manufacturer, self._model]
 
 
 class UnknownDevice(VivintDevice):

vivintpy/vivintskyapi.py

@@ -387,30 +387,3 @@ async def __call(
             data=data,
             allow_redirects=allow_redirects,
         )
-
-    async def get_zwave_details(self, manufacturer_id, product_id, product_type_id):
-        """Gets the zwave details by looking up the details on the openzwave device database."""
-        UNKNOWN_RESULT = ["Unknown", "Unknown"]
-
-        # zwave_lookup = f"{manufacturer_id}:{product_id}:{product_type_id}"
-        # device_info = self.__zwave_device_info.get(zwave_lookup)
-        # if device_info is not None:
-        #     return device_info
-
-        # async with aiohttp.ClientSession() as session:
-        #     async with session.get(
-        #         url=f"http://openzwave.net/device-database/{zwave_lookup}"
-        #     ) as response:
-        #         if response.status == 200:
-        #             text = await response.text()
-        #             title = re.search("<title>(.*)</title>", text, re.IGNORECASE)[1]
-        #             result = self.__zwave_device_info[zwave_lookup] = (
-        #                 UNKNOWN_RESULT
-        #                 if title == "Device Database"
-        #                 else title.split(" - ")
-        #             )
-        #             return result
-        #         else:
-        #             response.raise_for_status()
-        #             return UNKNOWN_RESULT
-        return UNKNOWN_RESULT

vivintpy/zjs_device_config_db.py

@@ -0,0 +1,158 @@
+import asyncio
+import concurrent.futures
+import json
+import logging
+import os
+import re
+import shutil
+import tarfile
+import threading
+from pathlib import Path
+
+import aiohttp
+import async_timeout
+
+_LOGGER = logging.getLogger(__name__)
+
+TMP_DIR = os.path.join(os.path.dirname(__file__), "./.tmp/")
+ZJS_TAR_FILE = os.path.join(TMP_DIR, "zjs.tar.gz")
+ZJS_TAR_URL = "http://github.com/zwave-js/node-zwave-js/archive/master.tar.gz"
+ZJS_TAR_CONFIG_BASE = "node-zwave-js-master/packages/config/config/"
+ZJS_DEVICE_CONFIG_DB_FILE = os.path.join(
+    os.path.dirname(__file__), "zjs_device_config_db.json"
+)
+
+__MUTEX = threading.Lock()
+
+
+def get_zwave_device_info(
+    manufacturer_id: int, product_type: int, product_id: int
+) -> dict:
+    """Lookup the Z-Wave device based on the manufacturer id, product type, and product id"""
+    key = f"0x{manufacturer_id:04x}:0x{product_type:04x}:0x{product_id:04x}"
+    return get_zjs_device_config_db().get(key, {})
+
+
+def get_zjs_device_config_db() -> dict:
+    """Returns the Z-Wave JS device config db as a dict."""
+    if not _device_config_db_file_exists():
+        pool = concurrent.futures.ThreadPoolExecutor()
+        result = pool.submit(asyncio.run, download_zjs_device_config_db()).result()
+        return result
+
+    return _load_db_from_file()
+
+
+def _device_config_db_file_exists() -> bool:
+    """Returns True if the device config db file exists."""
+    return (
+        os.path.isfile(ZJS_DEVICE_CONFIG_DB_FILE)
+        and os.path.getsize(ZJS_DEVICE_CONFIG_DB_FILE) > 0
+    )
+
+
+def _load_db_from_file() -> dict:
+    """Loads the Z-Wave JS device config from the saved JSON file."""
+    data = {}
+    if _device_config_db_file_exists():
+        with open(ZJS_DEVICE_CONFIG_DB_FILE) as f:
+            data = json.load(f)
+    return data
+
+
+async def download_zjs_device_config_db():
+    """Downloads the Z-Wave JS device config database."""
+    with __MUTEX:
+        if not _device_config_db_file_exists():
+            _LOGGER.info("Beginning download process")
+            _clean_temp_directory(create=True)
+            await _download_zjs_tarfile()
+            _extract_zjs_config_files()
+            data = _create_db_from_zjs_config_files()
+            _clean_temp_directory()
+            return data
+        else:
+            return _load_db_from_file()
+
+
+def _clean_temp_directory(create: bool = False) -> None:
+    """Ensures the temp directory is empty and creates it if specified."""
+    if os.path.exists(TMP_DIR):
+        _LOGGER.info("Removing temp directory")
+        shutil.rmtree(TMP_DIR)
+
+    if create:
+        _LOGGER.info("Creating temp directory")
+        os.mkdir(TMP_DIR)
+
+
+async def _download_zjs_tarfile() -> None:
+    """Downloads the Z-Wave JS tarfile from http://github.com/zwave-js/node-zwave-js."""
+    _LOGGER.info("Downloading tarfile from Z-Wave JS")
+    async with aiohttp.ClientSession() as session:
+        async with async_timeout.timeout(120):
+            async with session.get(ZJS_TAR_URL) as response:
+                with open(ZJS_TAR_FILE, "wb") as file:
+                    async for data in response.content.iter_chunked(1024):
+                        file.write(data)
+
+
+def _extract_zjs_config_files() -> None:
+    """Extracts the Z-Wave JSON config files."""
+    manufacturers_path = "".join([ZJS_TAR_CONFIG_BASE, "manufacturers.json"])
+    devices_path = "".join([ZJS_TAR_CONFIG_BASE, "devices/"])
+
+    def members(tf):
+        l = len(ZJS_TAR_CONFIG_BASE)
+        for member in tf.getmembers():
+            if member.path.startswith(manufacturers_path) or (
+                member.path.startswith(devices_path)
+                and member.path.endswith(".json")
+                and "/templates/" not in member.path
+            ):
+                member.path = member.path[l:]
+                yield member
+
+    _LOGGER.info("Extracting config files from download")
+    with tarfile.open(ZJS_TAR_FILE) as tar:
+        tar.extractall(members=members(tar), path=TMP_DIR)
+
+
+def _create_db_from_zjs_config_files() -> dict:
+    """Parses the Z-Wave JSON config files and creates a consolidated device db."""
+    _LOGGER.info("Parsing extracted config files")
+    json_files = Path(os.path.join(TMP_DIR, "devices")).glob("**/*.json")
+
+    device_db = {}
+
+    for file in json_files:
+        data = {}
+        try:
+            with open(file) as json_file:
+                json_string = "".join(
+                    re.sub("(([^:]|^)//[^a-zA-Z\d:].*)|(/\*.*\*/)", "", line)
+                    for line in json_file.readlines()
+                )
+                data = json.loads(json_string)
+        except:
+            print("oops, couldn't parse file %s", file)
+
+        manufacturer_id = data.get("manufacturerId")
+        manufacturer = data.get("manufacturer")
+        label = data.get("label")
+        description = data.get("description")
+
+        for device in data.get("devices", []):
+            product_type = device.get("productType")
+            product_id = device.get("productId")
+            device_db[f"{manufacturer_id}:{product_type}:{product_id}"] = {
+                "manufacturer": manufacturer,
+                "label": label,
+                "description": description,
+            }
+
+    _LOGGER.info("Creating consolidated device db")
+    with open(ZJS_DEVICE_CONFIG_DB_FILE, "w") as device_file:
+        device_file.write(json.dumps(device_db))
+
+    return device_db