1
0
mirror of https://github.com/home-assistant/core.git synced 2026-04-02 00:20:30 +01:00

Migrate motioneye to use runtime_data (#166848)

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
epenet
2026-03-30 15:56:08 +02:00
committed by GitHub
parent d1ccda18f7
commit cda52af178
7 changed files with 46 additions and 44 deletions

View File

@@ -45,7 +45,7 @@ from homeassistant.components.webhook import (
async_register as webhook_register,
async_unregister as webhook_unregister,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_DEVICE_ID, ATTR_NAME, CONF_URL, CONF_WEBHOOK_ID
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
@@ -80,7 +80,7 @@ from .const import (
WEB_HOOK_SENTINEL_KEY,
WEB_HOOK_SENTINEL_VALUE,
)
from .coordinator import MotionEyeUpdateCoordinator
from .coordinator import MotionEyeConfigEntry, MotionEyeUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [CAMERA_DOMAIN, SENSOR_DOMAIN, SWITCH_DOMAIN]
@@ -134,7 +134,7 @@ def is_acceptable_camera(camera: dict[str, Any] | None) -> bool:
@callback
def listen_for_new_cameras(
hass: HomeAssistant,
entry: ConfigEntry,
entry: MotionEyeConfigEntry,
add_func: Callable,
) -> None:
"""Listen for new cameras."""
@@ -168,7 +168,7 @@ def _add_camera(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
client: MotionEyeClient,
entry: ConfigEntry,
entry: MotionEyeConfigEntry,
camera_id: int,
camera: dict[str, Any],
device_identifier: tuple[str, str],
@@ -274,9 +274,8 @@ def _add_camera(
)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: MotionEyeConfigEntry) -> bool:
"""Set up motionEye from a config entry."""
hass.data.setdefault(DOMAIN, {})
client = create_motioneye_client(
entry.data[CONF_URL],
@@ -306,7 +305,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
)
coordinator = MotionEyeUpdateCoordinator(hass, entry, client)
hass.data[DOMAIN][entry.entry_id] = coordinator
entry.runtime_data = coordinator
current_cameras: set[tuple[str, str]] = set()
device_registry = dr.async_get(hass)
@@ -362,14 +361,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: MotionEyeConfigEntry) -> bool:
"""Unload a config entry."""
webhook_unregister(hass, entry.data[CONF_WEBHOOK_ID])
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
coordinator = hass.data[DOMAIN].pop(entry.entry_id)
await coordinator.client.async_client_close()
await entry.runtime_data.client.async_client_close()
return unload_ok
@@ -438,10 +436,14 @@ def _get_media_event_data(
event_file_type: int,
) -> dict[str, str]:
config_entry_id = next(iter(device.config_entries), None)
if not config_entry_id or config_entry_id not in hass.data[DOMAIN]:
if (
not config_entry_id
or not (entry := hass.config_entries.async_get_entry(config_entry_id))
or entry.state != ConfigEntryState.LOADED
):
return {}
coordinator = hass.data[DOMAIN][config_entry_id]
coordinator: MotionEyeUpdateCoordinator = entry.runtime_data
client = coordinator.client
for identifier in device.identifiers:

View File

@@ -30,7 +30,6 @@ from homeassistant.components.mjpeg import (
CONF_STILL_IMAGE_URL,
MjpegCamera,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_AUTHENTICATION,
CONF_NAME,
@@ -50,14 +49,13 @@ from .const import (
CONF_STREAM_URL_TEMPLATE,
CONF_SURVEILLANCE_PASSWORD,
CONF_SURVEILLANCE_USERNAME,
DOMAIN,
MOTIONEYE_MANUFACTURER,
SERVICE_ACTION,
SERVICE_SET_TEXT_OVERLAY,
SERVICE_SNAPSHOT,
TYPE_MOTIONEYE_MJPEG_CAMERA,
)
from .coordinator import MotionEyeUpdateCoordinator
from .coordinator import MotionEyeConfigEntry, MotionEyeUpdateCoordinator
from .entity import MotionEyeEntity
PLATFORMS = [Platform.CAMERA]
@@ -92,11 +90,11 @@ SCHEMA_SERVICE_SET_TEXT = vol.Schema(
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
entry: MotionEyeConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up motionEye from a config entry."""
coordinator = hass.data[DOMAIN][entry.entry_id]
coordinator = entry.runtime_data
@callback
def camera_add(camera: dict[str, Any]) -> None:

View File

@@ -14,7 +14,6 @@ import voluptuous as vol
from homeassistant.config_entries import (
SOURCE_REAUTH,
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
OptionsFlowWithReload,
@@ -39,6 +38,7 @@ from .const import (
DEFAULT_WEBHOOK_SET_OVERWRITE,
DOMAIN,
)
from .coordinator import MotionEyeConfigEntry
class MotionEyeConfigFlow(ConfigFlow, domain=DOMAIN):
@@ -180,7 +180,7 @@ class MotionEyeConfigFlow(ConfigFlow, domain=DOMAIN):
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
config_entry: MotionEyeConfigEntry,
) -> MotionEyeOptionsFlow:
"""Get the Hyperion Options flow."""
return MotionEyeOptionsFlow()

View File

@@ -16,13 +16,16 @@ from .const import DEFAULT_SCAN_INTERVAL, DOMAIN
_LOGGER = logging.getLogger(__name__)
type MotionEyeConfigEntry = ConfigEntry[MotionEyeUpdateCoordinator]
class MotionEyeUpdateCoordinator(DataUpdateCoordinator[dict[str, Any] | None]):
"""Coordinator for motionEye data."""
config_entry: ConfigEntry
config_entry: MotionEyeConfigEntry
def __init__(
self, hass: HomeAssistant, entry: ConfigEntry, client: MotionEyeClient
self, hass: HomeAssistant, entry: MotionEyeConfigEntry, client: MotionEyeClient
) -> None:
"""Initialize the coordinator."""
super().__init__(

View File

@@ -17,12 +17,13 @@ from homeassistant.components.media_source import (
PlayMedia,
Unresolvable,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from . import get_media_url, split_motioneye_device_identifier
from .const import DOMAIN
from .coordinator import MotionEyeConfigEntry
MIME_TYPE_MAP = {
"movies": "video/mp4",
@@ -74,7 +75,7 @@ class MotionEyeMediaSource(MediaSource):
self._verify_kind_or_raise(kind)
url = get_media_url(
self.hass.data[DOMAIN][config.entry_id].client,
config.runtime_data.client,
self._get_camera_id_or_raise(config, device),
self._get_path_or_raise(path),
kind == "images",
@@ -120,10 +121,10 @@ class MotionEyeMediaSource(MediaSource):
return self._build_media_devices(config)
return self._build_media_configs()
def _get_config_or_raise(self, config_id: str) -> ConfigEntry:
def _get_config_or_raise(self, config_id: str) -> MotionEyeConfigEntry:
"""Get a config entry from a URL."""
entry = self.hass.config_entries.async_get_entry(config_id)
if not entry:
if not entry or entry.state != ConfigEntryState.LOADED:
raise MediaSourceError(f"Unable to find config entry with id: {config_id}")
return entry
@@ -154,7 +155,7 @@ class MotionEyeMediaSource(MediaSource):
@classmethod
def _get_camera_id_or_raise(
cls, config: ConfigEntry, device: dr.DeviceEntry
cls, config: MotionEyeConfigEntry, device: dr.DeviceEntry
) -> int:
"""Get a config entry from a URL."""
for identifier in device.identifiers:
@@ -164,7 +165,7 @@ class MotionEyeMediaSource(MediaSource):
raise MediaSourceError(f"Could not find camera id for device id: {device.id}")
@classmethod
def _build_media_config(cls, config: ConfigEntry) -> BrowseMediaSource:
def _build_media_config(cls, config: MotionEyeConfigEntry) -> BrowseMediaSource:
return BrowseMediaSource(
domain=DOMAIN,
identifier=config.entry_id,
@@ -196,7 +197,7 @@ class MotionEyeMediaSource(MediaSource):
@classmethod
def _build_media_device(
cls,
config: ConfigEntry,
config: MotionEyeConfigEntry,
device: dr.DeviceEntry,
full_title: bool = True,
) -> BrowseMediaSource:
@@ -211,7 +212,7 @@ class MotionEyeMediaSource(MediaSource):
children_media_class=MediaClass.DIRECTORY,
)
def _build_media_devices(self, config: ConfigEntry) -> BrowseMediaSource:
def _build_media_devices(self, config: MotionEyeConfigEntry) -> BrowseMediaSource:
"""Build the media sources for device entries."""
device_registry = dr.async_get(self.hass)
devices = dr.async_entries_for_config_entry(device_registry, config.entry_id)
@@ -226,7 +227,7 @@ class MotionEyeMediaSource(MediaSource):
@classmethod
def _build_media_kind(
cls,
config: ConfigEntry,
config: MotionEyeConfigEntry,
device: dr.DeviceEntry,
kind: str,
full_title: bool = True,
@@ -251,7 +252,7 @@ class MotionEyeMediaSource(MediaSource):
)
def _build_media_kinds(
self, config: ConfigEntry, device: dr.DeviceEntry
self, config: MotionEyeConfigEntry, device: dr.DeviceEntry
) -> BrowseMediaSource:
base = self._build_media_device(config, device)
base.children = [
@@ -262,7 +263,7 @@ class MotionEyeMediaSource(MediaSource):
async def _build_media_path(
self,
config: ConfigEntry,
config: MotionEyeConfigEntry,
device: dr.DeviceEntry,
kind: str,
path: str,
@@ -276,7 +277,7 @@ class MotionEyeMediaSource(MediaSource):
base.children = []
client = self.hass.data[DOMAIN][config.entry_id].client
client = config.runtime_data.client
camera_id = self._get_camera_id_or_raise(config, device)
if kind == "movies":
@@ -286,7 +287,7 @@ class MotionEyeMediaSource(MediaSource):
sub_dirs: set[str] = set()
parts = parsed_path.parts
media_list = resp.get(KEY_MEDIA_LIST, [])
media_list = resp.get(KEY_MEDIA_LIST, []) if resp else []
def get_media_sort_key(media: dict) -> str:
"""Get media sort key."""

View File

@@ -9,24 +9,23 @@ from motioneye_client.client import MotionEyeClient
from motioneye_client.const import KEY_ACTIONS
from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from . import get_camera_from_cameras, listen_for_new_cameras
from .const import DOMAIN, TYPE_MOTIONEYE_ACTION_SENSOR
from .coordinator import MotionEyeUpdateCoordinator
from .const import TYPE_MOTIONEYE_ACTION_SENSOR
from .coordinator import MotionEyeConfigEntry, MotionEyeUpdateCoordinator
from .entity import MotionEyeEntity
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
entry: MotionEyeConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up motionEye from a config entry."""
coordinator = hass.data[DOMAIN][entry.entry_id]
coordinator = entry.runtime_data
@callback
def camera_add(camera: dict[str, Any]) -> None:

View File

@@ -16,14 +16,13 @@ from motioneye_client.const import (
)
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import get_camera_from_cameras, listen_for_new_cameras
from .const import DOMAIN, TYPE_MOTIONEYE_SWITCH_BASE
from .coordinator import MotionEyeUpdateCoordinator
from .const import TYPE_MOTIONEYE_SWITCH_BASE
from .coordinator import MotionEyeConfigEntry, MotionEyeUpdateCoordinator
from .entity import MotionEyeEntity
MOTIONEYE_SWITCHES = [
@@ -68,11 +67,11 @@ MOTIONEYE_SWITCHES = [
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
entry: MotionEyeConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up motionEye from a config entry."""
coordinator = hass.data[DOMAIN][entry.entry_id]
coordinator = entry.runtime_data
@callback
def camera_add(camera: dict[str, Any]) -> None: