jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 05:07:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Restructure device tracker (#23862)

Browse Source 
* Restructure device tracker

* Docstyle

* Fix typing

* Lint

* Lint

* Fix tests
This commit is contained in:
Paulus Schoutsen 2019-05-15 23:43:45 +02:00 committed by GitHub
parent 7a4238095d
commit 70ed58a78d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 978 additions and 757 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
homeassistant/components/bluetooth_le_tracker/device_tracker.py
View File

@ -2,12 +2,15 @@
import logging
from homeassistant.helpers.event import track_point_in_utc_time
from homeassistant.components.device_tracker import (
YAML_DEVICES, CONF_TRACK_NEW, CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL,
load_config, SOURCE_TYPE_BLUETOOTH_LE
from homeassistant.components.device_tracker.legacy import (
YAML_DEVICES, async_load_config
)
from homeassistant.components.device_tracker.const import (
CONF_TRACK_NEW, CONF_SCAN_INTERVAL, SCAN_INTERVAL, SOURCE_TYPE_BLUETOOTH_LE
)
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
import homeassistant.util.dt as dt_util
from homeassistant.util.async_ import run_coroutine_threadsafe
_LOGGER = logging.getLogger(__name__)
@ -79,7 +82,10 @@ def setup_scanner(hass, config, see, discovery_info=None):
# Load all known devices.
# We just need the devices so set consider_home and home range
# to 0
for device in load_config(yaml_path, hass, 0):
for device in run_coroutine_threadsafe(
async_load_config(yaml_path, hass, 0),
hass.loop
).result():
# check if device is a valid bluetooth device
if device.mac and device.mac[:4].upper() == BLE_PREFIX:
if device.track:
@ -97,7 +103,7 @@ def setup_scanner(hass, config, see, discovery_info=None):
_LOGGER.warning("No Bluetooth LE devices to track!")
return False
interval = config.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
interval = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL)
def update_ble(now):
"""Lookup Bluetooth LE devices and update status."""

20
homeassistant/components/bluetooth_tracker/device_tracker.py
View File

@ -5,11 +5,16 @@ import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.event import track_point_in_utc_time
from homeassistant.components.device_tracker import (
YAML_DEVICES, CONF_TRACK_NEW, CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL,
load_config, PLATFORM_SCHEMA, DEFAULT_TRACK_NEW, SOURCE_TYPE_BLUETOOTH,
DOMAIN)
from homeassistant.components.device_tracker import PLATFORM_SCHEMA
from homeassistant.components.device_tracker.legacy import (
YAML_DEVICES, async_load_config
)
from homeassistant.components.device_tracker.const import (
CONF_TRACK_NEW, CONF_SCAN_INTERVAL, SCAN_INTERVAL, DEFAULT_TRACK_NEW,
SOURCE_TYPE_BLUETOOTH, DOMAIN
)
import homeassistant.util.dt as dt_util
from homeassistant.util.async_ import run_coroutine_threadsafe
_LOGGER = logging.getLogger(__name__)
@ -60,7 +65,10 @@ def setup_scanner(hass, config, see, discovery_info=None):
# Load all known devices.
# We just need the devices so set consider_home and home range
# to 0
for device in load_config(yaml_path, hass, 0):
for device in run_coroutine_threadsafe(
async_load_config(yaml_path, hass, 0),
hass.loop
).result():
# Check if device is a valid bluetooth device
if device.mac and device.mac[:3].upper() == BT_PREFIX:
if device.track:
@ -77,7 +85,7 @@ def setup_scanner(hass, config, see, discovery_info=None):
devs_to_track.append(dev[0])
see_device(dev[0], dev[1])
interval = config.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
interval = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL)
request_rssi = config.get(CONF_REQUEST_RSSI, False)

709
homeassistant/components/device_tracker/__init__.py
View File

