Installation

pip install aioambient

Python Versions

API and Application Keys

Usage

REST API

importasynciofromdatetimeimportdatefromaiohttpimportClientSessionfromaioambientimportAPIasyncdefmain()->None:"""Create the aiohttp session and run the example."""api=API("<YOUR APPLICATION KEY>","<YOUR API KEY>")# Get all devices in an account:awaitapi.get_devices()# Get all stored readings from a device:awaitapi.get_device_details("<DEVICE MAC ADDRESS>")# Get all stored readings from a device (starting at a datetime):awaitapi.get_device_details("<DEVICE MAC ADDRESS>",end_date=date(2019,1,16))asyncio.run(main())

importasynciofromdatetimeimportdatefromaiohttpimportClientSessionfromaioambientimportAPIasyncdefmain()->None:"""Create the aiohttp session and run the example."""asyncwithClientSession()assession:api=API("<YOUR APPLICATION KEY>","<YOUR API KEY>")# Get all devices in an account:awaitapi.get_devices()# Get all stored readings from a device:awaitapi.get_device_details("<DEVICE MAC ADDRESS>")# Get all stored readings from a device (starting at a datetime):awaitapi.get_device_details("<DEVICE MAC ADDRESS>",end_date=date(2019,1,16))asyncio.run(main())

Websocket API

importasynciofromaiohttpimportClientSessionfromaioambientimportWebsocketasyncdefmain()->None:"""Create the aiohttp session and run the example."""websocket=Websocket("<YOUR APPLICATION KEY>","<YOUR API KEY>")# Note that you can watch multiple API keys at once:websocket=Websocket("YOUR APPLICATION KEY", ["<API KEY 1>","<API KEY 2>"])# Define a method that should be fired when the websocket client# connects:defconnect_method():"""Print a simple "hello" message."""print("Client has connected to the websocket")websocket.on_connect(connect_method)# Alternatively, define a coroutine handler:asyncdefconnect_coroutine():"""Waits for 3 seconds, then print a simple "hello" message."""awaitasyncio.sleep(3)print("Client has connected to the websocket")websocket.async_on_connect(connect_coroutine)# Define a method that should be run upon subscribing to the Ambient# Weather cloud:defsubscribed_method(data):"""Print the data received upon subscribing."""print(f"Subscription data received:{data}")websocket.on_subscribed(subscribed_method)# Alternatively, define a coroutine handler:asyncdefsubscribed_coroutine(data):"""Waits for 3 seconds, then print the incoming data."""awaitasyncio.sleep(3)print(f"Subscription data received:{data}")websocket.async_on_subscribed(subscribed_coroutine)# Define a method that should be run upon receiving data:defdata_method(data):"""Print the data received."""print(f"Data received:{data}")websocket.on_data(data_method)# Alternatively, define a coroutine handler:asyncdefdata_coroutine(data):"""Wait for 3 seconds, then print the data received."""awaitasyncio.sleep(3)print(f"Data received:{data}")websocket.async_on_data(data_coroutine)# Define a method that should be run when the websocket client# disconnects:defdisconnect_method(data):"""Print a simple "goodbye" message."""print("Client has disconnected from the websocket")websocket.on_disconnect(disconnect_method)# Alternatively, define a coroutine handler:asyncdefdisconnect_coroutine(data):"""Wait for 3 seconds, then print a simple "goodbye" message."""awaitasyncio.sleep(3)print("Client has disconnected from the websocket")websocket.async_on_disconnect(disconnect_coroutine)# Connect to the websocket:awaitwebsocket.connect()# At any point, disconnect from the websocket:awaitwebsocket.disconnect()asyncio.run(main())

Open REST API

importasynciofromdatetimeimportdatefromaiohttpimportClientSessionfromaioambientimportOpenAPIasyncdefmain()->None:"""Create the aiohttp session and run the example."""api=OpenAPI()# Get a list of all the devices that are located within a radius of# three miles from the given latitude/longitude. Each device lists its# MAC address.awaitapi.get_devices_by_location(32.5,-97.3,3.0)# Get the current data from a device:awaitapi.get_device_details("<DEVICE MAC ADDRESS>")asyncio.run(main())

Contributing