gronod/ha_ecovacs_bumper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

split api module out, split out api constants, refer to api constants in mqtt module, pulling logger back into individual modules

Browse Source
This commit is contained in:
bittles
2023-01-15 17:07:51 -05:00
parent 2b66ffa426
commit c7051bd51b
7 changed files with 244 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/ecovacs/sucks_api.py
+210
View File
@@ -0,0 +1,210 @@
import hashlib
import time
import requests
#import os
from base64 import b64decode, b64encode
from collections import OrderedDict
#from sleekxmppfs.xmlstream import ET
#from sleekxmppfs.exceptions import XMPPError
#from .sucks_mqtt import EcoVacsIOTMQ
#from .sucks_xmpp import EcoVacsXMPP
from .sucks_api_const import *
import logging
LOGGER = logging.getLogger(__name__)
class EcoVacsAPI:
CLIENT_KEY = API_CLIENT_KEY
SECRET = API_SECRET
PUBLIC_KEY = API_PUBLIC_KEY
MAIN_URL_FORMAT = API_MAIN_URL_FORMAT
USER_URL_FORMAT = API_USER_URL_FORMAT
PORTAL_URL_FORMAT = API_PORTAL_URL_FORMAT
USERSAPI = API_USERSAPI
IOTDEVMANAGERAPI = API_IOTDEVMANAGERAPI # IOT Device Manager - This provides control of "IOT" products via RestAPI, some bots use this instead of XMPP
PRODUCTAPI = API_PRODUCTAPI # Leaving this open, the only endpoint known currently is "Product IOT Map" - pim/product/getProductIotMap - This provides a list of "IOT" products. Not sure what this provides the app.
REALM = API_REALM
def __init__(self, device_id, account_id, password_hash, country, continent, verify_ssl=True):
self.meta = {
'country': country,
'lang': 'en',
'deviceId': device_id,
'appCode': 'i_eco_e',
#'appCode': 'i_eco_a' - iphone
'appVersion': '1.3.5',
#'appVersion': '1.4.6' - iphone
'channel': 'c_googleplay',
#'channel': 'c_iphone', - iphone
'deviceType': '1'
#'deviceType': '2' - iphone
}
self.verify_ssl = str_to_bool_or_cert(verify_ssl)
LOGGER.debug("Setting up EcoVacsAPI")
self.resource = device_id[0:8]
self.country = country
self.continent = continent
login_info = self.__call_main_api('user/login',
('account', self.encrypt(account_id)),
('password', self.encrypt(password_hash)))
self.uid = login_info['uid']
self.login_access_token = login_info['accessToken']
self.auth_code = self.__call_main_api('user/getAuthCode',
('uid', self.uid),
('accessToken', self.login_access_token))['authCode']
login_response = self.__call_login_by_it_token()
self.user_access_token = login_response['token']
if login_response['userId'] != self.uid:
LOGGER.debug("Switching to shorter UID " + login_response['userId'])
self.uid = login_response['userId']
LOGGER.debug("EcoVacsAPI connection complete")
def __sign(self, params):
result = params.copy()
result['authTimespan'] = int(time.time() * 1000)
result['authTimeZone'] = 'GMT-8'
sign_on = self.meta.copy()
sign_on.update(result)
sign_on_text = EcoVacsAPI.CLIENT_KEY + ''.join(
[k + '=' + str(sign_on[k]) for k in sorted(sign_on.keys())]) + EcoVacsAPI.SECRET
result['authAppkey'] = EcoVacsAPI.CLIENT_KEY
result['authSign'] = self.md5(sign_on_text)
return result
def __call_main_api(self, function, *args):
LOGGER.debug("calling main api {} with {}".format(function, args))
params = OrderedDict(args)
params['requestId'] = self.md5(time.time())
url = (EcoVacsAPI.MAIN_URL_FORMAT + "/" + function).format(**self.meta)
api_response = requests.get(url, self.__sign(params), verify=self.verify_ssl)
json = api_response.json()
LOGGER.debug("got {}".format(json))
if json['code'] == '0000':
return json['data']
elif json['code'] == '1005':
LOGGER.error("incorrect email or password")
raise ValueError("incorrect email or password")
else:
LOGGER.error("call to {} failed with {}".format(function, json))
raise RuntimeError("failure code {} ({}) for call {} and parameters {}".format(
json['code'], json['msg'], function, args))
def __call_user_api(self, function, args):
LOGGER.debug("calling user api {} with {}".format(function, args))
params = {'todo': function}
params.update(args)
response = requests.post(EcoVacsAPI.USER_URL_FORMAT.format(continent=self.continent), json=params, verify=self.verify_ssl)
json = response.json()
LOGGER.debug("got {}".format(json))
if json['result'] == 'ok':
return json
else:
LOGGER.error("call to {} failed with {}".format(function, json))
raise RuntimeError(
"failure {} ({}) for call {} and parameters {}".format(json['error'], json['errno'], function, params))
def __call_portal_api(self, api, function, args, verify_ssl=True, **kwargs):
if api == self.USERSAPI:
params = {'todo': function}
params.update(args)
else:
params = {}
params.update(args)
LOGGER.debug("calling portal api {} function {} with {}".format(api, function, params))
continent = self.continent
if 'continent' in kwargs:
continent = kwargs.get('continent')
url = (EcoVacsAPI.PORTAL_URL_FORMAT + "/" + api).format(continent=continent, **self.meta)
response = requests.post(url, json=params, verify=verify_ssl)
json = response.json()
LOGGER.debug("got {}".format(json))
if api == self.USERSAPI:
if json['result'] == 'ok':
return json
elif json['result'] == 'fail':
if json['error'] == 'set token error.': # If it is a set token error try again
if not 'set_token' in kwargs:
LOGGER.debug("loginByItToken set token error, trying again (2/3)")
return self.__call_portal_api(self.USERSAPI, function, args, verify_ssl=verify_ssl, set_token=1)
elif kwargs.get('set_token') == 1:
LOGGER.debug("loginByItToken set token error, trying again with ww (3/3)")
return self.__call_portal_api(self.USERSAPI, function, args, verify_ssl=verify_ssl, set_token=2, continent="ww")
else:
LOGGER.debug("loginByItToken set token error, failed after 3 attempts")
if api.startswith(self.PRODUCTAPI):
if json['code'] == 0:
return json
else:
LOGGER.error("call to {} failed with {}".format(function, json))
raise RuntimeError(
"failure {} ({}) for call {} and parameters {}".format(json['error'], json['errno'], function, params))
def __call_login_by_it_token(self):
return self.__call_portal_api(self.USERSAPI,'loginByItToken',
{'country': self.meta['country'].upper(),
'resource': self.resource,
'realm': EcoVacsAPI.REALM,
'userId': self.uid,
'token': self.auth_code}
, verify_ssl=self.verify_ssl)
def getdevices(self):
return self.__call_portal_api(self.USERSAPI,'GetDeviceList', {
'userid': self.uid,
'auth': {
'with': 'users',
'userid': self.uid,
'realm': EcoVacsAPI.REALM,
'token': self.user_access_token,
'resource': self.resource
}
}, verify_ssl=self.verify_ssl)['devices']
def getiotProducts(self):
return self.__call_portal_api(self.PRODUCTAPI + '/getProductIotMap','', {
'channel': '',
'auth': {
'with': 'users',
'userid': self.uid,
'realm': EcoVacsAPI.REALM,
'token': self.user_access_token,
'resource': self.resource
}
}, verify_ssl=self.verify_ssl)['data']
def SetIOTDevices(self, devices, iotproducts):
#Originally added for D900, and not actively used in code now - Not sure what the app checks the items in this list for
for device in devices: #Check if the device is part of iotProducts
device['iot_product'] = False
for iotProduct in iotproducts:
if device['class'] in iotProduct['classid']:
device['iot_product'] = True
return devices
def SetIOTMQDevices(self, devices):
#Added for devices that utilize MQTT instead of XMPP for communication
for device in devices:
device['iotmq'] = False
if device['company'] == 'eco-ng': #Check if the device is part of the list
device['iotmq'] = True
return devices
def devices(self):
return self.SetIOTMQDevices(self.getdevices())
@staticmethod
def md5(text):
return hashlib.md5(bytes(str(text), 'utf8')).hexdigest()
@staticmethod
def encrypt(text):
from Crypto.PublicKey import RSA
from Crypto.Cipher import PKCS1_v1_5
key = RSA.import_key(b64decode(EcoVacsAPI.PUBLIC_KEY))
cipher = PKCS1_v1_5.new(key)
result = cipher.encrypt(bytes(text, 'utf8'))
return str(b64encode(result), 'utf8')