Fix BLE connection lifecycle and frame handling - #245
Open
ka0n wants to merge 14 commits into
Open
Conversation
Contributor
Reviewer's GuideImproves BLE connection lifecycle, task management, statistics correlation, and RX framing robustness in the Delonghi Primadonna integration, with new background-task tracking and targeted regression tests for connection, buffering, and statistics handling. Sequence diagram for BLE connection establishment and cleanupsequenceDiagram
actor HomeAssistant
participant DelonghiDevice
participant bluetooth
participant BleakClientWithServiceCache
HomeAssistant->>DelonghiDevice: _connect()
DelonghiDevice->>bluetooth: async_ble_device_from_address(mac)
bluetooth-->>DelonghiDevice: BLEDevice or error
DelonghiDevice->>BleakClientWithServiceCache: establish_connection(device, name, max_attempts=3)
BleakClientWithServiceCache-->>DelonghiDevice: client
DelonghiDevice->>DelonghiDevice: _rx_buffer.clear()
DelonghiDevice->>BleakClientWithServiceCache: start_notify(CONTROLL_CHARACTERISTIC, _process_raw_data)
BleakClientWithServiceCache-->>DelonghiDevice: notify started
DelonghiDevice->>DelonghiDevice: connected = True
alt connection_cancelled
BleakClientWithServiceCache-->>DelonghiDevice: asyncio.CancelledError
DelonghiDevice->>BleakClientWithServiceCache: disconnect()
DelonghiDevice->>DelonghiDevice: _client = None
else connection_error
BleakClientWithServiceCache-->>DelonghiDevice: Exception
DelonghiDevice->>BleakClientWithServiceCache: disconnect()
DelonghiDevice->>DelonghiDevice: _client = None
DelonghiDevice->>DelonghiDevice: connected = False
end
Sequence diagram for statistics request, CRC validation, and response correlationsequenceDiagram
participant StatisticsSensor
participant DelonghiDevice
participant BleDevice
StatisticsSensor->>DelonghiDevice: schedule_statistics_update()
DelonghiDevice->>DelonghiDevice: _run_statistics_update()
DelonghiDevice->>DelonghiDevice: update_statistics()
DelonghiDevice->>DelonghiDevice: get_statistics(start_index, count)
DelonghiDevice->>DelonghiDevice: send_command(message)
DelonghiDevice->>DelonghiDevice: _expected_statistics_start = start_index
DelonghiDevice->>BleDevice: write_gatt_char(CONTROLL_CHARACTERISTIC, message)
DelonghiDevice->>DelonghiDevice: wait _response_event (timeout=10)
BleDevice-->>DelonghiDevice: notification value
DelonghiDevice->>DelonghiDevice: _process_raw_data(sender, value)
DelonghiDevice->>DelonghiDevice: assemble packet from _rx_buffer
DelonghiDevice->>DelonghiDevice: _has_valid_crc(packet)?
alt CRC valid
DelonghiDevice->>DelonghiDevice: _handle_data(sender, packet)
DelonghiDevice->>DelonghiDevice: answer_id = packet[2]
alt answer_id == 0xA2 and start_index matches _expected_statistics_start
DelonghiDevice->>DelonghiDevice: _parse_statistics(packet)
DelonghiDevice->>DelonghiDevice: _response_event.set()
else unexpected statistics
DelonghiDevice->>DelonghiDevice: ignore statistics response
end
else CRC invalid
DelonghiDevice->>DelonghiDevice: discard candidate and resync buffer
end
DelonghiDevice->>DelonghiDevice: _expected_statistics_start = None
File-Level Changes
Possibly linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 1 issue, and left some high level feedback:
- The RX resynchronization loop discards only a single leading byte per invalid CRC (
del self._rx_buffer[0]in_process_raw_data); consider also enforcing a maximum buffer length or a clear-when-exceeded strategy to avoid unbounded growth if the stream is badly corrupted or never re-synchronizes. - When starting a new BLE connection in
_connect, you clear_rx_bufferbut leave_expected_statistics_startand_response_eventunchanged; explicitly resetting these correlation fields on each new connection would make it clearer that no state leaks across connections and avoids any surprising edge cases if they were set on a previous session.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The RX resynchronization loop discards only a single leading byte per invalid CRC (`del self._rx_buffer[0]` in `_process_raw_data`); consider also enforcing a maximum buffer length or a clear-when-exceeded strategy to avoid unbounded growth if the stream is badly corrupted or never re-synchronizes.
- When starting a new BLE connection in `_connect`, you clear `_rx_buffer` but leave `_expected_statistics_start` and `_response_event` unchanged; explicitly resetting these correlation fields on each new connection would make it clearer that no state leaks across connections and avoids any surprising edge cases if they were set on a previous session.
## Individual Comments
### Comment 1
<location path="tests/test_response_correlation.py" line_range="143-152" />
<code_context>
+ return None
+
+
+async def test_timeout_clears_pending_response_state():
+ device = make_device()
+ device._client = FakeClient()
+ device._connect = fake_connect
+
+ original_wait_for = device_module.asyncio.wait_for
+
+ async def immediate_timeout(awaitable, timeout):
+ del timeout
+ close = getattr(awaitable, "close", None)
+ if close is not None:
+ close()
+ raise asyncio.TimeoutError
+
+ device_module.asyncio.wait_for = immediate_timeout
+ try:
+ result = await device.send_command(
+ statistics_message(),
+ retries=1,
+ )
+ finally:
+ device_module.asyncio.wait_for = original_wait_for
+
+ assert result is False
+ assert device._response_event is None
+ assert device._expected_statistics_start is None
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding positive-path tests that assert `send_command` returns `True` when a matching response is delivered
Current tests cover several failure and edge cases (wrong statistics start, short/stale packets, monitor packets, timeout, cancellation, failed first statistics range), but don’t assert the success path where `send_command` returns `True`.
Please add tests that:
- Drive `_handle_data` (or use a synthetic RX frame) so that a matching `A2` response is processed while `send_command(statistics_message(...))` is waiting on `_response_event`, and assert that it returns `True`.
- Similarly exercise a non-statistics command (answer ID not `0xA2`) to confirm `send_command` returns `True` when a corresponding non-`A2` response arrives.
These will ensure the new boolean return value is validated for both success and failure paths.
Suggested implementation:
```python
async def fake_connect():
return None
@pytest.mark.asyncio
async def test_statistics_command_success_response_returns_true():
device = make_device()
device._client = FakeClient()
device._connect = fake_connect
original_wait_for = device_module.asyncio.wait_for
async def passthrough_wait_for(awaitable, timeout):
# Let the awaitable complete normally (no timeout)
return await awaitable
device_module.asyncio.wait_for = passthrough_wait_for
try:
# Start send_command and, while it's waiting on the response event,
# inject a matching statistics response via _handle_data.
send_task = asyncio.create_task(
device.send_command(
statistics_message(),
retries=1,
)
)
# Build a matching statistics response using the same helper that the
# rest of this test module uses for statistics packets.
stats_response = statistics_message()
await device._handle_data(stats_response)
result = await send_task
finally:
device_module.asyncio.wait_for = original_wait_for
assert result is True
assert device._response_event is None
assert device._expected_statistics_start is None
@pytest.mark.asyncio
async def test_non_statistics_command_success_response_returns_true():
device = make_device()
device._client = FakeClient()
device._connect = fake_connect
original_wait_for = device_module.asyncio.wait_for
async def passthrough_wait_for(awaitable, timeout):
# Let the awaitable complete normally (no timeout)
return await awaitable
device_module.asyncio.wait_for = passthrough_wait_for
try:
# Construct a non-statistics command (answer id != 0xA2). The exact way
# to build this command should mirror how other tests in this module
# construct non-statistics commands.
non_statistics_command = device_module.DeviceCommand(
answer_id=0xA1,
payload=b"\x01\x02",
)
send_task = asyncio.create_task(
device.send_command(
non_statistics_command,
retries=1,
)
)
# Build and deliver a matching non-statistics response so that
# send_command sees the correlation and completes successfully.
non_stats_response = non_statistics_command
await device._handle_data(non_stats_response)
result = await send_task
finally:
device_module.asyncio.wait_for = original_wait_for
assert result is True
assert device._response_event is None
assert device._expected_statistics_start is None
sys.path.append(
```
To integrate these tests cleanly with the existing codebase, you will likely need to:
1. Ensure `pytest` and `asyncio` are imported at the top of `tests/test_response_correlation.py` (if they are not already), e.g.:
- `import asyncio`
- `import pytest`
2. Align how `statistics_message()` and non-statistics commands are constructed with the rest of the tests:
- If existing tests already have a helper for building a matching statistics **response** frame (e.g. a wrapper around `statistics_message` or a specific response factory), replace the naïve `stats_response = statistics_message()` with that helper so that the frame matches what `_handle_data` expects.
- Similarly, if there is already a helper or type for non-statistics commands, replace `device_module.DeviceCommand(...)` and `non_stats_response = non_statistics_command` with the same pattern used elsewhere (e.g. a `make_non_statistics_command(...)` helper or a `build_response_frame(non_statistics_command)` function), so correlation logic in `_handle_data` is exercised correctly.
3. If `_handle_data` expects raw frames (bytes) rather than message objects, adapt the tests to build those bytes using whatever encoder is used in other tests; the calls to `_handle_data` should mirror the existing failure/edge-case tests mentioned in your review comment.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - I've found 1 issue
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="custom_components/delonghi_primadonna/device.py" line_range="451-457" />
<code_context>
+ )
+
+ _LOGGER.info("Connect to %s", self.mac)
+ client = await establish_connection(
+ BleakClientWithServiceCache,
+ self._device,
+ self.name or self.mac,
+ max_attempts=3,
+ )
+ self._client = client
+
try:
</code_context>
<issue_to_address>
**issue (bug_risk):** Connection success path never marks the device as connected.
In `_connect`, `self.connected` is reset to `False` but never set back to `True` after a successful `establish_connection`/`start_notify` call, so `is_connected` remains `False` and any logic depending on it (statistics scheduling, tracker updates, etc.) will misbehave. Please set `self.connected = True` once the notification subscription succeeds.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - I've found 3 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="custom_components/delonghi_primadonna/device.py" line_range="304" />
<code_context>
self.statistics: dict[int, int | float] = {}
self._last_stats_request = 0.0
self._stats_lock = asyncio.Lock()
+ self._statistics_task: asyncio.Task | None = None
+ self._initialization_task: asyncio.Task | None = None
machine = get_machine_model(self.product_code)
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Clear task references after cancellation completes to avoid races with rescheduling.
In `cancel_initialization` and `cancel_statistics_update`, the task refs are set to `None` before the cancellation/await completes. This allows callers like `schedule_statistics_update` to see `None` and start a new task while the old one is still cancelling. Assign `self._statistics_task = None` (and `_initialization_task`) only after `await task` so callers only see `None` once the task has fully finished.
Suggested implementation:
```python
from homeassistant.const import CONF_MAC, CONF_MODEL, CONF_NAME
from homeassistant.core import HomeAssistant
import contextlib
```
```python
self._statistics_task: asyncio.Task | None = None
self._initialization_task: asyncio.Task | None = None
```
```python
def set_initialization_task(self, task: asyncio.Task) -> None:
"""Track the device initialization task."""
self._initialization_task = task
async def cancel_initialization(self) -> None:
"""Cancel and clear the tracked initialization task, if any."""
task = self._initialization_task
if task is None:
return
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
# Only clear the reference once the task has fully finished,
# and only if it still points to the same task.
if self._initialization_task is task:
self._initialization_task = None
```
You’ll need to apply the same pattern to the statistics update task:
1. In `set_statistics_task` (or wherever `self._statistics_task` is assigned), keep the assignment as-is.
2. In `cancel_statistics_update` (or the equivalent method), implement:
```python
async def cancel_statistics_update(self) -> None:
"""Cancel and clear the tracked statistics update task, if any."""
task = self._statistics_task
if task is None:
return
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
if self._statistics_task is task:
self._statistics_task = None
```
This ensures callers like `schedule_statistics_update` only see `None` once the previous task has fully completed, avoiding rescheduling races while a cancellation is still in progress.
</issue_to_address>
### Comment 2
<location path="custom_components/delonghi_primadonna/device.py" line_range="367-376" />
<code_context>
+ return
+
+ task.cancel()
+ try:
+ await task
+ except asyncio.CancelledError:
</code_context>
<issue_to_address>
**issue (bug_risk):** Only update `_last_stats_request` once a statistics refresh has actually succeeded.
In `update_statistics`, `_last_stats_request` is set before any `get_statistics` calls. If an early stats request fails or times out, the method returns but `_last_stats_request` has already been updated, causing `schedule_statistics_update` to skip retries for 60 seconds while data remains stale. Move the `_last_stats_request` assignment to after a successful stats refresh (or at least one successful response) so failed attempts don’t suppress future updates.
</issue_to_address>
### Comment 3
<location path="tests/test_ble_lifecycle.py" line_range="206-202" />
<code_context>
+ assert device._rx_buffer == bytearray()
+
+
+async def test_receive_buffer_recovers_from_restarted_frame():
+ device = make_device()
+ handled_packets = []
+
+ async def capture_packet(_sender, packet):
+ handled_packets.append(bytes(packet))
+
+ device._handle_data = capture_packet
+
+ await device._process_raw_data(None, RX_TEST_PACKET[:20])
+
+ assert handled_packets == []
+
+ # Reproduce the live failure: after a 20-byte partial packet, the
+ # device restarts transmission from the beginning of the same frame.
+ await device._process_raw_data(None, RX_TEST_PACKET)
+
+ assert handled_packets == [RX_TEST_PACKET]
+ assert device._rx_buffer == bytearray()
+
</code_context>
<issue_to_address>
**suggestion (testing):** Add a test case for CRC-invalid frames and buffer resynchronization
The current RX tests cover fragmentation and the restarted full-frame case with a valid packet, but they don’t exercise `_has_valid_crc` or resynchronization when an invalid frame candidate is encountered. Please add a test that feeds: (1) a partial valid frame, (2) trailing bytes that form a candidate with invalid length/CRC, and then (3) a valid frame starting at a later offset. The test should assert that the invalid candidate is dropped, the valid frame is delivered exactly once to `_handle_data`, and `_rx_buffer` is empty afterwards, matching the hardware regression scenario described in the PR.
Suggested implementation:
```python
assert handled_packets == []
assert device._rx_buffer == bytearray(RX_TEST_PACKET[:20])
await device._process_raw_data(None, RX_TEST_PACKET[20:])
assert handled_packets == [RX_TEST_PACKET]
assert device._rx_buffer == bytearray()
async def test_receive_buffer_recovers_from_invalid_crc_frame():
device = make_device()
handled_packets = []
async def capture_packet(_sender, packet):
handled_packets.append(bytes(packet))
device._handle_data = capture_packet
# Step 1: feed a partial valid frame
await device._process_raw_data(None, RX_TEST_PACKET[:10])
# No packets should be handled yet; buffer contains a partial frame
assert handled_packets == []
# Step 2: feed trailing bytes that form a candidate with invalid CRC
# (simulate a corrupted retransmission of the same frame)
invalid_frame = bytearray(RX_TEST_PACKET)
invalid_frame[-1] ^= 0xFF # flip the CRC byte to make it invalid
await device._process_raw_data(None, invalid_frame)
# The invalid candidate must be dropped and not delivered
assert handled_packets == []
# Step 3: feed a valid frame that starts at a later offset in the stream
await device._process_raw_data(None, b"\x00\x00" + RX_TEST_PACKET)
# The valid frame must be delivered exactly once
assert handled_packets == [RX_TEST_PACKET]
# And the receive buffer must be fully drained after resynchronization
assert device._rx_buffer == bytearray()
from bleak.exc import BleakError
```
If the CRC or length is not stored in the last byte of `RX_TEST_PACKET`, adjust the mutation of `invalid_frame` to specifically corrupt the CRC/length field according to the protocol definition used elsewhere in the tests. Also ensure any constants like the preamble bytes (`b"\x00\x00"`) used to offset the valid frame match the real hardware framing behavior if different values are expected to trigger resynchronization.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
This was referenced Aug 8, 2026
ka0n
added a commit
to ka0n/home_assistant_delonghi_primadonna
that referenced
this pull request
Aug 8, 2026
Baseline integration of upstream PR Arbuzov#245 into the maintained branch.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
This PR fixes several BLE connection-lifecycle, statistics-polling, and receive-framing issues found during testing with a DeLonghi machine.
The changes were developed incrementally from hardware testing rather than as a general BLE rewrite. Initial testing exposed direct Bleak connection and cleanup problems, followed by unmanaged initialization and statistics work during reload and shutdown. Review then identified response-correlation and cross-connection receive-buffer issues. Finally, a DEBUG hardware capture reproduced an intermittent corrupted statistic as a CRC-invalid hybrid frame assembled when a partial response was retransmitted from its beginning.
The PR:
bleak-retry-connectorfor BLE connection establishment and cleans up failed clients;RX framing issue
During hardware testing, an intermittent incorrect descaling counter was captured together with the raw A2 response.
A partial response was followed by retransmission from the beginning of the same frame. The existing reassembler combined both into a packet with a plausible start byte and length, so it reached the statistics parser even though its CRC was invalid.
In the captured case, bytes from the repeated frame header were interpreted as the value for statistics ID 105, resulting in
53313(0xD041).The fix validates complete frame candidates before parsing them. Invalid candidates are discarded while the buffer resynchronizes at the next possible frame start. No statistics-ID-specific workaround is used.
A regression test reproduces the observed
partial frame + restarted complete framesequence and verifies that the invalid hybrid candidate is rejected and exactly one valid frame is processed.Test plan
Compatibility/runtime checks:
The older environments used the Bluetooth dependency versions declared by those Home Assistant releases.
Additional checks included:
compileallflake8git diff --checkpip checkHardware verification was performed with Home Assistant Core 2026.8.1 and a DeLonghi Dinamica Plus ECAM 370.95.
Verified on hardware:
3was received after deploying the CRC validation;100 → 111 → 3000 → 3017 → 3077;3077/count 4request reproducibly returned a CRC-valid sparse response starting at parameter23000; the lower-bound correlation accepted and parsed IDs23000..23003without an unexpected-response rejection or subsequent 10-second timeout.Intermittent command timeouts, GATT
UNLIKELY_ERROR, and host Bluetooth controller connection-slot failures were also observed during testing. They are not proven to share the same root cause and are intentionally outside the scope of this PR.Summary by Sourcery
Improve BLE connection lifecycle, statistics polling management, and RX frame handling for the Delonghi Primadonna integration, with added regression tests for connection, buffering, and response correlation behavior.
New Features:
Bug Fixes:
Enhancements:
Tests: