Add functions to obtain miotspec information

micloud/cli.py

@@ -1,17 +1,68 @@
 import click
 import json
+import logging
+
 from micloud.micloud import MiCloud
+from .miotspec import MiotSpec, MIOT_STANDARD_TYPES
+
+
+@click.group()
+@click.option("-d", "--debug", is_flag=True)
+def cli(debug):
+    """Tool for fetching xiaomi cloud information."""
+    level = logging.INFO
+    if debug:
+        level = logging.DEBUG
+
+    logging.basicConfig(level=level)
 
-@click.command()
+@cli.command()
 @click.option('--username', '-u', prompt=True, help='Your Xiaomi username.')
 @click.option('--password', '-p', prompt=True, hide_input=True, confirmation_prompt=False)
 @click.option('--country', '-c', default='de', help='Language code of the server to query. Default: "de"')
 @click.option('--pretty', is_flag=True, help='Pretty print json output.')
 def get_devices(username, password, country, pretty):
+    """Get device information, including tokens."""
     mc = MiCloud(username, password)
     mc.login()
     devices = mc.get_devices(country=country)
     if pretty:
         click.echo(json.dumps(devices, indent=2, sort_keys=True))
     else:
         click.echo(json.dumps(devices))
+
+
+@cli.group()
+def miot():
+    """Commands for miotspec fetching."""
+
+
+@miot.command(name="specs")
+@click.option("--status", type=str, default="released")
+def miot_specs(status):
+    """Return all specs filtered by the given status."""
+    click.echo(json.dumps(MiotSpec.get_specs(status=status)))
+
+
+@miot.command(name="get-spec")
+@click.argument("urn")
+def miot_get_spec(urn):
+    """Return a device spec for the given URN."""
+    click.echo(json.dumps(MiotSpec.get_spec_for_urn(urn)))
+
+
+@miot.command("types")
+@click.argument("type", type=click.Choice(MIOT_STANDARD_TYPES))
+def miot_available_standard_types(type: str):
+    """Return available standard URNs for type. """
+    click.echo(json.dumps(MiotSpec.get_standard_types(type)))
+
+
+@miot.command("get-type-spec")
+@click.argument("urn", required=False)
+def miot_get_standard_type_spec(urn: str):
+    """Return a type spec for given type URN."""
+    click.echo(json.dumps(MiotSpec.get_standard_type_spec(urn)))
+
+if __name__ == "__main__":
+    cli()

micloud/micloud.py

@@ -6,7 +6,6 @@
 # email [email protected]
 # -----------------------------------------------------------
 
-import http.client, http.cookies
 import json
 import hashlib
 import logging
@@ -15,12 +14,14 @@
 import requests
 
 from micloud import miutils
-from micloud.micloudexception import MiCloudAccessDenied, MiCloudException
+from .miutils import get_session
+from .micloudexception import MiCloudAccessDenied, MiCloudException
 
 
-class MiCloud():
 
-    def __init__(self, username, password):
+class MiCloud:
+
+    def __init__(self, username=None, password=None):
         super().__init__()
         self.user_id =       None
         self.service_token = None
@@ -31,8 +32,6 @@ def __init__(self, username, password):
 
         self.failed_logins = 0 
 
-        self.agent_id = miutils.get_random_agent_id()
-        self.useragent = "Android-7.1.1-1.0.0-ONEPLUS A3010-136-" + self.agent_id + " APP/xiaomi.smarthome APPV/62830"
         self.locale = locale.getdefaultlocale()[0]
 
         timezone = datetime.datetime.now(tzlocal.get_localzone()).strftime('%z')
@@ -42,10 +41,6 @@ def __init__(self, username, password):
         self.default_server = 'de' # Sets default server to Europe.
         self.username = username
         self.password = password
-        if not self._check_credentials():
-            raise MiCloudException("username or password can't be empty")
-
-        self.client_id = miutils.get_random_string(6)
 
 
     def get_token(self):
