Coverage for src / bluetooth_sig / registry / profiles / permitted_characteristics.py: 85%

1"""Permitted Characteristics Registry for profile service constraints.""" 

2 

3from __future__ import annotations 

4 

5from pathlib import Path 

6from typing import Any, cast 

7 

8import msgspec 

9 

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 PermittedCharacteristicEntry 

13 

14# Profile subdirectories that contain ``*_permitted_characteristics.yaml``. 

15_PROFILE_DIRS: tuple[str, ...] = ("ess", "uds", "imds") 

16 

17 

18class PermittedCharacteristicsRegistry( 

19 BaseGenericRegistry["PermittedCharacteristicsRegistry"], 

20): 

21 """Registry for profile-specific permitted characteristic lists. 

22 

23 Loads ``permitted_characteristics`` YAML files from ESS, UDS and IMDS 

24 profile subdirectories under ``profiles_and_services/``. 

25 

26 Thread-safe: Multiple threads can safely access the registry concurrently. 

27 """ 

28 

29 def __init__(self) -> None: 

30 """Initialise the permitted characteristics registry.""" 

31 super().__init__() 

32 self._entries: dict[str, list[PermittedCharacteristicEntry]] = {} 

33 

34 # ------------------------------------------------------------------ 

35 # Loading 

36 # ------------------------------------------------------------------ 

37 

38 def _load_yaml_file(self, yaml_path: Path, profile: str) -> None: 

39 """Load a single permitted-characteristics YAML file.""" 

40 if not yaml_path.exists(): 

41 return 

42 

43 with yaml_path.open("r", encoding="utf-8") as fh: 

44 data = msgspec.yaml.decode(fh.read()) 

45 

46 if not isinstance(data, dict): 

47 return 

48 

49 data_dict = cast("dict[str, Any]", data) 

50 items_raw = data_dict.get("permitted_characteristics") 

51 if not isinstance(items_raw, list): 

52 return 

53 

54 entries: list[PermittedCharacteristicEntry] = [] 

55 for item in items_raw: 

56 if not isinstance(item, dict): 

57 continue 

58 service = item.get("service") 

59 chars_raw = item.get("characteristics") 

60 if not isinstance(service, str) or not isinstance(chars_raw, list): 

61 continue 

62 characteristics = tuple(str(c) for c in chars_raw if isinstance(c, str)) 

63 if characteristics: 

64 entries.append( 

65 PermittedCharacteristicEntry( 

66 service=service, 

67 characteristics=characteristics, 

68 ), 

69 ) 

70 

71 if entries: 

72 self._entries[profile] = entries 

73 

74 def _load(self) -> None: 

75 """Load all permitted-characteristics YAML files.""" 

76 uuids_path = find_bluetooth_sig_path() 

77 if not uuids_path: 

78 self._loaded = True 

79 return 

80 

81 profiles_path = uuids_path.parent / "profiles_and_services" 

82 if not profiles_path.exists(): 

83 self._loaded = True 

84 return 

85 

86 for profile_dir in _PROFILE_DIRS: 

87 dir_path = profiles_path / profile_dir 

88 if not dir_path.is_dir(): 

89 continue 

90 for yaml_file in sorted(dir_path.glob("*_permitted_characteristics.yaml")): 

91 self._load_yaml_file(yaml_file, profile_dir) 

92 

93 self._loaded = True 

94 

95 # ------------------------------------------------------------------ 

96 # Query API 

97 # ------------------------------------------------------------------ 

98 

99 def get_permitted_characteristics(self, profile: str) -> list[str]: 

100 """Get the flat list of permitted characteristic identifiers for a profile. 

101 

102 Args: 

103 profile: Profile key (e.g. ``"ess"``, ``"uds"``, ``"imds"``). 

104 

105 Returns: 

106 List of characteristic identifier strings, or an empty list. 

107 """ 

108 self._ensure_loaded() 

109 with self._lock: 

110 entries = self._entries.get(profile, []) 

111 return [c for entry in entries for c in entry.characteristics] 

112 

113 def get_entries(self, profile: str) -> list[PermittedCharacteristicEntry]: 

114 """Get the structured permitted-characteristic entries for a profile. 

115 

116 Args: 

117 profile: Profile key (e.g. ``"ess"``, ``"uds"``, ``"imds"``). 

118 

119 Returns: 

120 List of :class:`PermittedCharacteristicEntry` or an empty list. 

121 """ 

122 self._ensure_loaded() 

123 with self._lock: 

124 return list(self._entries.get(profile, [])) 

125 

126 def get_all_profiles(self) -> list[str]: 

127 """Return all loaded profile keys (sorted).""" 

128 self._ensure_loaded() 

129 with self._lock: 

130 return sorted(self._entries) 

131 

132 

133# Singleton instance for global use 

134permitted_characteristics_registry = PermittedCharacteristicsRegistry()