@ -1,78 +1,53 @@
"""Provide functionality to keep track of devices."""
import asyncio
from datetime import timedelta
import logging
from typing import Any, List, Sequence, Callable
import voluptuous as vol
from homeassistant.setup import async_prepare_setup_platform
from homeassistant.core import callback
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.loader import bind_hass
from homeassistant.components import group, zone
from homeassistant.components.group import (
ATTR_ADD_ENTITIES, ATTR_ENTITIES, ATTR_OBJECT_ID, ATTR_VISIBLE,
DOMAIN as DOMAIN_GROUP, SERVICE_SET)
from homeassistant.components.zone.zone import async_active_zone
from homeassistant.config import load_yaml_config_file, async_log_exception
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_per_platform, discovery
from homeassistant.components import group
from homeassistant.config import config_without_domain
from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import GPSType, ConfigType, HomeAssistantType
from homeassistant import util
from homeassistant.util.async_ import run_coroutine_threadsafe
import homeassistant.util.dt as dt_util
from homeassistant.util.yaml import dump
from homeassistant.helpers.event import async_track_utc_time_change
from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_GPS_ACCURACY, ATTR_ICON, ATTR_LATITUDE,
ATTR_LONGITUDE, ATTR_NAME, CONF_ICON, CONF_MAC, CONF_NAME,
DEVICE_DEFAULT_NAME, STATE_NOT_HOME, STATE_HOME)
from homeassistant.const import ATTR_GPS_ACCURACY, STATE_HOME
_LOGGER = logging.getLogger(__name__)
from . import legacy, setup
from .legacy import DeviceScanner # noqa # pylint: disable=unused-import
from .const import (
ATTR_ATTRIBUTES,
ATTR_BATTERY,
ATTR_CONSIDER_HOME,
ATTR_DEV_ID,
ATTR_GPS,
ATTR_HOST_NAME,
ATTR_LOCATION_NAME,
ATTR_MAC,
ATTR_SOURCE_TYPE,
CONF_AWAY_HIDE,
CONF_CONSIDER_HOME,
CONF_NEW_DEVICE_DEFAULTS,
CONF_SCAN_INTERVAL,
CONF_TRACK_NEW,
DEFAULT_AWAY_HIDE,
DEFAULT_CONSIDER_HOME,
DEFAULT_TRACK_NEW,
DOMAIN,
LOGGER,
PLATFORM_TYPE_LEGACY,
SCAN_INTERVAL,
SOURCE_TYPE_BLUETOOTH_LE,
SOURCE_TYPE_BLUETOOTH,
SOURCE_TYPE_GPS,
SOURCE_TYPE_ROUTER,
)
DOMAIN = 'device_tracker'
GROUP_NAME_ALL_DEVICES = 'all devices'
ENTITY_ID_ALL_DEVICES = group.ENTITY_ID_FORMAT.format('all_devices')
ENTITY_ID_FORMAT = DOMAIN + '.{}'
YAML_DEVICES = 'known_devices.yaml'
CONF_TRACK_NEW = 'track_new_devices'
DEFAULT_TRACK_NEW = True
CONF_NEW_DEVICE_DEFAULTS = 'new_device_defaults'
CONF_CONSIDER_HOME = 'consider_home'
DEFAULT_CONSIDER_HOME = timedelta(seconds=180)
CONF_SCAN_INTERVAL = 'interval_seconds'
DEFAULT_SCAN_INTERVAL = timedelta(seconds=12)
CONF_AWAY_HIDE = 'hide_if_away'
DEFAULT_AWAY_HIDE = False
EVENT_NEW_DEVICE = 'device_tracker_new_device'
SERVICE_SEE = 'see'
ATTR_ATTRIBUTES = 'attributes'
ATTR_BATTERY = 'battery'
ATTR_DEV_ID = 'dev_id'
ATTR_GPS = 'gps'
ATTR_HOST_NAME = 'host_name'
ATTR_LOCATION_NAME = 'location_name'
ATTR_MAC = 'mac'
ATTR_SOURCE_TYPE = 'source_type'
ATTR_CONSIDER_HOME = 'consider_home'
SOURCE_TYPE_GPS = 'gps'
SOURCE_TYPE_ROUTER = 'router'
SOURCE_TYPE_BLUETOOTH = 'bluetooth'
SOURCE_TYPE_BLUETOOTH_LE = 'bluetooth_le'
SOURCE_TYPES = (SOURCE_TYPE_GPS, SOURCE_TYPE_ROUTER,
SOURCE_TYPE_BLUETOOTH, SOURCE_TYPE_BLUETOOTH_LE)
@ -136,75 +111,52 @@ def see(hass: HomeAssistantType, mac: str = None, dev_id: str = None,
async def async_setup(hass: HomeAssistantType, config: ConfigType):
"""Set up the device tracker."""
yaml_path = hass.config.path(YAML_DEVICES)
tracker = await legacy.get_tracker(hass, config)
conf = config.get(DOMAIN, [])
conf = conf[0] if conf else {}
consider_home = conf.get(CONF_CONSIDER_HOME, DEFAULT_CONSIDER_HOME)
async def setup_entry_helper(entry):
"""Set up a config entry."""
platform = await setup.async_create_platform_type(
hass, config, entry.domain, entry)
defaults = conf.get(CONF_NEW_DEVICE_DEFAULTS, {})
track_new = conf.get(CONF_TRACK_NEW)
if track_new is None:
track_new = defaults.get(CONF_TRACK_NEW, DEFAULT_TRACK_NEW)
devices = await async_load_config(yaml_path, hass, consider_home)
tracker = DeviceTracker(
hass, consider_home, track_new, defaults, devices)
async def async_setup_platform(p_type, p_config, disc_info=None):
"""Set up a device tracker platform."""
platform = await async_prepare_setup_platform(
hass, config, DOMAIN, p_type)
if platform is None:
return
return False
_LOGGER.info("Setting up %s.%s", DOMAIN, p_type)
try:
scanner = None
setup = None
if hasattr(platform, 'async_get_scanner'):
scanner = await platform.async_get_scanner(
hass, {DOMAIN: p_config})
elif hasattr(platform, 'get_scanner'):
scanner = await hass.async_add_job(
platform.get_scanner, hass, {DOMAIN: p_config})
elif hasattr(platform, 'async_setup_scanner'):
setup = await platform.async_setup_scanner(
hass, p_config, tracker.async_see, disc_info)
elif hasattr(platform, 'setup_scanner'):
setup = await hass.async_add_job(
platform.setup_scanner, hass, p_config, tracker.see,
disc_info)
elif hasattr(platform, 'async_setup_entry'):
setup = await platform.async_setup_entry(
hass, p_config, tracker.async_see)
else:
raise HomeAssistantError("Invalid device_tracker platform.")
await platform.async_setup_legacy(hass, tracker)
if scanner:
async_setup_scanner_platform(
hass, p_config, scanner, tracker.async_see, p_type)
return
return True
if not setup:
_LOGGER.error("Error setting up platform %s", p_type)
return
hass.data[DOMAIN] = setup_entry_helper
component = EntityComponent(
LOGGER, DOMAIN, hass, SCAN_INTERVAL)
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Error setting up platform %s", p_type)
legacy_platforms, entity_platforms = \
await setup.async_extract_config(hass, config)
hass.data[DOMAIN] = async_setup_platform
setup_tasks = [
legacy_platform.async_setup_legacy(hass, tracker)
for legacy_platform in legacy_platforms
]
if entity_platforms:
setup_tasks.append(component.async_setup({
**config_without_domain(config, DOMAIN),
DOMAIN: [platform.config for platform in entity_platforms]
}))
setup_tasks = [async_setup_platform(p_type, p_config) for p_type, p_config
in config_per_platform(config, DOMAIN)]
if setup_tasks:
await asyncio.wait(setup_tasks, loop=hass.loop)
tracker.async_setup_group()
async def async_platform_discovered(platform, info):
async def async_platform_discovered(p_type, info):
"""Load a platform."""
await async_setup_platform(platform, {}, disc_info=info)
platform = await setup.async_create_platform_type(
hass, config, p_type, {})
if platform is None or platform.type != PLATFORM_TYPE_LEGACY:
return
await platform.async_setup_legacy(hass, tracker, info)
discovery.async_listen_platform(hass, DOMAIN, async_platform_discovered)
@ -230,533 +182,4 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType):
async def async_setup_entry(hass, entry):
"""Set up an entry."""
await hass.data[DOMAIN](entry.domain, entry)
return True
class DeviceTracker:
"""Representation of a device tracker."""
def __init__(self, hass: HomeAssistantType, consider_home: timedelta,
track_new: bool, defaults: dict,
devices: Sequence) -> None:
"""Initialize a device tracker."""
self.hass = hass
self.devices = {dev.dev_id: dev for dev in devices}
self.mac_to_dev = {dev.mac: dev for dev in devices if dev.mac}
self.consider_home = consider_home
self.track_new = track_new if track_new is not None \
else defaults.get(CONF_TRACK_NEW, DEFAULT_TRACK_NEW)
self.defaults = defaults
self.group = None
self._is_updating = asyncio.Lock(loop=hass.loop)
for dev in devices:
if self.devices[dev.dev_id] is not dev:
_LOGGER.warning('Duplicate device IDs detected %s', dev.dev_id)
if dev.mac and self.mac_to_dev[dev.mac] is not dev:
_LOGGER.warning('Duplicate device MAC addresses detected %s',
dev.mac)
def see(self, mac: str = None, dev_id: str = None, host_name: str = None,
location_name: str = None, gps: GPSType = None,
gps_accuracy: int = None, battery: int = None,
attributes: dict = None, source_type: str = SOURCE_TYPE_GPS,
picture: str = None, icon: str = None,
consider_home: timedelta = None):
"""Notify the device tracker that you see a device."""
self.hass.add_job(
self.async_see(mac, dev_id, host_name, location_name, gps,
gps_accuracy, battery, attributes, source_type,
picture, icon, consider_home)
)
async def async_see(
self, mac: str = None, dev_id: str = None, host_name: str = None,
location_name: str = None, gps: GPSType = None,
gps_accuracy: int = None, battery: int = None,
attributes: dict = None, source_type: str = SOURCE_TYPE_GPS,
picture: str = None, icon: str = None,
consider_home: timedelta = None):
"""Notify the device tracker that you see a device.
This method is a coroutine.
"""
if mac is None and dev_id is None:
raise HomeAssistantError('Neither mac or device id passed in')
if mac is not None:
mac = str(mac).upper()
device = self.mac_to_dev.get(mac)
if not device:
dev_id = util.slugify(host_name or '') or util.slugify(mac)
else:
dev_id = cv.slug(str(dev_id).lower())
device = self.devices.get(dev_id)
if device:
await device.async_seen(
host_name, location_name, gps, gps_accuracy, battery,
attributes, source_type, consider_home)
if device.track:
await device.async_update_ha_state()
return
# If no device can be found, create it
dev_id = util.ensure_unique_string(dev_id, self.devices.keys())
device = Device(
self.hass, consider_home or self.consider_home, self.track_new,
dev_id, mac, (host_name or dev_id).replace('_', ' '),
picture=picture, icon=icon,
hide_if_away=self.defaults.get(CONF_AWAY_HIDE, DEFAULT_AWAY_HIDE))
self.devices[dev_id] = device
if mac is not None:
self.mac_to_dev[mac] = device
await device.async_seen(
host_name, location_name, gps, gps_accuracy, battery, attributes,
source_type)
if device.track:
await device.async_update_ha_state()
# During init, we ignore the group
if self.group and self.track_new:
self.hass.async_create_task(
self.hass.async_call(
DOMAIN_GROUP, SERVICE_SET, {
ATTR_OBJECT_ID: util.slugify(GROUP_NAME_ALL_DEVICES),
ATTR_VISIBLE: False,
ATTR_NAME: GROUP_NAME_ALL_DEVICES,
ATTR_ADD_ENTITIES: [device.entity_id]}))
self.hass.bus.async_fire(EVENT_NEW_DEVICE, {
ATTR_ENTITY_ID: device.entity_id,
ATTR_HOST_NAME: device.host_name,
ATTR_MAC: device.mac,
})
# update known_devices.yaml
self.hass.async_create_task(
self.async_update_config(
self.hass.config.path(YAML_DEVICES), dev_id, device)
)
async def async_update_config(self, path, dev_id, device):
"""Add device to YAML configuration file.
This method is a coroutine.
"""
async with self._is_updating:
await self.hass.async_add_executor_job(
update_config, self.hass.config.path(YAML_DEVICES),
dev_id, device)
@callback
def async_setup_group(self):
"""Initialize group for all tracked devices.
This method must be run in the event loop.
"""
entity_ids = [dev.entity_id for dev in self.devices.values()
if dev.track]
self.hass.async_create_task(
self.hass.services.async_call(
DOMAIN_GROUP, SERVICE_SET, {
ATTR_OBJECT_ID: util.slugify(GROUP_NAME_ALL_DEVICES),
ATTR_VISIBLE: False,
ATTR_NAME: GROUP_NAME_ALL_DEVICES,
ATTR_ENTITIES: entity_ids}))
@callback
def async_update_stale(self, now: dt_util.dt.datetime):
"""Update stale devices.
This method must be run in the event loop.
"""
for device in self.devices.values():
if (device.track and device.last_update_home) and \
device.stale(now):
self.hass.async_create_task(device.async_update_ha_state(True))
async def async_setup_tracked_device(self):
"""Set up all not exists tracked devices.
This method is a coroutine.
"""
async def async_init_single_device(dev):
"""Init a single device_tracker entity."""
await dev.async_added_to_hass()
await dev.async_update_ha_state()
tasks = []
for device in self.devices.values():
if device.track and not device.last_seen:
tasks.append(self.hass.async_create_task(
async_init_single_device(device)))
if tasks:
await asyncio.wait(tasks, loop=self.hass.loop)
class Device(RestoreEntity):
"""Represent a tracked device."""
host_name = None # type: str
location_name = None # type: str
gps = None # type: GPSType
gps_accuracy = 0 # type: int
last_seen = None # type: dt_util.dt.datetime
consider_home = None # type: dt_util.dt.timedelta
battery = None # type: int
attributes = None # type: dict
icon = None # type: str
# Track if the last update of this device was HOME.
last_update_home = False
_state = STATE_NOT_HOME
def __init__(self, hass: HomeAssistantType, consider_home: timedelta,
track: bool, dev_id: str, mac: str, name: str = None,
picture: str = None, gravatar: str = None, icon: str = None,
hide_if_away: bool = False) -> None:
"""Initialize a device."""
self.hass = hass
self.entity_id = ENTITY_ID_FORMAT.format(dev_id)
# Timedelta object how long we consider a device home if it is not
# detected anymore.
self.consider_home = consider_home
# Device ID
self.dev_id = dev_id
self.mac = mac
# If we should track this device
self.track = track
# Configured name
self.config_name = name
# Configured picture
if gravatar is not None:
self.config_picture = get_gravatar_for_email(gravatar)
else:
self.config_picture = picture
self.icon = icon
self.away_hide = hide_if_away
self.source_type = None
self._attributes = {}
@property
def name(self):
"""Return the name of the entity."""
return self.config_name or self.host_name or DEVICE_DEFAULT_NAME
@property
def state(self):
"""Return the state of the device."""
return self._state
@property
def entity_picture(self):
"""Return the picture of the device."""
return self.config_picture
@property
def state_attributes(self):
"""Return the device state attributes."""
attr = {
ATTR_SOURCE_TYPE: self.source_type
}
if self.gps:
attr[ATTR_LATITUDE] = self.gps[0]
attr[ATTR_LONGITUDE] = self.gps[1]
attr[ATTR_GPS_ACCURACY] = self.gps_accuracy
if self.battery:
attr[ATTR_BATTERY] = self.battery
return attr
@property
def device_state_attributes(self):
"""Return device state attributes."""
return self._attributes
@property
def hidden(self):
"""If device should be hidden."""
return self.away_hide and self.state != STATE_HOME
async def async_seen(
self, host_name: str = None, location_name: str = None,
gps: GPSType = None, gps_accuracy=0, battery: int = None,
attributes: dict = None,
source_type: str = SOURCE_TYPE_GPS,
consider_home: timedelta = None):
"""Mark the device as seen."""
self.source_type = source_type
self.last_seen = dt_util.utcnow()
self.host_name = host_name
self.location_name = location_name
self.consider_home = consider_home or self.consider_home
if battery:
self.battery = battery
if attributes:
self._attributes.update(attributes)
self.gps = None
if gps is not None:
try:
self.gps = float(gps[0]), float(gps[1])
self.gps_accuracy = gps_accuracy or 0
except (ValueError, TypeError, IndexError):
self.gps = None
self.gps_accuracy = 0
_LOGGER.warning(
"Could not parse gps value for %s: %s", self.dev_id, gps)
# pylint: disable=not-an-iterable
await self.async_update()
def stale(self, now: dt_util.dt.datetime = None):
"""Return if device state is stale.
Async friendly.
"""
return self.last_seen is None or \
(now or dt_util.utcnow()) - self.last_seen > self.consider_home
def mark_stale(self):
"""Mark the device state as stale."""
self._state = STATE_NOT_HOME
self.gps = None
self.last_update_home = False
async def async_update(self):
"""Update state of entity.
This method is a coroutine.
"""
if not self.last_seen:
return
if self.location_name:
self._state = self.location_name
elif self.gps is not None and self.source_type == SOURCE_TYPE_GPS:
zone_state = async_active_zone(
self.hass, self.gps[0], self.gps[1], self.gps_accuracy)
if zone_state is None:
self._state = STATE_NOT_HOME
elif zone_state.entity_id == zone.ENTITY_ID_HOME:
self._state = STATE_HOME
else:
self._state = zone_state.name
elif self.stale():
self.mark_stale()
else:
self._state = STATE_HOME
self.last_update_home = True
async def async_added_to_hass(self):
"""Add an entity."""
await super().async_added_to_hass()
state = await self.async_get_last_state()
if not state:
return
self._state = state.state
self.last_update_home = (state.state == STATE_HOME)
self.last_seen = dt_util.utcnow()
for attr, var in (
(ATTR_SOURCE_TYPE, 'source_type'),
(ATTR_GPS_ACCURACY, 'gps_accuracy'),
(ATTR_BATTERY, 'battery'),
):
if attr in state.attributes:
setattr(self, var, state.attributes[attr])
if ATTR_LONGITUDE in state.attributes:
self.gps = (state.attributes[ATTR_LATITUDE],
state.attributes[ATTR_LONGITUDE])
class DeviceScanner:
"""Device scanner object."""
hass = None # type: HomeAssistantType
def scan_devices(self) -> List[str]:
"""Scan for devices."""
raise NotImplementedError()
def async_scan_devices(self) -> Any:
"""Scan for devices.
This method must be run in the event loop and returns a coroutine.
"""
return self.hass.async_add_job(self.scan_devices)
def get_device_name(self, device: str) -> str:
"""Get the name of a device."""
raise NotImplementedError()
def async_get_device_name(self, device: str) -> Any:
"""Get the name of a device.
This method must be run in the event loop and returns a coroutine.
"""
return self.hass.async_add_job(self.get_device_name, device)
def get_extra_attributes(self, device: str) -> dict:
"""Get the extra attributes of a device."""
raise NotImplementedError()
def async_get_extra_attributes(self, device: str) -> Any:
"""Get the extra attributes of a device.
This method must be run in the event loop and returns a coroutine.
"""
return self.hass.async_add_job(self.get_extra_attributes, device)
def load_config(path: str, hass: HomeAssistantType, consider_home: timedelta):
"""Load devices from YAML configuration file."""
return run_coroutine_threadsafe(
async_load_config(path, hass, consider_home), hass.loop).result()
async def async_load_config(path: str, hass: HomeAssistantType,
consider_home: timedelta):
"""Load devices from YAML configuration file.
This method is a coroutine.
"""
dev_schema = vol.Schema({
vol.Required(CONF_NAME): cv.string,
vol.Optional(CONF_ICON, default=None): vol.Any(None, cv.icon),
vol.Optional('track', default=False): cv.boolean,
vol.Optional(CONF_MAC, default=None):
vol.Any(None, vol.All(cv.string, vol.Upper)),
vol.Optional(CONF_AWAY_HIDE, default=DEFAULT_AWAY_HIDE): cv.boolean,
vol.Optional('gravatar', default=None): vol.Any(None, cv.string),
vol.Optional('picture', default=None): vol.Any(None, cv.string),
vol.Optional(CONF_CONSIDER_HOME, default=consider_home): vol.All(
cv.time_period, cv.positive_timedelta),
})
try:
result = []
try:
devices = await hass.async_add_job(
load_yaml_config_file, path)
except HomeAssistantError as err:
_LOGGER.error("Unable to load %s: %s", path, str(err))
return []
for dev_id, device in devices.items():
# Deprecated option. We just ignore it to avoid breaking change
device.pop('vendor', None)
try:
device = dev_schema(device)
device['dev_id'] = cv.slugify(dev_id)
except vol.Invalid as exp:
async_log_exception(exp, dev_id, devices, hass)
else:
result.append(Device(hass, **device))
return result
except (HomeAssistantError, FileNotFoundError):
# When YAML file could not be loaded/did not contain a dict
return []
@callback
def async_setup_scanner_platform(hass: HomeAssistantType, config: ConfigType,
scanner: Any, async_see_device: Callable,
platform: str):
"""Set up the connect scanner-based platform to device tracker.
This method must be run in the event loop.
"""
interval = config.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
update_lock = asyncio.Lock(loop=hass.loop)
scanner.hass = hass
# Initial scan of each mac we also tell about host name for config
seen = set() # type: Any
async def async_device_tracker_scan(now: dt_util.dt.datetime):
"""Handle interval matches."""
if update_lock.locked():
_LOGGER.warning(
"Updating device list from %s took longer than the scheduled "
"scan interval %s", platform, interval)
return
async with update_lock:
found_devices = await scanner.async_scan_devices()
for mac in found_devices:
if mac in seen:
host_name = None
else:
host_name = await scanner.async_get_device_name(mac)
seen.add(mac)
try:
extra_attributes = \
await scanner.async_get_extra_attributes(mac)
except NotImplementedError:
extra_attributes = dict()
kwargs = {
'mac': mac,
'host_name': host_name,
'source_type': SOURCE_TYPE_ROUTER,
'attributes': {
'scanner': scanner.__class__.__name__,
**extra_attributes
}
}
zone_home = hass.states.get(zone.ENTITY_ID_HOME)
if zone_home:
kwargs['gps'] = [zone_home.attributes[ATTR_LATITUDE],
zone_home.attributes[ATTR_LONGITUDE]]
kwargs['gps_accuracy'] = 0
hass.async_create_task(async_see_device(**kwargs))
async_track_time_interval(hass, async_device_tracker_scan, interval)
hass.async_create_task(async_device_tracker_scan(None))
def update_config(path: str, dev_id: str, device: Device):
"""Add device to YAML configuration file."""
with open(path, 'a') as out:
device = {device.dev_id: {
ATTR_NAME: device.name,
ATTR_MAC: device.mac,
ATTR_ICON: device.icon,
'picture': device.config_picture,
'track': device.track,
CONF_AWAY_HIDE: device.away_hide,
}}
out.write('\n')
out.write(dump(device))
def get_gravatar_for_email(email: str):
"""Return an 80px Gravatar for the given email address.
Async friendly.
"""
import hashlib
url = 'https://www.gravatar.com/avatar/{}.jpg?s=80&d=wavatar'
return url.format(hashlib.md5(email.encode('utf-8').lower()).hexdigest())
return await hass.data[DOMAIN](entry)

