Coverage for src/bluetooth_sig/registry/object_types.py: 94%

1"""Object types registry for Bluetooth SIG OTS object type definitions.""" 

2 

3from __future__ import annotations 

4 

5import threading 

6 

7import msgspec 

8 

9from bluetooth_sig.types.uuid import BluetoothUUID 

10 

11from .utils import find_bluetooth_sig_path, load_yaml_uuids, parse_bluetooth_uuid 

12 

13 

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

15 """Information about a Bluetooth SIG object type.""" 

16 

17 uuid: BluetoothUUID 

18 name: str 

19 id: str 

20 

21 

22class ObjectTypesRegistry: 

23 """Registry for Bluetooth SIG Object Transfer Service (OTS) object types.""" 

24 

25 def __init__(self) -> None: 

26 """Initialize the object types registry.""" 

27 self._lock = threading.RLock() 

28 self._object_types: dict[str, ObjectTypeInfo] = {} # normalized_uuid -> ObjectTypeInfo 

29 self._object_types_by_name: dict[str, ObjectTypeInfo] = {} # lower_name -> ObjectTypeInfo 

30 self._object_types_by_id: dict[str, ObjectTypeInfo] = {} # id -> ObjectTypeInfo 

31 

32 try: 

33 self._load_object_types() 

34 except (FileNotFoundError, Exception): # pylint: disable=broad-exception-caught 

35 # If YAML loading fails, continue with empty registry 

36 pass 

37 

38 def _load_object_types(self) -> None: 

39 """Load object type UUIDs from YAML file.""" 

40 base_path = find_bluetooth_sig_path() 

41 if not base_path: 

42 return 

43 

44 # Load object type UUIDs 

45 object_types_yaml = base_path / "object_types.yaml" 

46 if object_types_yaml.exists(): 

47 for object_type_info in load_yaml_uuids(object_types_yaml): 

48 uuid = object_type_info["uuid"] 

49 

50 bt_uuid = BluetoothUUID(uuid) 

51 info = ObjectTypeInfo(uuid=bt_uuid, name=object_type_info["name"], id=object_type_info["id"]) 

52 # Store using short form as key for easy lookup 

53 self._object_types[bt_uuid.short_form.upper()] = info 

54 # Also store by name and id for reverse lookup 

55 self._object_types_by_name[object_type_info["name"].lower()] = info 

56 self._object_types_by_id[object_type_info["id"]] = info 

57 

58 def get_object_type_info(self, uuid: str | int | BluetoothUUID) -> ObjectTypeInfo | None: 

59 """Get object type information by UUID. 

60 

61 Args: 

62 uuid: 16-bit UUID as string (with or without 0x), int, or BluetoothUUID 

63 

64 Returns: 

65 ObjectTypeInfo object, or None if not found 

66 """ 

67 with self._lock: 

68 try: 

69 bt_uuid = parse_bluetooth_uuid(uuid) 

70 

71 # Get the short form (16-bit) for lookup 

72 short_key = bt_uuid.short_form.upper() 

73 return self._object_types.get(short_key) 

74 except ValueError: 

75 return None 

76 

77 def get_object_type_info_by_name(self, name: str) -> ObjectTypeInfo | None: 

78 """Get object type information by name. 

79 

80 Args: 

81 name: Object type name (case-insensitive) 

82 

83 Returns: 

84 ObjectTypeInfo object, or None if not found 

85 """ 

86 with self._lock: 

87 return self._object_types_by_name.get(name.lower()) 

88 

89 def get_object_type_info_by_id(self, object_type_id: str) -> ObjectTypeInfo | None: 

90 """Get object type information by ID. 

91 

92 Args: 

93 object_type_id: Object type ID (e.g., "org.bluetooth.object.track") 

94 

95 Returns: 

96 ObjectTypeInfo object, or None if not found 

97 """ 

98 with self._lock: 

99 return self._object_types_by_id.get(object_type_id) 

100 

101 def is_object_type_uuid(self, uuid: str | int | BluetoothUUID) -> bool: 

102 """Check if a UUID is a registered object type UUID. 

103 

104 Args: 

105 uuid: UUID to check 

106 

107 Returns: 

108 True if the UUID is an object type UUID, False otherwise 

109 """ 

110 return self.get_object_type_info(uuid) is not None 

111 

112 def get_all_object_types(self) -> list[ObjectTypeInfo]: 

113 """Get all registered object types. 

114 

115 Returns: 

116 List of all ObjectTypeInfo objects 

117 """ 

118 with self._lock: 

119 return list(self._object_types.values()) 

120 

121 

122# Global instance 

123object_types_registry = ObjectTypesRegistry()