No project description provided
Project description
crownstone-python-lib-sse
Asynchronous Python library that listens to Crownstone SSE events.
Functionality
- Async: using asyncio and aiohttp, optimized for speed.
- Easy to use: simply pass your crownstone email and password to the constructor, and start the client!
- Complete: Fully integrated event bus that can be used to listen to events, and run callbacks.
- Independent: Client runs in a separate thread, your main thread will not be blocked!
Requirements
- Python 3.7 or higher
- Aiohttp 3.6.2
Standard installation
cd to the project folder and run:
$ python setup.py install
Install in a virtual environment
To install the library execute the following command:
$ python -m venv venv
Activate your venv using:
$ source venv/bin/activate
Once activated, the venv is used to executed python files, and libraries will be installed in the venv.
To install this library, cd to the project folder and run:
$ python setup.py install
Getting started
Example
from crownstone_sse.client import CrownstoneSSE
from crownstone_sse.events.switch_state_update_event import SwitchStateUpdateEvent
from crownstone_sse.events.system_event import SystemEvent
from crownstone_sse.events.presence_event import PresenceEvent
from crownstone_sse.events.ability_change_event import AbilityChangeEvent
from crownstone_sse.events.data_change_event import DataChangeEvent
from crownstone_sse.const import (
EVENT_SYSTEM_STREAM_START,
EVENT_SWITCH_STATE_UPDATE,
EVENT_PRESENCE_ENTER_LOCATION,
EVENT_ABILITY_CHANGE_DIMMING,
EVENT_DATA_CHANGE_CROWNSTONE,
OPERATION_CREATE,
OPERATION_DELETE,
OPERATION_UPDATE
)
import logging
import time
# enable logging
logging.basicConfig(format='%(levelname)s :%(message)s', level=logging.DEBUG)
def switch_update(event: SwitchStateUpdateEvent):
print("Crownstone {} switch state changed to {}".format(event.cloud_id, event.switch_state))
def notify_stream_start(event: SystemEvent):
print(event.message)
def notify_presence_changed(event: PresenceEvent):
print("User {} has entered location {}".format(event.user_id, event.location_id))
def notify_ability_changed(event: AbilityChangeEvent):
print("Ability {} changed to {}".format(event.ability_type, event.ability_enabled))
def notify_data_changed(event: DataChangeEvent):
if event.operation == OPERATION_CREATE:
print("New data is available: {}".format(event.changed_item_name))
if event.operation == OPERATION_UPDATE:
print("Name of id {} has been updated to {}".format(event.changed_item_id, event.changed_item_name))
if event.operation == OPERATION_DELETE:
print("Data {} has been deleted".format(event.changed_item_name))
# Create a sse client instance. Pass your crownstone account information.
# email and password are required for logging in again when an access token has expired.
sse_client = CrownstoneSSE('email', 'password')
# for usage with existing access token you can use this function:
# sse_client.set_access_token('myAccessToken')
# Start running the client
sse_client.start()
# Add listeners for event types of your liking, and the desired callback to be executed. see above.
sse_client.add_event_listener(EVENT_SYSTEM_STREAM_START, notify_stream_start)
sse_client.add_event_listener(EVENT_SWITCH_STATE_UPDATE, switch_update)
sse_client.add_event_listener(EVENT_PRESENCE_ENTER_LOCATION, notify_presence_changed)
sse_client.add_event_listener(EVENT_ABILITY_CHANGE_DIMMING, notify_ability_changed)
sse_client.add_event_listener(EVENT_DATA_CHANGE_CROWNSTONE, notify_data_changed)
# block for 120 seconds (let the client run for 120 second before stopping)
time.sleep(120)
# stop the client
sse_client.stop()
Creating callbacks
Callbacks are functions that will be executed everytime an event comes in of an specific event type.
Callback functions are standard functions, NOT coroutines!
The standard format for a callback is:
def callback(event: EventTypeClass):
# do something
Each event has it's own fields. It is recommended to provide the event type class hint to keep better track of which event it is, and what fields it has.
For example:
def callback(event: PresenceEvent):
print(event.user_id)
print(event.location_id)
Event types
Currently, there are 6 different event types:
- System event
- Command event
- Data change event
- Presence event
- Switch state update event
- Ability change event
System event
A system event is represented as:
- code
- message
Switch command event
A switch command event is represented as:
Sphere
- sphere_id
Crownstone
- cloud_id
- unique_id
- switch_val (as SwitchCommandValue)
Multi Switch command event
A multi switch command event is represented as:
Sphere
- sphere_id
Crownstone list
- crownstone_list
Each crownstone in the list:
- cloud_id
- unique_id
- switch_val (as SwitchCommandValue)
Data change event
A data change event is represented as:
- operation (update | delete | create)
Sphere
- sphere_id
Item
- changed_item_id
- changed_item_name
Presence event
A presence event is represented as:
Sphere
- sphere_id
Location
- location_id
User
- user_id
Switch state update event
A switch state update event is represented as:
Sphere
- sphere_id
Crownstone
- cloud_id
- unique_id
- switch_state (percentage)
Ability change event
An ability change event is represented as:
Sphere
- sphere_id
Crownstone
- cloud_id
- unique_id
Ability
- ability_type
- ability_enabled
- ability_synced_to_crownstone
Testing
To run the tests using tox install tox first by running:
$ pip install tox
To execute the tests cd to the project folder and run:
$ tox
To see which parts of the code are covered by the tests, a coverage report is generated after the tests have been successfull.
To see the coverage report run:
$ coverage report
If you like to get a better overview of the test you can generate a HTML file like so:
$ coverage html
To view your html file directly on Linux:
$ ./htmlcov/index.html
On Windows simply navigate to the htmlcov folder inside the project folder, and double-click index.html. It will be executed in your selected browser.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for crownstone_sse-1.2.5-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | e927327d143789bbc9ffc88e6e20dabd4aac87a44148f917294e4f7d219b52e0 |
|
MD5 | cae471240074302685c55e4f422c893c |
|
BLAKE2b-256 | 7e6667aba59daf896a108fa78948d5e3a92f7bb6a649ca4bd57ac5b988d714b8 |