API Reference

BlueTracker - Bluetooth Device Presence Tracking via MQTT.

This Python package provides classes for tracking the presence of Bluetooth devices and reporting their status (home/not_home) to Home Assistant using MQTT.

Introduction

Key Components

BlueScanner: Handles the actual Bluetooth scanning process.
BlueTracker: Manages the tracking of devices and interaction with Home Assistant via MQTT.
Device: Represents a Bluetooth device being tracked (name, MAC address, state).
MqttClient: Handles communication with the MQTT broker.

Functionality

BlueTracker periodically scans for Bluetooth devices using BlueScanner.
Device states (home/not_home) and attributes are published to Home Assistant via MQTT.
BlueTracker actively monitors the connection to Home Assistant. If the connection is lost, scanning will pause and automatically resume when the connection is restored.

Programmatic Usage

Create a BlueScanner instance for Bluetooth scanning.
Create an MqttClient instance to connect to your MQTT broker.
Create a list of Device objects representing the devices to track.
Instantiate BlueTracker with the scanner, MQTT client, and device list.
Call the BlueTracker.run() to start tracking.
When finished, call BlueTracker.stop() to gracefully stop the service.

Example Code

from bluetracker import BlueTracker, BlueScanner, MqttClient, Device, DeviceType

devices = [
    Device("My Phone", "AA:BB:CC:DD:EE:FF"),
    Device("My Tablet", "11:22:33:44:55:66")
]

scanner = BlueScanner(scan_interval=30, scan_timeout=5, consider_away=60)
mqttc = MqttClient(host="your_mqtt_broker", port=1883, ...)

tracker = BlueTracker(scanner, mqttc, devices)

try:
    tracker.run()
except KeyboardInterrupt:  # Graceful shutdown on Ctrl+C
    tracker.stop()

Automatic Setup and Execution with main.py

BlueTracker uses a bluetracker_config.toml file for configuration. If it doesn’t exist, a default one will be created on the first run.

Install BlueTracker: pip install bluetracker
Edit the Configuration: modify the bluetracker_config.toml file according to your MQTT broker settings and the devices you want to track.
Run BlueTracker: Execute bluetracker in your terminal.
A signal handler is provided for graceful shutdown on SIGINT (e.g., Ctrl+C).

"""Start BlueTracker application."""

import sys
from collections.abc import Callable
from importlib.resources import files
from logging import getLogger
from pathlib import Path
from shutil import copyfile
from signal import SIGINT, signal
from types import FrameType

from bluetracker import BlueTracker
from bluetracker.core import BlueScanner
from bluetracker.helpers.mqtt_client import MqttClient
from bluetracker.models.device import Device, DeviceType
from bluetracker.utils.config import BlueTrackerConfig, ConfigError, load_config
from bluetracker.utils.logging import set_logging

_LOGGER = getLogger(__name__)


def _create_bluescanner(config: dict[str, int]) -> BlueScanner:
    return BlueScanner(
        config['scan_interval'],
        config['scan_timeout'],
        config['consider_away'],
    )


def _create_mqtt_client(config: dict[str, str | int]) -> MqttClient:
    return MqttClient(
        str(config['host']),
        int(config['port']),
        str(config['username']),
        str(config['password']),
        str(config['homeassistant_token']),
        str(config['discovery_topic_prefix']),
    )


def _create_devices(devices: list[dict[str, str]]) -> list[Device]:
    return [
        Device(device['name'].title(), device['mac'].lower(), DeviceType.BLUETOOTH)
        for device in devices
    ]


def _config_path() -> str:
    src_config = str(files('bluetracker').joinpath('config.toml'))
    dst_config = Path.cwd().joinpath('bluetracker_config.toml')

    if not dst_config.exists():
        copyfile(src_config, dst_config)
        print(f'First run, configuration file copied to {dst_config}')
        print('Modify as required and restart.')
        sys.exit(0)
    else:
        print(f'Configuration file found at {dst_config}')

    return dst_config.as_posix()


def create_signal_handler(
    tracker_instance: BlueTracker,
) -> Callable[[int, FrameType | None], None]:
    """Creates a signal handler for graceful shutdown on SIGINT.

    This function returns a signal handler that is designed to be registered
    for the SIGINT signal (e.g., triggered by Ctrl+C). When the signal is
    received, the handler stops the BlueTracker instance, logs shutdown messages,
    and then exits the application.

    Args:
        tracker_instance: The BlueTracker instance to stop on shutdown.

    Returns:
        A signal handler function.
    """

    def signal_handler(_signum: int, _frame: FrameType | None) -> None:
        if signal_handler.called:  # type: ignore[attr-defined]
            _LOGGER.info(
                'BlueTracker already shutting down. This may take a few moments.',
            )
            return

        signal_handler.called = True  # type: ignore[attr-defined]
        _LOGGER.info('SIGINT received. Initiating graceful shutdown...')
        tracker_instance.stop()
        _LOGGER.info('%s shutdown complete', tracker_instance.__class__.__name__)
        sys.exit(0)

    signal_handler.called = False  # type: ignore[attr-defined]
    return signal_handler


