Skip to content

Always queue image delivery services #63

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
LordMike wants to merge 3 commits into
base: feat/clean-port
Choose a base branch
from
+224 −315
Open

Always queue image delivery services #63

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d58c5fb
Always queue image delivery services
LordMike Jul 16, 2026
b2917e0
Always start queued image delivery drain
LordMike Jul 16, 2026
ee661b4
Preserve newer queued uploads during delivery
LordMike Jul 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 48 additions & 8 deletions custom_components/opendisplay/delivery.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,9 @@ def __init__(self, hass: HomeAssistant, entry: OpenDisplayConfigEntry) -> None:
self._pending_config_resync: bool = False
self._last_error: str | None = None

self._active_upload: PendingUpload | None = None
self._delivering: bool = False
self._delivery_requested: bool = False
self._delivery_task: asyncio.Task[None] | None = None
self._unsub_device_seen: CALLBACK_TYPE | None = None

Expand Down Expand Up @@ -165,7 +167,9 @@ async def async_shutdown(self) -> None:
except Exception: # noqa: BLE001 - shutdown must not raise
_LOGGER.debug("Delivery task raised during shutdown", exc_info=True)
self._delivery_task = None
self._active_upload = None
self._delivering = False
self._delivery_requested = False

# -- public state -------------------------------------------------------

Expand Down Expand Up @@ -237,10 +241,17 @@ def _on_device_seen(self) -> None:
@callback
def notify_device_seen(self, source: str = "ble") -> None:
"""Start a delivery drain if there is pending work and none in flight."""
if self._delivering or not self._has_pending_work():
if not self._has_pending_work():
return
if self._delivering:
# A newer submission can arrive while an older frame is already
# streaming. Remember the request so latest-wins work is drained
# after the current BLE operation releases the single connection.
self._delivery_requested = True
return
_LOGGER.debug("%s: device seen (%s); starting delivery", self._address, source)
self._delivering = True
self._delivery_requested = False
self._delivery_task = self._entry.async_create_background_task(
self._hass, self._deliver(), f"opendisplay_delivery_{self._address}"
)
Expand All @@ -267,6 +278,7 @@ async def _deliver(self) -> None:
# read/connect timeouts as BLETimeoutError, so a bare TimeoutError here
# can only be the asyncio.timeout(DELIVERY_DEADLINE_S) firing.)
self._register_attempt_failure(
self._active_upload,
f"delivery deadline exceeded ({DELIVERY_DEADLINE_S:.0f}s)"
)
_LOGGER.error(
Expand All @@ -283,21 +295,27 @@ async def _deliver(self) -> None:
except (BLEConnectionError, BLETimeoutError) as err:
# Device slept mid-connect or the wake window was missed; keep the
# work queued and try again on the next wake.
self._register_attempt_failure(str(err))
self._register_attempt_failure(self._active_upload, str(err))
except (AuthenticationFailedError, AuthenticationRequiredError):
# Bad/rotated key: pause the upload and prompt the user to reauth.
_LOGGER.warning("%s: delivery auth failed; starting reauth", self._address)
if self._pending_upload is not None:
self._pending_upload.paused = True
upload = self._active_upload or self._pending_upload
if upload is not None and self._pending_upload is upload:
upload.paused = True
self._last_error = "auth"
self._entry.async_start_reauth(self._hass)
self._notify_state()
except OpenDisplayError as err:
_LOGGER.warning("%s: delivery failed: %s", self._address, err)
self._register_attempt_failure(str(err))
self._register_attempt_failure(self._active_upload, str(err))
finally:
delivery_requested = self._delivery_requested
self._active_upload = None
self._delivering = False
self._delivery_task = None
self._delivery_requested = False
if delivery_requested and self._has_pending_work():
self.notify_device_seen("follow-up")

async def _drain_once(self) -> None:
"""The connection + drain body (see `_deliver` for error handling)."""
Expand All @@ -313,7 +331,7 @@ async def _drain_once(self) -> None:
)
if ble_device is None:
# The advertisement that woke us has already aged out; retry next.
self._register_attempt_failure("device not connectable")
self._register_attempt_failure(None, "device not connectable")
return

upload = self._pending_upload
Expand All @@ -335,7 +353,9 @@ async def _drain_once(self) -> None:
) as device,
):
if upload is not None and not upload.paused:
self._active_upload = upload
await self._drain_upload(device, upload)
self._active_upload = None
if self._pending_config_resync:
await self._drain_resync(device)

