Coverage for src / bluetooth_sig / registry / company_identifiers / company_identifiers_registry.py: 82%

1"""Company Identifiers Registry for Bluetooth SIG manufacturer IDs.""" 

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 

12 

13 

14class CompanyIdentifierInfo(msgspec.Struct, frozen=True, kw_only=True): 

15 """Information about a Bluetooth SIG company identifier.""" 

16 

17 id: int 

18 name: str 

19 

20 

21class CompanyIdentifiersRegistry(BaseGenericRegistry[CompanyIdentifierInfo]): 

22 """Registry for Bluetooth SIG company identifiers with lazy loading. 

23 

24 This registry resolves manufacturer company IDs to company names from 

25 the official Bluetooth SIG assigned numbers. Data is lazily loaded on 

26 first access for performance. 

27 

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

29 """ 

30 

31 def __init__(self) -> None: 

32 """Initialize the company identifiers registry.""" 

33 super().__init__() 

34 self._companies: dict[int, CompanyIdentifierInfo] = {} 

35 

36 def _load_company_identifiers(self, yaml_path: Path) -> None: 

37 """Load company identifiers from YAML file. 

38 

39 Args: 

40 yaml_path: Path to the company_identifiers.yaml file 

41 """ 

42 if not yaml_path.exists(): 

43 return 

44 

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

46 data = msgspec.yaml.decode(file_handle.read()) 

47 

48 if not isinstance(data, dict): 

49 return 

50 

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

52 company_entries = data_dict.get("company_identifiers") 

53 if not isinstance(company_entries, list): 

54 return 

55 

56 # Load all company identifiers into cache 

57 for entry in company_entries: 

58 if isinstance(entry, dict): 

59 company_id = entry.get("value") 

60 company_name = entry.get("name") 

61 if company_id is not None and company_name: 

62 self._companies[company_id] = CompanyIdentifierInfo(id=company_id, name=company_name) 

63 

64 def _load(self) -> None: 

65 """Perform the actual loading of company identifiers data.""" 

66 # Use find_bluetooth_sig_path and navigate to company_identifiers 

67 uuids_path = find_bluetooth_sig_path() 

68 if not uuids_path: 

69 self._loaded = True 

70 return 

71 

72 # Navigate from uuids/ to company_identifiers/ 

73 base_path = uuids_path.parent / "company_identifiers" 

74 if not base_path.exists(): 

75 self._loaded = True 

76 return 

77 

78 yaml_path = base_path / "company_identifiers.yaml" 

79 if not yaml_path.exists(): 

80 self._loaded = True 

81 return 

82 

83 # Load company identifiers from YAML 

84 self._load_company_identifiers(yaml_path) 

85 self._loaded = True 

86 

87 def get_company_name(self, company_id: int) -> str | None: 

88 """Get company name by ID (lazy loads on first call). 

89 

90 Args: 

91 company_id: Manufacturer company identifier (e.g., 0x004C for Apple) 

92 

93 Returns: 

94 Company name or None if not found 

95 

96 Examples: 

97 >>> registry = CompanyIdentifiersRegistry() 

98 >>> registry.get_company_name(0x004C) 

99 'Apple, Inc.' 

100 >>> registry.get_company_name(0x0006) 

101 'Microsoft' 

102 >>> registry.get_company_name(0x00E0) 

103 'Google' 

104 >>> registry.get_company_name(0xFFFF) # Unknown ID 

105 None 

106 """ 

107 self._ensure_loaded() 

108 with self._lock: 

109 info = self._companies.get(company_id) 

110 return info.name if info else None 

111 

112 

113# Singleton instance for global use 

114company_identifiers_registry = CompanyIdentifiersRegistry()