"""Define MyQ accounts."""

import asyncio
import logging
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Dict, Optional

from .const import (
    DEVICE_FAMILY_GARAGEDOOR,
    DEVICE_FAMILY_GATEWAY,
    DEVICE_FAMILY_LAMP,
    DEVICE_FAMILY_LOCK,
    DEVICES_ENDPOINT,
)
from .device import MyQDevice
from .errors import MyQError
from .garagedoor import MyQGaragedoor
from .lamp import MyQLamp
from .lock import MyQLock

if TYPE_CHECKING:
    from .api import API

_LOGGER = logging.getLogger(__name__)

DEFAULT_STATE_UPDATE_INTERVAL = timedelta(seconds=5)


class MyQAccount:
    """Define an account."""

    def __init__(self, api: "API", account_json: dict, devices: Optional[dict] = None) -> None:

        self._api = api
        self.account_json = account_json
        if devices is None:
            devices = {}
        self._devices = devices
        self.last_state_update = None  # type: Optional[datetime]
        self._update = asyncio.Lock()  # type: asyncio.Lock

    @property
    def api(self) -> "API":
        """Return API object"""
        return self._api

    @property
    def id(self) -> Optional[str]:
        """Return account id """
        return self.account_json.get("id")

    @property
    def name(self) -> Optional[str]:
        """Return account name"""
        return self.account_json.get("name")

    @property
    def devices(self) -> Dict[str, MyQDevice]:
        """Return all devices within account"""
        return self._devices

    @property
    def covers(self) -> Dict[str, MyQGaragedoor]:
        """Return only those devices that are covers."""
        return {
            device_id: device
            for device_id, device in self.devices.items()
            if isinstance(device, MyQGaragedoor)
        }

    @property
    def lamps(self) -> Dict[str, MyQLamp]:
        """Return only those devices that are lamps."""
        return {
            device_id: device
            for device_id, device in self.devices.items()
            if isinstance(device, MyQLamp)
        }

    @property
    def gateways(self) -> Dict[str, MyQDevice]:
        """Return only those devices that are gateways."""
        return {
            device_id: device
            for device_id, device in self.devices.items()
            if device.device_json["device_family"] == DEVICE_FAMILY_GATEWAY
        }

    @property
    def locks(self) -> Dict[str, MyQDevice]:
        """Return only those devices that are locks."""
        return {
            device_id: device
            for device_id, device in self.devices.items()
            if device.device_json["device_family"] == DEVICE_FAMILY_LOCK
        }

    @property
    def other(self) -> Dict[str, MyQDevice]:
        """Return only those devices that are others."""
        return {
            device_id: device
            for device_id, device in self.devices.items()
            if type(device) is MyQDevice
            and device.device_json["device_family"] != DEVICE_FAMILY_GATEWAY
        }

    async def _get_devices(self) -> None:

        _LOGGER.debug("Retrieving devices for account %s", self.name or self.id)

        _, devices_resp = await self._api.request(
            method="get",
            returns="json",
            url=DEVICES_ENDPOINT.format(account_id=self.id),
        )

        if devices_resp is not None and not isinstance(devices_resp, dict):
            raise MyQError(
                f"Received object devices_resp of type {type(devices_resp)} but expecting type dict"
            )

        state_update_timestmp = datetime.utcnow()
        if devices_resp is not None and devices_resp.get("items") is not None:
            for device in devices_resp.get("items"):
                serial_number = device.get("serial_number")
                if serial_number is None:
                    _LOGGER.debug(
                        "No serial number for device with name %s.", device.get("name")
                    )
                    continue

                if serial_number in self._devices:
                    _LOGGER.debug(
                        "Updating information for device with serial number %s",
                        serial_number,
                    )
                    myqdevice = self._devices[serial_number]
                    myqdevice.device_json = device
                    myqdevice.last_state_update = state_update_timestmp

                else:
                    if device.get("device_family") == DEVICE_FAMILY_GARAGEDOOR:
                        _LOGGER.debug(
                            "Adding new garage door with serial number %s",
                            serial_number,
                        )
                        new_device = MyQGaragedoor(
                            account=self,
                            device_json=device,
                            state_update=state_update_timestmp,
                        )
                    elif device.get("device_family") == DEVICE_FAMILY_LAMP:
                        _LOGGER.debug(
                            "Adding new lamp with serial number %s", serial_number
                        )
                        new_device = MyQLamp(
                            account=self,
                            device_json=device,
                            state_update=state_update_timestmp,
                        )
                    elif device.get("device_family") == DEVICE_FAMILY_LOCK:
                        _LOGGER.debug(
                            "Adding new lock with serial number %s", serial_number
                        )
                        new_device = MyQLock(
                            account=self,
                            device_json=device,
                            state_update=state_update_timestmp,
                        )
                    else:
                        if device.get("device_family") == DEVICE_FAMILY_GATEWAY:
                            _LOGGER.debug(
                                "Adding new gateway with serial number %s",
                                serial_number,
                            )
                        else:
                            _LOGGER.debug(
                                "Adding unknown device family %s with serial number %s",
                                device.get("device_family"),
                                serial_number,
                            )

                        new_device = MyQDevice(
                            account=self,
                            device_json=device,
                            state_update=state_update_timestmp,
                        )

                    if new_device:
                        self._devices[serial_number] = new_device
        else:
            _LOGGER.debug("No devices found for account %s", self.name or self.id)

    async def update(self) -> None:
        """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
        # Ensure only 1 update task can run at a time.
        async with self._update:
            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
            )

            # Ensure we're within our minimum update interval
            if call_dt < next_available_call_dt:
                _LOGGER.debug(
                    "Ignoring device update request for account %s as it is within throttle window",
                    self.name or self.id,
                )
                return

            await self._get_devices()
            self.last_state_update = datetime.utcnow()

pymyq/account.py
Python-MyQ/Python-MyQ
"""Define the MyQ API."""
import asyncio
import base64
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple, Union
from urllib.parse import parse_qs, urlsplit

from aiohttp import ClientResponse, ClientSession
from aiohttp.client_exceptions import ClientError, ClientResponseError
from bs4 import BeautifulSoup
from pkce import generate_code_verifier, get_code_challenge
from yarl import URL

from .account import MyQAccount
from .const import (
    ACCOUNTS_ENDPOINT,
    AUTO_DOMAIN,
    DOMAINS,
    GET_ACCOUNT_INTERVAL,
    OAUTH_AUTHORIZE_URI,
    OAUTH_BASE_URI,
    OAUTH_CLIENT_ID,
    OAUTH_LOGIN_ADDITIONAL_DATA,
    OAUTH_REDIRECT_URI,
    OAUTH_TOKEN_URI,
)
from .device import MyQDevice
from .errors import (
    AuthenticationError,
    InvalidCredentialsError,
    MyQError,
    RequestError,
)
from .garagedoor import MyQGaragedoor
from .lamp import MyQLamp
from .lock import MyQLock
from .request import REQUEST_METHODS, MyQRequest

_LOGGER = logging.getLogger(__name__)

DEFAULT_STATE_UPDATE_INTERVAL = timedelta(seconds=10)
DEFAULT_TOKEN_REFRESH = 10 * 60  # 10 minutes


class API:  # pylint: disable=too-many-instance-attributes
    """Define a class for interacting with the MyQ iOS App API."""

    def __init__(
        self,
        username: str,
        password: str,
        websession: ClientSession = None,
    ) -> None:
        """Initialize."""
        self.__credentials = {"username": username, "password": password}
        self._myqrequests = MyQRequest(websession or ClientSession())
        self._authentication_task = None  # type:Optional[asyncio.Task]
        self._codeverifier = None  # type: Optional[str]
        self._invalid_credentials = False  # type: bool
        self._lock = asyncio.Lock()  # type: asyncio.Lock
        self._update = asyncio.Lock()  # type: asyncio.Lock
        self._security_token = (
            None,
            None,
            None,
        )  # type: Tuple[Optional[str], Optional[datetime], Optional[datetime]]

        self.accounts = {}  # type: Dict[str, MyQAccount]
        self.last_state_update = None  # type: Optional[datetime]
        self.last_account_update = None
        self.current_domain_index = 0

    @property
    def devices(self) -> Dict[str, Union[MyQDevice, MyQGaragedoor, MyQLamp, MyQLock]]:
        """Return all devices."""
        devices = {}
        for account in self.accounts.values():
            devices.update(account.devices)
        return devices

    @property
    def covers(self) -> Dict[str, MyQGaragedoor]:
        """Return only those devices that are covers."""
        covers = {}
        for account in self.accounts.values():
            covers.update(account.covers)
        return covers

    @property
    def lamps(self) -> Dict[str, MyQLamp]:
        """Return only those devices that are lamps."""
        lamps = {}
        for account in self.accounts.values():
            lamps.update(account.lamps)
        return lamps

    @property
    def locks(self) -> Dict[str, MyQLock]:
        """Return only those devices that are locks."""
        locks = {}
        for account in self.accounts.values():
            locks.update(account.locks)
        return locks

    @property
    def gateways(self) -> Dict[str, MyQDevice]:
        """Return only those devices that are gateways."""
        gateways = {}
        for account in self.accounts.values():
            gateways.update(account.gateways)
        return gateways

    @property
    def _code_verifier(self) -> str:
        if self._codeverifier is None:
            self._codeverifier = generate_code_verifier(length=43)
        return self._codeverifier

    @property
    def username(self) -> str:
        """Username used to authenticate with on MyQ

        Returns:
            str: username
        """
        return self.__credentials["username"]

    @username.setter
    def username(self, username: str):
        """Set username to use for authentication

        Args:
            username (str): Username to authenticate with
        """
        self._invalid_credentials = False
        self.__credentials["username"] = username

    @property
    def password(self) -> Optional[str]:
        """Will return None, password retrieval is not possible

        Returns:
            None
        """
        return None

    @password.setter
    def password(self, password: str):
        """Set password used to authenticate with

        Args:
            password (str): password
        """
        self._invalid_credentials = False
        self.__credentials["password"] = password

    async def _authentication_task_completed(self) -> None:
        # If we had something for an authentication task and
        # it is done then get the result and clear it out.
        if self._authentication_task is not None:
            authentication_task = await self.authenticate(wait=False)
            if authentication_task.done():
                _LOGGER.debug(
                    "Scheduled token refresh completed, ensuring no exception."
                )
                self._authentication_task = None
                try:
                    # Get the result so any exception is raised.
                    authentication_task.result()
                except asyncio.CancelledError:
                    pass
                except (RequestError, AuthenticationError) as auth_err:
                    message = f"Scheduled token refresh failed: {str(auth_err)}"
                    _LOGGER.debug(message)

    async def _refresh_token(self) -> None:
        # Check if token has to be refreshed.
        if (
            self._security_token[1] is None
            or self._security_token[1] <= datetime.utcnow()
        ):
            # Token has to be refreshed, get authentication task if running otherwise
            # start a new one.
            self._myqrequests.clear_useragent()
            # Wait for authentication task to be completed.
            _LOGGER.debug(
                "Waiting for updated token, last refresh was %s",
                self._security_token[2],
            )
            try:
                await self.authenticate(wait=True)
            except AuthenticationError as auth_err:
                message = f"Error trying to re-authenticate to myQ service: {str(auth_err)}"
                _LOGGER.debug(message)
                raise AuthenticationError(message) from auth_err

    def change_domain(self) -> None:
        """Change to a different domain"""
        old_domain = DOMAINS[self.current_domain_index]
        self.current_domain_index +=1
        if self.current_domain_index > len(DOMAINS) - 1:
            self.current_domain_index = 0
        _LOGGER.debug("Changing from %s to %s", old_domain, DOMAINS[self.current_domain_index])


    async def request(
        self,
        method: str,
        returns: str,
        url: Union[URL, str],
        websession: ClientSession = None,
        headers: dict = None,
        params: dict = None,
        data: dict = None,
        json: dict = None,
        allow_redirects: bool = True,
        login_request: bool = False,
    ) -> Tuple[Optional[ClientResponse], Optional[Union[dict, str]]]:
        """Make a request."""
        if isinstance(url, str):
            domain_specific_url = url.replace(AUTO_DOMAIN, DOMAINS[self.current_domain_index])
        else:
            domain_specific_url = url # I believe if we are getting a URL object, that means it was given to use by a response, so we should use that and not modify it.
        # Determine the method to call based on what is to be returned.
        call_method = REQUEST_METHODS.get(returns)
        if call_method is None:
            raise RequestError(f"Invalid return object requested: {returns}")

        call_method = getattr(self._myqrequests, call_method)

        # if this is a request as part of authentication to have it go through in parallel.
        if login_request:
            try:
                return await call_method(
                    method=method,
                    url=domain_specific_url,
                    websession=websession,
                    headers=headers,
                    params=params,
                    data=data,
                    json=json,
                    allow_redirects=allow_redirects,
                )
            except ClientResponseError as err:
                message = (
                    f"Error requesting data from {domain_specific_url}: {err.status} - {err.message}"
                )
                _LOGGER.debug(message)
                raise RequestError(message) from err

            except ClientError as err:
                message = f"Error requesting data from {domain_specific_url}: {str(err)}"
                _LOGGER.debug(message)
                raise RequestError(message) from err

        # The MyQ API can time out if multiple concurrent requests are made, so
        # ensure that only one gets through at a time.
        # Exception is when this is a login request AND there is already a lock, in that case
        # we're sending the request anyways as we know there is no active request now.
        async with self._lock:

            # Check if an authentication task was running and if so, if it has completed.
            await self._authentication_task_completed()

            # Check if token has to be refreshed and start task to refresh, wait if required now.
            await self._refresh_token()

            if not headers:
                headers = {}

            headers["Authorization"] = self._security_token[0]

            _LOGGER.debug("Sending %s request to %s.", method, domain_specific_url)
            # Do the request. We will try 2 times based on response.
            for attempt in range(2):
                try:
                    return await call_method(
                        method=method,
                        url=domain_specific_url,
                        websession=websession,
                        headers=headers,
                        params=params,
                        data=data,
                        json=json,
                        allow_redirects=allow_redirects,
                    )
                except ClientResponseError as err:
                    message = f"Error requesting data from {domain_specific_url}: {err.status} - {err.message}"

                    if err.status and err.status in (401, 403):
                        if attempt == 0:
                            self._security_token = (None, None, self._security_token[2])
                            _LOGGER.debug("Status 401 received, re-authenticating.")

                            await self._refresh_token()
                        else:
                            # Received unauthorized again,
                            # reset token and start task to get a new one.
                            _LOGGER.debug(message)
                            self._security_token = (None, None, self._security_token[2])
                            await self.authenticate(wait=False)
                            raise AuthenticationError(message) from err
                    else:
                        _LOGGER.debug(message)
                        raise RequestError(message) from err

                except ClientError as err:
                    message = f"Error requesting data from {url}: {str(err)}"
                    _LOGGER.debug(message)
                    raise RequestError(message) from err
        _LOGGER.debug("Failed to update token - we will change to a new domain.")
        self.change_domain()
        return None, None


    async def _oauth_authenticate(self) -> Tuple[str, int]:

        async with ClientSession() as session:
            # retrieve authentication page
            _LOGGER.debug("Retrieving authentication page")
            resp, html = await self.request(
                method="get",
                returns="text",
                url=OAUTH_AUTHORIZE_URI,
                websession=session,
                headers={
                    "redirect": "follow",
                },
                params={
                    "client_id": OAUTH_CLIENT_ID,
                    "code_challenge": get_code_challenge(self._code_verifier),
                    "code_challenge_method": "S256",
                    "redirect_uri": OAUTH_REDIRECT_URI,
                    "response_type": "code",
                    "scope": "MyQ_Residential offline_access",
                },
                login_request=True,
            )

            # Scanning returned web page for required fields.
            _LOGGER.debug("Scanning login page for fields to return")
            soup = BeautifulSoup(html, "html.parser")

            # Go through all potential forms in the page returned.
            # This is in case multiple forms are returned.
            forms = soup.find_all("form")
            data = {}
            for form in forms:
                have_email = False
                have_password = False
                have_submit = False
                have_verification_token = False
                # Go through all the input fields.
                for field in form.find_all("input"):
                    if field.get("type"):
                        # Hidden value, include so we return back
                        if field.get("type").lower() == "hidden":
                            data.update(
                                {
                                    field.get("name", "NONAME"): field.get(
                                        "value", "NOVALUE"
                                    )
                                }
                            )
                        # Email field
                        elif field.get("type").lower() == "email":
                            data.update({field.get("name", "Email"): self.username})
                            have_email = True
                        # Password field
                        elif field.get("type").lower() == "password":
                            data.update(
                                {
                                    field.get(
                                        "name", "Password"
                                    ): self.__credentials.get("password")
                                }
                            )
                            have_password = True
                        # To confirm this form also has a submit button
                        elif field.get("type").lower() == "submit":
                            have_submit = True
                if field.get("name") == "__RequestVerificationToken":
                    data.update({"__RequestVerificationToken": field.get("value") })
                    have_verification_token = True

                # Confirm we found email, password, and submit in the form to be submitted
                if have_email and have_password and have_submit and have_verification_token:
                    _LOGGER.debug("Missing one of the valid keys email: %s, password: %s, submit: %s, verfication: %s", have_email, have_password, have_submit, have_verification_token)
                    break

                # If we're here then this is not the form to submit.
                data = {}

            # If data is empty then we did not find the valid form and are unable to continue.
            if len(data) == 0:
                _LOGGER.debug("Form with required fields not found")
                raise RequestError(
                    "Form containing fields for email, password and submit not found."
                    "Unable to continue login process."
                )

            data.update(OAUTH_LOGIN_ADDITIONAL_DATA)

            # Perform login to MyQ
            _LOGGER.debug("Performing login to MyQ")
            resp, _ = await self.request(
                method="post",
                returns="response",
                url=resp.url,
                websession=session,
                headers={
                    "Content-Type": "application/x-www-form-urlencoded",
                    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9",
                },
                data=data,
                allow_redirects=False,
                login_request=True,
            )

            # We're supposed to receive back at least 2 cookies. If not then authentication failed.
            if len(resp.cookies) < 2:
                message = "Invalid MyQ credentials provided. Please recheck login and password."
                self._invalid_credentials = True
                _LOGGER.debug(message)
                raise InvalidCredentialsError(message)

            # Intercept redirect back to MyQ iOS app
            _LOGGER.debug("Calling redirect page")
            resp, _ = await self.request(
                method="get",
                returns="response",
                url=f"{OAUTH_BASE_URI}{resp.headers['Location']}",
                websession=session,
                headers={
                    "Cookie": resp.cookies.output(attrs=[]),
                },
                allow_redirects=False,
                login_request=True,
            )

            # Retrieve token
            _LOGGER.debug("Getting token")
            redirect_url = f"{OAUTH_BASE_URI}{resp.headers['Location']}"

            get_token_basic_auth = base64.b64encode(f"{OAUTH_CLIENT_ID}:".encode("ascii")).decode("ascii")
            resp, data = await self.request(
                returns="json",
                method="post",
                url=OAUTH_TOKEN_URI,
                websession=session,
                headers={
                    "Authorization": f"Basic {get_token_basic_auth}",
                    "Content-Type": "application/x-www-form-urlencoded",
                    "Accept": "*/*",
                },
                data={
                    "code": parse_qs(urlsplit(redirect_url).query).get("code", "")[0],
                    "code_verifier": self._code_verifier,
                    "grant_type": "authorization_code",
                    "redirect_uri": OAUTH_REDIRECT_URI,
                },
                login_request=True,
            )

            if not isinstance(data, dict):
                raise MyQError(
                    f"Received object data of type {type(data)} but expecting type dict"
                )

            token = f"{data.get('token_type')} {data.get('access_token')}"
            try:
                expires = int(data.get("expires_in", DEFAULT_TOKEN_REFRESH))
            except ValueError:
                _LOGGER.debug(
                    "Expires %s received is not an integer, using default.",
                    data.get("expires_in"),
                )
                expires = DEFAULT_TOKEN_REFRESH * 2

        if expires < DEFAULT_TOKEN_REFRESH * 2:
            _LOGGER.debug(
                "Expires %s is less then default %s, setting to default instead.",
                expires,
                DEFAULT_TOKEN_REFRESH,
            )
            expires = DEFAULT_TOKEN_REFRESH * 2

        return token, expires

    async def authenticate(self, wait: bool = True) -> Optional[asyncio.Task]:
        """Authenticate and get a security token."""
        if self.username is None or self.__credentials["password"] is None:
            message = "No username/password, most likely due to previous failed authentication."
            _LOGGER.debug(message)
            raise InvalidCredentialsError(message)

        if self._invalid_credentials:
            message = "Credentials are invalid, update username/password to re-try authentication."
            _LOGGER.debug(message)
            raise InvalidCredentialsError(message)

        if self._authentication_task is None:
            # No authentication task is currently running, start one
            _LOGGER.debug(
                "Scheduling token refresh, last refresh was %s", self._security_token[2]
            )
            self._authentication_task = asyncio.create_task(
                self._authenticate(), name="MyQ_Authenticate"
            )

        if wait:
            try:
                await self._authentication_task
            except (RequestError, AuthenticationError) as auth_err:
                # Raise authentication error, we need a new token to continue
                # and not getting it right now.
                self._authentication_task = None
                raise AuthenticationError(str(auth_err)) from auth_err
            self._authentication_task = None

        return self._authentication_task

    async def _authenticate(self) -> None:
        # Retrieve and store the initial security token:
        _LOGGER.debug("Initiating OAuth authentication")
        token, expires = await self._oauth_authenticate()

        if token is None:
            _LOGGER.debug("No security token received.")
            raise AuthenticationError(
                "Authentication response did not contain a security token yet one is expected."
            )

        _LOGGER.debug("Received token that will expire in %s seconds", expires)
        self._security_token = (
            token,
            datetime.utcnow() + timedelta(seconds=int(expires)),
            datetime.now(),
        )

    async def _get_accounts(self) -> List:

        _LOGGER.debug("Retrieving account information")

        # Retrieve the accounts
        _, accounts_resp = await self.request(
            method="get", returns="json", url=ACCOUNTS_ENDPOINT
        )

        if accounts_resp is not None and not isinstance(accounts_resp, dict):
            raise MyQError(
                f"Received object accounts_resp of type {type(accounts_resp)}"
                f"but expecting type dict"
            )

        return accounts_resp.get("accounts", []) if accounts_resp is not None else []

    async def update_device_info(self) -> None:
        """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
        # Ensure only 1 update task can run at a time.
        async with self._update:
            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
            )

            # Ensure we're within our minimum update interval AND
            # update request is not for a specific device
            if call_dt < next_available_call_dt:
                _LOGGER.debug("Ignoring update request as it is within throttle window")
                return

            _LOGGER.debug("Updating account information")
            if self.accounts and self.last_account_update is not None and (datetime.utcnow() - self.last_account_update).total_seconds() < GET_ACCOUNT_INTERVAL:
                # Don't update accounts as it hasn't been long enough.
                pass
            else:
                # If update request is for a specific account then do not retrieve account information.
                accounts = await self._get_accounts()

                if len(accounts) == 0:
                    _LOGGER.debug("No accounts found")
                    self.accounts = {}
                    return
                self.last_account_update = datetime.utcnow()
                for account in accounts:
                    account_id = account.get("id")
                    if account_id is not None:
                        if self.accounts.get(account_id):
                            # Account already existed, update information.
                            _LOGGER.debug(
                                "Updating account %s with name %s",
                                account_id,
                                account.get("name"),
                            )

                            self.accounts.get(account_id).account_json = account
                        else:
                            # This is a new account.
                            _LOGGER.debug(
                                "New account %s with name %s",
                                account_id,
                                account.get("name"),
                            )
                            self.accounts.update(
                                {account_id: MyQAccount(api=self, account_json=account)}
                            )
            for account in self.accounts.values():
                # Perform a device update for this account.
                await account.update()

            self.last_state_update = datetime.utcnow()


