Add OpenDisplay integration (#164048)

Co-authored-by: Norbert Rittel <norbert@rittel.de>

CODEOWNERS

@@ -1202,6 +1202,8 @@ build.json @home-assistant/supervisor
 /tests/components/open_meteo/ @frenck
 /homeassistant/components/open_router/ @joostlek
 /tests/components/open_router/ @joostlek
+/homeassistant/components/opendisplay/ @g4bri3lDev
+/tests/components/opendisplay/ @g4bri3lDev
 /homeassistant/components/openerz/ @misialq
 /tests/components/openerz/ @misialq
 /homeassistant/components/openevse/ @c00w @firstof9

homeassistant/components/opendisplay/__init__.py

@@ -0,0 +1,127 @@
+"""Integration for OpenDisplay BLE e-paper displays."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from opendisplay import (
+    BLEConnectionError,
+    BLETimeoutError,
+    GlobalConfig,
+    OpenDisplayDevice,
+    OpenDisplayError,
+)
+
+from homeassistant.components.bluetooth import async_ble_device_from_address
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers import config_validation as cv, device_registry as dr
+from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH
+from homeassistant.helpers.typing import ConfigType
+
+if TYPE_CHECKING:
+    from opendisplay.models import FirmwareVersion
+
+from .const import DOMAIN
+from .services import async_setup_services
+
+CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
+
+
+@dataclass
+class OpenDisplayRuntimeData:
+    """Runtime data for an OpenDisplay config entry."""
+
+    firmware: FirmwareVersion
+    device_config: GlobalConfig
+    is_flex: bool
+    upload_task: asyncio.Task | None = None
+
+
+type OpenDisplayConfigEntry = ConfigEntry[OpenDisplayRuntimeData]
+
+
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up the OpenDisplay integration."""
+    async_setup_services(hass)
+    return True
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: OpenDisplayConfigEntry) -> bool:
+    """Set up OpenDisplay from a config entry."""
+    address = entry.unique_id
+    if TYPE_CHECKING:
+        assert address is not None
+
+    ble_device = async_ble_device_from_address(hass, address, connectable=True)
+    if ble_device is None:
+        raise ConfigEntryNotReady(
+            f"Could not find OpenDisplay device with address {address}"
+        )
+
+    try:
+        async with OpenDisplayDevice(
+            mac_address=address, ble_device=ble_device
+        ) as device:
+            fw = await device.read_firmware_version()
+            is_flex = device.is_flex
+    except (BLEConnectionError, BLETimeoutError, OpenDisplayError) as err:
+        raise ConfigEntryNotReady(
+            f"Failed to connect to OpenDisplay device: {err}"
+        ) from err
+    device_config = device.config
+    if TYPE_CHECKING:
+        assert device_config is not None
+
+    entry.runtime_data = OpenDisplayRuntimeData(
+        firmware=fw,
+        device_config=device_config,
+        is_flex=is_flex,
+    )
+
+    # Will be moved to DeviceInfo object in entity.py once entities are added
+    manufacturer = device_config.manufacturer
+    display = device_config.displays[0]
+    color_scheme_enum = display.color_scheme_enum
+    color_scheme = (
+        str(color_scheme_enum)
+        if isinstance(color_scheme_enum, int)
+        else color_scheme_enum.name
+    )
+    size = (
+        f'{display.screen_diagonal_inches:.1f}"'
+        if display.screen_diagonal_inches is not None
+        else f"{display.pixel_width}x{display.pixel_height}"
+    )
+
+    dr.async_get(hass).async_get_or_create(
+        config_entry_id=entry.entry_id,
+        connections={(CONNECTION_BLUETOOTH, address)},
+        manufacturer=manufacturer.manufacturer_name,
+        model=f"{size} {color_scheme}",
+        sw_version=f"{fw['major']}.{fw['minor']}",
+        hw_version=f"{manufacturer.board_type_name or manufacturer.board_type} rev. {manufacturer.board_revision}"
+        if is_flex
+        else None,
+        configuration_url="https://opendisplay.org/firmware/config/"
+        if is_flex
+        else None,
+    )
+
+    return True
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, entry: OpenDisplayConfigEntry
+) -> bool:
+    """Unload a config entry."""
+    if (task := entry.runtime_data.upload_task) and not task.done():
+        task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await task
+
+    return True

homeassistant/components/opendisplay/config_flow.py

@@ -0,0 +1,130 @@
+"""Config flow for OpenDisplay integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from opendisplay import (
+    MANUFACTURER_ID,
+    BLEConnectionError,
+    OpenDisplayDevice,
+    OpenDisplayError,
+)
+import voluptuous as vol
+
+from homeassistant.components.bluetooth import (
+    BluetoothServiceInfoBleak,
+    async_ble_device_from_address,
+    async_discovered_service_info,
+)
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_ADDRESS
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class OpenDisplayConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for OpenDisplay."""
+
+    def __init__(self) -> None:
+        """Initialize the config flow."""
+        self._discovery_info: BluetoothServiceInfoBleak | None = None
+        self._discovered_devices: dict[str, BluetoothServiceInfoBleak] = {}
+
+    async def _async_test_connection(self, address: str) -> None:
+        """Connect to the device and verify it responds."""
+        ble_device = async_ble_device_from_address(self.hass, address, connectable=True)
+        if ble_device is None:
+            raise BLEConnectionError(f"Could not find connectable device for {address}")
+
+        async with OpenDisplayDevice(
+            mac_address=address, ble_device=ble_device
+        ) as device:
+            await device.read_firmware_version()
+
+    async def async_step_bluetooth(
+        self, discovery_info: BluetoothServiceInfoBleak
+    ) -> ConfigFlowResult:
+        """Handle the Bluetooth discovery step."""
+        await self.async_set_unique_id(discovery_info.address)
+        self._abort_if_unique_id_configured()
+        self._discovery_info = discovery_info
+        self.context["title_placeholders"] = {"name": discovery_info.name}
+
+        try:
+            await self._async_test_connection(discovery_info.address)
+        except OpenDisplayError:
+            return self.async_abort(reason="cannot_connect")
+        except Exception:
+            _LOGGER.exception("Unexpected error")
+            return self.async_abort(reason="unknown")
+
+        return await self.async_step_bluetooth_confirm()
+
+    async def async_step_bluetooth_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Confirm discovery."""
+        assert self._discovery_info is not None
+
+        if user_input is None:
+            self._set_confirm_only()
+            return self.async_show_form(
+                step_id="bluetooth_confirm",
+                description_placeholders=self.context["title_placeholders"],
+            )
+
+        return self.async_create_entry(title=self._discovery_info.name, data={})
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the user step to pick discovered device."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            address = user_input[CONF_ADDRESS]
+            await self.async_set_unique_id(address, raise_on_progress=False)
+            self._abort_if_unique_id_configured()
+
+            try:
+                await self._async_test_connection(address)
+            except OpenDisplayError:
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception("Unexpected error")
+                errors["base"] = "unknown"
+            else:
+                return self.async_create_entry(
+                    title=self._discovered_devices[address].name,
+                    data={},
+                )
+        else:
+            current_addresses = self._async_current_ids(include_ignore=False)
+            for discovery_info in async_discovered_service_info(self.hass):
+                address = discovery_info.address
+                if address in current_addresses or address in self._discovered_devices:
+                    continue
+                if MANUFACTURER_ID in discovery_info.manufacturer_data:
+                    self._discovered_devices[address] = discovery_info
+
+        if not self._discovered_devices:
+            return self.async_abort(reason="no_devices_found")
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_ADDRESS): vol.In(
+                        {
+                            addr: f"{info.name} ({addr})"
+                            for addr, info in self._discovered_devices.items()
+                        }
+                    )
+                }
+            ),
+            errors=errors,
+        )

homeassistant/components/opendisplay/const.py

@@ -0,0 +1,3 @@
+"""Constants for the OpenDisplay integration."""
+
+DOMAIN = "opendisplay"

homeassistant/components/opendisplay/icons.json

@@ -0,0 +1,7 @@
+{
+  "services": {
+    "upload_image": {
+      "service": "mdi:image-move"
+    }
+  }
+}

homeassistant/components/opendisplay/manifest.json

@@ -0,0 +1,18 @@
+{
+  "domain": "opendisplay",
+  "name": "OpenDisplay",
+  "bluetooth": [
+    {
+      "connectable": true,
+      "manufacturer_id": 9286
+    }
+  ],
+  "codeowners": ["@g4bri3lDev"],
+  "config_flow": true,
+  "dependencies": ["bluetooth_adapters", "http"],
+  "documentation": "https://www.home-assistant.io/integrations/opendisplay",
+  "integration_type": "device",
+  "iot_class": "local_push",
+  "quality_scale": "silver",
+  "requirements": ["py-opendisplay==5.2.0"]
+}

homeassistant/components/opendisplay/quality_scale.yaml

@@ -0,0 +1,103 @@
+rules:
+  # Bronze
+  action-setup: done
+  appropriate-polling:
+    status: exempt
+    comment: |
+      The `opendisplay` integration is a `local_push` integration that does not perform periodic polling.
+  brands: done
+  common-modules:
+    status: exempt
+    comment: Integration does not currently use entities or a DataUpdateCoordinator.
+  config-flow-test-coverage: done
+  config-flow: done
+  dependency-transparency: done
+  docs-actions: done
+  docs-high-level-description: done
+  docs-installation-instructions: done
+  docs-removal-instructions: done
+  entity-event-setup:
+    status: exempt
+    comment: Integration does not currently provide any entities.
+  entity-unique-id:
+    status: exempt
+    comment: Integration does not currently provide any entities.
+  has-entity-name:
+    status: exempt
+    comment: Integration does not currently provide any entities.
+  runtime-data: done
+  test-before-configure: done
+  test-before-setup: done
+  unique-config-entry: done
+
+  # Silver
+  action-exceptions: done
+  config-entry-unloading: done
+  docs-configuration-parameters:
+    status: exempt
+    comment: Integration has no options flow.
+  docs-installation-parameters: done
+  entity-unavailable:
+    status: exempt
+    comment: Integration does not currently provide any entities.
+  integration-owner: done
+  log-when-unavailable:
+    status: exempt
+    comment: Integration does not currently implement any entities or background polling.
+  parallel-updates:
+    status: exempt
+    comment: Integration does not provide any entities.
+  reauthentication-flow:
+    status: exempt
+    comment: Devices do not require authentication.
+  test-coverage: done
+
+  # Gold
+  devices: done
+  diagnostics: todo
+  discovery-update-info:
+    status: exempt
+    comment: The device's BLE MAC address is both its unique identifier and does not change.
+  discovery: done
+  docs-data-update:
+    status: exempt
+    comment: Integration does not poll or push data to entities.
+  docs-examples: todo
+  docs-known-limitations: todo
+  docs-supported-devices: todo
+  docs-supported-functions: todo
+  docs-troubleshooting: todo
+  docs-use-cases: todo
+  dynamic-devices:
+    status: exempt
+    comment: Only one device per config entry. New devices are set up as new entries.
+  entity-category:
+    status: exempt
+    comment: Integration does not provide any entities.
+  entity-device-class:
+    status: exempt
+    comment: Integration does not provide any entities.
+  entity-disabled-by-default:
+    status: exempt
+    comment: Integration does not provide any entities.
+  entity-translations:
+    status: exempt
+    comment: Integration does not provide any entities.
+  exception-translations: done
+  icon-translations: done
+  reconfiguration-flow:
+    status: exempt
+    comment: Reconfiguration would require selecting a new device, which is a new config entry.
+  repair-issues:
+    status: exempt
+    comment: Integration does not use repair issues.
+  stale-devices:
+    status: exempt
+    comment: Stale devices are removed with the config entry as there is only one device per entry.
+
+  # Platinum
+  async-dependency: done
+  inject-websession:
+    status: exempt
+    comment: The opendisplay library communicates over BLE and does not use HTTP.
+  strict-typing: todo

homeassistant/components/opendisplay/services.py

@@ -0,0 +1,228 @@
+"""Service registration for the OpenDisplay integration."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+import contextlib
+from datetime import timedelta
+from enum import IntEnum
+import io
+from typing import TYPE_CHECKING, Any
+
+import aiohttp
+from opendisplay import (
+    DitherMode,
+    FitMode,
+    OpenDisplayDevice,
+    OpenDisplayError,
+    RefreshMode,
+    Rotation,
+)
+from PIL import Image as PILImage, ImageOps
+import voluptuous as vol
+
+from homeassistant.components.bluetooth import async_ble_device_from_address
+from homeassistant.components.http.auth import async_sign_path
+from homeassistant.components.media_source import async_resolve_media
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.const import ATTR_DEVICE_ID
+from homeassistant.core import HomeAssistant, ServiceCall, callback
+from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
+from homeassistant.helpers import config_validation as cv, device_registry as dr
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH
+from homeassistant.helpers.network import get_url
+from homeassistant.helpers.selector import MediaSelector, MediaSelectorConfig
+
+if TYPE_CHECKING:
+    from . import OpenDisplayConfigEntry
+
+from .const import DOMAIN
+
+ATTR_IMAGE = "image"
+ATTR_ROTATION = "rotation"
+ATTR_DITHER_MODE = "dither_mode"
+ATTR_REFRESH_MODE = "refresh_mode"
+ATTR_FIT_MODE = "fit_mode"
+ATTR_TONE_COMPRESSION = "tone_compression"
+
+
+def _str_to_int_enum(enum_class: type[IntEnum]) -> Callable[[str], Any]:
+    """Return a validator that converts a lowercase enum name string to an enum member."""
+    members = {m.name.lower(): m for m in enum_class}
+
+    def validate(value: str) -> IntEnum:
+        if (result := members.get(value)) is None:
+            raise vol.Invalid(f"Invalid value: {value}")
+        return result
+
+    return validate
+
+
+SCHEMA_UPLOAD_IMAGE = vol.Schema(
+    {
+        vol.Required(ATTR_DEVICE_ID): cv.string,
+        vol.Required(ATTR_IMAGE): MediaSelector(
+            MediaSelectorConfig(accept=["image/*"])
+        ),
+        vol.Optional(ATTR_ROTATION, default=Rotation.ROTATE_0): vol.All(
+            vol.Coerce(int), vol.Coerce(Rotation)
+        ),
+        vol.Optional(ATTR_DITHER_MODE, default="burkes"): _str_to_int_enum(DitherMode),
+        vol.Optional(ATTR_REFRESH_MODE, default="full"): _str_to_int_enum(RefreshMode),
+        vol.Optional(ATTR_FIT_MODE, default="contain"): _str_to_int_enum(FitMode),
+        vol.Optional(ATTR_TONE_COMPRESSION): vol.All(
+            vol.Coerce(float), vol.Range(min=0.0, max=100.0)
+        ),
+    }
+)
+
+
+def _get_entry_for_device(call: ServiceCall) -> OpenDisplayConfigEntry:
+    """Return the config entry for the device targeted by a service call."""
+    device_id: str = call.data[ATTR_DEVICE_ID]
+    device_registry = dr.async_get(call.hass)
+
+    if (device := device_registry.async_get(device_id)) is None:
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="invalid_device_id",
+            translation_placeholders={"device_id": device_id},
+        )
+
+    mac_address = next(
+        (conn[1] for conn in device.connections if conn[0] == CONNECTION_BLUETOOTH),
+        None,
+    )
+    if mac_address is None:
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="invalid_device_id",
+            translation_placeholders={"device_id": device_id},
+        )
+
+    entry = call.hass.config_entries.async_entry_for_domain_unique_id(
+        DOMAIN, mac_address
+    )
+    if entry is None or entry.state is not ConfigEntryState.LOADED:
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="device_not_found",
+            translation_placeholders={"address": mac_address},
+        )
+
+    return entry
+
+
+def _load_image(path: str) -> PILImage.Image:
+    """Load an image from disk and apply EXIF orientation."""
+    image = PILImage.open(path)
+    image.load()
+    return ImageOps.exif_transpose(image)
+
+
+def _load_image_from_bytes(data: bytes) -> PILImage.Image:
+    """Load an image from bytes and apply EXIF orientation."""
+    image = PILImage.open(io.BytesIO(data))
+    image.load()
+    return ImageOps.exif_transpose(image)
+
+
+async def _async_download_image(hass: HomeAssistant, url: str) -> PILImage.Image:
+    """Download an image from a URL and return a PIL Image."""
+    if not url.startswith(("http://", "https://")):
+        url = get_url(hass) + async_sign_path(
+            hass, url, timedelta(minutes=5), use_content_user=True
+        )
+    session = async_get_clientsession(hass)
+    try:
+        async with session.get(url) as resp:
+            resp.raise_for_status()
+            data = await resp.read()
+    except aiohttp.ClientError as err:
+        raise HomeAssistantError(
+            translation_domain=DOMAIN,
+            translation_key="media_download_error",
+            translation_placeholders={"error": str(err)},
+        ) from err
+
+    return await hass.async_add_executor_job(_load_image_from_bytes, data)
+
+
+async def _async_upload_image(call: ServiceCall) -> None:
+    """Handle the upload_image service call."""
+    entry = _get_entry_for_device(call)
+    address = entry.unique_id
+    assert address is not None
+
+    image_data: dict[str, Any] = call.data[ATTR_IMAGE]
+    rotation: Rotation = call.data[ATTR_ROTATION]
+    dither_mode: DitherMode = call.data[ATTR_DITHER_MODE]
+    refresh_mode: RefreshMode = call.data[ATTR_REFRESH_MODE]
+    fit_mode: FitMode = call.data[ATTR_FIT_MODE]
+    tone_compression_pct: float | None = call.data.get(ATTR_TONE_COMPRESSION)
+    tone_compression: float | str = (
+        tone_compression_pct / 100.0 if tone_compression_pct is not None else "auto"
+    )
+
+    ble_device = async_ble_device_from_address(call.hass, address, connectable=True)
+    if ble_device is None:
+        raise HomeAssistantError(
+            translation_domain=DOMAIN,
+            translation_key="device_not_found",
+            translation_placeholders={"address": address},
+        )
+
+    current = asyncio.current_task()
+    if (prev := entry.runtime_data.upload_task) is not None and not prev.done():
+        prev.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await prev
+    entry.runtime_data.upload_task = current
+
+    try:
+        media = await async_resolve_media(
+            call.hass, image_data["media_content_id"], None
+        )
+
+        if media.path is not None:
+            pil_image = await call.hass.async_add_executor_job(
+                _load_image, str(media.path)
+            )
+        else:
+            pil_image = await _async_download_image(call.hass, media.url)
+
+        async with OpenDisplayDevice(
+            mac_address=address,
+            ble_device=ble_device,
+            config=entry.runtime_data.device_config,
+        ) as device:
+            await device.upload_image(
+                pil_image,
+                refresh_mode=refresh_mode,
+                dither_mode=dither_mode,
+                tone_compression=tone_compression,
+                fit=fit_mode,
+                rotate=rotation,
+            )
+    except asyncio.CancelledError:
+        return
+    except OpenDisplayError as err:
+        raise HomeAssistantError(
+            translation_domain=DOMAIN, translation_key="upload_error"
+        ) from err
+    finally:
+        if entry.runtime_data.upload_task is current:
+            entry.runtime_data.upload_task = None
+
+
+@callback
+def async_setup_services(hass: HomeAssistant) -> None:
+    """Register OpenDisplay services."""
+    hass.services.async_register(
+        DOMAIN,
+        "upload_image",
+        _async_upload_image,
+        schema=SCHEMA_UPLOAD_IMAGE,
+    )

homeassistant/components/opendisplay/services.yaml

@@ -0,0 +1,70 @@
+upload_image:
+  fields:
+    device_id:
+      required: true
+      selector:
+        device:
+          integration: opendisplay
+    image:
+      required: true
+      selector:
+        media:
+          accept:
+            - image/*
+    advanced_options:
+      collapsed: true
+      fields:
+        rotation:
+          required: false
+          default: 0
+          selector:
+            number:
+              min: 0
+              max: 270
+              step: 90
+              mode: slider
+        dither_mode:
+          required: false
+          default: "burkes"
+          selector:
+            select:
+              translation_key: dither_mode
+              options:
+                - "none"
+                - "burkes"
+                - "ordered"
+                - "floyd_steinberg"
+                - "atkinson"
+                - "stucki"
+                - "sierra"
+                - "sierra_lite"
+                - "jarvis_judice_ninke"
+        refresh_mode:
+          required: false
+          default: "full"
+          selector:
+            select:
+              translation_key: refresh_mode
+              options:
+                - "full"
+                - "fast"
+        fit_mode:
+          required: false
+          default: "contain"
+          selector:
+            select:
+              translation_key: fit_mode
+              options:
+                - "stretch"
+                - "contain"
+                - "cover"
+                - "crop"
+        tone_compression:
+          required: false
+          selector:
+            number:
+              min: 0
+              max: 100
+              step: 1
+              mode: slider
+              unit_of_measurement: "%"

homeassistant/components/opendisplay/strings.json

@@ -0,0 +1,114 @@
+{
+  "config": {
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
+      "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "flow_title": "{name}",
+    "step": {
+      "bluetooth_confirm": {
+        "description": "[%key:component::bluetooth::config::step::bluetooth_confirm::description%]"
+      },
+      "user": {
+        "data": {
+          "address": "[%key:common::config_flow::data::device%]"
+        },
+        "data_description": {
+          "address": "Select the Bluetooth device to set up."
+        },
+        "description": "[%key:component::bluetooth::config::step::user::description%]"
+      }
+    }
+  },
+  "exceptions": {
+    "device_not_found": {
+      "message": "Could not find Bluetooth device with address `{address}`."
+    },
+    "invalid_device_id": {
+      "message": "Device `{device_id}` is not a valid OpenDisplay device."
+    },
+    "media_download_error": {
+      "message": "Failed to download media: {error}"
+    },
+    "upload_error": {
+      "message": "Failed to upload image to the display."
+    }
+  },
+  "selector": {
+    "dither_mode": {
+      "options": {
+        "atkinson": "Atkinson",
+        "burkes": "Burkes",
+        "floyd_steinberg": "Floyd-Steinberg",
+        "jarvis_judice_ninke": "Jarvis, Judice & Ninke",
+        "none": "None",
+        "ordered": "Ordered",
+        "sierra": "Sierra",
+        "sierra_lite": "Sierra Lite",
+        "stucki": "Stucki"
+      }
+    },
+    "fit_mode": {
+      "options": {
+        "contain": "Contain",
+        "cover": "Cover",
+        "crop": "Crop",
+        "stretch": "Stretch"
+      }
+    },
+    "refresh_mode": {
+      "options": {
+        "fast": "Fast",
+        "full": "Full"
+      }
+    }
+  },
+  "services": {
+    "upload_image": {
+      "description": "Uploads an image to an OpenDisplay device.",
+      "fields": {
+        "device_id": {
+          "description": "The OpenDisplay device to upload the image to.",
+          "name": "Device"
+        },
+        "dither_mode": {
+          "description": "The dithering algorithm to use for converting the image to the display's color palette.",
+          "name": "Dither mode"
+        },
+        "fit_mode": {
+          "description": "How the image is fitted to the display dimensions.",
+          "name": "Fit mode"
+        },
+        "image": {
+          "description": "The image to upload to the display.",
+          "name": "Image"
+        },
+        "refresh_mode": {
+          "description": "The display refresh mode. Full refresh clears ghosting but is slower. Fast refresh is not supported on all displays.",
+          "name": "Refresh mode"
+        },
+        "rotation": {
+          "description": "The rotation angle in degrees, applied clockwise.",
+          "name": "Rotation"
+        },
+        "tone_compression": {
+          "description": "Dynamic range compression strength. Leave empty for automatic.",
+          "name": "Tone compression"
+        }
+      },
+      "name": "Upload image",
+      "sections": {
+        "advanced_options": {
+          "name": "Advanced options"
+        }
+      }
+    }
+  }
+}

homeassistant/generated/bluetooth.py

@@ -620,6 +620,11 @@ BLUETOOTH: Final[list[dict[str, bool | str | int | list[int]]]] = [
         "domain": "motionblinds_ble",
         "local_name": "MOTION_*",
     },
+    {
+        "connectable": True,
+        "domain": "opendisplay",
+        "manufacturer_id": 9286,
+    },
     {
         "domain": "oralb",
         "manufacturer_id": 220,

homeassistant/generated/config_flows.py

@@ -504,6 +504,7 @@ FLOWS = {
         "open_meteo",
         "open_router",
         "openai_conversation",
+        "opendisplay",
         "openevse",
         "openexchangerates",
         "opengarage",

homeassistant/generated/integrations.json

@@ -4875,6 +4875,12 @@
       "config_flow": false,
       "iot_class": "cloud_push"
     },
+    "opendisplay": {
+      "name": "OpenDisplay",
+      "integration_type": "device",
+      "config_flow": true,
+      "iot_class": "local_push"
+    },
     "openerz": {
       "name": "Open ERZ",
       "integration_type": "hub",

requirements_all.txt

@@ -1870,6 +1870,9 @@ py-nightscout==1.2.2
 # homeassistant.components.mta
 py-nymta==0.4.0
 
+# homeassistant.components.opendisplay
+py-opendisplay==5.2.0
+
 # homeassistant.components.schluter
 py-schluter==0.1.7
 

requirements_test_all.txt

@@ -1622,6 +1622,9 @@ py-nightscout==1.2.2
 # homeassistant.components.mta
 py-nymta==0.4.0
 
+# homeassistant.components.opendisplay
+py-opendisplay==5.2.0
+
 # homeassistant.components.ecovacs
 py-sucks==0.9.11
 

tests/components/opendisplay/__init__.py

@@ -0,0 +1,124 @@
+"""Tests for the OpenDisplay integration."""
+
+from time import time
+
+from bleak.backends.scanner import AdvertisementData
+from opendisplay import (
+    BoardManufacturer,
+    ColorScheme,
+    DisplayConfig,
+    GlobalConfig,
+    ManufacturerData,
+    PowerOption,
+    SystemConfig,
+)
+
+from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
+
+from tests.components.bluetooth import generate_ble_device
+
+OPENDISPLAY_MANUFACTURER_ID = 9286  # 0x2446
+
+# V1 advertisement payload (14 bytes): battery_mv=3700, temperature_c=25.0, loop_counter=1
+V1_ADVERTISEMENT_DATA = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x82\x72\x11"
+
+TEST_ADDRESS = "AA:BB:CC:DD:EE:FF"
+TEST_TITLE = "OpenDisplay 1234"
+
+# Firmware version response: major=1, minor=2, sha="abc123"
+FIRMWARE_VERSION = {"major": 1, "minor": 2, "sha": "abc123"}
+
+DEVICE_CONFIG = GlobalConfig(
+    system=SystemConfig(
+        ic_type=0,
+        communication_modes=0,
+        device_flags=0,
+        pwr_pin=0xFF,
+        reserved=b"\x00" * 17,
+    ),
+    manufacturer=ManufacturerData(
+        manufacturer_id=BoardManufacturer.SEEED,
+        board_type=1,
+        board_revision=0,
+        reserved=b"\x00" * 18,
+    ),
+    power=PowerOption(
+        power_mode=0,
+        battery_capacity_mah=0,
+        sleep_timeout_ms=0,
+        tx_power=0,
+        sleep_flags=0,
+        battery_sense_pin=0xFF,
+        battery_sense_enable_pin=0xFF,
+        battery_sense_flags=0,
+        capacity_estimator=0,
+        voltage_scaling_factor=0,
+        deep_sleep_current_ua=0,
+        deep_sleep_time_seconds=0,
+        reserved=b"\x00" * 12,
+    ),
+    displays=[
+        DisplayConfig(
+            instance_number=0,
+            display_technology=0,
+            panel_ic_type=0,
+            pixel_width=296,
+            pixel_height=128,
+            active_width_mm=67,
+            active_height_mm=29,
+            tag_type=0,
+            rotation=0,
+            reset_pin=0xFF,
+            busy_pin=0xFF,
+            dc_pin=0xFF,
+            cs_pin=0xFF,
+            data_pin=0,
+            partial_update_support=0,
+            color_scheme=ColorScheme.BWR.value,
+            transmission_modes=0x01,
+            clk_pin=0,
+            reserved_pins=b"\x00" * 7,
+            reserved=b"\x00" * 35,
+        )
+    ],
+)
+
+
+def make_service_info(
+    name: str | None = "OpenDisplay 1234",
+    address: str = "AA:BB:CC:DD:EE:FF",
+    manufacturer_data: dict[int, bytes] | None = None,
+) -> BluetoothServiceInfoBleak:
+    """Create a BluetoothServiceInfoBleak for testing."""
+    if manufacturer_data is None:
+        manufacturer_data = {OPENDISPLAY_MANUFACTURER_ID: V1_ADVERTISEMENT_DATA}
+    return BluetoothServiceInfoBleak(
+        name=name or "",
+        address=address,
+        rssi=-60,
+        manufacturer_data=manufacturer_data,
+        service_data={},
+        service_uuids=[],
+        source="local",
+        connectable=True,
+        time=time(),
+        device=generate_ble_device(address, name=name),
+        advertisement=AdvertisementData(
+            local_name=name,
+            manufacturer_data=manufacturer_data,
+            service_data={},
+            service_uuids=[],
+            rssi=-60,
+            tx_power=-127,
+            platform_data=(),
+        ),
+        tx_power=-127,
+    )
+
+
+VALID_SERVICE_INFO = make_service_info()
+
+NOT_OPENDISPLAY_SERVICE_INFO = make_service_info(
+    name="Other Device",
+    manufacturer_data={0x1234: b"\x00\x01"},
+)

tests/components/opendisplay/conftest.py

@@ -0,0 +1,75 @@
+"""OpenDisplay test fixtures."""
+
+from collections.abc import Generator
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from homeassistant.components.opendisplay.const import DOMAIN
+
+from . import DEVICE_CONFIG, FIRMWARE_VERSION, TEST_ADDRESS, TEST_TITLE
+
+from tests.common import MockConfigEntry
+from tests.components.bluetooth import generate_ble_device
+
+
+@pytest.fixture(autouse=True)
+def mock_bluetooth(enable_bluetooth: None) -> None:
+    """Auto mock bluetooth."""
+
+
+@pytest.fixture(autouse=True)
+def mock_ble_device() -> Generator[None]:
+    """Mock the BLE device being visible."""
+    ble_device = generate_ble_device(TEST_ADDRESS, TEST_TITLE)
+    with (
+        patch(
+            "homeassistant.components.opendisplay.async_ble_device_from_address",
+            return_value=ble_device,
+        ),
+        patch(
+            "homeassistant.components.opendisplay.config_flow.async_ble_device_from_address",
+            return_value=ble_device,
+        ),
+        patch(
+            "homeassistant.components.opendisplay.services.async_ble_device_from_address",
+            return_value=ble_device,
+        ),
+    ):
+        yield
+
+
+@pytest.fixture(autouse=True)
+def mock_opendisplay_device() -> Generator[MagicMock]:
+    """Mock the OpenDisplayDevice for setup entry."""
+    with (
+        patch(
+            "homeassistant.components.opendisplay.OpenDisplayDevice",
+            autospec=True,
+        ) as mock_device_init,
+        patch(
+            "homeassistant.components.opendisplay.config_flow.OpenDisplayDevice",
+            new=mock_device_init,
+        ),
+        patch(
+            "homeassistant.components.opendisplay.services.OpenDisplayDevice",
+            new=mock_device_init,
+        ),
+    ):
+        mock_device = mock_device_init.return_value
+        mock_device.__aenter__.return_value = mock_device
+        mock_device.read_firmware_version.return_value = FIRMWARE_VERSION
+        mock_device.config = DEVICE_CONFIG
+        mock_device.is_flex = True
+        yield mock_device
+
+
+@pytest.fixture
+def mock_config_entry() -> MockConfigEntry:
+    """Create a mock config entry."""
+    return MockConfigEntry(
+        domain=DOMAIN,
+        unique_id=TEST_ADDRESS,
+        title=TEST_TITLE,
+        data={},
+    )

tests/components/opendisplay/test_config_flow.py

@@ -0,0 +1,244 @@
+"""Test the OpenDisplay config flow."""
+
+from collections.abc import Generator
+from unittest.mock import MagicMock, patch
+
+from opendisplay import BLEConnectionError, BLETimeoutError, OpenDisplayError
+import pytest
+
+from homeassistant import config_entries
+from homeassistant.components.opendisplay.const import DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+
+from . import NOT_OPENDISPLAY_SERVICE_INFO, VALID_SERVICE_INFO
+
+from tests.common import MockConfigEntry
+
+
+@pytest.fixture(autouse=True)
+def mock_setup_entry() -> Generator[None]:
+    """Prevent the integration from actually setting up after config flow."""
+    with patch(
+        "homeassistant.components.opendisplay.async_setup_entry",
+        return_value=True,
+    ):
+        yield
+
+
+async def test_bluetooth_discovery(hass: HomeAssistant) -> None:
+    """Test discovery via Bluetooth with a valid device."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=VALID_SERVICE_INFO,
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "bluetooth_confirm"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input={}
+    )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["title"] == "OpenDisplay 1234"
+    assert result["data"] == {}
+    assert result["result"].unique_id == "AA:BB:CC:DD:EE:FF"
+
+
+async def test_bluetooth_discovery_already_configured(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry
+) -> None:
+    """Test discovery aborts when device is already configured."""
+    mock_config_entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=VALID_SERVICE_INFO,
+    )
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "already_configured"
+
+
+async def test_bluetooth_discovery_already_in_progress(hass: HomeAssistant) -> None:
+    """Test discovery aborts when same device flow is in progress."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=VALID_SERVICE_INFO,
+    )
+    assert result["type"] is FlowResultType.FORM
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=VALID_SERVICE_INFO,
+    )
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "already_in_progress"
+
+
+@pytest.mark.parametrize(
+    ("exception", "expected_reason"),
+    [
+        (BLEConnectionError("test"), "cannot_connect"),
+        (BLETimeoutError("test"), "cannot_connect"),
+        (OpenDisplayError("test"), "cannot_connect"),
+        (RuntimeError("test"), "unknown"),
+    ],
+)
+async def test_bluetooth_confirm_connection_error(
+    hass: HomeAssistant,
+    mock_opendisplay_device: MagicMock,
+    exception: Exception,
+    expected_reason: str,
+) -> None:
+    """Test confirm step aborts when connection fails before showing the form."""
+    mock_opendisplay_device.__aenter__.side_effect = exception
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=VALID_SERVICE_INFO,
+    )
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == expected_reason
+
+
+async def test_bluetooth_confirm_ble_device_not_found(
+    hass: HomeAssistant,
+) -> None:
+    """Test confirm step aborts when BLE device is not found."""
+    with patch(
+        "homeassistant.components.opendisplay.config_flow.async_ble_device_from_address",
+        return_value=None,
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_BLUETOOTH},
+            data=VALID_SERVICE_INFO,
+        )
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "cannot_connect"
+
+
+async def test_user_step_with_devices(hass: HomeAssistant) -> None:
+    """Test user step with discovered devices."""
+    with patch(
+        "homeassistant.components.opendisplay.config_flow.async_discovered_service_info",
+        return_value=[VALID_SERVICE_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_USER},
+        )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "user"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input={"address": "AA:BB:CC:DD:EE:FF"},
+    )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["title"] == "OpenDisplay 1234"
+    assert result["data"] == {}
+    assert result["result"].unique_id == "AA:BB:CC:DD:EE:FF"
+
+
+async def test_user_step_no_devices(hass: HomeAssistant) -> None:
+    """Test user step when no devices are discovered."""
+    with patch(
+        "homeassistant.components.opendisplay.config_flow.async_discovered_service_info",
+        return_value=[],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_USER},
+        )
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "no_devices_found"
+
+
+async def test_user_step_filters_unsupported(hass: HomeAssistant) -> None:
+    """Test user step filters out unsupported devices."""
+    with patch(
+        "homeassistant.components.opendisplay.config_flow.async_discovered_service_info",
+        return_value=[NOT_OPENDISPLAY_SERVICE_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_USER},
+        )
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "no_devices_found"
+
+
+@pytest.mark.parametrize(
+    ("exception", "expected_error"),
+    [
+        (BLEConnectionError("test"), "cannot_connect"),
+        (BLETimeoutError("test"), "cannot_connect"),
+        (OpenDisplayError("test"), "cannot_connect"),
+        (RuntimeError("test"), "unknown"),
+    ],
+)
+async def test_user_step_connection_error(
+    hass: HomeAssistant,
+    mock_opendisplay_device: MagicMock,
+    exception: Exception,
+    expected_error: str,
+) -> None:
+    """Test user step handles connection and unexpected errors."""
+    with patch(
+        "homeassistant.components.opendisplay.config_flow.async_discovered_service_info",
+        return_value=[VALID_SERVICE_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_USER},
+        )
+
+    assert result["type"] is FlowResultType.FORM
+
+    mock_opendisplay_device.__aenter__.side_effect = exception
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input={"address": "AA:BB:CC:DD:EE:FF"},
+    )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["errors"] == {"base": expected_error}
+
+    mock_opendisplay_device.__aenter__.side_effect = None
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input={"address": "AA:BB:CC:DD:EE:FF"},
+    )
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+
+
+async def test_user_step_already_configured(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry
+) -> None:
+    """Test user step aborts when device is already configured."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.opendisplay.config_flow.async_discovered_service_info",
+        return_value=[VALID_SERVICE_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_USER},
+        )
+
+    # Device is filtered out since it's already configured
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "no_devices_found"

tests/components/opendisplay/test_init.py

@@ -0,0 +1,137 @@
+"""Test the OpenDisplay integration setup and unload."""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from opendisplay import BLEConnectionError, BLETimeoutError, OpenDisplayError
+import pytest
+
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr
+
+from tests.common import MockConfigEntry
+
+
+async def test_setup_and_unload(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry
+) -> None:
+    """Test setting up and unloading a config entry."""
+    mock_config_entry.add_to_hass(hass)
+
+    assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert mock_config_entry.state is ConfigEntryState.LOADED
+
+    assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
+
+
+async def test_setup_device_not_found(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry
+) -> None:
+    """Test setup retries when device is not visible."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.opendisplay.async_ble_device_from_address",
+        return_value=None,
+    ):
+        await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
+
+
+@pytest.mark.parametrize(
+    "exception",
+    [
+        BLEConnectionError("connection failed"),
+        BLETimeoutError("timeout"),
+        OpenDisplayError("device error"),
+    ],
+)
+async def test_setup_connection_error(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    exception: Exception,
+) -> None:
+    """Test setup retries on BLE connection errors."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.opendisplay.OpenDisplayDevice",
+        return_value=AsyncMock(__aenter__=AsyncMock(side_effect=exception)),
+    ):
+        await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
+
+
+async def test_setup_device_registered(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test that a device is registered in the device registry after setup."""
+    mock_config_entry.add_to_hass(hass)
+
+    assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    devices = dr.async_entries_for_config_entry(
+        device_registry, mock_config_entry.entry_id
+    )
+    assert len(devices) == 1
+
+
+@pytest.mark.parametrize(
+    ("is_flex", "expect_hw_version", "expect_config_url"),
+    [
+        (True, True, True),
+        (False, False, False),
+    ],
+)
+async def test_setup_device_registry_fields(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_opendisplay_device: MagicMock,
+    device_registry: dr.DeviceRegistry,
+    is_flex: bool,
+    expect_hw_version: bool,
+    expect_config_url: bool,
+) -> None:
+    """Test that hw_version and configuration_url are only set for Flex devices."""
+    mock_opendisplay_device.is_flex = is_flex
+    mock_config_entry.add_to_hass(hass)
+
+    assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    devices = dr.async_entries_for_config_entry(
+        device_registry, mock_config_entry.entry_id
+    )
+    assert len(devices) == 1
+    device = devices[0]
+    assert (device.hw_version is not None) == expect_hw_version
+    assert (device.configuration_url is not None) == expect_config_url
+
+
+async def test_unload_cancels_active_upload_task(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry
+) -> None:
+    """Test that unloading the entry cancels an in-progress upload task."""
+    mock_config_entry.add_to_hass(hass)
+    assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    task = hass.async_create_task(asyncio.sleep(3600))
+    mock_config_entry.runtime_data.upload_task = task
+
+    assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+    assert task.cancelled()

tests/components/opendisplay/test_services.py

@@ -0,0 +1,290 @@
+"""Test the OpenDisplay upload_image service."""
+
+import asyncio
+from collections.abc import Generator
+import io
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import aiohttp
+from opendisplay import BLEConnectionError
+from PIL import Image as PILImage
+import pytest
+import voluptuous as vol
+
+from homeassistant.components.opendisplay.const import DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
+from homeassistant.helpers import device_registry as dr
+
+from tests.common import MockConfigEntry
+from tests.test_util.aiohttp import AiohttpClientMocker
+
+
+@pytest.fixture(autouse=True)
+async def setup_entry(hass: HomeAssistant, mock_config_entry: MockConfigEntry) -> None:
+    """Set up the config entry for service tests."""
+    mock_config_entry.add_to_hass(hass)
+    assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+
+@pytest.fixture
+def mock_upload_device(mock_opendisplay_device: MagicMock) -> MagicMock:
+    """Return the mock OpenDisplayDevice for upload service tests."""
+    return mock_opendisplay_device
+
+
+@pytest.fixture
+def mock_resolve_media(tmp_path: Path) -> Generator[MagicMock]:
+    """Mock async_resolve_media to return a local test image."""
+    image_path = tmp_path / "test.png"
+    PILImage.new("RGB", (10, 10)).save(image_path)
+    mock_media = MagicMock()
+    mock_media.path = image_path
+    with patch(
+        "homeassistant.components.opendisplay.services.async_resolve_media",
+        return_value=mock_media,
+    ):
+        yield mock_media
+
+
+def _device_id(hass: HomeAssistant, mock_config_entry: MockConfigEntry) -> str:
+    """Return the device registry ID for the config entry."""
+    registry = dr.async_get(hass)
+    devices = dr.async_entries_for_config_entry(registry, mock_config_entry.entry_id)
+    assert devices
+    return devices[0].id
+
+
+async def test_upload_image_local_file(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_upload_device: MagicMock,
+    mock_resolve_media: MagicMock,
+) -> None:
+    """Test successful upload from a local file with tone compression."""
+    device_id = _device_id(hass, mock_config_entry)
+
+    await hass.services.async_call(
+        DOMAIN,
+        "upload_image",
+        {
+            "device_id": device_id,
+            "image": {
+                "media_content_id": "media-source://local/test.png",
+                "media_content_type": "image/png",
+            },
+            "tone_compression": 50,
+        },
+        blocking=True,
+    )
+
+    mock_upload_device.upload_image.assert_called_once()
+
+
+async def test_upload_image_remote_url(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_upload_device: MagicMock,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test successful upload from a remote URL."""
+    device_id = _device_id(hass, mock_config_entry)
+
+    image = PILImage.new("RGB", (10, 10))
+    buf = io.BytesIO()
+    image.save(buf, format="PNG")
+    aioclient_mock.get("http://example.com/image.png", content=buf.getvalue())
+
+    mock_media = MagicMock()
+    mock_media.path = None
+    mock_media.url = "http://example.com/image.png"
+
+    with patch(
+        "homeassistant.components.opendisplay.services.async_resolve_media",
+        return_value=mock_media,
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            "upload_image",
+            {
+                "device_id": device_id,
+                "image": {
+                    "media_content_id": "media-source://local/test.png",
+                    "media_content_type": "image/png",
+                },
+            },
+            blocking=True,
+        )
+
+    mock_upload_device.upload_image.assert_called_once()
+
+
+async def test_upload_image_invalid_device_id(
+    hass: HomeAssistant,
+) -> None:
+    """Test that an invalid device_id raises ServiceValidationError."""
+    with pytest.raises(ServiceValidationError, match="not a valid OpenDisplay device"):
+        await hass.services.async_call(
+            DOMAIN,
+            "upload_image",
+            {
+                "device_id": "not-a-real-device-id",
+                "image": {
+                    "media_content_id": "media-source://local/test.png",
+                    "media_content_type": "image/png",
+                },
+            },
+            blocking=True,
+        )
+
+
+async def test_upload_image_device_not_in_range(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test that HomeAssistantError is raised if device is out of BLE range."""
+    device_id = _device_id(hass, mock_config_entry)
+
+    with (
+        patch(
+            "homeassistant.components.opendisplay.services.async_ble_device_from_address",
+            return_value=None,
+        ),
+        pytest.raises(HomeAssistantError),
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            "upload_image",
+            {
+                "device_id": device_id,
+                "image": {
+                    "media_content_id": "media-source://local/test.png",
+                    "media_content_type": "image/png",
+                },
+            },
+            blocking=True,
+        )
+
+
+async def test_upload_image_ble_error(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_opendisplay_device: MagicMock,
+    mock_resolve_media: MagicMock,
+) -> None:
+    """Test that HomeAssistantError is raised on BLE upload failure."""
+    device_id = _device_id(hass, mock_config_entry)
+
+    mock_opendisplay_device.__aenter__.side_effect = BLEConnectionError(
+        "connection lost"
+    )
+    with pytest.raises(HomeAssistantError):
+        await hass.services.async_call(
+            DOMAIN,
+            "upload_image",
+            {
+                "device_id": device_id,
+                "image": {
+                    "media_content_id": "media-source://local/test.png",
+                    "media_content_type": "image/png",
+                },
+            },
+            blocking=True,
+        )
+
+
+async def test_upload_image_download_error(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test that HomeAssistantError is raised on media download failure."""
+    device_id = _device_id(hass, mock_config_entry)
+
+    aioclient_mock.get(
+        "http://example.com/image.png",
+        exc=aiohttp.ClientError("connection refused"),
+    )
+
+    mock_media = MagicMock()
+    mock_media.path = None
+    mock_media.url = "http://example.com/image.png"
+
+    with (
+        patch(
+            "homeassistant.components.opendisplay.services.async_resolve_media",
+            return_value=mock_media,
+        ),
+        pytest.raises(HomeAssistantError),
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            "upload_image",
+            {
+                "device_id": device_id,
+                "image": {
+                    "media_content_id": "media-source://local/test.png",
+                    "media_content_type": "image/png",
+                },
+            },
+            blocking=True,
+        )
+
+
+@pytest.mark.parametrize(
+    "field",
+    ["dither_mode", "fit_mode", "refresh_mode"],
+)
+async def test_upload_image_invalid_mode(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    field: str,
+) -> None:
+    """Test that invalid mode strings are rejected by the schema."""
+    device_id = _device_id(hass, mock_config_entry)
+
+    with pytest.raises(vol.Invalid):
+        await hass.services.async_call(
+            DOMAIN,
+            "upload_image",
+            {
+                "device_id": device_id,
+                "image": {
+                    "media_content_id": "media-source://local/test.png",
+                    "media_content_type": "image/png",
+                },
+                field: "not_a_valid_value",
+            },
+            blocking=True,
+        )
+
+
+async def test_upload_image_cancels_previous_task(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_upload_device: MagicMock,
+    mock_resolve_media: MagicMock,
+) -> None:
+    """Test that starting a new upload cancels an in-progress upload task."""
+    device_id = _device_id(hass, mock_config_entry)
+
+    prev_task = hass.async_create_task(asyncio.sleep(3600))
+    mock_config_entry.runtime_data.upload_task = prev_task
+
+    await hass.services.async_call(
+        DOMAIN,
+        "upload_image",
+        {
+            "device_id": device_id,
+            "image": {
+                "media_content_id": "media-source://local/test.png",
+                "media_content_type": "image/png",
+            },
+        },
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+
+    assert prev_task.cancelled()