Source code for miio.miot_cloud

"""Module implementing handling of miot schema files."""

import json
import logging
from datetime import datetime, timedelta, timezone
from operator import attrgetter
from pathlib import Path
from typing import Dict, List, Optional

import appdirs
from micloud.miotspec import MiotSpec

    from pydantic.v1 import BaseModel, Field
except ImportError:
    from pydantic import BaseModel, Field

from miio import CloudException
from miio.miot_models import DeviceModel

_LOGGER = logging.getLogger(__name__)

[docs] class ReleaseInfo(BaseModel): """Information about individual miotspec release.""" model: str status: Optional[str] # only available on full listing type: str version: int @property def filename(self) -> str: return f"{self.model}_{self.status}_{self.version}.json"
[docs] class ReleaseList(BaseModel): """Model for miotspec release list.""" releases: List[ReleaseInfo] = Field(alias="instances")
[docs] def info_for_model(self, model: str, *, status_filter="released") -> ReleaseInfo: releases = [inst for inst in self.releases if inst.model == model] if not releases: raise CloudException( f"No releases found for {model=} with {status_filter=}" ) elif len(releases) > 1: _LOGGER.warning( "%s versions found for model %s: %s, using the newest one", len(releases), model, releases, ) newest_release = max(releases, key=attrgetter("version")) _LOGGER.debug("Using %s", newest_release) return newest_release
[docs] class MiotCloud: """Interface for miotspec data.""" MODEL_MAPPING_FILE = "model-to-urn.json" def __init__(self): self._cache_dir = Path(appdirs.user_cache_dir("python-miio"))
[docs] def get_release_list(self) -> ReleaseList: """Fetch a list of available releases.""" cache_file = self._cache_dir / MiotCloud.MODEL_MAPPING_FILE try: mapping = self._file_from_cache(cache_file) return ReleaseList.parse_obj(mapping) except FileNotFoundError: _LOGGER.debug("Did not found non-stale %s, trying to fetch", cache_file) specs = MiotSpec.get_specs() self._write_to_cache(cache_file, specs) return ReleaseList.parse_obj(specs)
[docs] def get_device_model(self, model: str) -> DeviceModel: """Get device model for model name.""" file = self._cache_dir / f"{model}.json" try: spec = self._file_from_cache(file) return DeviceModel.parse_obj(spec) except FileNotFoundError: _LOGGER.debug("Unable to find schema file %s, going to fetch" % file) return DeviceModel.parse_obj(self.get_model_schema(model))
[docs] def get_model_schema(self, model: str) -> Dict: """Get the preferred schema for the model.""" specs = self.get_release_list() release_info = specs.info_for_model(model) model_file = self._cache_dir / f"{release_info.model}.json" try: spec = self._file_from_cache(model_file) return spec except FileNotFoundError: _LOGGER.debug(f"Cached schema not found for {model}, going to fetch it") spec = MiotSpec.get_spec_for_urn(device_urn=release_info.type) self._write_to_cache(model_file, spec) return spec
def _write_to_cache(self, file: Path, data: Dict): """Write given *data* to cache file *file*.""" file.parent.mkdir(parents=True, exist_ok=True) written = file.write_text(json.dumps(data)) _LOGGER.debug("Written %s bytes to %s", written, file) def _file_from_cache(self, file, cache_hours=6) -> Dict: def _valid_cache(): expiration = timedelta(hours=cache_hours) if datetime.fromtimestamp( file.stat().st_mtime, tz=timezone.utc ) + expiration > return True return False if file.exists() and _valid_cache(): _LOGGER.debug("Cache hit, returning contents of %s", file) return json.loads(file.read_text()) raise FileNotFoundError("Cache file %s not found or it is stale" % file)