Coverage for src / bluetooth_sig / core / service_manager.py: 56%

1"""Discovered service lifecycle management. 

2 

3Owns the only mutable state from the original translator: the _services dict. 

4Handles process_services, get_service_by_uuid, discovered_services, clear_services. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import Any 

10 

11from ..gatt.services.base import BaseGattService 

12from ..gatt.services.registry import GattServiceRegistry 

13from ..types import CharacteristicInfo 

14from ..types.gatt_enums import WIRE_TYPE_MAP 

15from ..types.uuid import BluetoothUUID 

16 

17# Type alias for characteristic data in process_services 

18CharacteristicDataDict = dict[str, Any] 

19 

20 

21class ServiceManager: 

22 """Manages discovered GATT services. 

23 

24 This is the **only** delegate that holds mutable state — the dict of 

25 discovered services keyed by normalised UUID strings. 

26 """ 

27 

28 def __init__(self) -> None: 

29 """Initialise with an empty services dict.""" 

30 # Performance: Use str keys (normalised UUIDs) for fast dict lookups 

31 self._services: dict[str, BaseGattService] = {} 

32 

33 def process_services(self, services: dict[str, dict[str, CharacteristicDataDict]]) -> None: 

34 """Process discovered services and their characteristics. 

35 

36 Args: 

37 services: Dictionary of service UUIDs to their characteristics 

38 

39 """ 

40 for uuid_str, service_data in services.items(): 

41 uuid = BluetoothUUID(uuid_str) 

42 characteristics: dict[BluetoothUUID, CharacteristicInfo] = {} 

43 for char_uuid_str, char_data in service_data.get("characteristics", {}).items(): 

44 char_uuid = BluetoothUUID(char_uuid_str) 

45 vtype_raw = char_data.get("value_type", "bytes") 

46 python_type: type | None = None 

47 if isinstance(vtype_raw, str): 

48 python_type = WIRE_TYPE_MAP.get(vtype_raw.lower()) 

49 elif isinstance(vtype_raw, type): 

50 python_type = vtype_raw 

51 characteristics[char_uuid] = CharacteristicInfo( 

52 uuid=char_uuid, 

53 name=char_data.get("name", ""), 

54 unit=char_data.get("unit", ""), 

55 python_type=python_type, 

56 ) 

57 service = GattServiceRegistry.create_service(uuid, characteristics) 

58 if service: 

59 self._services[str(uuid)] = service 

60 

61 def get_service_by_uuid(self, uuid: str) -> BaseGattService | None: 

62 """Get a service instance by UUID. 

63 

64 Args: 

65 uuid: The service UUID 

66 

67 Returns: 

68 Service instance if found, None otherwise 

69 

70 """ 

71 return self._services.get(uuid) 

72 

73 @property 

74 def discovered_services(self) -> list[BaseGattService]: 

75 """Get list of discovered service instances. 

76 

77 Returns: 

78 List of discovered service instances 

79 

80 """ 

81 return list(self._services.values()) 

82 

83 def clear_services(self) -> None: 

84 """Clear all discovered services.""" 

85 self._services.clear()