Use micloud for miotspec cloud connectivity

Depends on Squachen/micloud#11

Author: rytilahti

miio/miot_cloud.py

@@ -1,22 +1,25 @@
 """Module implementing handling of miot schema files."""
+import json
 import logging
 from datetime import datetime, timedelta
 from operator import attrgetter
 from pathlib import Path
-from typing import List
+from typing import Dict, List, Optional
 
 import appdirs
-import requests  # TODO: externalize HTTP requests to avoid direct dependency
-from pydantic import BaseModel
+from micloud import MiotSpec
+from pydantic import BaseModel, Field
 
 from miio.miot_models import DeviceModel
 
 _LOGGER = logging.getLogger(__name__)
 
 
 class ReleaseInfo(BaseModel):
+    """Information about individual miotspec release."""
+
     model: str
-    status: str
+    status: Optional[str]  # only available on full listing
     type: str
     version: int
 
@@ -26,87 +29,87 @@ def filename(self) -> str:
 
 
 class ReleaseList(BaseModel):
-    instances: List[ReleaseInfo]
+    """Model for miotspec release list."""
+
+    releases: List[ReleaseInfo] = Field(alias="instances")
 
     def info_for_model(self, model: str, *, status_filter="released") -> ReleaseInfo:
-        matches = [inst for inst in self.instances if inst.model == model]
+        releases = [inst for inst in self.releases if inst.model == model]
 
-        if len(matches) > 1:
+        if not releases:
+            raise Exception(f"No releases found for {model=} with {status_filter=}")
+        elif len(releases) > 1:
             _LOGGER.warning(
-                "more than a single match for model %s: %s, filtering with status=%s",
+                "%s versions found for model %s: %s, using the newest one",
+                len(releases),
                 model,
-                matches,
+                releases,
                 status_filter,
             )
 
-        released_versions = [inst for inst in matches if inst.status == status_filter]
-        if not released_versions:
-            raise Exception(f"No releases for {model}, adjust status_filter?")
-
-        _LOGGER.debug("Got %s releases, picking the newest one", released_versions)
+        newest_release = max(releases, key=attrgetter("version"))
+        _LOGGER.debug("Using %s", newest_release)
 
-        match = max(released_versions, key=attrgetter("version"))
-        _LOGGER.debug("Using %s", match)
-
-        return match
+        return newest_release
 
 
 class MiotCloud:
+    """Interface for miotspec data."""
+
     def __init__(self):
         self._cache_dir = Path(appdirs.user_cache_dir("python-miio"))
 
     def get_device_model(self, model: str) -> DeviceModel:
         """Get device model for model name."""
         file = self._cache_dir / f"{model}.json"
-        if file.exists():
-            _LOGGER.debug("Using cached %s", file)
-            return DeviceModel.parse_raw(file.read_text())
+        spec = self.file_from_cache(file)
+        if spec is not None:
+            return DeviceModel.parse_obj(spec)
 
-        return DeviceModel.parse_raw(self.get_model_schema(model))
+        return DeviceModel.parse_obj(self.get_model_schema(model))
 
-    def get_model_schema(self, model: str) -> str:
+    def get_model_schema(self, model: str) -> Dict:
         """Get the preferred schema for the model."""
-        instances = self.fetch_release_list()
-        release_info = instances.info_for_model(model)
+        specs = self.get_release_list()
+        release_info = specs.info_for_model(model)
 
         model_file = self._cache_dir / f"{release_info.model}.json"
-        url = f"https://miot-spec.org/miot-spec-v2/instance?type={release_info.type}"
-
-        data = self._fetch(url, model_file)
+        spec = self.file_from_cache(model_file)
+        if spec is not None:
+            return spec
 
-        return data
+        spec = json.loads(MiotSpec.get_spec_for_urn(device_urn=release_info.type))
 
-    def fetch_release_list(self):
-        """Fetch a list of available schemas."""
-        mapping_file = "model-to-urn.json"
-        url = "http://miot-spec.org/miot-spec-v2/instances?status=all"
-        data = self._fetch(url, self._cache_dir / mapping_file)
-
-        return ReleaseList.parse_raw(data)
-
-    def _fetch(self, url: str, target_file: Path, cache_hours=6):
-        """Fetch the URL and cache results, if expired."""
+        return spec
 
-        def valid_cache():
+    def file_from_cache(self, file, cache_hours=6) -> Optional[Dict]:
+        def _valid_cache():
             expiration = timedelta(hours=cache_hours)
             if (
-                datetime.fromtimestamp(target_file.stat().st_mtime) + expiration
+                datetime.fromtimestamp(file.stat().st_mtime) + expiration
                 > datetime.utcnow()
             ):
                 return True
 
             return False
 
-        if target_file.exists() and valid_cache():
-            _LOGGER.debug("Returning data from cache: %s", target_file)
-            return target_file.read_text()
+        if file.exists() and _valid_cache():
+            _LOGGER.debug("Returning data from cache file %s", file)
+            return json.loads(file.read_text())
+
+        _LOGGER.debug("Cache file %s not found or it is stale", file)
+        return None
+
+    def get_release_list(self) -> ReleaseList:
+        """Fetch a list of available releases."""
+        mapping_file = "model-to-urn.json"
 
-        _LOGGER.debug("Going to download %s to %s", url, target_file)
-        content = requests.get(url)
-        content.raise_for_status()
+        cache_file = self._cache_dir / mapping_file
+        mapping = self.file_from_cache(cache_file)
+        if mapping is not None:
+            return ReleaseList.parse_obj(mapping)
 
-        response = content.text
-        written = target_file.write_text(response)
-        _LOGGER.debug("Written %s bytes to %s", written, target_file)
+        specs = MiotSpec.get_specs()
+        cache_file.write_text(json.dumps(specs))
 
-        return response
+        return ReleaseList.parse_obj(specs)