Expand All @@ -350,6 +370,15 @@ async def _drain_upload(
)
if upload.cancel_deadline:
upload.cancel_deadline()
if self._pending_upload is not upload:
# A newer image was queued while this frame was in flight. The BLE
# stream used the captured upload object, so the panel gets one whole
# old frame; keep the newer slot pending for the next drain.
_LOGGER.debug(
"%s: delivered superseded content; newer content remains queued",
self._address,
)
return
self._pending_upload = None
self._last_error = None
# Bump the image entity's timestamp to when the frame actually landed.
Expand Down Expand Up @@ -434,14 +463,25 @@ def _give_up_upload(self, slot: PendingUpload, reason: str) -> None:

# -- helpers ------------------------------------------------------------

def _register_attempt_failure(self, reason: str) -> None:
def _register_attempt_failure(
self, upload: PendingUpload | None, reason: str
) -> None:
"""Record a failed wake attempt and keep the work queued.

After ``MAX_DELIVERY_ATTEMPTS`` failures the upload is given up and
dropped (``_give_up_upload``) so a frame that can never be delivered does
not retry on every wake until ``queue_timeout``.
"""
upload = self._pending_upload
upload = upload or self._pending_upload
if upload is not None and self._pending_upload is not upload:
# The failed BLE operation belonged to a superseded frame. Do not
# increment attempts or set errors on the newer pending frame.
_LOGGER.debug(
"%s: superseded delivery attempt failed (%s); newer content remains queued",
self._address,
reason,
)
return
if upload is not None:
upload.attempts += 1
if upload.attempts >= MAX_DELIVERY_ATTEMPTS:
Expand Down
127 changes: 31 additions & 96 deletions custom_components/opendisplay/services.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -305,18 +305,6 @@ class _DeviceUnavailable(Exception):
"""


# Probe budget for a probably-asleep tag: one connect attempt, short timeout.
# 5 s is >2x the observed 1-3 s connect-during-window latency (plus proxy
# slack), so a genuinely awake tag connects comfortably, while a dark ESP32
# (radio fully off in timer deep sleep) costs at most ~5 s before queuing —
# vs the ~40 s default budget (4 attempts x 10 s). Deliberately below both the
# 10 s wake window and the 15 s freshness horizon (wake_window_s +
# FRESHNESS_SLACK_S), so a probe triggered by a just-missed advert still lands
# inside the window it is betting on.
PROBE_CONNECT_TIMEOUT_S = 5.0
PROBE_MAX_ATTEMPTS = 1


async def _async_connect_and_run(
hass: HomeAssistant,
entry: "OpenDisplayConfigEntry",
Expand Down Expand Up @@ -463,17 +451,7 @@ async def _async_send_image(
rotate: Rotation = Rotation.ROTATE_0,
use_measured_palettes: bool = False,
) -> DeliveryReceipt:
"""Upload a PIL image, delivering live or queuing it for the next wake.

