Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Migrate to new AWS endpoint #6

Merged
merged 1 commit into from
Feb 1, 2022
Merged

Migrate to new AWS endpoint #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
163 changes: 35 additions & 128 deletions aladdin_connect/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
import logging
from typing import Dict, Any

from aladdin_connect.session_manager import SessionManager

Expand All @@ -9,12 +8,22 @@ class AladdinConnectClient:
GET_USER_PORTALS_ENDPOINT = "/users/{user_id}/portals"
GET_PORTAL_DETAILS_ENDPOINT = "/portals/{portal_id}"

CONFIGURATION_ENDPOINT = "/configuration"

DOOR_STATUS_OPEN = 'open'
DOOR_STATUS_CLOSED = 'closed'
DOOR_STATUS_OPENING = 'opening'
DOOR_STATUS_CLOSING = 'closing'
DOOR_STATUS_UNKNOWN = 'unknown'

DOOR_COMMAND_CLOSE = "CloseDoor"
DOOR_COMMAND_OPEN = "OpenDoor"

DOOR_COMMANDS = {
'0': DOOR_COMMAND_CLOSE,
'1': DOOR_COMMAND_OPEN
}

DOOR_STATUS = {
0: DOOR_STATUS_UNKNOWN, # Unknown
1: DOOR_STATUS_OPEN, # open
Expand Down Expand Up @@ -70,85 +79,27 @@ def get_doors(self):
def _get_devices(self):
"""Get list of devices, i.e., Aladdin Door Controllers"""

# get the user id
try:
user = self._session.call_api(self.USER_DETAILS_ENDPOINT, method='get')
except ValueError as ex:
self._LOGGER.error("Aladdin Connect - Unable to retrieve user details %s", ex)
return

# get portals associated with user
try:
portals = self._session.call_api(self.GET_USER_PORTALS_ENDPOINT.format(user_id=user['id']), method='get')
except ValueError as ex:
self._LOGGER.error("Aladdin Connect - Unable to retrieve user portals %s", ex)
return

# for each portal get the list of devices
devices = []
self._device_portal.clear()
for portal in portals:
# only include portals that belong to user, i.e. not a portal that has been shared with user
if portal['UserEmail'] != self._user_email:
continue
try:
portal_details = self._session.call_api(self.GET_PORTAL_DETAILS_ENDPOINT.format(portal_id=portal["PortalID"]),
method='get')
except ValueError as ex:
self._LOGGER.error("Aladdin Connect - Unable to retrieve portal details %s", ex)
return

portal_id = portal_details["info"]["key"]

# we will need the portal id and device id to issue commands to doors connected to the device
for device_id in portal_details['devices']:
# save portal id in dict by device id, no need to expose this to users
self._device_portal[device_id] = portal_id
response = self._session.call_api(self.CONFIGURATION_ENDPOINT, method='get')
devices = []
for device in response["devices"]:
doors = []
for door in device["doors"]:
doors.append({
'device_id': device["id"],
'door_number': door["door_index"],
'name': door["name"],
'status': self.DOOR_STATUS[door["status"]],
'link_status': self.DOOR_LINK_STATUS[door["link_status"]]
})
devices.append({
'device_id': device_id,
'doors': self._get_doors_for_device(device_id)
'device_id': device["id"],
'doors': doors
})

return devices

def _get_doors_for_device(self, device_id):
payload = self._get_payload_auth_for_device(device_id)
payload['calls'] = [
self._get_read_rpc_call('dps1.link_status', 1),
self._get_read_rpc_call('dps1.name', 2),
self._get_read_rpc_call('dps1.door_status', 3),

self._get_read_rpc_call('dps2.link_status', 4),
self._get_read_rpc_call('dps2.name', 5),
self._get_read_rpc_call('dps2.door_status', 6),

self._get_read_rpc_call('dps3.link_status', 7),
self._get_read_rpc_call('dps3.name', 8),
self._get_read_rpc_call('dps3.door_status', 9)
]

try:
response = self._session.call_rpc(payload)
return devices
except ValueError as ex:
self._LOGGER.error("Aladdin Connect - Unable to get doors for device %s", ex)
return None

doors = []
for x in range(0, 3):
door_response = response[x*3:x*3+3]
if len(door_response[0]['result']) > 0:
link_status_id = door_response[0]['result'][0][1]
if self.DOOR_LINK_STATUS[link_status_id] is not self.STATUS_NOT_CONFIGURED:
name = door_response[1]['result'][0][1]
door_status_id = door_response[2]['result'][0][1]
doors.append({
'device_id': device_id,
'door_number': x + 1,
'name': name,
'status': self.DOOR_STATUS[door_status_id],
'link_status': self.DOOR_LINK_STATUS[link_status_id]
})
return doors
self._LOGGER.error("Aladdin Connect - Unable to retrieve configuration %s", ex)
return

def close_door(self, device_id, door_number):
self._set_door_status(device_id, door_number, self.REQUEST_DOOR_STATUS[self.DOOR_STATUS_CLOSED])
Expand All @@ -158,66 +109,22 @@ def open_door(self, device_id, door_number):

def _set_door_status(self, device_id, door_number, state):
"""Set door state"""
payload = self._get_payload_auth_for_device(device_id)
payload['calls'] = [
self._get_write_rpc_call('dps{}.desired_status'.format(door_number), 0, state),
self._get_write_rpc_call('dps{}.desired_status_user'.format(door_number), 1, self._user_email)
]
payload = {"command_key": self.DOOR_COMMANDS[state]}

try:
self._session.call_rpc(payload)
self._session.call_rpc(f"/devices/{device_id}/door/{door_number}/command", payload)
except ValueError as ex:
self._LOGGER.error("Aladdin Connect - Unable to set door status %s", ex)
return False

return True

def get_door_status(self, device_id, door_number):
payload = self._get_payload_auth_for_device(device_id)
payload['calls'] = [
self._get_read_rpc_call('dps{}.door_status'.format(door_number), 1),
]

try:
response = self._session.call_rpc(payload)
doors = self.get_doors()
for door in doors:
if door["device_id"] == device_id and door["door_number"] == door_number:
return door["status"]
except ValueError as ex:
self._LOGGER.error("Aladdin Connect - Unable to get doors status %s", ex)
return None

status = response[0]['result'][0][1]
return self.DOOR_STATUS[status]

def _get_payload_auth_for_device(self, device_id) -> Dict[str, Any]:
portal_id = self._device_portal[device_id]
return {
'auth': {
'cik': portal_id,
'client_id': device_id
}
}

@staticmethod
def _get_read_rpc_call(alias, index):
return {
'arguments': [
{
'alias': alias
},
{}
],
'id': index,
'procedure': 'read'
}

@staticmethod
def _get_write_rpc_call(alias, index, val):
return {
'arguments': [
{
'alias': alias
},
val
],
'id': index,
'procedure': 'write'
}
self._LOGGER.error("Aladdin Connect - Unable to get door status %s", ex)
return self.DOOR_STATUS_UNKNOWN
65 changes: 38 additions & 27 deletions aladdin_connect/session_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,80 +5,91 @@


class SessionManager:
HEADER_USER_AGENT = "Aladdin Connect Android v2.10.1'"
HEADER_CONTENT_TYPE_URLENCODED = 'application/x-www-form-urlencoded'
HEADER_USER_AGENT = "okhttp/3.12.1"
HEADER_BUNDLE_NAME = "com.geniecompany.AladdinConnect"
HEADER_BUILD_VERSION = "131"
HEADER_APP_VERSION = "2.10.1"
HEADER_APP_VERSION = "5.25"

API_BASE_URL = "https://genie.exosite.com/api/portals/v1"
RPC_URL = "https://genie.m2.exosite.com/onep:v1/rpc/process"
API_BASE_URL = "https://pxdqkls7aj.execute-api.us-east-1.amazonaws.com/Android"
RPC_URL = API_BASE_URL

LOGIN_ENDPOINT = "/users/_this/token"
LOGIN_ENDPOINT = "/oauth/token"
X_API_KEY = "fkowarQ0dX9Gj1cbB9Xkx1yXZkd6bzVn5x24sECW"

_LOGGER = logging.getLogger(__name__)

def __init__(self, email, password):
self._session = requests.Session()
self._session.headers.update({'Content-Type': 'application/json',
self._session.headers.update({'Content-Type': self.HEADER_CONTENT_TYPE_URLENCODED,
'AppVersion': self.HEADER_APP_VERSION,
'BundleName': self.HEADER_BUNDLE_NAME,
'User-Agent': self.HEADER_USER_AGENT,
'BuildVersion': self.HEADER_BUILD_VERSION})
self._login_token = base64.b64encode("{}:{}".format(email, password).encode('utf-8')).decode('utf-8')
'BuildVersion': self.HEADER_BUILD_VERSION,
'X-Api-Key': self.X_API_KEY})
self._auth_token = None
self._user_email = email
self._password = password
self._logged_in = False

def login(self):
self._auth_token = None
self._logged_in = False

self._session.headers.update({'Authorization': 'Basic {}'.format(self._login_token)})

try:
response = self.call_api(self.LOGIN_ENDPOINT, response_type='text')
if response:
password_base64 = base64.b64encode(self._password.encode('utf-8')).decode('utf-8')
response = self.call_api(self.LOGIN_ENDPOINT, method="post",
payload={"grant_type": "password",
"client_id": "1000",
"brand": "ALADDIN",
"username": self._user_email,
"password": password_base64,
"platform": "platform",
"model": "Google Pixel 6",
"app_version": "5.25",
"build_number": "2038",
"os_version": "12.0.0"})
if response and "access_token" in response:
self._logged_in = True
self._auth_token = response
self._session.headers.update({'Authorization': 'Token: {}'.format(self._auth_token)})
self._auth_token = response["access_token"]
self._session.headers.update({'Authorization': 'Bearer {}'.format(self._auth_token)})
return True
except ValueError as ex:
self._LOGGER.error("Aladdin Connect - Unable to login %s", ex)

return False

def call_api(self, api, payload=None, method='get', response_type='json'):
return self._rest_call(self.API_BASE_URL + api, payload, method, response_type)
self._session.headers.update({'Content-Type': 'application/x-www-form-urlencoded'})
return self._rest_call(self.API_BASE_URL + api, payload, method, request_type='text', response_type=response_type)

def call_rpc(self, payload=None, method='post', response_type='json'):
return self._rest_call(self.RPC_URL, payload, method, response_type)
def call_rpc(self, api, payload=None, method='post', response_type='json'):
self._session.headers.update({'Content-Type': 'application/json'})
return self._rest_call(self.RPC_URL + api, payload, method, request_type='json', response_type=response_type)

def _rest_call(self, uri, payload=None, method='get', response_type='json'):
def _rest_call(self, uri, payload=None, method='get', request_type='text', response_type='json'):
"""Generic method for calling REST APIs."""
# Sanity check parameters first...
if (method != 'get' and method != 'post' and
method != 'put' and method != 'delete'):
msg = "Tried call_api with bad method: {0}"
raise ValueError(msg.format(method))

# Payload is always JSON
if payload is not None:
payload_json = json.dumps(payload)
else:
payload_json = ''

try:
response = getattr(self._session, method)(uri, data=payload_json)
if request_type == 'json':
response = getattr(self._session, method)(uri, json=payload)
else:
response = getattr(self._session, method)(uri, data=payload)
except requests.exceptions.HTTPError as ex:
self._LOGGER.error("Aladding Connect - API Error %s", ex)
self._LOGGER.error("Aladdin Connect - API Error %s", ex)
return None

# Unauthorized
if response.status_code == 401 or response.status_code == 403:
# Maybe we got logged out? Let's try logging in if we've been logged in previously.
if self._logged_in and self.login():
# Retry the request...
response = getattr(self._session, method)(uri, data=payload_json)
response = getattr(self._session, method)(uri, data=payload)

if response.status_code != 200 and response.status_code != 204:
msg = "Aladdin API call ({0}) failed: {1}, {2}".format(
Expand Down