Coverage for src / bluetooth_sig / core / async_context.py: 85%

1"""Async context manager for device parsing sessions.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Mapping 

6from types import TracebackType 

7from typing import Any, TypeVar, cast, overload 

8 

9from ..gatt.characteristics.base import BaseCharacteristic 

10from ..types import CharacteristicContext 

11from ..types.uuid import BluetoothUUID 

12from .translator import BluetoothSIGTranslator 

13 

14# Type variable for generic characteristic return types 

15T = TypeVar("T") 

16 

17 

18class AsyncParsingSession: 

19 """Async context manager for parsing sessions. 

20 

21 Maintains parsing context across multiple async operations. 

22 

23 Example:: 

24 

25 async with AsyncParsingSession() as session: 

26 result1 = await session.parse("2A19", data1) 

27 result2 = await session.parse("2A6E", data2) 

28 # Context automatically shared between parses 

29 """ 

30 

31 def __init__( 

32 self, 

33 translator: BluetoothSIGTranslator, 

34 ctx: CharacteristicContext | None = None, 

35 ) -> None: 

36 """Initialize parsing session. 

37 

38 Args: 

39 translator: Translator instance to use for parsing 

40 ctx: Optional initial context 

41 """ 

42 self.translator = translator 

43 self.context = ctx 

44 # Store parsed characteristic values for context sharing 

45 self.results: dict[str, Any] = {} 

46 

47 async def __aenter__(self) -> AsyncParsingSession: 

48 """Enter async context.""" 

49 return self 

50 

51 async def __aexit__( 

52 self, 

53 exc_type: type[BaseException] | None, 

54 exc_val: BaseException | None, 

55 exc_tb: TracebackType | None, 

56 ) -> bool: 

57 """Exit async context.""" 

58 # Cleanup if needed 

59 return False 

60 

61 @overload 

62 async def parse( 

63 self, 

64 char: type[BaseCharacteristic[T]], 

65 data: bytes, 

66 ) -> T: ... 

67 

68 @overload 

69 async def parse( 

70 self, 

71 char: str | BluetoothUUID, 

72 data: bytes, 

73 ) -> Any: ... # noqa: ANN401 # Runtime UUID dispatch cannot be type-safe 

74 

75 async def parse( 

76 self, 

77 char: str | BluetoothUUID | type[BaseCharacteristic[T]], 

78 data: bytes, 

79 ) -> T | Any: # noqa: ANN401 # Runtime UUID dispatch cannot be type-safe 

80 """Parse characteristic with accumulated context. 

81 

82 Args: 

83 char: Characteristic class (type-safe) or UUID string/BluetoothUUID (not type-safe). 

84 data: Raw bytes 

85 

86 Returns: 

87 Parsed characteristic value. Return type is inferred when passing characteristic class. 

88 """ 

89 # Update context with previous results 

90 # Cast dict to Mapping to satisfy CharacteristicContext type requirements 

91 results_as_mapping = cast(Mapping[str, Any], self.results) 

92 

93 if self.context is None: 

94 self.context = CharacteristicContext(other_characteristics=results_as_mapping) 

95 else: 

96 self.context = CharacteristicContext( 

97 device_info=self.context.device_info, 

98 advertisement=self.context.advertisement, 

99 other_characteristics=results_as_mapping, 

100 raw_service=self.context.raw_service, 

101 ) 

102 

103 # Handle characteristic class input (type-safe path) 

104 if isinstance(char, type) and issubclass(char, BaseCharacteristic): 

105 result = await self.translator.parse_characteristic_async(char, data, self.context) 

106 # Store result using UUID string key 

107 char_instance = char() 

108 uuid_str = str(char_instance.uuid) 

109 self.results[uuid_str] = result 

110 return result 

111 

112 # Parse with context (not type-safe path) 

113 uuid_str = str(char) if isinstance(char, BluetoothUUID) else char 

114 result = await self.translator.parse_characteristic_async(uuid_str, data, self.context) 

115 

116 # Store result for future context using string UUID key 

117 self.results[uuid_str] = result 

118 

119 return result