feat: first version

.github/workflows/ci.yml

@@ -36,15 +36,12 @@ jobs:
       fail-fast: false
       matrix:
         python-version:
-          - "3.8"
           - "3.9"
           - "3.10"
           - "3.11"
           - "3.12"
         os:
           - ubuntu-latest
-          - windows-latest
-          - macOS-latest
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v4

.pre-commit-config.yaml

@@ -45,10 +45,6 @@ repos:
     rev: 24.2.0
     hooks:
       - id: black
-  - repo: https://github.com/codespell-project/codespell
-    rev: v2.2.6
-    hooks:
-      - id: codespell
   - repo: https://github.com/pre-commit/mirrors-mypy
     rev: v1.8.0
     hooks:

build_oui.py

@@ -0,0 +1,12 @@
+"""Build oui table."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from generate_oui_data import generate
+
+
+def build(setup_kwargs: dict[str, Any]) -> None:
+    """Build the OUI data."""
+    generate()

generate_oui_data.py

@@ -0,0 +1,25 @@
+"""Build oui table."""
+
+import pathlib
+
+import requests  # type: ignore
+
+
+def generate() -> None:
+    """Generate the OUI data."""
+    resp = requests.get("https://standards-oui.ieee.org/oui.txt", timeout=10)
+    resp.raise_for_status()
+    oui_bytes = resp.content
+    oui_to_vendor = {}
+    for line in oui_bytes.splitlines():
+        if b"(base 16)" in line:
+            oui, _, vendor = line.partition(b"(base 16)")
+            oui_to_vendor[oui.strip()] = vendor.strip()
+    file = pathlib.Path(__file__)
+    target_file = file.parent.joinpath("src").joinpath("aiooui").joinpath("oui.data")
+    with open(target_file, "wb") as f:
+        f.write(b"\n".join(b"=".join((o, v)) for o, v in oui_to_vendor.items()))
+
+
+if __name__ == "__main__":
+    generate()

pyproject.toml

@@ -22,11 +22,13 @@ packages = [
 "Changelog" = "https://github.com/bluetooth-devices/aiooui/blob/main/CHANGELOG.md"
 
 [tool.poetry.dependencies]
-python = "^3.8"
+python = "^3.9"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.0"
 pytest-cov = "^3.0"
+pytest-asyncio = "^0.23.5"
+requests = "^2.31.0"
 
 [tool.semantic_release]
 version_toml = ["pyproject.toml:tool.poetry.version"]
@@ -125,5 +127,9 @@ module = "tests.*"
 allow_untyped_defs = true
 
 [build-system]
-requires = ["poetry-core>=1.0.0"]
+requires = ["setuptools>=65.4.1", "poetry-core>=1.0.0", 'requests']
 build-backend = "poetry.core.masonry.api"
+
+[tool.poetry.build]
+generate-setup-file = true
+script = "build_oui.py"

src/aiooui/__init__.py

@@ -1 +1,73 @@
+from __future__ import annotations
+
 __version__ = "0.0.0"
+
+import asyncio
+import pathlib
+
+_OUI_DATA_FILE = pathlib.Path(__file__).parent.joinpath("oui.data")
+
+
+class OUIManager:
+    """Manages the OUI data."""
+
+    def __init__(self) -> None:
+        """Initialize the OUIManager."""
+        self._oui_to_vendor: dict[str, str] = {}
+        self._load_future: asyncio.Future[None] | None = None
+
+    def get_vendor(self, mac: str) -> str | None:
+        """Get the vendor for a MAC address."""
+        if not self._oui_to_vendor:
+            raise RuntimeError("OUI data not loaded, call async_load first")
+        return self._oui_to_vendor.get(mac.replace(":", "")[:6].upper())
+
+    async def async_load(self) -> None:
+        """Load the OUI data."""
+        if self._oui_to_vendor:
+            return
+        if self._load_future:
+            await self._load_future
+            return
+        loop = asyncio.get_running_loop()
+        self._load_future = loop.create_future()
+        try:
+            self._oui_to_vendor = await loop.run_in_executor(None, self._load_oui_data)
+        except Exception as err:
+            self._load_future.set_exception(err)
+            raise
+        else:
+            self._load_future.set_result(None)
+        finally:
+            self._load_future = None
+
+    def _load_oui_data(self) -> dict[str, str]:
+        """Load the OUI data."""
+        with open(_OUI_DATA_FILE) as f:
+            oui_to_vendor: dict[str, str] = {}
+            for line in f.read().splitlines():
+                oui, _, vendor = line.partition("=")
+                oui_to_vendor[oui] = vendor
+
+        return oui_to_vendor
+
+
+_OUI_MANAGER = OUIManager()
+
+
+def is_loaded() -> bool:
+    """Return if the OUI data is loaded."""
+    return bool(_OUI_MANAGER._oui_to_vendor)
+
+
+def get_vendor(mac: str) -> str | None:
+    """Get the vendor for a MAC address."""
+    return _OUI_MANAGER.get_vendor(mac)
+
+
+async def async_load() -> None:
+    """Load the OUI data."""
+    await _OUI_MANAGER.async_load()
+
+
+__all__ = ["async_load", "get_vendor", "is_loaded"]