Coverage for src/bluetooth_sig/device/peripheral_device.py: 100%
91 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-28 01:26 +0000
1"""High-level peripheral (GATT server) abstraction.
3Provides :class:`PeripheralDevice`, a server-side helper that hosts GATT
4services and encodes values for remote centrals to read.
5"""
7from __future__ import annotations
9import logging
10import warnings
12# Any is required: BaseCharacteristic is generic over its value type (T), but
13# PeripheralDevice hosts heterogeneous characteristics with different T types
14# in a single dict, so the container must erase the type parameter to Any.
15from typing import Any
17from bluetooth_sig.gatt.characteristics.base import BaseCharacteristic
18from bluetooth_sig.types.gatt_enums import GattProperty
19from bluetooth_sig.types.peripheral_types import CharacteristicDefinition, ServiceDefinition
20from bluetooth_sig.types.uuid import BluetoothUUID
22from .peripheral import PeripheralManagerProtocol
24logger = logging.getLogger(__name__)
27class HostedCharacteristic:
28 """Tracks a hosted characteristic with its definition and class instance.
30 Attributes:
31 definition: The GATT characteristic definition registered on the peripheral.
32 characteristic: The SIG characteristic class instance used for encoding/decoding.
33 last_value: The last Python value that was encoded and set on this characteristic.
35 """
37 __slots__ = ("characteristic", "definition", "last_value")
39 def __init__(
40 self,
41 definition: CharacteristicDefinition,
42 characteristic: BaseCharacteristic[Any],
43 initial_value: Any = None, # noqa: ANN401
44 ) -> None:
45 """Initialise a hosted characteristic record.
47 Args:
48 definition: The GATT characteristic definition registered on the peripheral.
49 characteristic: The SIG characteristic class instance for encoding/decoding.
50 initial_value: Optional initial Python value set on this characteristic.
52 """
53 self.definition = definition
54 self.characteristic = characteristic
55 self.last_value: Any = initial_value
58class PeripheralDevice:
59 """High-level BLE peripheral abstraction using composition pattern.
61 .. deprecated:: 0.5.0
62 Scheduled for removal; see ``docs/source/explanation/limitations.md``.
64 Coordinates between :class:`PeripheralManagerProtocol` (backend) and
65 ``BaseCharacteristic`` instances (encoding) so callers work with typed
66 Python values.
68 Encoding is handled directly by the characteristic's ``build_value()``
69 method — no translator is needed on the peripheral (server) side.
71 The workflow mirrors :class:`Device` but for the server role:
73 1. Create a ``PeripheralDevice`` wrapping a backend.
74 2. Add services with :meth:`add_service` (typed helpers encode initial values).
75 3. Start advertising with :meth:`start`.
76 4. Update characteristic values with :meth:`update_value`.
77 5. Stop with :meth:`stop`.
79 Example::
81 >>> from bluetooth_sig.gatt.characteristics import BatteryLevelCharacteristic
82 >>> from bluetooth_sig.gatt.services import BatteryService
83 >>>
84 >>> peripheral = PeripheralDevice(backend)
85 >>> battery_char = BatteryLevelCharacteristic()
86 >>> peripheral.add_characteristic(
87 ... service_uuid=BatteryService.get_class_uuid(),
88 ... characteristic=battery_char,
89 ... initial_value=85,
90 ... )
91 >>> await peripheral.start()
92 >>> await peripheral.update_value(battery_char, 72)
94 """
96 def __init__(
97 self,
98 peripheral_manager: PeripheralManagerProtocol,
99 ) -> None:
100 """Initialise PeripheralDevice.
102 Args:
103 peripheral_manager: Backend implementing PeripheralManagerProtocol.
105 """
106 warnings.warn(
107 "PeripheralDevice is deprecated and scheduled for removal; "
108 "use client-side Device + parse/encode APIs instead.",
109 DeprecationWarning,
110 stacklevel=2,
111 )
112 self._manager = peripheral_manager
114 # UUID (normalised upper-case) → HostedCharacteristic
115 self._hosted: dict[str, HostedCharacteristic] = {}
117 # Service UUID → ServiceDefinition (tracks services added via helpers)
118 self._pending_services: dict[str, ServiceDefinition] = {}
120 # ------------------------------------------------------------------
121 # Properties
122 # ------------------------------------------------------------------
124 @property
125 def name(self) -> str:
126 """Advertised device name."""
127 return self._manager.name
129 @property
130 def is_advertising(self) -> bool:
131 """Whether the peripheral is currently advertising."""
132 return self._manager.is_advertising
134 @property
135 def services(self) -> list[ServiceDefinition]:
136 """Registered GATT services."""
137 return self._manager.services
139 @property
140 def hosted_characteristics(self) -> dict[str, HostedCharacteristic]:
141 """Map of UUID → HostedCharacteristic for all hosted characteristics."""
142 return dict(self._hosted)
144 # ------------------------------------------------------------------
145 # Service & Characteristic Registration
146 # ------------------------------------------------------------------
148 def add_characteristic(
149 self,
150 service_uuid: str | BluetoothUUID,
151 characteristic: BaseCharacteristic[Any],
152 initial_value: Any, # noqa: ANN401
153 *,
154 properties: GattProperty | None = None,
155 ) -> CharacteristicDefinition:
156 """Register a characteristic on a service, encoding the initial value.
158 If the service has not been seen before, a new primary
159 :class:`ServiceDefinition` is created automatically.
161 Args:
162 service_uuid: UUID of the parent service.
163 characteristic: SIG characteristic class instance.
164 initial_value: Python value to encode as the initial value.
165 properties: GATT properties. Defaults to ``READ | NOTIFY``.
167 Returns:
168 The created :class:`CharacteristicDefinition`.
170 Raises:
171 RuntimeError: If the peripheral has already started advertising.
173 """
174 if self._manager.is_advertising:
175 raise RuntimeError("Cannot add characteristics after peripheral has started")
177 char_def = CharacteristicDefinition.from_characteristic(
178 characteristic,
179 initial_value,
180 properties=properties,
181 )
183 svc_key = str(service_uuid).upper()
184 if svc_key not in self._pending_services:
185 self._pending_services[svc_key] = ServiceDefinition(
186 uuid=BluetoothUUID(svc_key),
187 characteristics=[],
188 )
189 self._pending_services[svc_key].characteristics.append(char_def)
191 uuid_key = str(char_def.uuid).upper()
192 self._hosted[uuid_key] = HostedCharacteristic(
193 definition=char_def,
194 characteristic=characteristic,
195 initial_value=initial_value,
196 )
198 return char_def
200 async def add_service(self, service: ServiceDefinition) -> None:
201 """Register a pre-built service definition directly.
203 For full control over the service definition. If you prefer typed
204 helpers, use :meth:`add_characteristic` instead.
206 Args:
207 service: Complete service definition.
209 Raises:
210 RuntimeError: If the peripheral has already started advertising.
212 """
213 await self._manager.add_service(service)
215 # ------------------------------------------------------------------
216 # Lifecycle
217 # ------------------------------------------------------------------
219 async def start(self) -> None:
220 """Register pending services on the backend and start advertising.
222 All services added via :meth:`add_characteristic` are flushed to the
223 backend before ``start()`` is called on the manager.
225 Raises:
226 RuntimeError: If the peripheral has already started.
228 """
229 # Flush pending services to the backend
230 for service_def in self._pending_services.values():
231 await self._manager.add_service(service_def)
232 self._pending_services.clear()
234 await self._manager.start()
236 async def stop(self) -> None:
237 """Stop advertising and disconnect all clients."""
238 await self._manager.stop()
240 # ------------------------------------------------------------------
241 # Value Updates
242 # ------------------------------------------------------------------
244 async def update_value(
245 self,
246 characteristic: BaseCharacteristic[Any] | str | BluetoothUUID,
247 value: Any, # noqa: ANN401
248 *,
249 notify: bool = True,
250 ) -> None:
251 """Encode a typed value and push it to the hosted characteristic.
253 Args:
254 characteristic: The characteristic instance, UUID string, or BluetoothUUID.
255 value: Python value to encode via ``build_value()``.
256 notify: Whether to notify subscribed centrals. Default ``True``.
258 Raises:
259 KeyError: If the characteristic is not hosted on this peripheral.
260 RuntimeError: If the peripheral has not started.
262 """
263 uuid_key = self._resolve_uuid_key(characteristic)
264 hosted = self._hosted.get(uuid_key)
265 if hosted is None:
266 raise KeyError(f"Characteristic {uuid_key} is not hosted on this peripheral")
268 encoded = hosted.characteristic.build_value(value)
269 hosted.last_value = value
271 await self._manager.update_characteristic(uuid_key, encoded, notify=notify)
273 async def update_raw(
274 self,
275 char_uuid: str | BluetoothUUID,
276 raw_value: bytearray,
277 *,
278 notify: bool = True,
279 ) -> None:
280 """Push pre-encoded bytes to a hosted characteristic.
282 Use this when you already have the encoded value or the
283 characteristic does not have a SIG class registered.
285 Args:
286 char_uuid: UUID of the characteristic.
287 raw_value: Pre-encoded bytes to set.
288 notify: Whether to notify subscribed centrals.
290 Raises:
291 KeyError: If the characteristic UUID is not hosted.
292 RuntimeError: If the peripheral has not started.
294 """
295 uuid_key = str(char_uuid).upper()
296 await self._manager.update_characteristic(uuid_key, raw_value, notify=notify)
298 async def get_current_value(
299 self,
300 characteristic: BaseCharacteristic[Any] | str | BluetoothUUID,
301 ) -> Any: # noqa: ANN401
302 """Get the last Python value set for a hosted characteristic.
304 Args:
305 characteristic: The characteristic instance, UUID string, or BluetoothUUID.
307 Returns:
308 The last value passed to :meth:`update_value`, or the initial value.
310 Raises:
311 KeyError: If the characteristic is not hosted.
313 """
314 uuid_key = self._resolve_uuid_key(characteristic)
315 hosted = self._hosted.get(uuid_key)
316 if hosted is None:
317 raise KeyError(f"Characteristic {uuid_key} is not hosted on this peripheral")
318 return hosted.last_value
320 # ------------------------------------------------------------------
321 # Fluent Configuration Delegation
322 # ------------------------------------------------------------------
324 def with_manufacturer_data(self, company_id: int, payload: bytes) -> PeripheralDevice:
325 """Configure manufacturer data for advertising.
327 Args:
328 company_id: Bluetooth SIG company identifier.
329 payload: Manufacturer-specific payload bytes.
331 Returns:
332 Self for method chaining.
334 """
335 self._manager.with_manufacturer_id(company_id, payload)
336 return self
338 def with_tx_power(self, power_dbm: int) -> PeripheralDevice:
339 """Set TX power level.
341 Args:
342 power_dbm: Transmission power in dBm.
344 Returns:
345 Self for method chaining.
347 """
348 self._manager.with_tx_power(power_dbm)
349 return self
351 def with_connectable(self, connectable: bool) -> PeripheralDevice:
352 """Set whether the peripheral accepts connections.
354 Args:
355 connectable: True to accept connections.
357 Returns:
358 Self for method chaining.
360 """
361 self._manager.with_connectable(connectable)
362 return self
364 def with_discoverable(self, discoverable: bool) -> PeripheralDevice:
365 """Set whether the peripheral is discoverable.
367 Args:
368 discoverable: True to be discoverable.
370 Returns:
371 Self for method chaining.
373 """
374 self._manager.with_discoverable(discoverable)
375 return self
377 # ------------------------------------------------------------------
378 # Internals
379 # ------------------------------------------------------------------
381 def _resolve_uuid_key(self, characteristic: BaseCharacteristic[Any] | str | BluetoothUUID) -> str:
382 """Normalise a characteristic reference to an upper-case UUID string."""
383 if isinstance(characteristic, BaseCharacteristic):
384 return str(characteristic.uuid).upper()
385 return str(characteristic).upper()
387 def __repr__(self) -> str:
388 """Return a developer-friendly representation."""
389 state = "advertising" if self.is_advertising else "stopped"
390 return (
391 f"PeripheralDevice(name={self.name!r}, "
392 f"state={state}, "
393 f"services={len(self.services)}, "
394 f"characteristics={len(self._hosted)})"
395 )
398__all__ = [
399 "HostedCharacteristic",
400 "PeripheralDevice",
401]