async def login(
    username: str,
    password: str,
    websession: ClientSession = None,
    auth_only: bool = False,
) -> API:
    """Log in to the API."""

    # Set the user agent in the headers.
    api = API(username=username, password=password, websession=websession)
    _LOGGER.debug("Performing initial authentication into MyQ")
    try:
        await api.authenticate(wait=True)
    except InvalidCredentialsError as err:
        _LOGGER.error("Username and/or password are invalid. Update username/password.")
        raise err
    except AuthenticationError as err:
        _LOGGER.error("Authentication failed: %s", str(err))
        raise err

    if not auth_only:
        # Retrieve and store initial set of devices:
        _LOGGER.debug("Retrieving MyQ information")
        await api.update_device_info()

    return api

pymyq/api.py
Python-MyQ/Python-MyQ
"""The myq constants."""

OAUTH_CLIENT_ID = "ANDROID_CGI_MYQ"
OAUTH_CLIENT_SECRET = "VUQ0RFhuS3lQV3EyNUJTdw=="
OAUTH_BASE_URI = "https://partner-identity.myq-cloud.com"
OAUTH_AUTHORIZE_URI = f"{OAUTH_BASE_URI}/connect/authorize"
OAUTH_REDIRECT_URI = "com.myqops://android"
OAUTH_TOKEN_URI = f"{OAUTH_BASE_URI}/connect/token"
OAUTH_LOGIN_ADDITIONAL_DATA = {
    "brand": "myq",
    "UnifiedFlowRequested": "True"
}

ACCOUNTS_ENDPOINT = "https://accounts.myq-cloud.com/api/v6.0/accounts"
DEVICES_ENDPOINT = (
    "https://devices.myq-cloud.com/api/v6.0/Accounts/{account_id}/Devices"
)
AUTO_DOMAIN = ".myq-cloud.com"
EAST_DOMAIN = "-east.myq-cloud.com"
WEST_DOMAIN = "-west.myq-cloud.com"
DOMAINS = [AUTO_DOMAIN, EAST_DOMAIN, WEST_DOMAIN]
WAIT_TIMEOUT = 60
# When we get a error for too many request, we should pause updating for 45 minutes.
TOO_MANY_REQUEST_TIMEOUT = 45 * 60
# How frequently we want to check for account changes.
GET_ACCOUNT_INTERVAL = 60 * 60
DEVICE_TYPE = "device_type"
DEVICE_TYPE_GATE = "gate"
DEVICE_FAMILY = "device_family"
DEVICE_FAMILY_GATEWAY = "gateway"
DEVICE_FAMILY_GARAGEDOOR = "garagedoor"
DEVICE_FAMILY_LAMP = "lamp"
DEVICE_FAMILY_LOCK = "locks"
DEVICE_STATE = "state"
DEVICE_STATE_ONLINE = "online"

MANUFACTURER = "The Chamberlain Group Inc."

KNOWN_MODELS = {
    "00": "Chamberlain Ethernet Gateway",
    "01": "LiftMaster Ethernet Gateway",
    "02": "Craftsman Ethernet Gateway",
    "03": "Chamberlain Wi-Fi hub",
    "04": "LiftMaster Wi-Fi hub",
    "05": "Craftsman Wi-Fi hub",
    "08": "LiftMaster Wi-Fi GDO DC w/Battery Backup",
    "09": "Chamberlain Wi-Fi GDO DC w/Battery Backup",
    "10": "Craftsman Wi-Fi GDO DC 3/4HP",
    "11": "MyQ Replacement Logic Board Wi-Fi GDO DC 3/4HP",
    "12": "Chamberlain Wi-Fi GDO DC 1.25HP",
    "13": "LiftMaster Wi-Fi GDO DC 1.25HP",
    "14": "Craftsman Wi-Fi GDO DC 1.25HP",
    "15": "MyQ Replacement Logic Board Wi-Fi GDO DC 1.25HP",
    "0A": "Chamberlain Wi-Fi GDO or Gate Operator AC",
    "0B": "LiftMaster Wi-Fi GDO or Gate Operator AC",
    "0C": "Craftsman Wi-Fi GDO or Gate Operator AC",
    "0D": "MyQ Replacement Logic Board Wi-Fi GDO or Gate Operator AC",
    "0E": "Chamberlain Wi-Fi GDO DC 3/4HP",
    "0F": "LiftMaster Wi-Fi GDO DC 3/4HP",
    "20": "Chamberlain MyQ Home Bridge",
    "21": "LiftMaster MyQ Home Bridge",
    "23": "Chamberlain Smart Garage Hub",
    "24": "LiftMaster Smart Garage Hub",
    "27": "LiftMaster Wi-Fi Wall Mount opener",
    "28": "LiftMaster Commercial Wi-Fi Wall Mount operator",
    "80": "EU LiftMaster Ethernet Gateway",
    "81": "EU Chamberlain Ethernet Gateway",
}

pymyq/const.py
Python-MyQ/Python-MyQ
"""Define MyQ devices."""
import asyncio
import logging
from datetime import datetime
from typing import TYPE_CHECKING, List, Optional, Union

from .const import DEVICE_TYPE, WAIT_TIMEOUT
from .errors import MyQError, RequestError

if TYPE_CHECKING:
    from .account import MyQAccount

_LOGGER = logging.getLogger(__name__)


class MyQDevice:
    """Define a generic device."""

    def __init__(
        self,
        device_json: dict,
        account: "MyQAccount",
        state_update: datetime,
    ) -> None:
        """Initialize.
        :type account: str
        """
        self._account = account
        self.device_json = device_json
        self.last_state_update = state_update
        self.state_update = None
        self._device_state = None  # Type: Optional[str]
        self._send_command_lock = asyncio.Lock()  # type: asyncio.Lock
        self._wait_for_state_task = None

    @property
    def account(self) -> "MyQAccount":
        """Return account associated with device"""
        return self._account

    @property
    def device_family(self) -> Optional[str]:
        """Return the family in which this device lives."""
        return self.device_json.get("device_family")

    @property
    def device_id(self) -> Optional[str]:
        """Return the device ID (serial number)."""
        return self.device_json.get("serial_number")

    @property
    def device_platform(self) -> Optional[str]:
        """Return the device platform."""
        return self.device_json.get("device_platform")

    @property
    def device_type(self) -> Optional[str]:
        """Return the device type."""
        return self.device_json.get(DEVICE_TYPE)

    @property
    def firmware_version(self) -> Optional[str]:
        """Return the family in which this device lives."""
        return self.device_json["state"].get("firmware_version")

    @property
    def name(self) -> Optional[str]:
        """Return the device name."""
        return self.device_json.get("name")

    @property
    def online(self) -> bool:
        """Return whether the device is online."""
        state = self.device_json.get("state")
        if state is None:
            return False

        return state.get("online") is True

    @property
    def parent_device_id(self) -> Optional[str]:
        """Return the device ID (serial number) of this device's parent."""
        return self.device_json.get("parent_device_id")

    @property
    def href(self) -> Optional[str]:
        """Return the hyperlinks of the device."""
        return self.device_json.get("href")

    @property
    def state(self) -> Optional[str]:
        """Return current state

        Returns:
            Optional[str]: State for the device
        """
        return self._device_state or self.device_state

    @state.setter
    def state(self, value: str):
        """Set the current state of the device."""
        self._device_state = value

    @property
    def device_state(self) -> Optional[str]:
        return None

    @property
    def close_allowed(self) -> bool:
        """Return whether the device can be closed unattended."""
        return False

    @property
    def open_allowed(self) -> bool:
        """Return whether the device can be opened unattended."""
        return False

    async def close(self, wait_for_state: bool = False) -> Union[asyncio.Task, bool]:
        raise NotImplementedError

    async def open(self, wait_for_state: bool = False) -> Union[asyncio.Task, bool]:
        raise NotImplementedError

    async def turnoff(self, wait_for_state: bool = False) -> Union[asyncio.Task, bool]:
        raise NotImplementedError

    async def turnon(self, wait_for_state: bool = False) -> Union[asyncio.Task, bool]:
        raise NotImplementedError

    async def update_device(self, device_json: dict, state_update_timestmp: datetime):
        """Update state of device depending on last update in MyQ is after last state set

        by us

        Args:
            device_json (dict): device json
            state_update_timestmp (datetime): [description]
        """
        # When performing commands we might update the state temporary, need to ensure
        # that the state is not set back to something else if MyQ does not yet have updated
        # state
        last_update = self.device_json["state"].get("last_update")
        self.device_json = device_json

        if (
            self.device_json["state"].get("last_update") is not None
            and self.device_json["state"].get("last_update") != last_update
        ):
            # MyQ has updated device state, reset ours ensuring we have the one from MyQ.
            self._device_state = None
            _LOGGER.debug(
                "State for device %s was updated to %s", self.name, self.state
            )

        self.state_update = state_update_timestmp

    async def _send_state_command(
        self,
        to_state: str,
        intermediate_state: str,
        url: str,
        command: str,
        wait_for_state: bool = False,
    ) -> Union[asyncio.Task, bool]:
        """Send command to device to change state."""

        # If the user tries to open or close, say, a gateway, throw an exception:
        if not self.state:
            raise RequestError(
                f"Cannot change state of device type: {self.device_type}"
            )

        # If currently there is a wait_for_state task running,
        # then wait until it completes first.
        if self._wait_for_state_task is not None:
            # Return wait task if we're currently waiting for same task to be completed
            if self.state == intermediate_state and not wait_for_state:
                _LOGGER.debug(
                    "Command %s for %s was already send, returning wait task for it instead",
                    command,
                    self.name,
                )
                return self._wait_for_state_task

            _LOGGER.debug(
                "Another command for %s is still in progress, waiting for it to complete first before issuing command %s",
                self.name,
                command,
            )
            await self._wait_for_state_task

        # We return true if current state is already in new state.
        if self.state == to_state:
            _LOGGER.debug(
                "Device %s is in state %s, nothing to do.", self.name, to_state
            )
            return True

        async with self._send_command_lock:
            _LOGGER.debug("Sending command %s for %s", command, self.name)
            await self.account.api.request(
                method="put",
                returns="response",
                url=url,
            )

            self.state = intermediate_state

            self._wait_for_state_task = asyncio.create_task(
                self.wait_for_state(
                    current_state=[self.state],
                    new_state=[to_state],
                    last_state_update=self.device_json["state"].get("last_update"),
                    timeout=60,
                ),
                name="MyQ_WaitFor" + to_state,
            )

            # Make sure our wait task starts
            await asyncio.sleep(0)

        if not wait_for_state:
            return self._wait_for_state_task

        _LOGGER.debug("Waiting till device is %s", to_state)
        return await self._wait_for_state_task

    async def update(self) -> None:
        """Get the latest info for this device."""
        await self.account.update()

    async def wait_for_state(
        self,
        current_state: List,
        new_state: List,
        last_state_update: datetime,
        timeout: int = WAIT_TIMEOUT,
    ) -> bool:
        """Wait until device has reached new state

        Args:
            current_state (List): List of possible current states
            new_state (List): List of new states to wait for
            last_state_update (datetime): Last time state was updated
            timeout (int, optional): Timeout in seconds to wait for new state.
                                     Defaults to WAIT_TIMEOUT.

        Returns:
            bool: True if new state reached, False if new state was not reached
        """
        # First wait until door state is actually updated.
        _LOGGER.debug("Waiting until device state has been updated for %s", self.name)
        wait_timeout = timeout
        while (
            last_state_update
            == self.device_json["state"].get("last_update", datetime.utcnow())
            and wait_timeout > 0
        ):
            wait_timeout = wait_timeout - 5
            try:
                await self._account.update()
            except MyQError:
                # Ignoring
                pass
            await asyncio.sleep(5)

        # Wait until the state is to what we want it to be
        _LOGGER.debug("Waiting until device state for %s is %s", self.name, new_state)
        wait_timeout = timeout
        while self.device_state not in new_state and wait_timeout > 0:
            wait_timeout = wait_timeout - 5
            try:
                await self._account.update()
            except MyQError:
                # Ignoring
                pass
            await asyncio.sleep(5)

        # Reset self.state ensuring it reflects actual device state.
        # Only do this if state is still what it would have been,
        # this to ensure if something else had updated it to something else we don't override.
        if self._device_state in current_state or self._device_state in new_state:
            self._device_state = None

        self._wait_for_state_task = None
        return self.state in new_state

pymyq/device.py
Python-MyQ/Python-MyQ
"""Define exceptions."""


class MyQError(Exception):
    """Define a base exception."""

    pass


class InvalidCredentialsError(MyQError):
    """Define an exception related to invalid credentials."""

    pass


class AuthenticationError(MyQError):
    """Define an exception related to invalid credentials."""

    pass


class RequestError(MyQError):
    """Define an exception related to bad HTTP requests."""

    pass


class UserRateLimit(MyQError):
    """An Exception that is thrown when a user is currently timed out."""

pymyq/errors.py
Python-MyQ/Python-MyQ
"""Define MyQ devices."""
import asyncio
import logging
from datetime import datetime
from typing import TYPE_CHECKING, Optional, Union

from .device import MyQDevice

if TYPE_CHECKING:
    from .account import MyQAccount

_LOGGER = logging.getLogger(__name__)

COMMAND_URI = (
    "https://account-devices-gdo.myq-cloud.com/api/v6.0/Accounts/{account_id}"
    "/door_openers/{device_serial}/{command}"
)
COMMAND_CLOSE = "close"
COMMAND_OPEN = "open"
STATE_CLOSED = "closed"
STATE_CLOSING = "closing"
STATE_OPEN = "open"
STATE_OPENING = "opening"
STATE_STOPPED = "stopped"
STATE_UNKNOWN = "unknown"


class MyQGaragedoor(MyQDevice):
    """Define a generic device."""

    def __init__(
        self,
        device_json: dict,
        account: "MyQAccount",
        state_update: datetime,
    ) -> None:
        """Initialize.
        :type account: str
        """
        super().__init__(
            account=account, device_json=device_json, state_update=state_update
        )

    @property
    def close_allowed(self) -> bool:
        """Return whether the device can be closed unattended."""
        return self.device_json["state"].get("is_unattended_close_allowed") is True

    @property
    def open_allowed(self) -> bool:
        """Return whether the device can be opened unattended."""
        return self.device_json["state"].get("is_unattended_open_allowed") is True

    @property
    def low_battery(self) -> bool:
        """Return whether the device has low battery."""
        return self.device_json["state"].get("dps_low_battery_mode") is True

    @property
    def device_state(self) -> Optional[str]:
        """Return the current state of the device."""
        return (
            self.device_json["state"].get("door_state")
            if self.device_json.get("state") is not None
            else None
        )

    async def close(self, wait_for_state: bool = False) -> Union[asyncio.Task, bool]:
        """Close the device."""

        return await self._send_state_command(
            to_state=STATE_CLOSED,
            intermediate_state=STATE_CLOSING,
            url=COMMAND_URI.format(
                account_id=self.account.id,
                device_serial=self.device_id,
                command=COMMAND_CLOSE,
            ),
            command=COMMAND_CLOSE,
            wait_for_state=wait_for_state,
        )

    async def open(self, wait_for_state: bool = False) -> Union[asyncio.Task, bool]:
        """Open the device."""

        return await self._send_state_command(
            to_state=STATE_OPEN,
            intermediate_state=STATE_OPENING,
            url=COMMAND_URI.format(
                account_id=self.account.id,
                device_serial=self.device_id,
                command=COMMAND_OPEN,
            ),
            command=COMMAND_OPEN,
            wait_for_state=wait_for_state,
        )

pymyq/garagedoor.py
Python-MyQ/Python-MyQ
"""Define MyQ devices."""
import asyncio
import logging
from datetime import datetime
from typing import TYPE_CHECKING, Optional, Union

from .device import MyQDevice

if TYPE_CHECKING:
    from .account import MyQAccount

_LOGGER = logging.getLogger(__name__)

