mirror of
https://github.com/home-assistant/core.git
synced 2026-08-24 10:13:52 -05:00
Add select entity to Harbor (#177556)
This commit is contained in:
@@ -6,7 +6,7 @@ DOMAIN = "harbor"
|
||||
MANUFACTURER = "Harbor"
|
||||
MODEL = "Harbor Camera"
|
||||
|
||||
PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.SWITCH]
|
||||
PLATFORMS: list[Platform] = [Platform.SELECT, Platform.SENSOR, Platform.SWITCH]
|
||||
|
||||
CONF_CERT_PEM = "cert_pem"
|
||||
CONF_KEY_PEM = "key_pem"
|
||||
|
||||
@@ -7,7 +7,7 @@ from uuid import uuid4
|
||||
|
||||
from harbor.config import HarborCameraConfig
|
||||
from harbor.devices.camera import HarborCamera
|
||||
from harbor.mqtt import DEFAULT_INITIAL_COMMANDS, HarborMQTTClient
|
||||
from harbor.mqtt import DEFAULT_INITIAL_COMMANDS, HarborMQTTClient, NightMode
|
||||
from harbor.state import HarborDeviceState
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
@@ -180,6 +180,10 @@ class HarborCoordinator(DataUpdateCoordinator[HarborDeviceState]):
|
||||
"""Show or hide the clock overlay burned into the video."""
|
||||
await self._client.set_clock_display(clock_display)
|
||||
|
||||
async def async_set_night_mode(self, night_mode: NightMode) -> None:
|
||||
"""Set the camera night-mode preference."""
|
||||
await self._client.set_night_mode(night_mode)
|
||||
|
||||
def _handle_device_update(self, state: HarborDeviceState) -> None:
|
||||
"""Mirror a library device update into Home Assistant."""
|
||||
self._data_event.set()
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
{
|
||||
"entity": {
|
||||
"select": {
|
||||
"night_mode_preference": {
|
||||
"default": "mdi:weather-night"
|
||||
}
|
||||
},
|
||||
"sensor": {
|
||||
"num_viewers": {
|
||||
"default": "mdi:account-eye"
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
"""Select entities for Harbor."""
|
||||
|
||||
from collections.abc import Callable, Coroutine
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, cast, override
|
||||
|
||||
from harbor import HarborCommandError
|
||||
from harbor.mqtt import NIGHT_MODE_MODES, NightMode
|
||||
|
||||
from homeassistant.components.select import SelectEntity, SelectEntityDescription
|
||||
from homeassistant.const import EntityCategory
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
|
||||
from .const import DOMAIN
|
||||
from .coordinator import HarborConfigEntry, HarborCoordinator
|
||||
from .entity import HarborEntity
|
||||
|
||||
# Commands are sent over a single MQTT session to one camera, and each settings
|
||||
# write is followed by a settings refresh, so they are serialized.
|
||||
PARALLEL_UPDATES = 1
|
||||
|
||||
|
||||
@dataclass(frozen=True, kw_only=True)
|
||||
class HarborSelectEntityDescription(SelectEntityDescription):
|
||||
"""Describes a Harbor select entity."""
|
||||
|
||||
select_fn: Callable[[HarborCoordinator, str], Coroutine[Any, Any, None]]
|
||||
|
||||
|
||||
CAMERA_SELECTS: tuple[HarborSelectEntityDescription, ...] = (
|
||||
HarborSelectEntityDescription(
|
||||
key="night_mode_preference",
|
||||
translation_key="night_mode_preference",
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
options=list(NIGHT_MODE_MODES),
|
||||
select_fn=lambda coordinator, option: coordinator.async_set_night_mode(
|
||||
cast(NightMode, option)
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: HarborConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up Harbor selects from a config entry."""
|
||||
coordinator = entry.runtime_data
|
||||
async_add_entities(
|
||||
HarborSelect(coordinator, description) for description in CAMERA_SELECTS
|
||||
)
|
||||
|
||||
|
||||
class HarborSelect(HarborEntity, SelectEntity):
|
||||
"""A Harbor select entity."""
|
||||
|
||||
entity_description: HarborSelectEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: HarborCoordinator,
|
||||
description: HarborSelectEntityDescription,
|
||||
) -> None:
|
||||
"""Initialize the Harbor select."""
|
||||
self.entity_description = description
|
||||
super().__init__(coordinator, description.key)
|
||||
|
||||
@override
|
||||
@property
|
||||
def current_option(self) -> str | None:
|
||||
"""Return the currently selected option."""
|
||||
option = self.coordinator.data.values.get(self.entity_description.key)
|
||||
# The library falls back to the literal string "unknown" for any value
|
||||
# it doesn't recognize.
|
||||
return option if option in self.options else None
|
||||
|
||||
@override
|
||||
async def async_select_option(self, option: str) -> None:
|
||||
"""Change the selected option."""
|
||||
try:
|
||||
await self.entity_description.select_fn(self.coordinator, option)
|
||||
except (HarborCommandError, TimeoutError, ConnectionError) as err:
|
||||
raise HomeAssistantError(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="select_option_failed",
|
||||
translation_placeholders={
|
||||
"option": option,
|
||||
"select": self.entity_description.key,
|
||||
},
|
||||
) from err
|
||||
@@ -28,6 +28,16 @@
|
||||
}
|
||||
},
|
||||
"entity": {
|
||||
"select": {
|
||||
"night_mode_preference": {
|
||||
"name": "Night mode",
|
||||
"state": {
|
||||
"auto": "[%key:common::state::auto%]",
|
||||
"off": "[%key:common::state::off%]",
|
||||
"on": "[%key:common::state::on%]"
|
||||
}
|
||||
}
|
||||
},
|
||||
"sensor": {
|
||||
"bitrate": {
|
||||
"name": "Bitrate"
|
||||
@@ -66,6 +76,9 @@
|
||||
"cannot_connect": {
|
||||
"message": "Could not connect to the Harbor camera. It may be offline or unreachable."
|
||||
},
|
||||
"select_option_failed": {
|
||||
"message": "Failed to select {option} for {select}."
|
||||
},
|
||||
"switch_turn_off_failed": {
|
||||
"message": "Failed to turn off {switch}."
|
||||
},
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
# serializer version: 1
|
||||
# name: test_selects[select.harbor_camera_1234567890_night_mode-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
None,
|
||||
]),
|
||||
'area_id': None,
|
||||
'capabilities': dict({
|
||||
<SelectEntityCapabilityAttribute.OPTIONS: 'options'>: list([
|
||||
'auto',
|
||||
'on',
|
||||
'off',
|
||||
]),
|
||||
}),
|
||||
'config_entry_id': <ANY>,
|
||||
'config_subentry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'select',
|
||||
'entity_category': <EntityCategory.CONFIG: 'config'>,
|
||||
'entity_id': 'select.harbor_camera_1234567890_night_mode',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'object_id_base': 'Night mode',
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': None,
|
||||
'original_icon': None,
|
||||
'original_name': 'Night mode',
|
||||
'platform': 'harbor',
|
||||
'previous_unique_id': None,
|
||||
'suggested_object_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': 'night_mode_preference',
|
||||
'unique_id': '1234567890_night_mode_preference',
|
||||
'unit_of_measurement': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_selects[select.harbor_camera_1234567890_night_mode-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Harbor Camera 1234567890 Night mode',
|
||||
<SelectEntityCapabilityAttribute.OPTIONS: 'options'>: list([
|
||||
'auto',
|
||||
'on',
|
||||
'off',
|
||||
]),
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'select.harbor_camera_1234567890_night_mode',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': 'auto',
|
||||
})
|
||||
# ---
|
||||
@@ -0,0 +1,132 @@
|
||||
"""Test the Harbor selects."""
|
||||
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from harbor import HarborCommandError
|
||||
import pytest
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
|
||||
from homeassistant.components.select import (
|
||||
ATTR_OPTION,
|
||||
DOMAIN as SELECT_DOMAIN,
|
||||
SERVICE_SELECT_OPTION,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from . import setup_integration
|
||||
from .conftest import SETTINGS_TOPIC, emit_message
|
||||
|
||||
from tests.common import MockConfigEntry, snapshot_platform
|
||||
|
||||
NIGHT_MODE_ENTITY = "select.harbor_camera_1234567890_night_mode"
|
||||
|
||||
SELECT_SETTINGS_PAYLOAD: dict[str, Any] = {
|
||||
"settings": {"preference_video_night_mode": "auto"},
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
|
||||
async def test_selects(
|
||||
hass: HomeAssistant,
|
||||
entity_registry: er.EntityRegistry,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_mqtt_client: AsyncMock,
|
||||
snapshot: SnapshotAssertion,
|
||||
) -> None:
|
||||
"""Test the Harbor selects report their current option."""
|
||||
with patch("homeassistant.components.harbor.PLATFORMS", [Platform.SELECT]):
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
await emit_message(mock_mqtt_client, SETTINGS_TOPIC, SELECT_SETTINGS_PAYLOAD)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"option",
|
||||
[
|
||||
pytest.param("auto", id="auto"),
|
||||
pytest.param("on", id="on"),
|
||||
pytest.param("off", id="off"),
|
||||
],
|
||||
)
|
||||
async def test_select_option(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_mqtt_client: AsyncMock,
|
||||
option: str,
|
||||
) -> None:
|
||||
"""Test selecting an option passes the device value to the library."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
await hass.services.async_call(
|
||||
SELECT_DOMAIN,
|
||||
SERVICE_SELECT_OPTION,
|
||||
{ATTR_ENTITY_ID: NIGHT_MODE_ENTITY, ATTR_OPTION: option},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
mock_mqtt_client.return_value.set_night_mode.assert_awaited_once_with(option)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"error",
|
||||
[
|
||||
pytest.param(
|
||||
HarborCommandError("command", {"error": "rejected"}), id="command"
|
||||
),
|
||||
pytest.param(TimeoutError, id="timeout"),
|
||||
pytest.param(ConnectionError, id="connection"),
|
||||
],
|
||||
)
|
||||
async def test_select_option_failure_raises(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_mqtt_client: AsyncMock,
|
||||
error: Exception | type[Exception],
|
||||
) -> None:
|
||||
"""Test a failed camera command surfaces as a HomeAssistantError."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
mock_mqtt_client.return_value.set_night_mode.side_effect = error
|
||||
|
||||
with pytest.raises(HomeAssistantError):
|
||||
await hass.services.async_call(
|
||||
SELECT_DOMAIN,
|
||||
SERVICE_SELECT_OPTION,
|
||||
{ATTR_ENTITY_ID: NIGHT_MODE_ENTITY, ATTR_OPTION: "off"},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
async def test_unexpected_option_stays_valid(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_mqtt_client: AsyncMock,
|
||||
) -> None:
|
||||
"""Test a night mode outside the declared options surfaces as unknown.
|
||||
|
||||
The library maps unrecognized enum values onto its own "unknown" member;
|
||||
the select treats that as no selection rather than exposing "unknown" as a
|
||||
literal option.
|
||||
"""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
await emit_message(mock_mqtt_client, SETTINGS_TOPIC, SELECT_SETTINGS_PAYLOAD)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(NIGHT_MODE_ENTITY).state == "auto"
|
||||
|
||||
await emit_message(
|
||||
mock_mqtt_client,
|
||||
SETTINGS_TOPIC,
|
||||
{"settings": {"preference_video_night_mode": "sunset"}},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(NIGHT_MODE_ENTITY).state == STATE_UNKNOWN
|
||||
Reference in New Issue
Block a user