Asynchronous library to communicate with the Automower Connect API
from aioautomower import GetAccessTokenClientCredentials, GetMowerData, Return
from aiohttp import ClientError
from aiohttp.client_exceptions import ClientConnectorError
import asyncio
api_key = "12345678-abcd-1234-a1a1-efghijklmnop" ## Your API-Key
client_secret = "client_secret" ## Your client_secret
class ExampleToken:
"""Returns the access token as dict."""
def __init__(self, api_key, client_secret):
self.client_id = api_key
self.client_secret = client_secret
async def token(self):
try:
get_token = GetAccessTokenClientCredentials(
self.client_id,
self.client_secret,
)
access_token_raw = await get_token.async_get_access_token()
except (ClientConnectorError, ClientError):
raise KeyError
return access_token_raw
class MowerData:
"""Returns the data of all mowers as dict."""
def __init__(self, api_key, access_token, provider, token_type):
self.api_key = api_key
self.access_token = access_token
self.provider = provider
self.token_type = token_type
async def mowers(self):
try:
get_mower_data = GetMowerData(
self.api_key,
self.access_token,
self.provider,
self.token_type,
)
mower_data = await get_mower_data.async_mower_state()
except (ClientConnectorError, ClientError):
return "Make sure, you are connected to the Authentication API and the Automower API"
return mower_data
class SendingCommand:
"""Returns the data of all mowers as dict."""
def __init__(self, api_key, access_token, provider, token_type, mower_id, payload, command_type):
self.api_key = api_key
self.access_token = access_token
self.provider = provider
self.token_type = token_type
self.mower_id = mower_id
self.payload = payload
self.command_type = command_type
async def mowers(self):
try:
send = Return(
self.api_key,
self.access_token,
self.provider,
self.token_type,
self.mower_id,
self.payload
self.command_type
)
send = await send.async_mower_command()
except Exception:
return "Something went wrong"
return send
example = ExampleToken(api_key, client_id)
token_output = asyncio.run(example.token())
print(token_output)
access_token = token_output["access_token"]
provider = token_output["provider"]
token_type = token_output["token_type"]
example2 = MowerData(api_key, access_token, provider, token_type)
mower_output = asyncio.run(example2.mowers())
print(mower_output)
mower_id = mower_output["data"][0]["id"] ## '0' is your first mower
print ("Mower ID:", mower_id)
command_type = "actions"
payload = '{"data": {"type": "ResumeSchedule"}}' ## For more commands see: https://developer.husqvarnagroup.cloud/apis/Automower+Connect+API#/swagger
example3 = SendingCommand(api_key, access_token, provider, token_type, mower_id, payload, command_type)
result = asyncio.run(example3.mowers())
print (result) ## if, 202, then okay
An AutomowerSession keeps track of the access token, refreshing it whenever needed and monitors a websocket for updates, whose data is sent to callbacks provided by the user.
import asyncio
import logging
import aioautomower
CLIENT_SECRET = "mystringpassword"
API_KEY = "12312312-0126-6222-2662-3e6c49f0012c"
async def main():
sess = aioautomower.AutomowerSession(API_KEY, token=None)
# Add a callback, can be done at any point in time and
# multiple callbacks can be added.
sess.register_cb(lambda data:print(data))
# If no token was passed to the constructor, we need to call login()
# before connect(). The token can be stored somewhere and passed to
# the constructor later on.
token = await sess.logincc(CLIENT_SECRET)
if not await sess.connect():
# If the token is still None or too old, the connect will fail.
print("Connect failed")
return
await asyncio.sleep(5)
status = await sess.get_status()
print(status)
await asyncio.sleep(30)
# The close() will stop the websocket and the token refresh tasks
await sess.close()
asyncio.run(main())
An AutomowerSession that provides you with the data in a CLI
automower --client_secret 12312312-12ec-486b-a7a7-9d9b06644a14 --api-key 12312312-0126-6222-2662-3e6c49f0012c