40
homeassistant/components/device_tracker/const.py Normal file
View File

@ -0,0 +1,40 @@
"""Device tracker constants."""
from datetime import timedelta
import logging
LOGGER = logging.getLogger(__package__)
DOMAIN = 'device_tracker'
ENTITY_ID_FORMAT = DOMAIN + '.{}'
PLATFORM_TYPE_LEGACY = 'legacy'
PLATFORM_TYPE_ENTITY = 'entity_platform'
SOURCE_TYPE_GPS = 'gps'
SOURCE_TYPE_ROUTER = 'router'
SOURCE_TYPE_BLUETOOTH = 'bluetooth'
SOURCE_TYPE_BLUETOOTH_LE = 'bluetooth_le'
CONF_SCAN_INTERVAL = 'interval_seconds'
SCAN_INTERVAL = timedelta(seconds=12)
CONF_TRACK_NEW = 'track_new_devices'
DEFAULT_TRACK_NEW = True
CONF_AWAY_HIDE = 'hide_if_away'
DEFAULT_AWAY_HIDE = False
CONF_CONSIDER_HOME = 'consider_home'
DEFAULT_CONSIDER_HOME = timedelta(seconds=180)
CONF_NEW_DEVICE_DEFAULTS = 'new_device_defaults'
ATTR_ATTRIBUTES = 'attributes'
ATTR_BATTERY = 'battery'
ATTR_DEV_ID = 'dev_id'
ATTR_GPS = 'gps'
ATTR_HOST_NAME = 'host_name'
ATTR_LOCATION_NAME = 'location_name'
ATTR_MAC = 'mac'
ATTR_SOURCE_TYPE = 'source_type'
ATTR_CONSIDER_HOME = 'consider_home'

528
homeassistant/components/device_tracker/legacy.py Normal file
View File

@ -0,0 +1,528 @@
"""Legacy device tracker classes."""
import asyncio
from datetime import timedelta
from typing import Any, List, Sequence
import voluptuous as vol
from homeassistant.core import callback
from homeassistant.components import zone
from homeassistant.components.group import (
ATTR_ADD_ENTITIES, ATTR_ENTITIES, ATTR_OBJECT_ID, ATTR_VISIBLE,
DOMAIN as DOMAIN_GROUP, SERVICE_SET)
from homeassistant.components.zone.zone import async_active_zone
from homeassistant.config import load_yaml_config_file, async_log_exception
from homeassistant.exceptions import HomeAssistantError
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import GPSType, HomeAssistantType
from homeassistant import util
import homeassistant.util.dt as dt_util
from homeassistant.util.yaml import dump
from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_GPS_ACCURACY, ATTR_ICON, ATTR_LATITUDE,
ATTR_LONGITUDE, ATTR_NAME, CONF_ICON, CONF_MAC, CONF_NAME,
DEVICE_DEFAULT_NAME, STATE_NOT_HOME, STATE_HOME)
from .const import (
ATTR_BATTERY,
ATTR_HOST_NAME,
ATTR_MAC,
ATTR_SOURCE_TYPE,
CONF_AWAY_HIDE,
CONF_CONSIDER_HOME,
CONF_NEW_DEVICE_DEFAULTS,
CONF_TRACK_NEW,
DEFAULT_AWAY_HIDE,
DEFAULT_CONSIDER_HOME,
DEFAULT_TRACK_NEW,
DOMAIN,
ENTITY_ID_FORMAT,
LOGGER,
SOURCE_TYPE_GPS,
)
YAML_DEVICES = 'known_devices.yaml'
GROUP_NAME_ALL_DEVICES = 'all devices'
EVENT_NEW_DEVICE = 'device_tracker_new_device'
async def get_tracker(hass, config):
"""Create a tracker."""
yaml_path = hass.config.path(YAML_DEVICES)
conf = config.get(DOMAIN, [])
conf = conf[0] if conf else {}
consider_home = conf.get(CONF_CONSIDER_HOME, DEFAULT_CONSIDER_HOME)
defaults = conf.get(CONF_NEW_DEVICE_DEFAULTS, {})
track_new = conf.get(CONF_TRACK_NEW)
if track_new is None:
track_new = defaults.get(CONF_TRACK_NEW, DEFAULT_TRACK_NEW)
devices = await async_load_config(yaml_path, hass, consider_home)
tracker = DeviceTracker(
hass, consider_home, track_new, defaults, devices)
return tracker
class DeviceTracker:
"""Representation of a device tracker."""
def __init__(self, hass: HomeAssistantType, consider_home: timedelta,
track_new: bool, defaults: dict,
devices: Sequence) -> None:
"""Initialize a device tracker."""
self.hass = hass
self.devices = {dev.dev_id: dev for dev in devices}
self.mac_to_dev = {dev.mac: dev for dev in devices if dev.mac}
self.consider_home = consider_home
self.track_new = track_new if track_new is not None \
else defaults.get(CONF_TRACK_NEW, DEFAULT_TRACK_NEW)
self.defaults = defaults
self.group = None
self._is_updating = asyncio.Lock(loop=hass.loop)
for dev in devices:
if self.devices[dev.dev_id] is not dev:
LOGGER.warning('Duplicate device IDs detected %s', dev.dev_id)
if dev.mac and self.mac_to_dev[dev.mac] is not dev:
LOGGER.warning('Duplicate device MAC addresses detected %s',
dev.mac)
def see(self, mac: str = None, dev_id: str = None, host_name: str = None,
location_name: str = None, gps: GPSType = None,
gps_accuracy: int = None, battery: int = None,
attributes: dict = None, source_type: str = SOURCE_TYPE_GPS,
picture: str = None, icon: str = None,
consider_home: timedelta = None):
"""Notify the device tracker that you see a device."""
self.hass.add_job(
self.async_see(mac, dev_id, host_name, location_name, gps,
gps_accuracy, battery, attributes, source_type,
picture, icon, consider_home)
)
async def async_see(
self, mac: str = None, dev_id: str = None, host_name: str = None,
location_name: str = None, gps: GPSType = None,
gps_accuracy: int = None, battery: int = None,
attributes: dict = None, source_type: str = SOURCE_TYPE_GPS,
picture: str = None, icon: str = None,
consider_home: timedelta = None):
"""Notify the device tracker that you see a device.
This method is a coroutine.
"""
if mac is None and dev_id is None:
raise HomeAssistantError('Neither mac or device id passed in')
if mac is not None:
mac = str(mac).upper()
device = self.mac_to_dev.get(mac)
if not device:
dev_id = util.slugify(host_name or '') or util.slugify(mac)
else:
dev_id = cv.slug(str(dev_id).lower())
device = self.devices.get(dev_id)
if device:
await device.async_seen(
host_name, location_name, gps, gps_accuracy, battery,
attributes, source_type, consider_home)
if device.track:
await device.async_update_ha_state()
return
# If no device can be found, create it
dev_id = util.ensure_unique_string(dev_id, self.devices.keys())
device = Device(
self.hass, consider_home or self.consider_home, self.track_new,
dev_id, mac, (host_name or dev_id).replace('_', ' '),
picture=picture, icon=icon,
hide_if_away=self.defaults.get(CONF_AWAY_HIDE, DEFAULT_AWAY_HIDE))
self.devices[dev_id] = device
if mac is not None:
self.mac_to_dev[mac] = device
await device.async_seen(
host_name, location_name, gps, gps_accuracy, battery, attributes,
source_type)
if device.track:
await device.async_update_ha_state()
# During init, we ignore the group
if self.group and self.track_new:
self.hass.async_create_task(
self.hass.async_call(
DOMAIN_GROUP, SERVICE_SET, {
ATTR_OBJECT_ID: util.slugify(GROUP_NAME_ALL_DEVICES),
ATTR_VISIBLE: False,
ATTR_NAME: GROUP_NAME_ALL_DEVICES,
ATTR_ADD_ENTITIES: [device.entity_id]}))
self.hass.bus.async_fire(EVENT_NEW_DEVICE, {
ATTR_ENTITY_ID: device.entity_id,
ATTR_HOST_NAME: device.host_name,
ATTR_MAC: device.mac,
})
# update known_devices.yaml
self.hass.async_create_task(
self.async_update_config(
self.hass.config.path(YAML_DEVICES), dev_id, device)
)
async def async_update_config(self, path, dev_id, device):
"""Add device to YAML configuration file.
This method is a coroutine.
"""
async with self._is_updating:
await self.hass.async_add_executor_job(
update_config, self.hass.config.path(YAML_DEVICES),
dev_id, device)
@callback
def async_setup_group(self):
"""Initialize group for all tracked devices.
This method must be run in the event loop.
"""
entity_ids = [dev.entity_id for dev in self.devices.values()
if dev.track]
self.hass.async_create_task(
self.hass.services.async_call(
DOMAIN_GROUP, SERVICE_SET, {
ATTR_OBJECT_ID: util.slugify(GROUP_NAME_ALL_DEVICES),
ATTR_VISIBLE: False,
ATTR_NAME: GROUP_NAME_ALL_DEVICES,
ATTR_ENTITIES: entity_ids}))
@callback
def async_update_stale(self, now: dt_util.dt.datetime):
"""Update stale devices.
This method must be run in the event loop.
"""
for device in self.devices.values():
if (device.track and device.last_update_home) and \
device.stale(now):
self.hass.async_create_task(device.async_update_ha_state(True))
async def async_setup_tracked_device(self):
"""Set up all not exists tracked devices.
This method is a coroutine.
"""
async def async_init_single_device(dev):
"""Init a single device_tracker entity."""
await dev.async_added_to_hass()
await dev.async_update_ha_state()
tasks = []
for device in self.devices.values():
if device.track and not device.last_seen:
tasks.append(self.hass.async_create_task(
async_init_single_device(device)))
if tasks:
await asyncio.wait(tasks, loop=self.hass.loop)
class Device(RestoreEntity):
"""Represent a tracked device."""
host_name = None # type: str
location_name = None # type: str
gps = None # type: GPSType
gps_accuracy = 0 # type: int
last_seen = None # type: dt_util.dt.datetime
consider_home = None # type: dt_util.dt.timedelta
battery = None # type: int
attributes = None # type: dict
icon = None # type: str
# Track if the last update of this device was HOME.
last_update_home = False
_state = STATE_NOT_HOME
def __init__(self, hass: HomeAssistantType, consider_home: timedelta,
track: bool, dev_id: str, mac: str, name: str = None,
picture: str = None, gravatar: str = None, icon: str = None,
hide_if_away: bool = False) -> None:
"""Initialize a device."""
self.hass = hass
self.entity_id = ENTITY_ID_FORMAT.format(dev_id)
# Timedelta object how long we consider a device home if it is not
# detected anymore.
self.consider_home = consider_home
# Device ID
self.dev_id = dev_id
self.mac = mac
# If we should track this device
self.track = track
# Configured name
self.config_name = name
# Configured picture
if gravatar is not None:
self.config_picture = get_gravatar_for_email(gravatar)
else:
self.config_picture = picture
self.icon = icon
self.away_hide = hide_if_away
self.source_type = None
self._attributes = {}
@property
def name(self):
"""Return the name of the entity."""
return self.config_name or self.host_name or DEVICE_DEFAULT_NAME
@property
def state(self):
"""Return the state of the device."""
return self._state
@property
def entity_picture(self):
"""Return the picture of the device."""
return self.config_picture
@property
def state_attributes(self):
"""Return the device state attributes."""
attr = {
ATTR_SOURCE_TYPE: self.source_type
}
if self.gps:
attr[ATTR_LATITUDE] = self.gps[0]
attr[ATTR_LONGITUDE] = self.gps[1]
attr[ATTR_GPS_ACCURACY] = self.gps_accuracy
if self.battery:
attr[ATTR_BATTERY] = self.battery
return attr
@property
def device_state_attributes(self):
"""Return device state attributes."""
return self._attributes
@property
def hidden(self):
"""If device should be hidden."""
return self.away_hide and self.state != STATE_HOME
async def async_seen(
self, host_name: str = None, location_name: str = None,
gps: GPSType = None, gps_accuracy=0, battery: int = None,
attributes: dict = None,
source_type: str = SOURCE_TYPE_GPS,
consider_home: timedelta = None):
"""Mark the device as seen."""
self.source_type = source_type
self.last_seen = dt_util.utcnow()
self.host_name = host_name
self.location_name = location_name
self.consider_home = consider_home or self.consider_home
if battery:
self.battery = battery
if attributes:
self._attributes.update(attributes)
self.gps = None
if gps is not None:
try:
self.gps = float(gps[0]), float(gps[1])
self.gps_accuracy = gps_accuracy or 0
except (ValueError, TypeError, IndexError):
self.gps = None
self.gps_accuracy = 0
LOGGER.warning(
"Could not parse gps value for %s: %s", self.dev_id, gps)
# pylint: disable=not-an-iterable
await self.async_update()
def stale(self, now: dt_util.dt.datetime = None):
"""Return if device state is stale.
Async friendly.
"""
return self.last_seen is None or \
(now or dt_util.utcnow()) - self.last_seen > self.consider_home
def mark_stale(self):
"""Mark the device state as stale."""
self._state = STATE_NOT_HOME
self.gps = None
self.last_update_home = False
async def async_update(self):
"""Update state of entity.
This method is a coroutine.
"""
if not self.last_seen:
return
if self.location_name:
self._state = self.location_name
elif self.gps is not None and self.source_type == SOURCE_TYPE_GPS:
zone_state = async_active_zone(
self.hass, self.gps[0], self.gps[1], self.gps_accuracy)
if zone_state is None:
self._state = STATE_NOT_HOME
elif zone_state.entity_id == zone.ENTITY_ID_HOME:
self._state = STATE_HOME
else:
self._state = zone_state.name
elif self.stale():
self.mark_stale()
else:
self._state = STATE_HOME
self.last_update_home = True
async def async_added_to_hass(self):
"""Add an entity."""
await super().async_added_to_hass()
state = await self.async_get_last_state()
if not state:
return
self._state = state.state
self.last_update_home = (state.state == STATE_HOME)
self.last_seen = dt_util.utcnow()
for attr, var in (
(ATTR_SOURCE_TYPE, 'source_type'),
(ATTR_GPS_ACCURACY, 'gps_accuracy'),
(ATTR_BATTERY, 'battery'),
):
if attr in state.attributes:
setattr(self, var, state.attributes[attr])
if ATTR_LONGITUDE in state.attributes:
self.gps = (state.attributes[ATTR_LATITUDE],
state.attributes[ATTR_LONGITUDE])
class DeviceScanner:
"""Device scanner object."""
hass = None # type: HomeAssistantType
def scan_devices(self) -> List[str]:
"""Scan for devices."""
raise NotImplementedError()
def async_scan_devices(self) -> Any:
"""Scan for devices.
This method must be run in the event loop and returns a coroutine.
"""
return self.hass.async_add_job(self.scan_devices)
def get_device_name(self, device: str) -> str:
"""Get the name of a device."""
raise NotImplementedError()
def async_get_device_name(self, device: str) -> Any:
"""Get the name of a device.
This method must be run in the event loop and returns a coroutine.
"""
return self.hass.async_add_job(self.get_device_name, device)
def get_extra_attributes(self, device: str) -> dict:
"""Get the extra attributes of a device."""
raise NotImplementedError()
def async_get_extra_attributes(self, device: str) -> Any:
"""Get the extra attributes of a device.
This method must be run in the event loop and returns a coroutine.
"""
return self.hass.async_add_job(self.get_extra_attributes, device)
async def async_load_config(path: str, hass: HomeAssistantType,
consider_home: timedelta):
"""Load devices from YAML configuration file.
This method is a coroutine.
"""
dev_schema = vol.Schema({
vol.Required(CONF_NAME): cv.string,
vol.Optional(CONF_ICON, default=None): vol.Any(None, cv.icon),
vol.Optional('track', default=False): cv.boolean,
vol.Optional(CONF_MAC, default=None):
vol.Any(None, vol.All(cv.string, vol.Upper)),
vol.Optional(CONF_AWAY_HIDE, default=DEFAULT_AWAY_HIDE): cv.boolean,
vol.Optional('gravatar', default=None): vol.Any(None, cv.string),
vol.Optional('picture', default=None): vol.Any(None, cv.string),
vol.Optional(CONF_CONSIDER_HOME, default=consider_home): vol.All(
cv.time_period, cv.positive_timedelta),
})
try:
result = []
try:
devices = await hass.async_add_job(
load_yaml_config_file, path)
except HomeAssistantError as err:
LOGGER.error("Unable to load %s: %s", path, str(err))
return []
for dev_id, device in devices.items():
# Deprecated option. We just ignore it to avoid breaking change
device.pop('vendor', None)
try:
device = dev_schema(device)
device['dev_id'] = cv.slugify(dev_id)
except vol.Invalid as exp:
async_log_exception(exp, dev_id, devices, hass)
else:
result.append(Device(hass, **device))
return result
except (HomeAssistantError, FileNotFoundError):
# When YAML file could not be loaded/did not contain a dict
return []
def update_config(path: str, dev_id: str, device: Device):
"""Add device to YAML configuration file."""
with open(path, 'a') as out:
device = {device.dev_id: {
ATTR_NAME: device.name,
ATTR_MAC: device.mac,
ATTR_ICON: device.icon,
'picture': device.config_picture,
'track': device.track,
CONF_AWAY_HIDE: device.away_hide,
}}
out.write('\n')
out.write(dump(device))
def get_gravatar_for_email(email: str):
"""Return an 80px Gravatar for the given email address.
Async friendly.
"""
import hashlib
url = 'https://www.gravatar.com/avatar/{}.jpg?s=80&d=wavatar'
return url.format(hashlib.md5(email.encode('utf-8').lower()).hexdigest())