COMMAND_URI = (
    "https://account-devices-lamp.myq-cloud.com/api/v6.0/Accounts/{account_id}"
    "/lamps/{device_serial}/{command}"
)
COMMAND_ON = "on"
COMMAND_OFF = "off"
STATE_ON = "on"
STATE_OFF = "off"


class MyQLamp(MyQDevice):
    """Define a generic device."""

    def __init__(
        self,
        device_json: dict,
        account: "MyQAccount",
        state_update: datetime,
    ) -> None:
        """Initialize.
        :type account: str
        """
        super().__init__(
            account=account, device_json=device_json, state_update=state_update
        )

    @property
    def device_state(self) -> Optional[str]:
        """Return the current state of the device."""
        return (
            self.device_json["state"].get("lamp_state")
            if self.device_json.get("state") is not None
            else None
        )

    async def turnoff(self, wait_for_state: bool = False) -> Union[asyncio.Task, bool]:
        """Turn light off."""

        return await self._send_state_command(
            to_state=STATE_OFF,
            intermediate_state=STATE_OFF,
            url=COMMAND_URI.format(
                account_id=self.account.id,
                device_serial=self.device_id,
                command=COMMAND_OFF,
            ),
            command=COMMAND_OFF,
            wait_for_state=wait_for_state,
        )

    async def turnon(self, wait_for_state: bool = False) -> Union[asyncio.Task, bool]:
        """Turn light on."""

        return await self._send_state_command(
            to_state=STATE_ON,
            intermediate_state=STATE_ON,
            url=COMMAND_URI.format(
                account_id=self.account.id,
                device_serial=self.device_id,
                command=COMMAND_ON,
            ),
            command=COMMAND_ON,
            wait_for_state=wait_for_state,
        )

pymyq/lamp.py
Python-MyQ/Python-MyQ
"""Define MyQ devices."""
import logging
from datetime import datetime
from typing import TYPE_CHECKING, Optional

from .device import MyQDevice

if TYPE_CHECKING:
    from .account import MyQAccount

_LOGGER = logging.getLogger(__name__)

COMMAND_URI = (
    "https://account-devices-lock.myq-cloud.com/api/v6.0/Accounts/{account_id}"
    "/locks/{device_serial}/{command}"
)


class MyQLock(MyQDevice):
    """Define a generic device."""

    def __init__(
        self,
        device_json: dict,
        account: "MyQAccount",
        state_update: datetime,
    ) -> None:
        """Initialize.
        :type account: str
        """
        super().__init__(
            account=account, device_json=device_json, state_update=state_update
        )

    @property
    def device_state(self) -> Optional[str]:
        """Return the current state of the device."""
        return (
            self.device_json["state"].get("lock_state")
            if self.device_json.get("state") is not None
            else None
        )

pymyq/lock.py
Python-MyQ/Python-MyQ
"""Handle requests to MyQ."""
import asyncio
import logging
import string
from datetime import datetime, timedelta
from json import JSONDecodeError
from random import choices
from typing import Optional, Tuple

from aiohttp import ClientResponse, ClientSession
from aiohttp.client_exceptions import (
    ClientError,
    ClientOSError,
    ClientResponseError,
    ServerDisconnectedError,
)

from .const import TOO_MANY_REQUEST_TIMEOUT
from .errors import RequestError, UserRateLimit

_LOGGER = logging.getLogger(__name__)

REQUEST_METHODS = {
    "json": "request_json", "text": "request_text", "response": "request_response"
}
DEFAULT_REQUEST_RETRIES = 5
USER_AGENT_REFRESH = timedelta(hours=1)


