Skip to content
158 changes: 92 additions & 66 deletions homeassistant/components/bluetooth_tracker/device_tracker.py
Original file line number Diff line number Diff line change
@@ -1,25 +1,30 @@
"""Tracking for bluetooth devices."""
import logging
from typing import List, Set, Tuple

# pylint: disable=import-error
import bluetooth
from bt_proximity import BluetoothRSSI
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 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,
CONF_TRACK_NEW,
DEFAULT_TRACK_NEW,
SOURCE_TYPE_BLUETOOTH,
DOMAIN,
SCAN_INTERVAL,
SOURCE_TYPE_BLUETOOTH,
)
import homeassistant.util.dt as dt_util
from homeassistant.components.device_tracker.legacy import (
YAML_DEVICES,
async_load_config,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.event import track_point_in_utc_time
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util.async_ import run_coroutine_threadsafe
import homeassistant.util.dt as dt_util

_LOGGER = logging.getLogger(__name__)

Expand All @@ -42,66 +47,86 @@
)


def setup_scanner(hass, config, see, discovery_info=None):
"""Set up the Bluetooth Scanner."""
# pylint: disable=import-error
import bluetooth
from bt_proximity import BluetoothRSSI

def see_device(mac, name, rssi=None):
"""Mark a device as seen."""
attributes = {}
if rssi is not None:
attributes["rssi"] = rssi
see(
mac=f"{BT_PREFIX}{mac}",
host_name=name,
attributes=attributes,
source_type=SOURCE_TYPE_BLUETOOTH,
)

device_id = config.get(CONF_DEVICE_ID)

def discover_devices():
"""Discover Bluetooth devices."""
result = bluetooth.discover_devices(
duration=8,
lookup_names=True,
flush_cache=True,
lookup_class=False,
device_id=device_id,
)
_LOGGER.debug("Bluetooth devices discovered = %d", len(result))
return result

yaml_path = hass.config.path(YAML_DEVICES)
devs_to_track = []
devs_donot_track = []

# Load all known devices.
# We just need the devices so set consider_home and home range
# to 0
def is_bluetooth_device(device) -> bool:
"""Check whether a device is a bluetooth device by its mac."""
return device.mac and device.mac[:3].upper() == BT_PREFIX


def discover_devices(device_id: int) -> List[Tuple[str, str]]:
"""Discover Bluetooth devices."""
result = bluetooth.discover_devices(
duration=8,
lookup_names=True,
flush_cache=True,
lookup_class=False,
device_id=device_id,
)
_LOGGER.debug("Bluetooth devices discovered = %d", len(result))
return result


def see_device(see, mac: str, device_name: str, rssi=None) -> None:
"""Mark a device as seen."""
attributes = {}
if rssi is not None:
attributes["rssi"] = rssi
see(
mac=f"{BT_PREFIX}{mac}",
host_name=device_name,
attributes=attributes,
source_type=SOURCE_TYPE_BLUETOOTH,
)


def get_tracking_devices(hass: HomeAssistantType) -> Tuple[Set[str], Set[str]]:
"""
Load all known devices.

We just need the devices so set consider_home and home range to 0
"""
yaml_path: str = hass.config.path(YAML_DEVICES)
devices_to_track: Set[str] = set()
devices_to_not_track: Set[str] = set()

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:
devs_to_track.append(device.mac[3:])
else:
devs_donot_track.append(device.mac[3:])
if not is_bluetooth_device(device):
continue

normalized_mac: str = device.mac[3:]
if device.track:
devices_to_track.add(normalized_mac)
else:
devices_to_not_track.add(normalized_mac)

return devices_to_track, devices_to_not_track


def setup_scanner(hass: HomeAssistantType, config: dict, see, discovery_info=None):
"""Set up the Bluetooth Scanner."""
device_id: int = config.get(CONF_DEVICE_ID)
devices_to_track, devices_to_not_track = get_tracking_devices(hass)

# If track new devices is true discover new devices on startup.
track_new = config.get(CONF_TRACK_NEW, DEFAULT_TRACK_NEW)
track_new: bool = config.get(CONF_TRACK_NEW, DEFAULT_TRACK_NEW)
_LOGGER.debug("Tracking new devices = %s", track_new)

if not devices_to_track and not track_new:
_LOGGER.debug("No Bluetooth devices to track and not tracking new devices")

if track_new:
for dev in discover_devices():
if dev[0] not in devs_to_track and dev[0] not in devs_donot_track:
devs_to_track.append(dev[0])
see_device(dev[0], dev[1])
for mac, device_name in discover_devices(device_id):
if mac not in devices_to_track and mac not in devices_to_not_track:
devices_to_track.add(mac)
see_device(see, mac, device_name)

interval = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL)

request_rssi = config.get(CONF_REQUEST_RSSI, False)
if request_rssi:
_LOGGER.debug("Detecting RSSI for devices")

def update_bluetooth(_):
"""Update Bluetooth and set timer for the next update."""
Expand All @@ -112,21 +137,22 @@ def update_bluetooth_once():
"""Lookup Bluetooth device and update status."""
try:
if track_new:
for dev in discover_devices():
if dev[0] not in devs_to_track and dev[0] not in devs_donot_track:
devs_to_track.append(dev[0])
for mac in devs_to_track:
for mac, device_name in discover_devices(device_id):
if mac not in devices_to_track and mac not in devices_to_not_track:
devices_to_track.add(mac)

for mac in devices_to_track:
_LOGGER.debug("Scanning %s", mac)
result = bluetooth.lookup_name(mac, timeout=5)
device_name = bluetooth.lookup_name(mac, timeout=5)
rssi = None
if request_rssi:
client = BluetoothRSSI(mac)
rssi = client.request_rssi()
client.close()
if result is None:
if device_name is None:
# Could not lookup device name
continue
see_device(mac, result, rssi)
see_device(see, mac, device_name, rssi)
except bluetooth.BluetoothError:
_LOGGER.exception("Error looking up Bluetooth device")

Expand Down