Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
_LOGGER = logging.getLogger(__name__)
BASE_API_VERSION = 5
API_BASE = "https://api.myqdevice.com/api/v{0}"
DEFAULT_APP_ID = "JVM/G9Nwih5BwKgNCjLxiFUQxQijAebyyg8QUHr7JOrP+tuPb8iHfRHKwTmDzHOu"
DEFAULT_USER_AGENT = "okhttp/3.10.0"
DEFAULT_BRAND_ID = 2
DEFAULT_REQUEST_RETRIES = 5
DEFAULT_CULTURE = "en"
MYQ_HEADERS = {
"Content-Type": "application/json",
"MyQApplicationId": DEFAULT_APP_ID,
"User-Agent": DEFAULT_USER_AGENT,
"ApiVersion": str(DEVICES_API_VERSION),
"BrandId": str(DEFAULT_BRAND_ID),
"Culture": DEFAULT_CULTURE
}
DEFAULT_STATE_UPDATE_INTERVAL = timedelta(seconds=5)
NON_COVER_DEVICE_FAMILIES = "gateway"
class API: # pylint: disable=too-many-instance-attributes
"""Define a class for interacting with the MyQ iOS App API."""
def __init__(self, websession: ClientSession = None) -> None:
"""Initialize."""
self._account_info = {}
self._last_state_update = None # type: Optional[datetime]
self._lock = asyncio.Lock()
self._security_token = None # type: Optional[str]
async def _send_state_command(self, state_command: str) -> None:
"""Instruct the API to change the state of the device."""
# If the user tries to open or close, say, a gateway, throw an exception:
if not self.state:
raise RequestError(
"Cannot change state of device type: {0}".format(self.device_type)
)
await self._api.request(
"put",
"Accounts/{0}/Devices/{1}/actions".format(
self._api.account_id, self.device_id
),
json={"action_type": state_command},
api_version=DEVICES_API_VERSION
)
async def update_device_info(self) -> dict:
"""Get up-to-date device info."""
# The MyQ API can time out if state updates are too frequent; therefore,
# if back-to-back requests occur within a threshold, respond to only the first:
call_dt = datetime.utcnow()
if not self._last_state_update:
self._last_state_update = call_dt - DEFAULT_STATE_UPDATE_INTERVAL
next_available_call_dt = self._last_state_update + DEFAULT_STATE_UPDATE_INTERVAL
if call_dt < next_available_call_dt:
_LOGGER.debug("Ignoring subsequent request within throttle window")
return
devices_resp = await self.request(
"get", "Accounts/{0}/Devices".format(self.account_id), api_version=DEVICES_API_VERSION
)
if devices_resp is None or devices_resp.get("items") is None:
_LOGGER.debug("Response did not contain any devices, no updates.")
return
for device_json in devices_resp["items"]:
serial_number = device_json.get("serial_number")
if serial_number is None:
_LOGGER.debug("No serial number for device with name {name}.".format(name=device_json.get("name")))
continue
if serial_number in self.devices:
device = self.devices[serial_number]
device.device_json = device_json
else: