-
Notifications
You must be signed in to change notification settings - Fork 18
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'rytilahti-feat/add_miotspec'
- Loading branch information
Showing
5 changed files
with
199 additions
and
23 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -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', | ||
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -22,6 +22,6 @@ | |
], | ||
entry_points=''' | ||
[console_scripts] | ||
micloud=micloud.cli:get_devices | ||
micloud=micloud.cli:cli | ||
''', | ||
) |