199
homeassistant/components/device_tracker/setup.py Normal file
View File

@ -0,0 +1,199 @@
"""Device tracker helpers."""
import asyncio
from typing import Dict, Any, Callable, Optional
from types import ModuleType
import attr
from homeassistant.core import callback
from homeassistant.setup import async_prepare_setup_platform
from homeassistant.helpers import config_per_platform
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.util import dt as dt_util
from homeassistant.const import (
ATTR_LATITUDE,
ATTR_LONGITUDE,
)
from .const import (
DOMAIN,
PLATFORM_TYPE_ENTITY,
PLATFORM_TYPE_LEGACY,
CONF_SCAN_INTERVAL,
SCAN_INTERVAL,
SOURCE_TYPE_ROUTER,
LOGGER,
)
@attr.s
class DeviceTrackerPlatform:
"""Class to hold platform information."""
LEGACY_SETUP = (
'async_get_scanner',
'get_scanner',
'async_setup_scanner',
'setup_scanner',
# Small steps, initially just legacy setup supported.
'async_setup_entry'
)
# ENTITY_PLATFORM_SETUP = (
# 'setup_platform',
# 'async_setup_platform',
# 'async_setup_entry'
# )
name = attr.ib(type=str)
platform = attr.ib(type=ModuleType)
config = attr.ib(type=Dict)
@property
def type(self):
"""Return platform type."""
for methods, platform_type in (
(self.LEGACY_SETUP, PLATFORM_TYPE_LEGACY),
# (self.ENTITY_PLATFORM_SETUP, PLATFORM_TYPE_ENTITY),
):
for meth in methods:
if hasattr(self.platform, meth):
return platform_type
return None
async def async_setup_legacy(self, hass, tracker, discovery_info=None):
"""Set up a legacy platform."""
LOGGER.info("Setting up %s.%s", DOMAIN, self.type)
try:
scanner = None
setup = None
if hasattr(self.platform, 'async_get_scanner'):
scanner = await self.platform.async_get_scanner(
hass, {DOMAIN: self.config})
elif hasattr(self.platform, 'get_scanner'):
scanner = await hass.async_add_job(
self.platform.get_scanner, hass, {DOMAIN: self.config})
elif hasattr(self.platform, 'async_setup_scanner'):
setup = await self.platform.async_setup_scanner(
hass, self.config, tracker.async_see, discovery_info)
elif hasattr(self.platform, 'setup_scanner'):
setup = await hass.async_add_job(
self.platform.setup_scanner, hass, self.config,
tracker.see, discovery_info)
elif hasattr(self.platform, 'async_setup_entry'):
setup = await self.platform.async_setup_entry(
hass, self.config, tracker.async_see)
else:
raise HomeAssistantError(
"Invalid legacy device_tracker platform.")
if scanner:
async_setup_scanner_platform(
hass, self.config, scanner, tracker.async_see, self.type)
return
if not setup:
LOGGER.error("Error setting up platform %s", self.type)
return
except Exception: # pylint: disable=broad-except
LOGGER.exception("Error setting up platform %s", self.type)
async def async_extract_config(hass, config):
"""Extract device tracker config and split between legacy and modern."""
legacy = []
entity_platform = []
for platform in await asyncio.gather(*[
async_create_platform_type(hass, config, p_type, p_config)
for p_type, p_config in config_per_platform(config, DOMAIN)
]):
if platform is None:
continue
if platform.type == PLATFORM_TYPE_ENTITY:
entity_platform.append(platform)
elif platform.type == PLATFORM_TYPE_LEGACY:
legacy.append(platform)
else:
raise ValueError("Unable to determine type for {}: {}".format(
platform.name, platform.type))
return (legacy, entity_platform)
async def async_create_platform_type(hass, config, p_type, p_config) \
-> Optional[DeviceTrackerPlatform]:
"""Determine type of platform."""
platform = await async_prepare_setup_platform(
hass, config, DOMAIN, p_type)
if platform is None:
return None
return DeviceTrackerPlatform(p_type, platform, p_config)
@callback
def async_setup_scanner_platform(hass: HomeAssistantType, config: ConfigType,
scanner: Any, async_see_device: Callable,
platform: str):
"""Set up the connect scanner-based platform to device tracker.
This method must be run in the event loop.
"""
interval = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL)
update_lock = asyncio.Lock(loop=hass.loop)
scanner.hass = hass
# Initial scan of each mac we also tell about host name for config
seen = set() # type: Any
async def async_device_tracker_scan(now: dt_util.dt.datetime):
"""Handle interval matches."""
if update_lock.locked():
LOGGER.warning(
"Updating device list from %s took longer than the scheduled "
"scan interval %s", platform, interval)
return
async with update_lock:
found_devices = await scanner.async_scan_devices()
for mac in found_devices:
if mac in seen:
host_name = None
else:
host_name = await scanner.async_get_device_name(mac)
seen.add(mac)
try:
extra_attributes = \
await scanner.async_get_extra_attributes(mac)
except NotImplementedError:
extra_attributes = dict()
kwargs = {
'mac': mac,
'host_name': host_name,
'source_type': SOURCE_TYPE_ROUTER,
'attributes': {
'scanner': scanner.__class__.__name__,
**extra_attributes
}
}
zone_home = hass.states.get(hass.components.zone.ENTITY_ID_HOME)
if zone_home:
kwargs['gps'] = [zone_home.attributes[ATTR_LATITUDE],
zone_home.attributes[ATTR_LONGITUDE]]
kwargs['gps_accuracy'] = 0
hass.async_create_task(async_see_device(**kwargs))
async_track_time_interval(hass, async_device_tracker_scan, interval)
hass.async_create_task(async_device_tracker_scan(None))

