Upstream/aioesphomeapi
1
0
Fork 0
mirror of https://github.com/esphome/aioesphomeapi.git synced 2024-11-23 12:15:13 +01:00
Code Issues Projects Releases Wiki Activity

Refactor bluetooth GATT writes to reduce duplicate code (#771)

Browse Source
This commit is contained in:
J. Nick Koston 2023-11-28 08:16:53 -06:00 committed by GitHub
parent 15e6279246
commit 5e0b6c1b07
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 33 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
aioesphomeapi/client.py
View File

@ -809,30 +809,6 @@ class APIClient:
timeout,
)
async def bluetooth_gatt_write(
self,
address: int,
handle: int,
data: bytes,
response: bool,
timeout: float = DEFAULT_BLE_TIMEOUT,
) -> None:
req = BluetoothGATTWriteRequest(
address=address, handle=handle, response=response, data=data
)
if not response:
self._get_connection().send_message(req)
return
await self._send_bluetooth_message_await_response(
address,
handle,
req,
BluetoothGATTWriteResponse,
timeout,
)
async def bluetooth_gatt_read_descriptor(
self,
address: int,
@ -868,6 +844,22 @@ class APIClient:
assert isinstance(resp, BluetoothGATTReadResponse)
return bytearray(resp.data)
async def bluetooth_gatt_write(
self,
address: int,
handle: int,
data: bytes,
response: bool,
timeout: float = DEFAULT_BLE_TIMEOUT,
) -> None:
await self._bluetooth_gatt_write(
address,
handle,
BluetoothGATTWriteRequest(response=response, data=data),
timeout,
response,
)
async def bluetooth_gatt_write_descriptor(
self,
address: int,
@ -876,14 +868,28 @@ class APIClient:
timeout: float = DEFAULT_BLE_TIMEOUT,
wait_for_response: bool = True,
) -> None:
req = BluetoothGATTWriteDescriptorRequest(
address=address, handle=handle, data=data
await self._bluetooth_gatt_write(
address,
handle,
BluetoothGATTWriteDescriptorRequest(data=data),
timeout,
wait_for_response,
)
async def _bluetooth_gatt_write(
self,
address: int,
handle: int,
req: BluetoothGATTWriteDescriptorRequest | BluetoothGATTWriteRequest,
timeout: float,
wait_for_response: bool,
) -> None:
"""Perform a GATT write to a char or descriptor."""
req.address = address
req.handle = handle
if not wait_for_response:
self._get_connection().send_message(req)
return
await self._send_bluetooth_message_await_response(
address,
handle,