@@ -63,6 +58,7 @@ def login(self):
         :return: True if login successful, False otherwise.
         """
         if not self._check_credentials():
+            logging.error("You need to define username and password to log in")
             return False
 
         if self.user_id and self.service_token:
@@ -81,15 +77,15 @@ def login(self):
             self.service_token = None
             if self.failed_logins > 10:
                 logging.info("Repeated errors logging on to Xiaomi cloud. Cleaning stored cookies")
-                self.self._init_session(reset=True)
+                self._init_session(reset=True)
             return False
         except MiCloudAccessDenied as e:
             logging.info("Access denied when logging on to Xiaomi cloud (%s): %s", self.failed_logins, str(e))
             self.failed_logins += 1
             self.service_token = None
             if self.failed_logins > 10:
                 logging.info("Repeated errors logging on to Xiaomi cloud. Cleaning stored cookies")
-                self.self._init_session(reset=True)
+                self._init_session(reset=True)
             raise e
         except:
             logging.exception("Unknown exception occurred!")
@@ -125,12 +121,9 @@ def _login_request(self):
 
     def _init_session(self, reset=False):
         if not self.session or reset:
-            self.session = requests.Session()
-            self.session.headers.update({'User-Agent': self.useragent})
-            self.session.cookies.update({
-                'sdkVersion': '3.8.6',
-                'deviceId': self.client_id
-            })
+            if self.session is not None:
+                self.session.close()
+            self.session = get_session()
 
 
     def _login_step1(self):
@@ -288,9 +281,8 @@ def request(self, url, params):
 
         logging.debug("Send request: %s to %s", params['data'], url)
 
-        self.session = requests.Session()
+        self._init_session(reset=True)
         self.session.headers.update({
-            'User-Agent': self.useragent,
             'Accept-Encoding': 'identity',
             'x-xiaomi-protocal-flag-cli': 'PROTOCAL-HTTP2',
             'content-type': 'application/x-www-form-urlencoded',

micloud/miotspec.py

@@ -0,0 +1,116 @@
+import logging
+
+from .micloudexception import MiCloudException
+from .miutils import get_session
+
+# A dictionary consisting of miot standard types and their singular form
+MIOT_STANDARD_TYPES = {
+    "devices": "device",
+    "services": "service",
+    "properties": "property",
+    "actions": "action",
+    "events": "event",
+}
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class MiotSpec:
+    """Simple wrapper for accessing miot-spec.org web service.
+
+    You can pass an existing requests.Session object using *session* to each class method,
+    otherwise a new session is created using :meth:`micloud.miutils.get_session`.
+
+    :meth:`get_specs` returns a list of all available specs with their version and urn information.
+    The latter can be used with :meth:`get_spec_for_urn` to obtain a full miotspec schema.
+    """
+
+    BASE_URL = "https://miot-spec.org/miot-spec-v2"
+
+    @classmethod
+    def get_specs(cls, status="released", session=None):
+        """Return information about all available miotspec implementations.
+
+        Note that every model may appear multiple times using different version in the response.
+
+        Use :meth:`miot_get_schema_for_urn` to download the miotspec schema file based on the urn.
+
+        :param status: filter by status, "released", "debug", "preview", "all", defaults to "released"
+        """
+        if session is None:
+            session = get_session()
+
+        url = f"{cls.BASE_URL}/instances?status={status}"
+
+        AVAILABLE_STATUSES = ["released", "debug", "preview", "all"]
+        if status not in AVAILABLE_STATUSES:
+            raise MiCloudException("Unknown release status %s" % status)
+
+        _LOGGER.debug("Going to download specs listing with status %s" % status)
+
+        response = session.get(url)
+        return response.json()
+
+    @classmethod
+    def get_spec_for_urn(cls, device_urn: str, session=None):
+        """Return miotspec device schema for the given device URN.
+
+        The returned dict contains information about all services, properties, actions, events etc.
+        the device has been reported to support.
+
+        :meth:`miot_get_available_schemas` can be used to return a list of all available URNs.
+        """
+        if session is None:
+            session = get_session()
+
+        _LOGGER.debug("Going to download a spec for %s" % device_urn)
+
+        url = f"{cls.BASE_URL}/instance?type={device_urn}"
+
+        response = session.get(url)
+        response.raise_for_status()
+        return response.json()
+
+    @classmethod
+    def get_standard_types(cls, type_: str, session=None):
+        """Return standardized URNs for a given type.
+
+        The type can be either devices, services, properties, actions, or events.
+        """
+        if type_ not in MIOT_STANDARD_TYPES:
+            raise MiCloudException("Invalid schema type requested: %s" % type_)
+
+        if session is None:
+            session = get_session()
+
+        url = f"{cls.BASE_URL}/spec/{type_}"
+
+        _LOGGER.debug("Going to download definition for type %s" % type_)
+
+        response = session.get(url)
+        response.raise_for_status()
+        return response.json()
+
+    @classmethod
+    def get_standard_type_spec(cls, type_urn: str, session=None):
+        """Return a schema for a standard type URN.
+
+        The response depends on the requested type and contains metadata about
+        the elements the given standard type must and can implement.
+        """
+        splitted_urn = type_urn.split(":")
+        spec_type = splitted_urn[2]
+        namespace = splitted_urn[1]
+        if namespace != "miot-spec-v2":
+            raise MiCloudException(
+                "Tried to fetch spec for non-standard namespace %s" % namespace
+            )
+
+        if session is None:
+            session = get_session()
+
+        url = f"{cls.BASE_URL}/spec/{spec_type}?type={type_urn}"
+
+        response = session.get(url)
+        response.raise_for_status()
+        return response.json()

micloud/miutils.py

@@ -10,10 +10,11 @@
 import random
 import hashlib, hmac, base64
 import string
+import requests
 from urllib.parse import urlparse
 from Crypto.Cipher import ARC4
 
-from micloud.micloudexception import MiCloudException
+from .micloudexception import MiCloudException
 
 
 def get_random_agent_id():
@@ -113,3 +114,19 @@ def decrypt_rc4(password, payload):
     r = ARC4.new(base64.b64decode(password))
     r.encrypt(bytes(1024))
     return r.encrypt(base64.b64decode(payload))
+
+
+def get_session():
+    """Create client session with pre-defined common headers."""
+    agent_id = get_random_agent_id()
+    client_id = get_random_string(6)
+    useragent = "Android-7.1.1-1.0.0-ONEPLUS A3010-136-" + agent_id + " APP/xiaomi.smarthome APPV/62830"
+
+    session = requests.Session()
+    session.headers.update({'User-Agent': useragent})
+    session.cookies.update({
+        'sdkVersion': '3.8.6',
+        'deviceId': client_id
+    })
+
+    return session

setup.py

@@ -22,6 +22,6 @@
     ],
     entry_points='''
         [console_scripts]
-        micloud=micloud.cli:get_devices
+        micloud=micloud.cli:cli
     ''',
 )