class MyQRequest:  # pylint: disable=too-many-instance-attributes
    """Define a class to handle requests to MyQ"""
    # Global until I know if I can set a time frame in HA.
    _block_request_until: datetime | None = None
    def __init__(self, websession: ClientSession = None) -> None:
        self._websession = websession or ClientSession()
        self._useragent = None
        self._last_useragent_update = None
        self._request_made = 0

    def clear_useragent(self) -> None:
        """Remove the user agent."""
        self._useragent = None
        self._last_useragent_update = None

    async def _get_useragent(self) -> None:
        """Retrieve a user agent to use in headers."""

        # Only see to retrieve a user agent if currently we do not have one,
        # we do not have an datetime on when we last retrieved one,
        # or we're passed the minimum time between requests.
        if (
            self._useragent is not None
            and self._last_useragent_update is not None
            and self._last_useragent_update + USER_AGENT_REFRESH > datetime.utcnow()
        ):
            _LOGGER.debug(
                "Ignoring user agent update request as it is within throttle window"
            )
            return

        self._last_useragent_update = datetime.utcnow()

        # Retrieve user agent from GitHub if not provided for login.
        _LOGGER.debug("Retrieving user agent from GitHub.")
        url = "https://raw.githubusercontent.com/Python-MyQ/Python-MyQ/master/.USER_AGENT"

        try:
            async with self._websession.get(url) as resp:
                useragent = await resp.text()
                resp.raise_for_status()
                _LOGGER.debug("Retrieved user agent %s from GitHub.", useragent)

        except ClientError as exc:
            # Default user agent to random string with length of 5
            # if failure to retrieve it from GitHub.
            useragent = "#RANDOM:5"
            _LOGGER.warning(
                "Failed retrieving user agent from GitHub, will use randomized user agent "
                "instead: %s",
                str(exc),
            )

        useragent = useragent.strip()
        # Check if value for useragent is to create a random user agent.
        useragent_list = useragent.split(":")
        if useragent_list[0] == "#RANDOM":
            # Create a random string, check if length is provided for the random string,
            # if not then default is 5.
            try:
                randomlength = int(useragent_list[1]) if len(useragent_list) == 2 else 5
            except ValueError:
                _LOGGER.debug(
                    "Random length value %s in user agent %s is not an integer. "
                    "Setting to 5 instead.",
                    useragent_list[1],
                    useragent,
                )
                randomlength = 5

            # Create the random user agent.
            useragent = "".join(
                choices(string.ascii_letters + string.digits, k=randomlength)
            )
            _LOGGER.debug("User agent set to randomized value: %s.", useragent)

        self._useragent = useragent

    async def _send_request(
        self,
        method: str,
        url: str,
        websession: ClientSession,
        headers: dict = None,
        params: dict = None,
        data: dict = None,
        json: dict = None,
        allow_redirects: bool = False,
    ) -> Optional[ClientResponse]:
        if headers and 'Authorization' in headers.keys():
            headers.update({"User-Agent": ""})
        if self._block_request_until is not None:
            # If we are currently blocked on this domain
            # Check if this domain can be unblocked
            if datetime.utcnow() >= self._block_request_until:
                self._block_request_until = None
            else:
                _LOGGER.debug("Skipping sending request because we are currently timed out.")
                raise UserRateLimit(f"Timed out until {str(self._block_request_until)}")
        resp = None
        resp_exc = None
        last_status = ""
        last_error = ""

        for attempt in range(DEFAULT_REQUEST_RETRIES):
            self._request_made +=1
            if self._useragent is not None and self._useragent != "":
                headers.update({"User-Agent": self._useragent})

            if attempt != 0:
                wait_for = min(2 ** attempt, 5)
                _LOGGER.debug(
                    'Request failed with "%s %s" (attempt #%s/%s)"; trying again in %s seconds',
                    last_status,
                    last_error,
                    attempt,
                    DEFAULT_REQUEST_RETRIES,
                    wait_for,
                )
                await asyncio.sleep(wait_for)
            try:
                _LOGGER.debug(
                    "Sending myq api request %s and headers %s with connection pooling",
                    url,
                    headers,
                )
                resp = await websession.request(
                    method,
                    url,
                    headers=headers,
                    params=params,
                    data=data,
                    json=json,
                    skip_auto_headers={"USER-AGENT"},
                    allow_redirects=allow_redirects,
                    raise_for_status=True,
                )

                _LOGGER.debug("Response:")
                _LOGGER.debug("    Response Code: %s", resp.status)
                _LOGGER.debug("    Headers: %s", resp.raw_headers)
                _LOGGER.debug("    Body: %s", await resp.text())
                return resp
            except ClientResponseError as err:
                _LOGGER.debug(
                    "Attempt %s request failed with exception : %s - %s",
                    attempt + 1,
                    err.status,
                    err.message,
                )
                if err.status == 401:
                    raise err
                if err.status == 429:
                    timeout = int(err.headers.get("Retry-After", TOO_MANY_REQUEST_TIMEOUT)) if err.headers is not None else TOO_MANY_REQUEST_TIMEOUT
                    _LOGGER.debug("Too many request have been made - putting a temporary pause on sending any requests for %d minutes", timeout/60)
                    self._block_request_until = datetime.utcnow() + timedelta(seconds=timeout)
                    raise UserRateLimit(f"Got 429 error - stopping request  until {str(self._block_request_until)}. there were {self._request_made} request") from err
                last_status = err.status
                last_error = err.message
                resp_exc = err

                if 400 <= err.status < 500 and attempt == 0:
                    _LOGGER.debug(
                        "Received error status %d, bad request. Will refresh user agent.", err.status
                    )
                    await self._get_useragent()

            except (ClientOSError, ServerDisconnectedError) as err:
                errno = getattr(err, "errno", -1)
                if errno in (-1, 54, 104) and attempt == 0:
                    _LOGGER.debug(
                        "Received error status %s, connection reset. Will refresh user agent.",
                        errno,
                    )
                    await self._get_useragent()
                else:
                    _LOGGER.debug(
                        "Attempt %s request failed with exception: %s",
                        attempt,
                        str(err),
                    )
                last_status = ""
                last_error = str(err)
                resp_exc = err

            except ClientError as err:
                _LOGGER.debug(
                    "Attempt %s request failed with exception: %s",
                    attempt,
                    str(err),
                )
                last_status = ""
                last_error = str(err)
                resp_exc = err

        if resp_exc is not None:
            raise resp_exc

        return resp

    async def request_json(
        self,
        method: str,
        url: str,
        websession: ClientSession = None,
        headers: dict = None,
        params: dict = None,
        data: dict = None,
        json: dict = None,
        allow_redirects: bool = False,
    ) -> Tuple[Optional[ClientResponse], Optional[dict]]:
        """Send request and retrieve json response

        Args:
            method (str): [description]
            url (str): [description]
            websession (ClientSession, optional): [description]. Defaults to None.
            headers (dict, optional): [description]. Defaults to None.
            params (dict, optional): [description]. Defaults to None.
            data (dict, optional): [description]. Defaults to None.
            json (dict, optional): [description]. Defaults to None.
            allow_redirects (bool, optional): [description]. Defaults to False.

        Raises:
            RequestError: [description]

        Returns:
            Tuple[Optional[ClientResponse], Optional[dict]]: [description]
        """

        websession = websession or self._websession
        json_data = None

        resp = await self._send_request(
            method=method,
            url=url,
            headers=headers,
            params=params,
            data=data,
            json=json,
            allow_redirects=allow_redirects,
            websession=websession,
        )

        if resp is not None:
            try:
                json_data = await resp.json(content_type=None)
            except JSONDecodeError as err:
                message = (
                    f"JSON Decoder error {err.msg} in response at line {err.lineno}"
                    f" column {err.colno}. Response received was:\n{err.doc}"
                )
                _LOGGER.error(message)
                raise RequestError(message) from err

        return resp, json_data

    async def request_text(
        self,
        method: str,
        url: str,
        websession: ClientSession = None,
        headers: dict = None,
        params: dict = None,
        data: dict = None,
        json: dict = None,
        allow_redirects: bool = False,
    ) -> Tuple[Optional[ClientResponse], Optional[str]]:
        """Send request and retrieve text

        Args:
            method (str): [description]
            url (str): [description]
            websession (ClientSession, optional): [description]. Defaults to None.
            headers (dict, optional): [description]. Defaults to None.
            params (dict, optional): [description]. Defaults to None.
            data (dict, optional): [description]. Defaults to None.
            json (dict, optional): [description]. Defaults to None.
            allow_redirects (bool, optional): [description]. Defaults to False.

        Returns:
            Tuple[Optional[ClientResponse], Optional[str]]: [description]
        """

        websession = websession or self._websession
        data_text = None
        resp = await self._send_request(
            method=method,
            url=url,
            headers=headers,
            params=params,
            data=data,
            json=json,
            allow_redirects=allow_redirects,
            websession=websession,
        )

        if resp is not None:
            data_text = await resp.text()

        return resp, data_text

    async def request_response(
        self,
        method: str,
        url: str,
        websession: ClientSession = None,
        headers: dict = None,
        params: dict = None,
        data: dict = None,
        json: dict = None,
        allow_redirects: bool = False,
    ) -> Tuple[Optional[ClientResponse], None]:
        """Send request and just receive the ClientResponse object

        Args:
            method (str): [description]
            url (str): [description]
            websession (ClientSession, optional): [description]. Defaults to None.
            headers (dict, optional): [description]. Defaults to None.
            params (dict, optional): [description]. Defaults to None.
            data (dict, optional): [description]. Defaults to None.
            json (dict, optional): [description]. Defaults to None.
            allow_redirects (bool, optional): [description]. Defaults to False.

        Returns:
            Tuple[Optional[ClientResponse], None]: [description]
        """

        websession = websession or self._websession

        return (
            await self._send_request(
                method=method,
                url=url,
                headers=headers,
                params=params,
                data=data,
                json=json,
                allow_redirects=allow_redirects,
                websession=websession,
            ),
            None,
        )

pymyq/request.py
Python-MyQ/Python-MyQ