6
homeassistant/components/icloud/device_tracker.py
View File

@ -6,8 +6,10 @@ import os
import voluptuous as vol
from homeassistant.const import CONF_USERNAME, CONF_PASSWORD
from homeassistant.components.device_tracker import (
PLATFORM_SCHEMA, DOMAIN, ATTR_ATTRIBUTES, ENTITY_ID_FORMAT, DeviceScanner)
from homeassistant.components.device_tracker import PLATFORM_SCHEMA
from homeassistant.components.device_tracker.const import (
DOMAIN, ATTR_ATTRIBUTES, ENTITY_ID_FORMAT)
from homeassistant.components.device_tracker.legacy import DeviceScanner
from homeassistant.components.zone.zone import active_zone
from homeassistant.helpers.event import track_utc_time_change
import homeassistant.helpers.config_validation as cv

7
homeassistant/components/ping/device_tracker.py
View File

@ -8,8 +8,9 @@ import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.components.device_tracker import (
PLATFORM_SCHEMA, CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL,
SOURCE_TYPE_ROUTER)
PLATFORM_SCHEMA)
from homeassistant.components.device_tracker.const import (
CONF_SCAN_INTERVAL, SCAN_INTERVAL, SOURCE_TYPE_ROUTER)
from homeassistant import util
from homeassistant import const
@ -68,7 +69,7 @@ def setup_scanner(hass, config, see, discovery_info=None):
interval = config.get(CONF_SCAN_INTERVAL,
timedelta(seconds=len(hosts) *
config[CONF_PING_COUNT])
+ DEFAULT_SCAN_INTERVAL)
+ SCAN_INTERVAL)
_LOGGER.debug("Started ping tracker with interval=%s on hosts: %s",
interval, ",".join([host.ip_address for host in hosts]))

14
homeassistant/config.py
View File

