feat: add initial implementation (#1)

Author: bdraco

.github/workflows/ci.yml

@@ -40,6 +40,7 @@ jobs:
           - "3.10"
           - "3.11"
           - "3.12"
+          - "3.13"
         os:
           - ubuntu-latest
           - windows-latest

pyproject.toml

@@ -26,6 +26,7 @@ classifiers = [
 ]
 
 dependencies = [
+    "asyncinotify>=4.2.0",
 ]
 urls."Bug Tracker" = "https://github.com/bluetooth-devices/aiousbwatcher/issues"
 urls.Changelog = "https://github.com/bluetooth-devices/aiousbwatcher/blob/main/CHANGELOG.md"
@@ -159,3 +160,9 @@ match = "main"
 [tool.semantic_release.branches.noop]
 match = "(?!main$)"
 prerelease = true
+
+[tool.uv]
+dev-dependencies = [
+    "pytest-asyncio>=0.25.2",
+    "pytest-cov>=6.0.0",
+]

src/aiousbwatcher/__init__.py

@@ -1 +1,5 @@
 __version__ = "0.0.0"
+
+from .impl import AIOUSBWatcher, InotifyNotAvailableError
+
+__all__ = ["AIOUSBWatcher", "InotifyNotAvailableError"]

src/aiousbwatcher/impl.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import Generator
+from functools import partial
+from pathlib import Path
+from typing import Callable
+
+_PATH = "/dev/bus/usb"
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class InotifyNotAvailableError(Exception):
+    """Raised when inotify is not available on the platform."""
+
+
+def _get_directories_recursive_gen(path: Path) -> Generator[Path, None, None]:
+    if path.is_dir():
+        yield path
+        for child in path.iterdir():
+            yield from _get_directories_recursive_gen(child)
+
+
+def _get_directories_recursive(path: Path) -> list[Path]:
+    return list(_get_directories_recursive_gen(path))
+
+
+async def _async_get_directories_recursive(
+    loop: asyncio.AbstractEventLoop, path: Path
+) -> list[Path]:
+    return await loop.run_in_executor(None, _get_directories_recursive, path)
+
+
+class AIOUSBWatcher:
+    """A watcher for USB devices that uses asyncio."""
+
+    def __init__(self) -> None:
+        self._path = Path(_PATH)
+        self._loop = asyncio.get_running_loop()
+        self._task: asyncio.Task[None] | None = None
+        self._callbacks: set[Callable[[], None]] = set()
+
+    def async_start(self) -> Callable[[], None]:
+        """Start the watcher."""
+        if self._task is not None:
+            raise RuntimeError("Watcher already started")
+        try:
+            from asyncinotify import Inotify  # noqa
+        except Exception as ex:
+            raise InotifyNotAvailableError(
+                "Inotify not available on this platform"
+            ) from ex
+        self._task = self._loop.create_task(self._watcher())
+        return self._async_stop
+
+    def async_register_callback(
+        self, callback: Callable[[], None]
+    ) -> Callable[[], None]:
+        """Register callback that will be called when a USB device is added/removed."""
+        self._callbacks.add(callback)
+        return partial(self._async_unregister_callback, callback)
+
+    def _async_stop(self) -> None:
+        """Stop the watcher."""
+        assert self._task is not None  # noqa
+        self._task.cancel()
+        self._task = None
+
+    async def _watcher(self) -> None:
+        from asyncinotify import Inotify, Mask
+
+        mask = (
+            Mask.CREATE
+            | Mask.MOVED_FROM
+            | Mask.MOVED_TO
+            | Mask.CREATE
+            | Mask.DELETE_SELF
+            | Mask.IGNORED
+        )
+
+        with Inotify() as inotify:
+            for directory in await _async_get_directories_recursive(
+                self._loop, self._path
+            ):
+                inotify.add_watch(directory, mask)
+
+            async for event in inotify:
+                # Add subdirectories to watch if a new directory is added.
+                if Mask.CREATE in event.mask and event.path is not None:
+                    for directory in await _async_get_directories_recursive(
+                        self._loop, event.path
+                    ):
+                        inotify.add_watch(directory, mask)
+
+                # If there is at least some overlap, assume the user wants this event.
+                if event.mask & mask:
+                    self._async_call_callbacks()
+
+    def _async_unregister_callback(self, callback: Callable[[], None]) -> None:
+        self._callbacks.remove(callback)
+
+    def _async_call_callbacks(self) -> None:
+        for callback in self._callbacks:
+            try:
+                callback()
+            except Exception as e:
+                _LOGGER.exception("Error calling callback %s", callback, exc_info=e)

src/aiousbwatcher/main.py

@@ -0,0 +1,43 @@
+import asyncio
+from pathlib import Path
+from sys import platform
+from unittest.mock import patch
+
+import pytest
+
+from aiousbwatcher import AIOUSBWatcher, InotifyNotAvailableError
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(platform == "linux", reason="Inotify is available on this platform")
+async def test_aiousbwatcher_not_available() -> None:
+    with pytest.raises(InotifyNotAvailableError):
+        watcher = AIOUSBWatcher()
+        watcher.async_start()
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(
+    platform != "linux", reason="Inotify not available on this platform"
+)
+async def test_aiousbwatcher_callbacks(tmp_path: Path) -> None:
+    called: bool = False
+
+    def callback() -> None:
+        nonlocal called
+        called = True
+
+    with patch("aiousbwatcher.impl._PATH", str(tmp_path)):
+        watcher = AIOUSBWatcher()
+        unregister = watcher.async_register_callback(callback)
+        stop = watcher.async_start()
+        await asyncio.sleep(0.1)
+        assert not called
+        (tmp_path / "test").touch()
+        await asyncio.sleep(0.1)
+        assert called
+        called = False  # type: ignore[unreachable]
+        unregister()
+        stop()
+        await asyncio.sleep(0.1)
+        assert not called