def main() -> None:
    """Main entry point of the BlueTracker application.

    - Loads configuration,
    - sets up logging,
    - creates scanner, MQTT client and devices,
    - starts the BlueTracker instance.
    """
    config_path = _config_path()

    try:
        config: BlueTrackerConfig = load_config(config_path)
    except ConfigError as error:
        print(f'Fatal error: {error}')
        sys.exit(1)

    set_logging(config.environment)

    scanner = _create_bluescanner(config.bluetooth)
    mqtt_client = _create_mqtt_client(config.mqtt)
    devices: list[Device] = _create_devices(config.devices)

    bluetracker = BlueTracker(scanner, mqtt_client, devices)

    signal(SIGINT, create_signal_handler(bluetracker))

    bluetracker.run()


if __name__ == '__main__':
    main()  # pragma: no cover

class BlueScanner

Scan bluetooth classic devices.

__init__(scan_interval: int, scan_timeout: int, consider_away: int) → None

Initialize the scanner.

Parameters:

scan_interval – Seconds to wait between scans.
scan_timeout – Seconds to wait for a device response.
consider_away – Seconds to wait to mark a device as away.

config_as_dict() → dict[str, int]

Get the bluetooth config as a dictionary.

Returns:: A dictionary representation of the config.

scan(device: Device) → Device

Scan a device.

Parameters:: device – The device to scan.
Returns:: The device with its new state.

stop(device: Device) → Device

Stop a device scan.

Sets a device state to NOT_HOME.

Parameters:: device – The device to stop.
Returns:: The device with its new state.

class BlueTracker

Tracks Bluetooth devices and reports their status to Home Assistant via MQTT.

__init__(scanner: BlueScanner, mqtt_client: MqttClient, devices: list[Device]) → None

Initializes the BlueTracker.

Parameters:

scanner – Object responsible for Bluetooth scanning.
mqtt_client – Object responsible for MQTT communication with an MQTT broker.
devices – A list of objects representing tracked devices.

Raises:

BlueTrackerTypeError – If any argument has an unexpected type.

__str__() → str

Returns a string representation of tracked devices.

Returns:: The string representation.

run() → None

Main execution loop for the BlueTracker service.

Initializes tracked entities and continuously scans for Bluetooth devices when Home Assistant is online.

If Home Assistant is unavailable, the loop waits for it to come online before resuming scanning.

Discovered devices are published via MQTT to Home Assistant. The scan interval is determined by the configured scan_interval in the BlueScanner instance.

stop() → None: Gracefully stops BlueTracker service and closes MQTT client connection.

exception BlueTrackerTypeError

Custom exception when invalid type is encountered in BlueTracker.

__init__(value: Any) → None

Initialize.

Parameters:: value – The object with an invalid type that triggered the exception.

class Device

Device model.

__init__(name: str, mac: str, source_type: ~bluetracker.models.device.DeviceType, last_seen: ~datetime.datetime = <factory>, state: ~bluetracker.models.device.DeviceState = DeviceState.NOT_HOME, reason: ~bluetracker.models.device.DeviceResponse = DeviceResponse.SETUP) → None

__repr__(): Return repr(self).

__str__() → str

The device as a string.

Returns:: The device as a string.

last_seen: datetime

mac: str

name: str

reason: DeviceResponse = 'server setup'

source_type: DeviceType

state: DeviceState = 'not_home'

to_dict() → dict[str, str]

Get the device as a dictionary.

Returns:: A dictionary representation of the device.

class DeviceResponse

Device responses.

CONSIDERED_HOME = 'considered home'

NO_RESPONSE = 'no response'

RESPONDED = 'responded'

SETUP = 'server setup'

SHUTDOWN = 'server shutdown'

class DeviceState

Device states.

HOME = 'home'

NOT_HOME = 'not_home'

class DeviceType

Device types.

BLUETOOTH = 'bluetooth'

class MqttClient

A client to connect to an MQTT broker to publish messages.

__init__(host: str, port: int, username: str, password: str, homeassistant_token: str, discovery_topic_prefix: str = 'homeassistant') → None

Initialize the MQTT client.

Parameters:

host – The MQTT host.
port – The MQTT port.
username – The MQTT username.
password – The MQTT password.
homeassistant_token – The Home Assistant token.
discovery_topic_prefix – The MQTT discovery prefix for Home Assistant.

is_connected: bool: Indicates whether the client is currently connected to the MQTT broker.

is_homeassistant_online: bool: Determines Home Assistant’s online status by checking both MQTT messages and Home Assistant’s Rest API.

publish(topic: str, payload: str, *, retain: bool = False) → bool

Publish a message to the MQTT broker.

Waits for connection if not already connected and returns success based on the publish result code.

Parameters:

topic – The MQTT topic to publish to.
payload – The message to publish.
retain – Whether the message should be retained by the broker.

Returns:

True if the message was published successfully, False otherwise.

start() → None: Start a connection to the MQTT broker.

stop() → None: Stop the connection to the MQTT broker.

API Reference

Introduction

Key Components

Functionality

Programmatic Usage

Example Code

Automatic Setup and Execution with __main__.py

Automatic Setup and Execution with main.py