-
Notifications
You must be signed in to change notification settings - Fork 14
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #110 from pyalarmdotcom/websocket-consolidated
Websocket consolidated
- Loading branch information
Showing
19 changed files
with
509 additions
and
305 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
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,8 +1,26 @@ | ||
| { | ||
| "python.testing.pytestArgs": [ | ||
| "tests" | ||
| ], | ||
| "python.testing.unittestEnabled": false, | ||
| "[python]": { | ||
| "editor.codeActionsOnSave": { | ||
| "source.fixAll.ruff": true, | ||
| "source.organizeImports.ruff": true | ||
| }, | ||
| "editor.defaultFormatter": "ms-python.black-formatter" | ||
| }, | ||
| "[yaml]": { | ||
| "editor.defaultFormatter": "esbenp.prettier-vscode" | ||
| }, | ||
| "editor.formatOnPaste": false, | ||
| "editor.formatOnSave": true, | ||
| "editor.formatOnType": true, | ||
| "python.analysis.autoSearchPaths": false, | ||
| "python.analysis.extraPaths": ["/workspaces/pyalarmdotcomajax"], | ||
| "python.formatting.provider": "black", | ||
| "python.linting.enabled": true, | ||
| "python.testing.pytestEnabled": true, | ||
| "python.analysis.typeCheckingMode": "off" | ||
| "terminal.integrated.defaultProfile.linux": "bash", | ||
| "terminal.integrated.profiles.linux": { | ||
| "zsh": { | ||
| "path": "/bin/bash" | ||
| } | ||
| } | ||
| } |
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
This file was deleted.
Oops, something went wrong.
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,115 @@ | ||
| """Basic example for logging in using time-based one-time password via pyalarmdotcomajax.""" | ||
|
|
||
| import asyncio | ||
| import logging | ||
| import os | ||
| import sys | ||
|
|
||
| import aiohttp | ||
|
|
||
| from pyalarmdotcomajax import AlarmController, OtpType | ||
| from pyalarmdotcomajax.errors import TwoFactor_ConfigurationRequired, TwoFactor_OtpRequired | ||
| from pyalarmdotcomajax.errors import AuthenticationFailed, DataFetchFailed | ||
|
|
||
| USERNAME = os.environ.get("ADC_USERNAME") | ||
| PASSWORD = os.environ.get("ADC_PASSWORD") | ||
| TWOFA_COOKIE = os.environ.get("ADC_2FA_TOKEN") | ||
|
|
||
| log = logging.getLogger(__name__) | ||
| logging.basicConfig(level=logging.DEBUG) | ||
|
|
||
|
|
||
| async def main() -> None: | ||
| """Request Alarm.com sensor data.""" | ||
|
|
||
| if not (USERNAME and PASSWORD): | ||
| sys.exit("Missing account credentials.") | ||
|
|
||
| async with aiohttp.ClientSession() as session: | ||
| # | ||
| # CREATE ALARM CONTROLLER | ||
| # | ||
|
|
||
| alarm = AlarmController( | ||
| username=USERNAME, | ||
| password=PASSWORD, | ||
| websession=session, | ||
| twofactorcookie=TWOFA_COOKIE, | ||
| ) | ||
|
|
||
| try: | ||
| await alarm.async_login() | ||
|
|
||
| except TwoFactor_OtpRequired: | ||
| print("Two factor authentication is enabled for this user.") | ||
| await async_handle_otp_workflow(alarm) | ||
|
|
||
| except TwoFactor_ConfigurationRequired: | ||
| sys.exit("Unable to log in. Please set up two-factor authentication for this account.") | ||
|
|
||
| except (ConnectionError, DataFetchFailed): | ||
| sys.exit("Could not connect to Alarm.com.") | ||
|
|
||
| except AuthenticationFailed: | ||
| sys.exit("Invalid credentials.") | ||
|
|
||
| # | ||
| # PULL DEVICE DATA FROM ALARM.COM | ||
| # | ||
|
|
||
| await alarm.async_update() | ||
|
|
||
| # | ||
| # OPEN WEBSOCKET CONNECTION | ||
| # | ||
|
|
||
| ws_client = alarm.get_websocket_client() | ||
| await ws_client.async_connect() | ||
|
|
||
|
|
||
| async def async_handle_otp_workflow(alarm: AlarmController) -> None: | ||
| """Handle two-factor authentication workflow.""" | ||
|
|
||
| selected_otp_method: OtpType | ||
|
|
||
| # | ||
| # Determine which OTP method to use | ||
| # | ||
|
|
||
| # Get list of enabled OTP methods. | ||
| if len(enabled_2fa_methods := await alarm.async_get_enabled_2fa_methods()) == 1: | ||
| # If only one OTP method is enabled, use it without prompting user. | ||
| selected_otp_method = enabled_2fa_methods[0] | ||
| print(f"Using {selected_otp_method.value} for One-Time Password.") | ||
| else: | ||
| # If multiple OTP methods are enabled, let the user pick. | ||
| print("\nAvailable one-time password methods:") | ||
| for otp_method in enabled_2fa_methods: | ||
| print(f"{otp_method.value}: {otp_method.name}") | ||
| if not ( | ||
| selected_otp_method := OtpType( | ||
| int(input("\nWhich OTP method would you like to use? Enter the method's number: ")) | ||
| ) | ||
| ): | ||
| sys.exit("Valid OTP method was not entered.") | ||
|
|
||
| # | ||
| # Request OTP | ||
| # | ||
|
|
||
| if selected_otp_method in (OtpType.EMAIL, OtpType.SMS): | ||
| # Ask Alarm.com to send OTP if selected method is EMAIL or SMS. | ||
| print(f"Requesting One-Time Password via {selected_otp_method.name}...") | ||
| await alarm.async_request_otp(selected_otp_method) | ||
|
|
||
| # | ||
| # Prompt user for OTP | ||
| # | ||
|
|
||
| if not (code := input("Enter One-Time Password: ")): | ||
| sys.exit("Requested OTP was not entered.") | ||
|
|
||
| await alarm.async_submit_otp(code=code, method=selected_otp_method) | ||
|
|
||
|
|
||
| asyncio.run(main()) |
Oops, something went wrong.