adafruit-mirror/Adafruit_IO_Python
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Add support for dashboards, blocks and layouts

Browse source
This commit is contained in:
Doug Zobel 2021-11-16 12:06:01 -06:00
parent d047655176
commit 11bddacc67
6 changed files with 348 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Adafruit_IO/__init__.py
View file

@ -21,5 +21,5 @@
from .client import Client
from .mqtt_client import MQTTClient
from .errors import AdafruitIOError, RequestError, ThrottlingError, MQTTError
from .model import Data, Feed, Group
from .model import Data, Feed, Group, Dashboard, Block, Layout
from ._version import __version__

78
Adafruit_IO/client.py
View file

@ -27,7 +27,7 @@ import pkg_resources
import requests
from .errors import RequestError, ThrottlingError
from .model import Data, Feed, Group
from .model import Data, Feed, Group, Dashboard, Block, Layout
# set outgoing version, pulled from setup.py
version = pkg_resources.require("Adafruit_IO")[0].version
@ -278,11 +278,13 @@ class Client(object):
:param string feed: Key of Adafruit IO feed.
:param group_key group: Group to place new feed in.
"""
f = feed._asdict()
del f['id'] # Don't pass id on create call
path = "feeds/"
if group_key is not None: # create feed in a group
path="/groups/%s/feeds"%group_key
return Feed.from_dict(self._post(path, {"feed": feed._asdict()}))
return Feed.from_dict(self._post(path, {"feed": feed._asdict()}))
return Feed.from_dict(self._post(path, {"feed": f}))
return Feed.from_dict(self._post(path, {"feed": f}))
def delete_feed(self, feed):
"""Delete the specified feed.
@ -315,3 +317,73 @@ class Client(object):
"""
path = "groups/{0}".format(group)
self._delete(path)
# Dashboard functionality.
def dashboards(self, dashboard=None):
"""Retrieve a list of all dashboards, or the specified dashboard.
:param string dashboard: Key of Adafruit IO Dashboard. Defaults to None.
"""
if dashboard is None:
path = "dashboards/"
return list(map(Dashboard.from_dict, self._get(path)))
path = "dashboards/{0}".format(dashboard)
return Dashboard.from_dict(self._get(path))
def create_dashboard(self, dashboard):
"""Create the specified dashboard.
:param Dashboard dashboard: Dashboard object to create
"""
path = "dashboards/"
return Dashboard.from_dict(self._post(path, dashboard._asdict()))
def delete_dashboard(self, dashboard):
"""Delete the specified dashboard.
:param string dashboard: Key of Adafruit IO Dashboard.
"""
path = "dashboards/{0}".format(dashboard)
self._delete(path)
# Block functionality.
def blocks(self, dashboard, block=None):
"""Retrieve a list of all blocks from a dashboard, or the specified block.
:param string dashboard: Key of Adafruit IO Dashboard.
:param string block: id of Adafruit IO Block. Defaults to None.
"""
if block is None:
path = "dashboards/{0}/blocks".format(dashboard)
return list(map(Block.from_dict, self._get(path)))
path = "dashboards/{0}/blocks/{1}".format(dashboard, block)
return Block.from_dict(self._get(path))
def create_block(self, dashboard, block):
"""Create the specified block under the specified dashboard.
:param string dashboard: Key of Adafruit IO Dashboard.
:param Block block: Block object to create under dashboard
"""
path = "dashboards/{0}/blocks".format(dashboard)
return Block.from_dict(self._post(path, block._asdict()))
def delete_block(self, dashboard, block):
"""Delete the specified block.
:param string dashboard: Key of Adafruit IO Dashboard.
:param string block: id of Adafruit IO Block.
"""
path = "dashboards/{0}/blocks/{1}".format(dashboard, block)
self._delete(path)
# Layout functionality.
def layouts(self, dashboard):
"""Retrieve the layouts array from a dashboard
:param string dashboard: key of Adafruit IO Dashboard.
"""
path = "dashboards/{0}".format(dashboard)
dashboard = self._get(path)
return Layout.from_dict(dashboard['layouts'])
def update_layout(self, dashboard, layout):
"""Update the layout of the specified dashboard.
:param string dashboard: Key of Adafruit IO Dashboard.
:param Layout layout: Layout object to update under dashboard
"""
path = "dashboards/{0}/update_layouts".format(dashboard)
return Layout.from_dict(self._post(path, {'layouts': layout._asdict()}))

44
Adafruit_IO/model.py
View file

@ -43,6 +43,7 @@ DATA_FIELDS = [ 'created_epoch',
FEED_FIELDS = [ 'name',
'key',
'id',
'description',
'unit_type',
'unit_symbol',
@ -61,6 +62,26 @@ GROUP_FIELDS = [ 'description',
'properties',
'name' ]
DASHBOARD_FIELDS = [ 'name',
'key',
'description',
'show_header',
'color_mode',
'block_borders',
'header_image_url',
'blocks' ]
BLOCK_FIELDS = [ 'name',
'id',
'visual_type',
'properties',
'block_feeds' ]
LAYOUT_FIELDS = ['xl',
'lg',
'md',
'sm',
'xs' ]
# These are very simple data model classes that are based on namedtuple. This is
# to keep the classes simple and prevent any confusion around updating data
@ -71,15 +92,24 @@ GROUP_FIELDS = [ 'description',
Data = namedtuple('Data', DATA_FIELDS)
Feed = namedtuple('Feed', FEED_FIELDS)
Group = namedtuple('Group', GROUP_FIELDS)
Dashboard = namedtuple('Dashboard', DASHBOARD_FIELDS)
Block = namedtuple('Block', BLOCK_FIELDS)
Layout = namedtuple('Layout', LAYOUT_FIELDS)
# Magic incantation to make all parameters to the initializers optional with a
# default value of None.
Group.__new__.__defaults__ = tuple(None for x in GROUP_FIELDS)
Data.__new__.__defaults__ = tuple(None for x in DATA_FIELDS)
Layout.__new__.__defaults__ = tuple(None for x in LAYOUT_FIELDS)
# explicitly set dashboard values so that 'color_mode' is 'dark'
Dashboard.__new__.__defaults__ = (None, None, None, False, "dark", True, None, None)
# explicitly set block values so 'properties' is a dictionary
Block.__new__.__defaults__ = (None, None, None, {}, None)
# explicitly set feed values
Feed.__new__.__defaults__ = (None, None, None, None, None, 'ON', 'Private', None, None, None)
Feed.__new__.__defaults__ = (None, None, None, None, None, None, 'ON', 'Private', None, None, None)
# Define methods to convert from dicts to the data types.
def _from_dict(cls, data):
@ -103,7 +133,17 @@ def _group_from_dict(cls, data):
return cls(**params)
def _dashboard_from_dict(cls, data):
params = {x: data.get(x, None) for x in cls._fields}
# Parse the blocks if they're provided and generate block instances.
params['blocks'] = tuple(map(Feed.from_dict, data.get('blocks', [])))
return cls(**params)
# Now add the from_dict class methods defined above to the data types.
Data.from_dict = classmethod(_from_dict)
Feed.from_dict = classmethod(_feed_from_dict)
Group.from_dict = classmethod(_group_from_dict)
Dashboard.from_dict = classmethod(_dashboard_from_dict)
Block.from_dict = classmethod(_from_dict)
Layout.from_dict = classmethod(_from_dict)

88
examples/basics/dashboard.py Normal file
View file

@ -0,0 +1,88 @@
"""
'dashboard.py'
=========================================
Creates a dashboard with 3 blocks and feed it data
Author(s): Doug Zobel
"""
from time import sleep
from random import randrange
from Adafruit_IO import Client, Feed, Block, Dashboard, Layout
# Set to your Adafruit IO key.
# Remember, your key is a secret,
# so make sure not to publish it when you publish this code!
ADAFRUIT_IO_USERNAME = ''
# Set to your Adafruit IO username.
# (go to https://accounts.adafruit.com to find your username)
ADAFRUIT_IO_KEY = ''
# Create an instance of the REST client.
aio = Client(ADAFRUIT_IO_USERNAME, ADAFRUIT_IO_KEY)
# Create a new feed named 'Dashboard Data' under the default group
feed = aio.create_feed(Feed(name="Dashboard Data"), "default")
# Fetch group info (group.id needed when adding feeds to blocks)
group = aio.groups("default")
# Create a new dasbhoard named 'Example Dashboard'
dashboard = aio.create_dashboard(Dashboard(name="Example Dashboard"))
# Create a line_chart
linechart = Block(name="Linechart Data",
visual_type = 'line_chart',
properties = {
"gridLines": True,
"historyHours": "2"},
block_feeds = [{
"group_id": group.id,
"feed_id": feed.id
}])
linechart = aio.create_block(dashboard.key, linechart)
# Create a gauge
gauge = Block(name="Gauge Data",
visual_type = 'gauge',
block_feeds = [{
"group_id": group.id,
"feed_id": feed.id
}])
gauge = aio.create_block(dashboard.key, gauge)
# Create a text stream
stream = Block(name="Stream Data",
visual_type = 'stream',
properties = {
"fontSize": "12",
"fontColor": "#63de00",
"showGroupName": "no"},
block_feeds = [{
"group_id": group.id,
"feed_id": feed.id
}])
stream = aio.create_block(dashboard.key, stream)
# Update the large layout to:
# |----------------|
# | Line Chart |
# |----------------|
# | Gauge | Stream |
# |----------------|
layout = Layout(lg = [
{'x': 0, 'y': 0, 'w': 16, 'h': 4, 'i': str(linechart.id)},
{'x': 0, 'y': 4, 'w': 8, 'h': 4, 'i': str(gauge.id)},
{'x': 8, 'y': 4, 'w': 8, 'h': 4, 'i': str(stream.id)}])
aio.update_layout(dashboard.key, layout)
print("Dashboard created at: " +
"https://io.adafruit.com/{0}/dashboards/{1}".format(ADAFRUIT_IO_USERNAME,
dashboard.key))
# Now send some data
value = 0
while True:
value = (value + randrange(0, 10)) % 100
print('sending data: ', value)
aio.send_data(feed.key, value)
sleep(3)

105
tests/test_client.py
View file

@ -3,7 +3,7 @@
import time
import unittest
from Adafruit_IO import Client, Data, Feed, Group, RequestError
from Adafruit_IO import Client, Data, Feed, Group, Dashboard, Block, Layout, RequestError
import base
@ -46,6 +46,22 @@ class TestClient(base.IOTestCase):
# Swallow the error if the group doesn't exist.
pass
def ensure_dashboard_deleted(self, client, dashboard):
# Delete the specified dashboard if it exists.
try:
client.delete_dashboard(dashboard)
except RequestError:
# Swallow the error if the dashboard doesn't exist.
pass
def ensure_block_deleted(self, client, dashboard, block):
# Delete the specified block if it exists.
try:
client.delete_block(dashboard, block)
except RequestError:
# Swallow the error if the block doesn't exist.
pass
def empty_feed(self, client, feed):
# Remove all the data from a specified feed (but don't delete the feed).
data = client.data(feed)
@ -269,3 +285,90 @@ class TestClient(base.IOTestCase):
group = io.create_group(Group(name='grouprx'))
response = io.groups(group.key)
self.assertEqual(response.key, 'grouprx')
# Test Dashboard Functionality
def test_dashboard_create_dashboard(self):
io = self.get_client()
self.ensure_dashboard_deleted(io, 'dashtest')
response = io.create_dashboard(Dashboard(name='dashtest'))
self.assertEqual(response.name, 'dashtest')
def test_dashboard_returns_all_dashboards(self):
io = self.get_client()
self.ensure_dashboard_deleted(io, 'dashtest')
dashboard = io.create_dashboard(Dashboard(name='dashtest'))
response = io.dashboards()
self.assertGreaterEqual(len(response), 1)
def test_dashboard_returns_requested_feed(self):
io = self.get_client()
self.ensure_dashboard_deleted(io, 'dashtest')
dashboard = io.create_dashboard(Dashboard(name='dashtest'))
response = io.dashboards('dashtest')
self.assertEqual(response.name, 'dashtest')
# Test Block Functionality
def test_block_create_block(self):
io = self.get_client()
self.ensure_block_deleted(io, 'dashtest', 'blocktest')
self.ensure_dashboard_deleted(io, 'dashtest')
dash = io.create_dashboard(Dashboard(name='dashtest'))
block = io.create_block(dash.key, Block(name='blocktest',
visual_type = 'line_chart'))
self.assertEqual(block.name, 'blocktest')
io.delete_block(dash.key, block.id)
io.delete_dashboard(dash.key)
def test_dashboard_returns_all_blocks(self):
io = self.get_client()
self.ensure_block_deleted(io, 'dashtest', 'blocktest')
self.ensure_dashboard_deleted(io, 'dashtest')
dash = io.create_dashboard(Dashboard(name='dashtest'))
block = io.create_block(dash.key, Block(name='blocktest',
visual_type = 'line_chart'))
response = io.blocks(dash.key)
self.assertEqual(len(response), 1)
io.delete_block(dash.key, block.id)
io.delete_dashboard(dash.key)
def test_dashboard_returns_requested_block(self):
io = self.get_client()
self.ensure_block_deleted(io, 'dashtest', 'blocktest')
self.ensure_dashboard_deleted(io, 'dashtest')
dash = io.create_dashboard(Dashboard(name='dashtest'))
block = io.create_block(dash.key, Block(name='blocktest',
visual_type = 'line_chart'))
response = io.blocks(dash.key, block.id)
self.assertEqual(response.name, 'blocktest')
io.delete_block(dash.key, block.id)
io.delete_dashboard(dash.key)
# Test Layout Functionality
def test_layout_returns_all_layouts(self):
io = self.get_client()
self.ensure_block_deleted(io, 'dashtest', 'blocktest')
self.ensure_dashboard_deleted(io, 'dashtest')
dash = io.create_dashboard(Dashboard(name='dashtest'))
block = io.create_block(dash.key, Block(name='blocktest',
visual_type = 'line_chart'))
response = io.layouts(dash.key)
self.assertEqual(len(response), 5) # 5 layouts: xs, sm, md, lg, xl
self.assertEqual(len(response.lg), 1)
io.delete_block(dash.key, block.id)
io.delete_dashboard(dash.key)
def test_layout_update_layout(self):
io = self.get_client()
self.ensure_block_deleted(io, 'dashtest', 'blocktest')
self.ensure_dashboard_deleted(io, 'dashtest')
dash = io.create_dashboard(Dashboard(name='dashtest'))
block = io.create_block(dash.key, Block(name='blocktest',
visual_type = 'line_chart'))
layout = Layout(lg = [
{'x': 0, 'y': 0, 'w': 16, 'h': 4, 'i': str(block.id)}])
io.update_layout(dash.key, layout)
response = io.layouts(dash.key)
self.assertEqual(len(response.lg), 1)
self.assertEqual(response.lg[0]['w'], 16)
io.delete_block(dash.key, block.id)
io.delete_dashboard(dash.key)

39
tests/test_model.py
View file

@ -18,7 +18,7 @@
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
from Adafruit_IO import Data, Feed, Group
from Adafruit_IO import Data, Feed, Group, Dashboard, Block, Layout
import base
@ -45,11 +45,12 @@ class TestData(base.IOTestCase):
def test_feeds_have_explicitly_set_values(self):
""" Let's make sure feeds are explicitly set from within the model:
Feed.__new__.__defaults__ = (None, None, None, None, None, 'ON', 'Private', None, None, None)
Feed.__new__.__defaults__ = (None, None, None, None, None, None, 'ON', 'Private', None, None, None)
"""
feed = Feed(name='foo')
self.assertEqual(feed.name, 'foo')
self.assertIsNone(feed.key)
self.assertIsNone(feed.id)
self.assertIsNone(feed.description)
self.assertIsNone(feed.unit_type)
self.assertIsNone(feed.unit_symbol)
@ -69,6 +70,40 @@ class TestData(base.IOTestCase):
self.assertIsNone(group.feeds)
self.assertIsNone(group.properties)
""" Let's make sure feeds are explicitly set from within the model:
Dashboard.__new__.__defaults__ = (None, None, None, False, "dark", True, None, None)
"""
def test_dashboard_have_explicitly_set_values(self):
dashboard = Dashboard(name="foo")
self.assertEqual(dashboard.name, 'foo')
self.assertIsNone(dashboard.key)
self.assertIsNone(dashboard.description)
self.assertFalse(dashboard.show_header)
self.assertEqual(dashboard.color_mode, 'dark')
self.assertTrue(dashboard.block_borders)
self.assertIsNone(dashboard.header_image_url)
self.assertIsNone(dashboard.blocks)
""" Let's make sure feeds are explicitly set from within the model:
Block.__new__.__defaults__ = (None, None, None {}, None)
"""
def test_block_have_explicitly_set_values(self):
block = Block(name="foo")
self.assertEqual(block.name, 'foo')
self.assertIsNone(block.id)
self.assertIsNone(block.visual_type)
self.assertEqual(type(block.properties), dict)
self.assertEqual(len(block.properties), 0)
self.assertIsNone(block.block_feeds)
def test_layout_properties_are_optional(self):
layout = Layout()
self.assertIsNone(layout.xl)
self.assertIsNone(layout.lg)
self.assertIsNone(layout.md)
self.assertIsNone(layout.sm)
self.assertIsNone(layout.xs)
def test_from_dict_ignores_unknown_items(self):
data = Data.from_dict({'value': 'foo', 'feed_id': 10, 'unknown_param': 42})