aioautomatic 0.4.0

Asyncio library for the Automatic API

Asyncio library for the Automatic API

• Free software: Apache Software License 2.0

All methods are python wrappers of the API definitions defined by Automatic.

Usage

It is recommended to manage the aiohttp ClientSession object externally and pass it to the Client constructor. (See the aiohttp documentation.) If not passed to Server, a ClientSession object will be created automatically.

Query for information from the users account.

import asyncio
import aioautomatic
import aiohttp
from datetime import datetime
from datetime import timedelta

SCOPE = ['location', 'vehicle:profile', 'user:profile', 'trip']

CLIENT_ID = '<client_id>'
SECRET_ID = '<secret>'
USER_EMAIL = '<user_email>'
USER_PASSWORD = '<user_password>'


@asyncio.coroutine
def loop():
    aiohttp_session = aiohttp.ClientSession()
    try:
        client = aioautomatic.Client(
            CLIENT_ID,
            SECRET_ID,
            aiohttp_session)
        session = yield from client.create_session_from_password(
                SCOPE, USER_EMAIL, USER_PASSWORD)

        # Fetch information about the authorized user
        user = yield from session.get_user()
        user_profile = yield from user.get_profile()
        user_metadata = yield from user.get_metadata()
        print("***USER***")
        print(user)
        print(user.email)
        print(user.first_name)
        print(user.last_name)
        print(user_profile.date_joined)
        print(user_metadata.firmware_version)
        print(user_metadata.device_type)
        print(user_metadata.phone_platform)

        # Fetch all devices associated with the user account
        devices = yield from session.get_devices()
        print("***DEVICES***")
        print(devices)

        # Fetch a list of vehicles associated with the user account
        vehicles = yield from session.get_vehicles()
        print("***VEHICLES***")
        print(vehicles)
        print(vehicles[0].make)
        print(vehicles[0].model)
        print(vehicles[0].fuel_level_percent)

        # Fetch a list of all trips in the last 10 days
        min_end_time = datetime.utcnow() - timedelta(days=10)
        trips = yield from session.get_trips(ended_at__gte=min_end_time, limit=10)
        print("***TRIPS***")
        print(trips)
        print(trips[0].start_location.lat)
        print(trips[0].start_location.lon)
        print(trips[0].start_address.name)
        print(trips[0].distance_m)
        print(trips[0].duration_s)

        # If more than 10 trips exist, get the next page of results
        if trips.next is not None:
            trips = yield from trips.get_next()
            print(trips)

    finally:
        yield from aiohttp_session.close()

asyncio.get_event_loop().run_until_complete(loop())

Create a session using an oauth handshake authorization code.

import asyncio
import aioautomatic
import aiohttp

CLIENT_ID = '<client_id>'
SECRET_ID = '<secret>'
# The user is redirected to Automatic's website, and after they authorize
# the app, they are redirected back to the Redirect URL, with the required
# code in the query parameters.
# See: https://developer.automatic.com/api-reference/#oauth-workflow
AUTH_CODE = '<code>'


@asyncio.coroutine
def loop():
    aiohttp_session = aiohttp.ClientSession()
    try:
        client = aioautomatic.Client(
            CLIENT_ID,
            SECRET_ID,
            aiohttp_session)
        session = yield from client.create_session_from_oauth_code(
            AUTH_CODE)

        # Fetch information about the authorized user
        user = yield from session.get_user()
        user_profile = yield from user.get_profile()
        user_metadata = yield from user.get_metadata()
        print("***USER***")
        print(user)
        print(user.email)
        print(user.first_name)
        print(user.last_name)

        # Save the refresh token from the session for use next time
        # a session needs to be created.
        refresh_token = session.refresh_token

        # Create a new session with the refresh token.
        session = yield from client.create_session_from_refresh_token(
            refresh_token)

    finally:
        yield from aiohttp_session.close()

asyncio.get_event_loop().run_until_complete(loop())

Open a websocket connection for realtime updates

import asyncio
import aioautomatic
import aiohttp

SCOPE = ['current_location', 'location', 'vehicle:profile', 'user:profile', 'trip']

CLIENT_ID = '<client_id>'
SECRET_ID = '<secret>'


def error_callback(name, message):
    print(message)


def event_callback(name, data):
    print(name)
    if data.location:
        print(data.location.lat)
        print(data.location.lon)


def speeding_callback(name, data):
    print("Speeding! Velocity: {:1.2f} KPH".format(data.velocity_kph))


@asyncio.coroutine
def loop():
    aiohttp_session = aiohttp.ClientSession()
    try:
        client = aioautomatic.Client(
            CLIENT_ID,
            SECRET_ID,
            aiohttp_session)

        client.on('closed', closed_callback)
        client.on('notification:speeding', speeding_callback)
        client.on_app_event(callback)
        future = yield from client.ws_connect()

        # Run until websocket is closed
        yield from future

    finally:
        yield from aiohttp_session.close()

asyncio.get_event_loop().run_until_complete(loop())

Credits

This package is built on aiohttp, which provides the foundation for async HTTP and websocket calls.

This package was created with Cookiecutter and the audreyr/cookiecutter-pypackage project template.