Returns a receipt describing whether the frame was delivered immediately or
queued (for a sleeping device). Non-sleepy devices always deliver live and
keep the original strict-failure behavior.
"""
# Split the upload into its heavy CPU half and its BLE-I/O half. The CPU
# work (rotate + fit + dither + encode + zlib on a full frame) is offloaded
# to an executor thread so it never blocks the event loop; only the BLE
# transfer runs on the loop. This mirrors what device.upload_image() does
# internally, but that call ran _prepare_image synchronously on the loop.
"""Prepare and queue a PIL image for asynchronous delivery."""
config = entry.runtime_data.device_config
display_cfg = config.displays[0] if config and config.displays else None
# Match upload_image(): only ask prepare_image() to build compressed data
Expand Down Expand Up @@ -513,77 +491,36 @@ async def _async_send_image(
# image entity immediately (D6), not only after a successful delivery.
jpeg = await hass.async_add_executor_job(_pil_to_jpeg, img)

profile = runtime.sleep_profile
# Image services are queue-only: a successful service call means "accepted
# for delivery", never "visible on the panel". Every set-up entry should
# have a DeliveryManager; fail loudly if setup invariants are broken.
manager = runtime.delivery
sleepy = manager is not None and profile.is_sleepy

def _queue() -> DeliveryReceipt:
assert manager is not None
return manager.submit_upload(
prepared=prepared,
refresh_mode=refresh_mode,
partial_state=state,
use_measured_palettes=use_measured_palettes,
preview_jpeg=jpeg,
device_id=device_id,
)

async def _upload(device: OpenDisplayDevice) -> None:
await device.upload_prepared_image(
prepared, refresh_mode=refresh_mode, state=state
if manager is None:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="upload_error",
translation_placeholders={"error": "delivery manager unavailable"},
)

# Freshness gate: a probably-asleep tag will not usually answer a live
# connect. Instead of queuing blind, spend one short connect attempt (the
# "probe") in case the tag is actually awake: its wake adverts may have
# been missed by the scanner, and Silabs tags advertise continuously even
# when their power config looks sleepy. A dark ESP32 costs at most
# ~PROBE_CONNECT_TIMEOUT_S before falling back to the queue. HA drops the
# connectable BLEDevice ~3-5 min after the last advert, so a long-dark or
# never-seen tag short-circuits to the queue at near-zero cost (no
# BLEDevice -> _DeviceUnavailable without any radio traffic).
probing = False
connect_kwargs: dict[str, Any] = {}
if sleepy:
last_seen = (
runtime.coordinator.data.last_seen if runtime.coordinator.data else None
)
if profile.probably_asleep(last_seen):
if not profile.probe_before_queue:
return _queue()
probing = True
connect_kwargs = {
"connect_timeout": PROBE_CONNECT_TIMEOUT_S,
"max_attempts": PROBE_MAX_ATTEMPTS,
}
# DeliveryManager owns latest-wins replacement, queue expiry, preview state,
# retries, and delivered/expired events. The service does not open BLE.
receipt = manager.submit_upload(
prepared=prepared,
refresh_mode=refresh_mode,
partial_state=state,
use_measured_palettes=use_measured_palettes,
preview_jpeg=jpeg,
device_id=device_id,
)

try:
await _async_connect_and_run(
hass,
entry,
_upload,
use_measured_palettes=use_measured_palettes,
reraise_ble=sleepy,
**connect_kwargs,
)
except _DeviceUnavailable:
# The device was dark, dropped, or refused the link; defer.
receipt = _queue()
if probing:
# Race: a wake advert arriving DURING the failed probe found no
# pending work (nothing queued yet), so no drain started. If the
# tag now looks fresh, kick a drain instead of waiting out a full
# sleep cycle. notify_device_seen is a no-op while one is running.
last_seen = (
runtime.coordinator.data.last_seen if runtime.coordinator.data else None
)
if not profile.probably_asleep(last_seen):
assert manager is not None
manager.notify_device_seen("post-probe")
return receipt
# Always ask the queue to drain in the background after a new submission.
# The service does not decide whether the device is currently reachable; the
# drain path owns reachability checks and leaves the frame queued if there is
# no connectable device yet. "queued-submit" is only an internal source label
# for logging/diagnostics, not a Home Assistant or BLE protocol term.
manager.notify_device_seen("queued-submit")

async_dispatcher_send(hass, f"{SIGNAL_IMAGE_UPDATED}_{entry.unique_id}", jpeg)
return DeliveryReceipt(status="delivered", expires_at=None)
return receipt


def _receipt_response(receipt: DeliveryReceipt) -> ServiceResponse:
Expand Down Expand Up @@ -623,12 +560,10 @@ async def _async_upload_image(call: ServiceCall) -> ServiceResponse:
translation_placeholders={"url": image_data},
)

# Latest-wins for upload_image specifically: a newer upload cancels an
# older still-running one instead of queuing behind it (an automation
# pushing a fresh snapshot supersedes the stale one). This composes with the
# per-MAC ble_lock without deadlocking: the cancel + await below happens
# before _async_send_image acquires the lock, so the cancelled task releases
# the lock (its `async with ble_lock` unwinds) before this one takes it.
# Latest-wins for upload_image specifically: a newer upload cancels older
# still-running download/prepare work before it can replace the queue slot.
# Once a frame reaches DeliveryManager.submit_upload(), the queue's own
# latest-wins behavior supersedes the previous pending frame.
current = asyncio.current_task()
if (prev := entry.runtime_data.upload_task) is not None and not prev.done():
prev.cancel()
Expand Down Expand Up @@ -997,4 +932,4 @@ def async_setup_services(hass: HomeAssistant) -> None:
supports_response=SupportsResponse.OPTIONAL,
)
hass.services.async_register(DOMAIN, "activate_led", _async_activate_led, schema=SCHEMA_ACTIVATE_LED)
hass.services.async_register(DOMAIN, "activate_buzzer", _async_activate_buzzer, schema=SCHEMA_ACTIVATE_BUZZER)
hass.services.async_register(DOMAIN, "activate_buzzer", _async_activate_buzzer, schema=SCHEMA_ACTIVATE_BUZZER)
Loading
Loading