JSON MQTT Device tracker ()

* ready for PR

* minor fix

* another minor fix

* new platform mqtt_json

* using ATTR constants

* voluptuous check on JSON payload

* voluptuous check on JSON payload
This commit is contained in:
Gianluca Barbaro 2017-04-20 07:53:07 +02:00 committed by Paulus Schoutsen
parent e020d5114a
commit f59b3da5fe
2 changed files with 213 additions and 0 deletions
homeassistant/components/device_tracker
tests/components/device_tracker

View file

@ -0,0 +1,85 @@
"""
Support for GPS tracking MQTT enabled devices.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/device_tracker.mqtt_json/
"""
import asyncio
import json
import logging
import voluptuous as vol
import homeassistant.components.mqtt as mqtt
from homeassistant.core import callback
from homeassistant.components.mqtt import CONF_QOS
from homeassistant.components.device_tracker import PLATFORM_SCHEMA
import homeassistant.helpers.config_validation as cv
from homeassistant.const import (
CONF_DEVICES, ATTR_GPS_ACCURACY, ATTR_LATITUDE,
ATTR_LONGITUDE, ATTR_BATTERY_LEVEL)
DEPENDENCIES = ['mqtt']
_LOGGER = logging.getLogger(__name__)
GPS_JSON_PAYLOAD_SCHEMA = vol.Schema({
vol.Required(ATTR_LATITUDE): vol.Coerce(float),
vol.Required(ATTR_LONGITUDE): vol.Coerce(float),
vol.Optional(ATTR_GPS_ACCURACY, default=None): vol.Coerce(int),
vol.Optional(ATTR_BATTERY_LEVEL, default=None): vol.Coerce(str),
}, extra=vol.ALLOW_EXTRA)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(mqtt.SCHEMA_BASE).extend({
vol.Required(CONF_DEVICES): {cv.string: mqtt.valid_subscribe_topic},
})
@asyncio.coroutine
def async_setup_scanner(hass, config, async_see, discovery_info=None):
"""Setup the MQTT tracker."""
devices = config[CONF_DEVICES]
qos = config[CONF_QOS]
dev_id_lookup = {}
@callback
def async_tracker_message_received(topic, payload, qos):
"""MQTT message received."""
dev_id = dev_id_lookup[topic]
try:
data = GPS_JSON_PAYLOAD_SCHEMA(json.loads(payload))
except vol.MultipleInvalid:
_LOGGER.error('Skipping update for following data '
'because of missing or malformatted data: %s',
payload)
return
except ValueError:
_LOGGER.error('Error parsing JSON payload: %s', payload)
return
kwargs = _parse_see_args(dev_id, data)
hass.async_add_job(
async_see(**kwargs))
for dev_id, topic in devices.items():
dev_id_lookup[topic] = dev_id
yield from mqtt.async_subscribe(
hass, topic, async_tracker_message_received, qos)
return True
def _parse_see_args(dev_id, data):
"""Parse the payload location parameters, into the format see expects."""
kwargs = {
'gps': (data[ATTR_LATITUDE], data[ATTR_LONGITUDE]),
'dev_id': dev_id
}
if ATTR_GPS_ACCURACY in data:
kwargs[ATTR_GPS_ACCURACY] = data[ATTR_GPS_ACCURACY]
if ATTR_BATTERY_LEVEL in data:
kwargs['battery'] = data[ATTR_BATTERY_LEVEL]
return kwargs

View file

@ -0,0 +1,128 @@
"""The tests for the JSON MQTT device tracker platform."""
import asyncio
import json
import unittest
from unittest.mock import patch
import logging
import os
from homeassistant.setup import setup_component
from homeassistant.components import device_tracker
from homeassistant.const import CONF_PLATFORM
from tests.common import (
get_test_home_assistant, mock_mqtt_component, fire_mqtt_message)
_LOGGER = logging.getLogger(__name__)
LOCATION_MESSAGE = {
'longitude': 1.0,
'gps_accuracy': 60,
'latitude': 2.0,
'battery_level': 99.9}
LOCATION_MESSAGE_INCOMPLETE = {
'longitude': 2.0}
class TestComponentsDeviceTrackerJSONMQTT(unittest.TestCase):
"""Test JSON MQTT device tracker platform."""
def setUp(self): # pylint: disable=invalid-name
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_mqtt_component(self.hass)
def tearDown(self): # pylint: disable=invalid-name
"""Stop everything that was started."""
self.hass.stop()
try:
os.remove(self.hass.config.path(device_tracker.YAML_DEVICES))
except FileNotFoundError:
pass
def test_ensure_device_tracker_platform_validation(self): \
# pylint: disable=invalid-name
"""Test if platform validation was done."""
@asyncio.coroutine
def mock_setup_scanner(hass, config, see, discovery_info=None):
"""Check that Qos was added by validation."""
self.assertTrue('qos' in config)
with patch('homeassistant.components.device_tracker.mqtt_json.'
'async_setup_scanner', autospec=True,
side_effect=mock_setup_scanner) as mock_sp:
dev_id = 'paulus'
topic = 'location/paulus'
assert setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'mqtt_json',
'devices': {dev_id: topic}
}
})
assert mock_sp.call_count == 1
def test_json_message(self):
"""Test json location message."""
dev_id = 'zanzito'
topic = 'location/zanzito'
location = json.dumps(LOCATION_MESSAGE)
self.hass.config.components = set(['mqtt_json', 'zone'])
assert setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'mqtt_json',
'devices': {dev_id: topic}
}
})
fire_mqtt_message(self.hass, topic, location)
self.hass.block_till_done()
state = self.hass.states.get('device_tracker.zanzito')
self.assertEqual(state.attributes.get('latitude'), 2.0)
self.assertEqual(state.attributes.get('longitude'), 1.0)
def test_non_json_message(self):
"""Test receiving a non JSON message."""
dev_id = 'zanzito'
topic = 'location/zanzito'
location = 'home'
self.hass.config.components = set(['mqtt_json'])
assert setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'mqtt_json',
'devices': {dev_id: topic}
}
})
with self.assertLogs(level='ERROR') as test_handle:
fire_mqtt_message(self.hass, topic, location)
self.hass.block_till_done()
self.assertIn(
"ERROR:homeassistant.components.device_tracker.mqtt_json:"
"Error parsing JSON payload: home",
test_handle.output[0])
def test_incomplete_message(self):
"""Test receiving an incomplete message."""
dev_id = 'zanzito'
topic = 'location/zanzito'
location = json.dumps(LOCATION_MESSAGE_INCOMPLETE)
self.hass.config.components = set(['mqtt_json'])
assert setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'mqtt_json',
'devices': {dev_id: topic}
}
})
with self.assertLogs(level='ERROR') as test_handle:
fire_mqtt_message(self.hass, topic, location)
self.hass.block_till_done()
self.assertIn(
"ERROR:homeassistant.components.device_tracker.mqtt_json:"
"Skipping update for following data because of missing "
"or malformatted data: {\"longitude\": 2.0}",
test_handle.output[0])