@ -827,14 +827,22 @@ async def async_process_component_config(
# Create a copy of the configuration with all config for current
# component removed and add validated config back in.
filter_keys = extract_domain_configs(config, domain)
config = {key: value for key, value in config.items()
if key not in filter_keys}
config = config_without_domain(config, domain)
config[domain] = platforms
return config
@callback
def config_without_domain(config: Dict, domain: str) -> Dict:
"""Return a config with all configuration for a domain removed."""
filter_keys = extract_domain_configs(config, domain)
return {
key: value for key, value in config.items()
if key not in filter_keys
}
async def async_check_ha_config_file(hass: HomeAssistant) -> Optional[str]:
"""Check if Home Assistant configuration file is valid.

5
tests/components/demo/test_init.py
View File

@ -5,7 +5,8 @@ import os
import pytest
from homeassistant.setup import async_setup_component
from homeassistant.components import demo, device_tracker
from homeassistant.components import demo
from homeassistant.components.device_tracker.legacy import YAML_DEVICES
from homeassistant.helpers.json import JSONEncoder
@ -20,7 +21,7 @@ def demo_cleanup(hass):
"""Clean up device tracker demo file."""
yield
try:
os.remove(hass.config.path(device_tracker.YAML_DEVICES))
os.remove(hass.config.path(YAML_DEVICES))
except FileNotFoundError:
pass

6
tests/components/device_sun_light_trigger/test_init.py
View File

@ -8,6 +8,8 @@ from homeassistant.setup import async_setup_component
from homeassistant.const import CONF_PLATFORM, STATE_HOME, STATE_NOT_HOME
from homeassistant.components import (
device_tracker, light, device_sun_light_trigger)
from homeassistant.components.device_tracker.const import (
ENTITY_ID_FORMAT as DT_ENTITY_ID_FORMAT)
from homeassistant.util import dt as dt_util
from tests.common import async_fire_time_changed
@ -26,7 +28,7 @@ def scanner(hass):
getattr(hass.components, 'test.light').init()
with patch(
'homeassistant.components.device_tracker.load_yaml_config_file',
'homeassistant.components.device_tracker.legacy.load_yaml_config_file',
return_value={
'device_1': {
'hide_if_away': False,
@ -102,7 +104,7 @@ async def test_lights_turn_on_when_coming_home_after_sun_set(hass, scanner):
device_sun_light_trigger.DOMAIN: {}})
hass.states.async_set(
device_tracker.ENTITY_ID_FORMAT.format('device_2'), STATE_HOME)
DT_ENTITY_ID_FORMAT.format('device_2'), STATE_HOME)
await hass.async_block_till_done()
assert light.is_on(hass)

113
tests/components/device_tracker/test_init.py
View File

@ -10,9 +10,11 @@ import pytest
from homeassistant.components import zone
import homeassistant.components.device_tracker as device_tracker
from homeassistant.components.device_tracker import const, legacy
from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_ENTITY_PICTURE, ATTR_FRIENDLY_NAME, ATTR_HIDDEN,
ATTR_ICON, CONF_PLATFORM, STATE_HOME, STATE_NOT_HOME)
ATTR_ICON, CONF_PLATFORM, STATE_HOME, STATE_NOT_HOME,
ATTR_LATITUDE, ATTR_LONGITUDE, ATTR_GPS_ACCURACY)
from homeassistant.core import State, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import discovery
@ -33,7 +35,7 @@ _LOGGER = logging.getLogger(__name__)
@pytest.fixture(name='yaml_devices')
def mock_yaml_devices(hass):
"""Get a path for storing yaml devices."""
yaml_devices = hass.config.path(device_tracker.YAML_DEVICES)
yaml_devices = hass.config.path(legacy.YAML_DEVICES)
if os.path.isfile(yaml_devices):
os.remove(yaml_devices)
yield yaml_devices
@ -43,7 +45,7 @@ def mock_yaml_devices(hass):
async def test_is_on(hass):
"""Test is_on method."""
entity_id = device_tracker.ENTITY_ID_FORMAT.format('test')
entity_id = const.ENTITY_ID_FORMAT.format('test')
hass.states.async_set(entity_id, STATE_HOME)
@ -65,21 +67,21 @@ async def test_reading_broken_yaml_config(hass):
'bad_device:\n nme: Device')}
args = {'hass': hass, 'consider_home': timedelta(seconds=60)}
with patch_yaml_files(files):
assert await device_tracker.async_load_config(
assert await legacy.async_load_config(
'empty.yaml', **args) == []
assert await device_tracker.async_load_config(
assert await legacy.async_load_config(
'nodict.yaml', **args) == []
assert await device_tracker.async_load_config(
assert await legacy.async_load_config(
'noname.yaml', **args) == []
assert await device_tracker.async_load_config(
assert await legacy.async_load_config(
'badkey.yaml', **args) == []
res = await device_tracker.async_load_config('allok.yaml', **args)
res = await legacy.async_load_config('allok.yaml', **args)
assert len(res) == 1
assert res[0].name == 'Device'
assert res[0].dev_id == 'my_device'
res = await device_tracker.async_load_config('oneok.yaml', **args)
res = await legacy.async_load_config('oneok.yaml', **args)
assert len(res) == 1
assert res[0].name == 'Device'
assert res[0].dev_id == 'my_device'
@ -88,17 +90,16 @@ async def test_reading_broken_yaml_config(hass):
async def test_reading_yaml_config(hass, yaml_devices):
"""Test the rendering of the YAML configuration."""
dev_id = 'test'
device = device_tracker.Device(
device = legacy.Device(
hass, timedelta(seconds=180), True, dev_id,
'AB:CD:EF:GH:IJ', 'Test name', picture='http://test.picture',
hide_if_away=True, icon='mdi:kettle')
await hass.async_add_executor_job(
device_tracker.update_config, yaml_devices, dev_id, device)
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN,
TEST_PLATFORM)
config = (await device_tracker.async_load_config(yaml_devices, hass,
device.consider_home))[0]
legacy.update_config, yaml_devices, dev_id, device)
assert await async_setup_component(hass, device_tracker.DOMAIN,
TEST_PLATFORM)
config = (await legacy.async_load_config(yaml_devices, hass,
device.consider_home))[0]
assert device.dev_id == config.dev_id
assert device.track == config.track
assert device.mac == config.mac
@ -108,15 +109,15 @@ async def test_reading_yaml_config(hass, yaml_devices):
assert device.icon == config.icon
@patch('homeassistant.components.device_tracker._LOGGER.warning')
@patch('homeassistant.components.device_tracker.const.LOGGER.warning')
async def test_duplicate_mac_dev_id(mock_warning, hass):
"""Test adding duplicate MACs or device IDs to DeviceTracker."""
devices = [
device_tracker.Device(hass, True, True, 'my_device', 'AB:01',
'My device', None, None, False),
device_tracker.Device(hass, True, True, 'your_device',
'AB:01', 'Your device', None, None, False)]
device_tracker.DeviceTracker(hass, False, True, {}, devices)
legacy.Device(hass, True, True, 'my_device', 'AB:01',
'My device', None, None, False),
legacy.Device(hass, True, True, 'your_device',
'AB:01', 'Your device', None, None, False)]
legacy.DeviceTracker(hass, False, True, {}, devices)
_LOGGER.debug(mock_warning.call_args_list)
assert mock_warning.call_count == 1, \
"The only warning call should be duplicates (check DEBUG)"
@ -126,11 +127,11 @@ async def test_duplicate_mac_dev_id(mock_warning, hass):
mock_warning.reset_mock()
devices = [
device_tracker.Device(hass, True, True, 'my_device',
'AB:01', 'My device', None, None, False),
device_tracker.Device(hass, True, True, 'my_device',
None, 'Your device', None, None, False)]
device_tracker.DeviceTracker(hass, False, True, {}, devices)
legacy.Device(hass, True, True, 'my_device',
'AB:01', 'My device', None, None, False),
legacy.Device(hass, True, True, 'my_device',
None, 'Your device', None, None, False)]
legacy.DeviceTracker(hass, False, True, {}, devices)
_LOGGER.debug(mock_warning.call_args_list)
assert mock_warning.call_count == 1, \
@ -150,7 +151,7 @@ async def test_setup_without_yaml_file(hass):
async def test_gravatar(hass):
"""Test the Gravatar generation."""
dev_id = 'test'
device = device_tracker.Device(
device = legacy.Device(
hass, timedelta(seconds=180), True, dev_id,
'AB:CD:EF:GH:IJ', 'Test name', gravatar='test@example.com')
gravatar_url = ("https://www.gravatar.com/avatar/"
@ -161,7 +162,7 @@ async def test_gravatar(hass):
async def test_gravatar_and_picture(hass):
"""Test that Gravatar overrides picture."""
dev_id = 'test'
device = device_tracker.Device(
device = legacy.Device(
hass, timedelta(seconds=180), True, dev_id,
'AB:CD:EF:GH:IJ', 'Test name', picture='http://test.picture',
gravatar='test@example.com')
@ -171,7 +172,7 @@ async def test_gravatar_and_picture(hass):
@patch(
'homeassistant.components.device_tracker.DeviceTracker.see')
'homeassistant.components.device_tracker.legacy.DeviceTracker.see')
@patch(
'homeassistant.components.demo.device_tracker.setup_scanner',
autospec=True)
@ -196,7 +197,7 @@ async def test_update_stale(hass, mock_device_tracker_conf):
register_time = datetime(2015, 9, 15, 23, tzinfo=dt_util.UTC)
scan_time = datetime(2015, 9, 15, 23, 1, tzinfo=dt_util.UTC)
with patch('homeassistant.components.device_tracker.dt_util.utcnow',
with patch('homeassistant.components.device_tracker.legacy.dt_util.utcnow',
return_value=register_time):
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN, {
@ -211,7 +212,7 @@ async def test_update_stale(hass, mock_device_tracker_conf):
scanner.leave_home('DEV1')
with patch('homeassistant.components.device_tracker.dt_util.utcnow',
with patch('homeassistant.components.device_tracker.legacy.dt_util.utcnow',
return_value=scan_time):
async_fire_time_changed(hass, scan_time)
await hass.async_block_till_done()
@ -224,12 +225,12 @@ async def test_entity_attributes(hass, mock_device_tracker_conf):
"""Test the entity attributes."""
devices = mock_device_tracker_conf
dev_id = 'test_entity'
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
entity_id = const.ENTITY_ID_FORMAT.format(dev_id)
friendly_name = 'Paulus'
picture = 'http://placehold.it/200x200'
icon = 'mdi:kettle'
device = device_tracker.Device(
device = legacy.Device(
hass, timedelta(seconds=180), True, dev_id, None,
friendly_name, picture, hide_if_away=True, icon=icon)
devices.append(device)
@ -249,8 +250,8 @@ async def test_device_hidden(hass, mock_device_tracker_conf):
"""Test hidden devices."""
devices = mock_device_tracker_conf
dev_id = 'test_entity'
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
device = device_tracker.Device(
entity_id = const.ENTITY_ID_FORMAT.format(dev_id)
device = legacy.Device(
hass, timedelta(seconds=180), True, dev_id, None,
hide_if_away=True)
devices.append(device)
@ -269,8 +270,8 @@ async def test_group_all_devices(hass, mock_device_tracker_conf):
"""Test grouping of devices."""
devices = mock_device_tracker_conf
dev_id = 'test_entity'
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
device = device_tracker.Device(
entity_id = const.ENTITY_ID_FORMAT.format(dev_id)
device = legacy.Device(
hass, timedelta(seconds=180), True, dev_id, None,
hide_if_away=True)
devices.append(device)
@ -288,7 +289,8 @@ async def test_group_all_devices(hass, mock_device_tracker_conf):
assert (entity_id,) == state.attributes.get(ATTR_ENTITY_ID)
@patch('homeassistant.components.device_tracker.DeviceTracker.async_see')
@patch('homeassistant.components.device_tracker.legacy.'
'DeviceTracker.async_see')
async def test_see_service(mock_see, hass):
"""Test the see service with a unicode dev_id and NO MAC."""
with assert_setup_component(1, device_tracker.DOMAIN):
@ -401,8 +403,8 @@ async def test_see_state(hass, yaml_devices):
common.async_see(hass, **params)
await hass.async_block_till_done()
config = await device_tracker.async_load_config(yaml_devices, hass,
timedelta(seconds=0))
config = await legacy.async_load_config(
yaml_devices, hass, timedelta(seconds=0))
assert len(config) == 1
state = hass.states.get('device_tracker.example_com')
@ -442,7 +444,7 @@ async def test_see_passive_zone_state(hass, mock_device_tracker_conf):
scanner.reset()
scanner.come_home('dev1')
with patch('homeassistant.components.device_tracker.dt_util.utcnow',
with patch('homeassistant.components.device_tracker.legacy.dt_util.utcnow',
return_value=register_time):
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN, {
@ -466,7 +468,7 @@ async def test_see_passive_zone_state(hass, mock_device_tracker_conf):
scanner.leave_home('dev1')
with patch('homeassistant.components.device_tracker.dt_util.utcnow',
with patch('homeassistant.components.device_tracker.legacy.dt_util.utcnow',
return_value=scan_time):
async_fire_time_changed(hass, scan_time)
await hass.async_block_till_done()
@ -484,11 +486,11 @@ async def test_see_passive_zone_state(hass, mock_device_tracker_conf):
device_tracker.SOURCE_TYPE_ROUTER
@patch('homeassistant.components.device_tracker._LOGGER.warning')
@patch('homeassistant.components.device_tracker.const.LOGGER.warning')
async def test_see_failures(mock_warning, hass, mock_device_tracker_conf):
"""Test that the device tracker see failures."""
devices = mock_device_tracker_conf
tracker = device_tracker.DeviceTracker(
tracker = legacy.DeviceTracker(
hass, timedelta(seconds=60), 0, {}, [])
# MAC is not a string (but added)
@ -512,16 +514,15 @@ async def test_see_failures(mock_warning, hass, mock_device_tracker_conf):
async def test_async_added_to_hass(hass):
"""Test restoring state."""
attr = {
device_tracker.ATTR_LONGITUDE: 18,
device_tracker.ATTR_LATITUDE: -33,
device_tracker.ATTR_LATITUDE: -33,
device_tracker.ATTR_SOURCE_TYPE: 'gps',
device_tracker.ATTR_GPS_ACCURACY: 2,
device_tracker.ATTR_BATTERY: 100
ATTR_LONGITUDE: 18,
ATTR_LATITUDE: -33,
const.ATTR_SOURCE_TYPE: 'gps',
ATTR_GPS_ACCURACY: 2,
const.ATTR_BATTERY: 100
}
mock_restore_cache(hass, [State('device_tracker.jk', 'home', attr)])
path = hass.config.path(device_tracker.YAML_DEVICES)
path = hass.config.path(legacy.YAML_DEVICES)
files = {
path: 'jk:\n name: JK Phone\n track: True',
@ -570,7 +571,7 @@ async def test_adding_unknown_device_to_config(mock_device_tracker_conf, hass):
async def test_picture_and_icon_on_see_discovery(mock_device_tracker_conf,
hass):
"""Test that picture and icon are set in initial see."""
tracker = device_tracker.DeviceTracker(
tracker = legacy.DeviceTracker(
hass, timedelta(seconds=60), False, {}, [])
await tracker.async_see(dev_id=11, picture='pic_url', icon='mdi:icon')
await hass.async_block_till_done()
@ -581,7 +582,7 @@ async def test_picture_and_icon_on_see_discovery(mock_device_tracker_conf,
async def test_default_hide_if_away_is_used(mock_device_tracker_conf, hass):
"""Test that default track_new is used."""
tracker = device_tracker.DeviceTracker(
tracker = legacy.DeviceTracker(
hass, timedelta(seconds=60), False,
{device_tracker.CONF_AWAY_HIDE: True}, [])
await tracker.async_see(dev_id=12)
@ -593,7 +594,7 @@ async def test_default_hide_if_away_is_used(mock_device_tracker_conf, hass):
async def test_backward_compatibility_for_track_new(mock_device_tracker_conf,
hass):
"""Test backward compatibility for track new."""
tracker = device_tracker.DeviceTracker(
tracker = legacy.DeviceTracker(
hass, timedelta(seconds=60), False,
{device_tracker.CONF_TRACK_NEW: True}, [])
await tracker.async_see(dev_id=13)
@ -604,7 +605,7 @@ async def test_backward_compatibility_for_track_new(mock_device_tracker_conf,
async def test_old_style_track_new_is_skipped(mock_device_tracker_conf, hass):
"""Test old style config is skipped."""
tracker = device_tracker.DeviceTracker(
tracker = legacy.DeviceTracker(
hass, timedelta(seconds=60), None,
{device_tracker.CONF_TRACK_NEW: False}, [])
await tracker.async_see(dev_id=14)

2
tests/components/geofency/test_init.py
View File

@ -125,7 +125,7 @@ async def geofency_client(loop, hass, aiohttp_client):
}})
await hass.async_block_till_done()
with patch('homeassistant.components.device_tracker.update_config'):
with patch('homeassistant.components.device_tracker.legacy.update_config'):
return await aiohttp_client(hass.http.app)

2
tests/components/gpslogger/test_init.py
View File

@ -38,7 +38,7 @@ async def gpslogger_client(loop, hass, aiohttp_client):
await hass.async_block_till_done()
with patch('homeassistant.components.device_tracker.update_config'):
with patch('homeassistant.components.device_tracker.legacy.update_config'):
return await aiohttp_client(hass.http.app)

2
tests/components/locative/test_init.py
View File

@ -30,7 +30,7 @@ async def locative_client(loop, hass, hass_client):
})
await hass.async_block_till_done()
with patch('homeassistant.components.device_tracker.update_config'):
with patch('homeassistant.components.device_tracker.legacy.update_config'):
return await hass_client()

11
tests/components/mqtt/test_device_tracker.py
View File

@ -3,6 +3,7 @@ from asynctest import patch
import pytest
from homeassistant.components import device_tracker
from homeassistant.components.device_tracker.const import ENTITY_ID_FORMAT
from homeassistant.const import CONF_PLATFORM
from homeassistant.setup import async_setup_component
@ -39,7 +40,7 @@ async def test_ensure_device_tracker_platform_validation(hass):
async def test_new_message(hass, mock_device_tracker_conf):
"""Test new message."""
dev_id = 'paulus'
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
entity_id = ENTITY_ID_FORMAT.format(dev_id)
topic = '/location/paulus'
location = 'work'
@ -58,7 +59,7 @@ async def test_new_message(hass, mock_device_tracker_conf):
async def test_single_level_wildcard_topic(hass, mock_device_tracker_conf):
"""Test single level wildcard topic."""
dev_id = 'paulus'
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
entity_id = ENTITY_ID_FORMAT.format(dev_id)
subscription = '/location/+/paulus'
topic = '/location/room/paulus'
location = 'work'
@ -78,7 +79,7 @@ async def test_single_level_wildcard_topic(hass, mock_device_tracker_conf):
async def test_multi_level_wildcard_topic(hass, mock_device_tracker_conf):
"""Test multi level wildcard topic."""
dev_id = 'paulus'
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
entity_id = ENTITY_ID_FORMAT.format(dev_id)
subscription = '/location/#'
topic = '/location/room/paulus'
location = 'work'
@ -99,7 +100,7 @@ async def test_single_level_wildcard_topic_not_matching(
hass, mock_device_tracker_conf):
"""Test not matching single level wildcard topic."""
dev_id = 'paulus'
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
entity_id = ENTITY_ID_FORMAT.format(dev_id)
subscription = '/location/+/paulus'
topic = '/location/paulus'
location = 'work'
@ -120,7 +121,7 @@ async def test_multi_level_wildcard_topic_not_matching(
hass, mock_device_tracker_conf):
"""Test not matching multi level wildcard topic."""
dev_id = 'paulus'
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
entity_id = ENTITY_ID_FORMAT.format(dev_id)
subscription = '/location/#'
topic = '/somewhere/room/paulus'
location = 'work'

43
tests/components/mqtt_json/test_device_tracker.py
View File

@ -1,12 +1,13 @@
"""The tests for the JSON MQTT device tracker platform."""
import json
from asynctest import patch
import logging
import os
from asynctest import patch
import pytest
from homeassistant.setup import async_setup_component
from homeassistant.components import device_tracker
from homeassistant.components.device_tracker.legacy import (
YAML_DEVICES, ENTITY_ID_FORMAT, DOMAIN as DT_DOMAIN)
from homeassistant.const import CONF_PLATFORM
from tests.common import async_mock_mqtt_component, async_fire_mqtt_message
@ -27,7 +28,7 @@ LOCATION_MESSAGE_INCOMPLETE = {
def setup_comp(hass):
"""Initialize components."""
hass.loop.run_until_complete(async_mock_mqtt_component(hass))
yaml_devices = hass.config.path(device_tracker.YAML_DEVICES)
yaml_devices = hass.config.path(YAML_DEVICES)
yield
if os.path.isfile(yaml_devices):
os.remove(yaml_devices)
@ -45,8 +46,8 @@ async def test_ensure_device_tracker_platform_validation(hass):
dev_id = 'paulus'
topic = 'location/paulus'
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
assert await async_setup_component(hass, DT_DOMAIN, {
DT_DOMAIN: {
CONF_PLATFORM: 'mqtt_json',
'devices': {dev_id: topic}
}
@ -60,8 +61,8 @@ async def test_json_message(hass):
topic = 'location/zanzito'
location = json.dumps(LOCATION_MESSAGE)
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
assert await async_setup_component(hass, DT_DOMAIN, {
DT_DOMAIN: {
CONF_PLATFORM: 'mqtt_json',
'devices': {dev_id: topic}
}
@ -79,8 +80,8 @@ async def test_non_json_message(hass, caplog):
topic = 'location/zanzito'
location = 'home'
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
assert await async_setup_component(hass, DT_DOMAIN, {
DT_DOMAIN: {
CONF_PLATFORM: 'mqtt_json',
'devices': {dev_id: topic}
}
@ -100,8 +101,8 @@ async def test_incomplete_message(hass, caplog):
topic = 'location/zanzito'
location = json.dumps(LOCATION_MESSAGE_INCOMPLETE)
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
assert await async_setup_component(hass, DT_DOMAIN, {
DT_DOMAIN: {
CONF_PLATFORM: 'mqtt_json',
'devices': {dev_id: topic}
}
@ -123,8 +124,8 @@ async def test_single_level_wildcard_topic(hass):
topic = 'location/room/zanzito'
location = json.dumps(LOCATION_MESSAGE)
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
assert await async_setup_component(hass, DT_DOMAIN, {
DT_DOMAIN: {
CONF_PLATFORM: 'mqtt_json',
'devices': {dev_id: subscription}
}
@ -143,8 +144,8 @@ async def test_multi_level_wildcard_topic(hass):
topic = 'location/zanzito'
location = json.dumps(LOCATION_MESSAGE)
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
assert await async_setup_component(hass, DT_DOMAIN, {
DT_DOMAIN: {
CONF_PLATFORM: 'mqtt_json',
'devices': {dev_id: subscription}
}
@ -159,13 +160,13 @@ async def test_multi_level_wildcard_topic(hass):
async def test_single_level_wildcard_topic_not_matching(hass):
"""Test not matching single level wildcard topic."""
dev_id = 'zanzito'
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
entity_id = ENTITY_ID_FORMAT.format(dev_id)
subscription = 'location/+/zanzito'
topic = 'location/zanzito'
location = json.dumps(LOCATION_MESSAGE)
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
assert await async_setup_component(hass, DT_DOMAIN, {
DT_DOMAIN: {
CONF_PLATFORM: 'mqtt_json',
'devices': {dev_id: subscription}
}
@ -178,13 +179,13 @@ async def test_single_level_wildcard_topic_not_matching(hass):
async def test_multi_level_wildcard_topic_not_matching(hass):
"""Test not matching multi level wildcard topic."""
dev_id = 'zanzito'
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
entity_id = ENTITY_ID_FORMAT.format(dev_id)
subscription = 'location/#'
topic = 'somewhere/zanzito'
location = json.dumps(LOCATION_MESSAGE)
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
assert await async_setup_component(hass, DT_DOMAIN, {
DT_DOMAIN: {
CONF_PLATFORM: 'mqtt_json',
'devices': {dev_id: subscription}
}

4
tests/components/tplink/test_device_tracker.py
View File

@ -3,7 +3,7 @@
import os
import pytest
from homeassistant.components import device_tracker
from homeassistant.components.device_tracker.legacy import YAML_DEVICES
from homeassistant.components.tplink.device_tracker import Tplink4DeviceScanner
from homeassistant.const import (CONF_PLATFORM, CONF_PASSWORD, CONF_USERNAME,
CONF_HOST)
@ -13,7 +13,7 @@ import requests_mock
@pytest.fixture(autouse=True)
def setup_comp(hass):
"""Initialize components."""
yaml_devices = hass.config.path(device_tracker.YAML_DEVICES)
yaml_devices = hass.config.path(YAML_DEVICES)
yield
if os.path.isfile(yaml_devices):
os.remove(yaml_devices)

4
tests/components/unifi_direct/test_device_tracker.py
View File

@ -7,7 +7,7 @@ import pytest
import voluptuous as vol
from homeassistant.setup import async_setup_component
from homeassistant.components import device_tracker
from homeassistant.components.device_tracker.legacy import YAML_DEVICES
from homeassistant.components.device_tracker import (
CONF_CONSIDER_HOME, CONF_TRACK_NEW, CONF_AWAY_HIDE,
CONF_NEW_DEVICE_DEFAULTS)
@ -27,7 +27,7 @@ scanner_path = 'homeassistant.components.unifi_direct.device_tracker.' + \
def setup_comp(hass):
"""Initialize components."""
mock_component(hass, 'zone')
yaml_devices = hass.config.path(device_tracker.YAML_DEVICES)
yaml_devices = hass.config.path(YAML_DEVICES)
yield
if os.path.isfile(yaml_devices):
os.remove(yaml_devices)

4
tests/conftest.py
View File

@ -102,11 +102,11 @@ def mock_device_tracker_conf():
devices.append(entity)
with patch(
'homeassistant.components.device_tracker'
'homeassistant.components.device_tracker.legacy'
'.DeviceTracker.async_update_config',
side_effect=mock_update_config
), patch(
'homeassistant.components.device_tracker.async_load_config',
'homeassistant.components.device_tracker.legacy.async_load_config',
side_effect=lambda *args: mock_coro(devices)
):
yield devices