Coverage for src / bluetooth_sig / registry / profiles / profile_lookup.py: 92%
120 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-18 11:17 +0000
1"""Profile Lookup Registry for simple name/value profile parameters."""
3from __future__ import annotations
5from pathlib import Path
6from typing import Any, cast
8import msgspec
10from bluetooth_sig.registry.base import BaseGenericRegistry
11from bluetooth_sig.registry.utils import find_bluetooth_sig_path
12from bluetooth_sig.types.registry.profile_types import ProfileLookupEntry
14# Field names tried (in order) when extracting the integer value from a YAML entry.
15_VALUE_FIELDS: tuple[str, ...] = ("value", "id", "identifier", "attribute", "MDEP_data_type")
17# Field names tried (in order) when extracting the human-readable name.
18_NAME_FIELDS: tuple[str, ...] = (
19 "name",
20 "label",
21 "codec",
22 "description",
23 "audio_location",
24 "mnemonic",
25 "client_name",
26 "data_type",
27 "document_name",
28)
30# Directories containing LTV / codec-capability structures — deferred.
31_DEFERRED_DIRS: frozenset[str] = frozenset(
32 {
33 "ltv_structures",
34 "metadata_ltv",
35 "codec_capabilities",
36 "codec_configuration_ltv",
37 },
38)
41class ProfileLookupRegistry(BaseGenericRegistry["ProfileLookupRegistry"]):
42 """Registry for simple profile parameter lookup tables.
44 Loads non-LTV, non-permitted-characteristics YAML files from
45 ``profiles_and_services/`` and normalises each entry into a
46 :class:`ProfileLookupEntry` keyed by the YAML top-level key.
48 Thread-safe: Multiple threads can safely access the registry concurrently.
49 """
51 def __init__(self) -> None:
52 """Initialise the profile lookup registry."""
53 super().__init__()
54 self._tables: dict[str, list[ProfileLookupEntry]] = {}
56 # ------------------------------------------------------------------
57 # Helpers
58 # ------------------------------------------------------------------
60 @staticmethod
61 def _extract_int_value(entry: dict[str, Any]) -> int | None:
62 """Return the first integer-coercible value from *entry*."""
63 for field in _VALUE_FIELDS:
64 raw = entry.get(field)
65 if raw is None:
66 continue
67 if isinstance(raw, int):
68 return raw
69 if isinstance(raw, str):
70 try:
71 return int(raw, 16) if raw.startswith("0x") else int(raw)
72 except ValueError:
73 continue
74 return None
76 @staticmethod
77 def _extract_name(entry: dict[str, Any]) -> str | None:
78 """Return the first usable name string from *entry*."""
79 for field in _NAME_FIELDS:
80 raw = entry.get(field)
81 if isinstance(raw, str) and raw:
82 return raw
83 return None
85 @staticmethod
86 def _build_metadata(entry: dict[str, Any], used_keys: set[str]) -> dict[str, str]:
87 """Collect remaining string-coercible fields as metadata."""
88 meta: dict[str, str] = {}
89 for key, val in entry.items():
90 if key in used_keys:
91 continue
92 if isinstance(val, (str, int, float, bool)):
93 meta[key] = str(val)
94 return meta
96 # ------------------------------------------------------------------
97 # Loading
98 # ------------------------------------------------------------------
100 def _load_yaml_file(self, yaml_path: Path) -> None:
101 """Load a single YAML file and store entries keyed by top-level key."""
102 with yaml_path.open("r", encoding="utf-8") as fh:
103 data = msgspec.yaml.decode(fh.read())
105 if not isinstance(data, dict):
106 return
108 data_dict = cast("dict[str, Any]", data)
109 for top_key, entries_raw in data_dict.items():
110 if not isinstance(entries_raw, list):
111 continue
113 entries: list[ProfileLookupEntry] = []
114 for entry in entries_raw:
115 if not isinstance(entry, dict):
116 continue
118 value = self._extract_int_value(entry)
119 name = self._extract_name(entry)
120 if value is None or name is None:
121 continue
123 # Determine which keys were consumed for name and value
124 used: set[str] = set()
125 for field in _VALUE_FIELDS:
126 raw = entry.get(field)
127 if raw is not None:
128 if isinstance(raw, int):
129 used.add(field)
130 break
131 if isinstance(raw, str):
132 try:
133 int(raw, 16) if raw.startswith("0x") else int(raw)
134 used.add(field)
135 break
136 except ValueError:
137 continue
138 for field in _NAME_FIELDS:
139 raw = entry.get(field)
140 if isinstance(raw, str) and raw:
141 used.add(field)
142 break
144 metadata = self._build_metadata(entry, used)
145 entries.append(ProfileLookupEntry(name=name, value=value, metadata=metadata))
147 if entries:
148 self._tables[top_key] = entries
150 @staticmethod
151 def _is_deferred(path: Path) -> bool:
152 """Return True if *path* is inside a deferred subdirectory."""
153 return any(part in _DEFERRED_DIRS for part in path.parts)
155 @staticmethod
156 def _is_permitted_characteristics(path: Path) -> bool:
157 """Return True if *path* is a permitted-characteristics file."""
158 return "permitted_characteristics" in path.name
160 def _load(self) -> None:
161 """Load all non-LTV, non-permitted-characteristics profile YAMLs."""
162 uuids_path = find_bluetooth_sig_path()
163 if not uuids_path:
164 self._loaded = True
165 return
167 profiles_path = uuids_path.parent / "profiles_and_services"
168 if not profiles_path.exists():
169 self._loaded = True
170 return
172 for yaml_file in sorted(profiles_path.rglob("*.yaml")):
173 if self._is_deferred(yaml_file) or self._is_permitted_characteristics(yaml_file):
174 continue
175 self._load_yaml_file(yaml_file)
177 self._loaded = True
179 # ------------------------------------------------------------------
180 # Query API
181 # ------------------------------------------------------------------
183 def get_entries(self, table_key: str) -> list[ProfileLookupEntry]:
184 """Get all entries for a named lookup table.
186 Args:
187 table_key: The YAML top-level key, e.g. ``"audio_codec_id"``,
188 ``"bearer_technology"``, ``"display_types"``.
190 Returns:
191 List of :class:`ProfileLookupEntry` or an empty list if not found.
192 """
193 self._ensure_loaded()
194 with self._lock:
195 return list(self._tables.get(table_key, []))
197 def get_all_table_keys(self) -> list[str]:
198 """Return all loaded table key names (sorted)."""
199 self._ensure_loaded()
200 with self._lock:
201 return sorted(self._tables)
203 def resolve_name(self, table_key: str, value: int) -> str | None:
204 """Look up the name for a given numeric value within a table.
206 Args:
207 table_key: Table key (e.g. ``"bearer_technology"``).
208 value: The numeric identifier.
210 Returns:
211 The entry name or ``None`` if not found.
212 """
213 self._ensure_loaded()
214 with self._lock:
215 for entry in self._tables.get(table_key, []):
216 if entry.value == value:
217 return entry.name
218 return None
221# Singleton instance for global use
222profile_lookup_registry = ProfileLookupRegistry()