230 lines
7.6 KiB
Python
230 lines
7.6 KiB
Python
"""The USB Discovery integration."""
|
|
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
import fnmatch
|
|
import logging
|
|
import os
|
|
import sys
|
|
|
|
from serial.tools.list_ports import comports
|
|
from serial.tools.list_ports_common import ListPortInfo
|
|
import voluptuous as vol
|
|
|
|
from homeassistant import config_entries
|
|
from homeassistant.components import websocket_api
|
|
from homeassistant.components.websocket_api.connection import ActiveConnection
|
|
from homeassistant.const import EVENT_HOMEASSISTANT_STARTED, EVENT_HOMEASSISTANT_STOP
|
|
from homeassistant.core import Event, HomeAssistant, callback
|
|
from homeassistant.helpers import system_info
|
|
from homeassistant.helpers.debounce import Debouncer
|
|
from homeassistant.helpers.typing import ConfigType
|
|
from homeassistant.loader import async_get_usb
|
|
|
|
from .const import DOMAIN
|
|
from .flow import FlowDispatcher, USBFlow
|
|
from .models import USBDevice
|
|
from .utils import usb_device_from_port
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
REQUEST_SCAN_COOLDOWN = 60 # 1 minute cooldown
|
|
|
|
|
|
def human_readable_device_name(
|
|
device: str,
|
|
serial_number: str | None,
|
|
manufacturer: str | None,
|
|
description: str | None,
|
|
vid: str | None,
|
|
pid: str | None,
|
|
) -> str:
|
|
"""Return a human readable name from USBDevice attributes."""
|
|
device_details = f"{device}, s/n: {serial_number or 'n/a'}"
|
|
manufacturer_details = f" - {manufacturer}" if manufacturer else ""
|
|
vendor_details = f" - {vid}:{pid}" if vid else ""
|
|
full_details = f"{device_details}{manufacturer_details}{vendor_details}"
|
|
|
|
if not description:
|
|
return full_details
|
|
return f"{description[:26]} - {full_details}"
|
|
|
|
|
|
def get_serial_by_id(dev_path: str) -> str:
|
|
"""Return a /dev/serial/by-id match for given device if available."""
|
|
by_id = "/dev/serial/by-id"
|
|
if not os.path.isdir(by_id):
|
|
return dev_path
|
|
|
|
for path in (entry.path for entry in os.scandir(by_id) if entry.is_symlink()):
|
|
if os.path.realpath(path) == dev_path:
|
|
return path
|
|
return dev_path
|
|
|
|
|
|
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
|
"""Set up the USB Discovery integration."""
|
|
usb = await async_get_usb(hass)
|
|
usb_discovery = USBDiscovery(hass, FlowDispatcher(hass), usb)
|
|
await usb_discovery.async_setup()
|
|
hass.data[DOMAIN] = usb_discovery
|
|
websocket_api.async_register_command(hass, websocket_usb_scan)
|
|
|
|
return True
|
|
|
|
|
|
def _fnmatch_lower(name: str | None, pattern: str) -> bool:
|
|
"""Match a lowercase version of the name."""
|
|
if name is None:
|
|
return False
|
|
return fnmatch.fnmatch(name.lower(), pattern)
|
|
|
|
|
|
class USBDiscovery:
|
|
"""Manage USB Discovery."""
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistant,
|
|
flow_dispatcher: FlowDispatcher,
|
|
usb: list[dict[str, str]],
|
|
) -> None:
|
|
"""Init USB Discovery."""
|
|
self.hass = hass
|
|
self.flow_dispatcher = flow_dispatcher
|
|
self.usb = usb
|
|
self.seen: set[tuple[str, ...]] = set()
|
|
self.observer_active = False
|
|
self._request_debouncer: Debouncer | None = None
|
|
|
|
async def async_setup(self) -> None:
|
|
"""Set up USB Discovery."""
|
|
await self._async_start_monitor()
|
|
self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STARTED, self.async_start)
|
|
|
|
async def async_start(self, event: Event) -> None:
|
|
"""Start USB Discovery and run a manual scan."""
|
|
self.flow_dispatcher.async_start()
|
|
await self._async_scan_serial()
|
|
|
|
async def _async_start_monitor(self) -> None:
|
|
"""Start monitoring hardware with pyudev."""
|
|
if not sys.platform.startswith("linux"):
|
|
return
|
|
info = await system_info.async_get_system_info(self.hass)
|
|
if info.get("docker"):
|
|
return
|
|
|
|
from pyudev import ( # pylint: disable=import-outside-toplevel
|
|
Context,
|
|
Monitor,
|
|
MonitorObserver,
|
|
)
|
|
|
|
try:
|
|
context = Context()
|
|
except (ImportError, OSError):
|
|
return
|
|
|
|
monitor = Monitor.from_netlink(context)
|
|
try:
|
|
monitor.filter_by(subsystem="tty")
|
|
except ValueError as ex: # this fails on WSL
|
|
_LOGGER.debug(
|
|
"Unable to setup pyudev filtering; This is expected on WSL: %s", ex
|
|
)
|
|
return
|
|
observer = MonitorObserver(
|
|
monitor, callback=self._device_discovered, name="usb-observer"
|
|
)
|
|
observer.start()
|
|
self.hass.bus.async_listen_once(
|
|
EVENT_HOMEASSISTANT_STOP, lambda event: observer.stop()
|
|
)
|
|
self.observer_active = True
|
|
|
|
def _device_discovered(self, device):
|
|
"""Call when the observer discovers a new usb tty device."""
|
|
if device.action != "add":
|
|
return
|
|
_LOGGER.debug(
|
|
"Discovered Device at path: %s, triggering scan serial",
|
|
device.device_path,
|
|
)
|
|
self.scan_serial()
|
|
|
|
@callback
|
|
def _async_process_discovered_usb_device(self, device: USBDevice) -> None:
|
|
"""Process a USB discovery."""
|
|
_LOGGER.debug("Discovered USB Device: %s", device)
|
|
device_tuple = dataclasses.astuple(device)
|
|
if device_tuple in self.seen:
|
|
return
|
|
self.seen.add(device_tuple)
|
|
for matcher in self.usb:
|
|
if "vid" in matcher and device.vid != matcher["vid"]:
|
|
continue
|
|
if "pid" in matcher and device.pid != matcher["pid"]:
|
|
continue
|
|
if "serial_number" in matcher and not _fnmatch_lower(
|
|
device.serial_number, matcher["serial_number"]
|
|
):
|
|
continue
|
|
if "manufacturer" in matcher and not _fnmatch_lower(
|
|
device.manufacturer, matcher["manufacturer"]
|
|
):
|
|
continue
|
|
if "description" in matcher and not _fnmatch_lower(
|
|
device.description, matcher["description"]
|
|
):
|
|
continue
|
|
flow: USBFlow = {
|
|
"domain": matcher["domain"],
|
|
"context": {"source": config_entries.SOURCE_USB},
|
|
"data": dataclasses.asdict(device),
|
|
}
|
|
self.flow_dispatcher.async_create(flow)
|
|
|
|
@callback
|
|
def _async_process_ports(self, ports: list[ListPortInfo]) -> None:
|
|
"""Process each discovered port."""
|
|
for port in ports:
|
|
if port.vid is None and port.pid is None:
|
|
continue
|
|
self._async_process_discovered_usb_device(usb_device_from_port(port))
|
|
|
|
def scan_serial(self) -> None:
|
|
"""Scan serial ports."""
|
|
self.hass.add_job(self._async_process_ports, comports())
|
|
|
|
async def _async_scan_serial(self) -> None:
|
|
"""Scan serial ports."""
|
|
self._async_process_ports(await self.hass.async_add_executor_job(comports))
|
|
|
|
async def async_request_scan_serial(self) -> None:
|
|
"""Request a serial scan."""
|
|
if not self._request_debouncer:
|
|
self._request_debouncer = Debouncer(
|
|
self.hass,
|
|
_LOGGER,
|
|
cooldown=REQUEST_SCAN_COOLDOWN,
|
|
immediate=True,
|
|
function=self._async_scan_serial,
|
|
)
|
|
await self._request_debouncer.async_call()
|
|
|
|
|
|
@websocket_api.require_admin
|
|
@websocket_api.websocket_command({vol.Required("type"): "usb/scan"})
|
|
@websocket_api.async_response
|
|
async def websocket_usb_scan(
|
|
hass: HomeAssistant,
|
|
connection: ActiveConnection,
|
|
msg: dict,
|
|
) -> None:
|
|
"""Scan for new usb devices."""
|
|
usb_discovery: USBDiscovery = hass.data[DOMAIN]
|
|
if not usb_discovery.observer_active:
|
|
await usb_discovery.async_request_scan_serial()
|
|
connection.send_result(msg["id"])
|