Add config flow to Dune HD (#36345)

* Add config_flow to the dunehd integration

* Add tests

* Run gen_requirements_all

* Fix pylint error

* Better hostname validation

* Build device info in the class
This commit is contained in:
Maciej Bieniek 2020-06-03 14:01:56 +02:00 committed by GitHub
parent 95563e04e8
commit 465b98513b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 325 additions and 43 deletions

View File

@ -176,6 +176,8 @@ omit =
homeassistant/components/dsmr_reader/*
homeassistant/components/dte_energy_bridge/sensor.py
homeassistant/components/dublin_bus_transport/sensor.py
homeassistant/components/dunehd/__init__.py
homeassistant/components/dunehd/const.py
homeassistant/components/dunehd/media_player.py
homeassistant/components/dwd_weather_warnings/sensor.py
homeassistant/components/dweet/*

View File

@ -98,6 +98,7 @@ homeassistant/components/directv/* @ctalkington
homeassistant/components/discogs/* @thibmaek
homeassistant/components/doorbird/* @oblogic7 @bdraco
homeassistant/components/dsmr_reader/* @depl0y
homeassistant/components/dunehd/* @bieniu
homeassistant/components/dweet/* @fabaff
homeassistant/components/dynalite/* @ziv1234
homeassistant/components/dyson/* @etheralm

View File

@ -1 +1,49 @@
"""The dunehd component."""
"""The Dune HD component."""
import asyncio
from pdunehd import DuneHDPlayer
from homeassistant.const import CONF_HOST
from .const import DOMAIN
PLATFORMS = ["media_player"]
async def async_setup(hass, config):
"""Set up the Dune HD component."""
return True
async def async_setup_entry(hass, config_entry):
"""Set up a config entry."""
host = config_entry.data[CONF_HOST]
player = DuneHDPlayer(host)
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][config_entry.entry_id] = player
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, component)
)
return True
async def async_unload_entry(hass, config_entry):
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(config_entry, component)
for component in PLATFORMS
]
)
)
if unload_ok:
hass.data[DOMAIN].pop(config_entry.entry_id)
return unload_ok

View File

@ -0,0 +1,101 @@
"""Adds config flow for Dune HD integration."""
import ipaddress
import logging
import re
from pdunehd import DuneHDPlayer
import voluptuous as vol
from homeassistant import config_entries, exceptions
from homeassistant.const import CONF_HOST
from .const import DOMAIN # pylint:disable=unused-import
_LOGGER = logging.getLogger(__name__)
def host_valid(host):
"""Return True if hostname or IP address is valid."""
try:
if ipaddress.ip_address(host).version == (4 or 6):
return True
except ValueError:
if len(host) > 253:
return False
allowed = re.compile(r"(?!-)[A-Z\d\-\_]{1,63}(?<!-)$", re.IGNORECASE)
return all(allowed.match(x) for x in host.split("."))
class DuneHDConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Dune HD integration."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
def __init__(self):
"""Initialize."""
self.host = None
async def init_device(self, host):
"""Initialize Dune HD player."""
player = DuneHDPlayer(host)
state = await self.hass.async_add_executor_job(player.update_state)
if not state:
raise CannotConnect()
async def async_step_user(self, user_input=None):
"""Handle the initial step."""
errors = {}
if user_input is not None:
if host_valid(user_input[CONF_HOST]):
self.host = user_input[CONF_HOST]
try:
if self.host_already_configured(self.host):
raise AlreadyConfigured()
await self.init_device(self.host)
except CannotConnect:
errors[CONF_HOST] = "cannot_connect"
except AlreadyConfigured:
errors[CONF_HOST] = "already_configured"
else:
return self.async_create_entry(title=self.host, data=user_input)
else:
errors[CONF_HOST] = "invalid_host"
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({vol.Required(CONF_HOST, default=""): str}),
errors=errors,
)
async def async_step_import(self, user_input=None):
"""Handle configuration by yaml file."""
self.host = user_input[CONF_HOST]
if self.host_already_configured(self.host):
return self.async_abort(reason="already_configured")
try:
await self.init_device(self.host)
except CannotConnect:
_LOGGER.error("Import aborted, cannot connect to %s", self.host)
return self.async_abort(reason="cannot_connect")
else:
return self.async_create_entry(title=self.host, data=user_input)
def host_already_configured(self, host):
"""See if we already have a dunehd entry matching user input configured."""
existing_hosts = {
entry.data[CONF_HOST] for entry in self._async_current_entries()
}
return host in existing_hosts
class CannotConnect(exceptions.HomeAssistantError):
"""Error to indicate we cannot connect."""
class AlreadyConfigured(exceptions.HomeAssistantError):
"""Error to indicate device is already configured."""

View File

@ -0,0 +1,4 @@
"""Constants for Dune HD integration."""
ATTR_MANUFACTURER = "Dune"
DOMAIN = "dunehd"
DEFAULT_NAME = "Dune HD"

View File

@ -1,7 +1,8 @@
{
"domain": "dunehd",
"name": "DuneHD",
"name": "Dune HD",
"documentation": "https://www.home-assistant.io/integrations/dunehd",
"requirements": ["pdunehd==1.3.1"],
"codeowners": []
"codeowners": ["@bieniu"],
"config_flow": true
}

View File

@ -1,5 +1,4 @@
"""DuneHD implementation of the media player."""
from pdunehd import DuneHDPlayer
"""Dune HD implementation of the media player."""
import voluptuous as vol
from homeassistant.components.media_player import PLATFORM_SCHEMA, MediaPlayerEntity
@ -8,10 +7,10 @@ from homeassistant.components.media_player.const import (
SUPPORT_PAUSE,
SUPPORT_PLAY,
SUPPORT_PREVIOUS_TRACK,
SUPPORT_SELECT_SOURCE,
SUPPORT_TURN_OFF,
SUPPORT_TURN_ON,
)
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
@ -22,7 +21,7 @@ from homeassistant.const import (
)
import homeassistant.helpers.config_validation as cv
DEFAULT_NAME = "DuneHD"
from .const import ATTR_MANUFACTURER, DEFAULT_NAME, DOMAIN
CONF_SOURCES = "sources"
@ -38,7 +37,6 @@ DUNEHD_PLAYER_SUPPORT = (
SUPPORT_PAUSE
| SUPPORT_TURN_ON
| SUPPORT_TURN_OFF
| SUPPORT_SELECT_SOURCE
| SUPPORT_PREVIOUS_TRACK
| SUPPORT_NEXT_TRACK
| SUPPORT_PLAY
@ -46,25 +44,35 @@ DUNEHD_PLAYER_SUPPORT = (
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the DuneHD media player platform."""
sources = config.get(CONF_SOURCES, {})
"""Set up the Dune HD media player platform."""
host = config.get(CONF_HOST)
name = config.get(CONF_NAME)
add_entities([DuneHDPlayerEntity(DuneHDPlayer(host), name, sources)], True)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data={CONF_HOST: host}
)
)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Add Dune HD entities from a config_entry."""
unique_id = config_entry.entry_id
player = hass.data[DOMAIN][config_entry.entry_id]
async_add_entities([DuneHDPlayerEntity(player, DEFAULT_NAME, unique_id)], True)
class DuneHDPlayerEntity(MediaPlayerEntity):
"""Implementation of the Dune HD player."""
def __init__(self, player, name, sources):
def __init__(self, player, name, unique_id):
"""Initialize entity to control Dune HD."""
self._player = player
self._name = name
self._sources = sources
self._media_title = None
self._state = None
self._unique_id = unique_id
def update(self):
"""Update internal status of the entity."""
@ -78,7 +86,7 @@ class DuneHDPlayerEntity(MediaPlayerEntity):
state = STATE_OFF
if "playback_position" in self._state:
state = STATE_PLAYING
if self._state.get("player_state") in ("playing", "buffering"):
if self._state.get("player_state") in ("playing", "buffering", "photo_viewer"):
state = STATE_PLAYING
if int(self._state.get("playback_speed", 1234)) == 0:
state = STATE_PAUSED
@ -96,6 +104,20 @@ class DuneHDPlayerEntity(MediaPlayerEntity):
"""Return True if entity is available."""
return bool(self._state)
@property
def unique_id(self):
"""Return a unique_id for this entity."""
return self._unique_id
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._unique_id)},
"name": DEFAULT_NAME,
"manufacturer": ATTR_MANUFACTURER,
}
@property
def volume_level(self):
"""Return the volume level of the media player (0..1)."""
@ -106,11 +128,6 @@ class DuneHDPlayerEntity(MediaPlayerEntity):
"""Return a boolean if volume is currently muted."""
return int(self._state.get("playback_mute", 0)) == 1
@property
def source_list(self):
"""Return a list of available input sources."""
return list(self._sources.keys())
@property
def supported_features(self):
"""Flag media player features that are supported."""
@ -132,22 +149,18 @@ class DuneHDPlayerEntity(MediaPlayerEntity):
"""Turn off media player."""
self._media_title = None
self._state = self._player.turn_off()
self.schedule_update_ha_state()
def turn_on(self):
"""Turn off media player."""
self._state = self._player.turn_on()
self.schedule_update_ha_state()
def media_play(self):
"""Play media player."""
self._state = self._player.play()
self.schedule_update_ha_state()
def media_pause(self):
"""Pause media player."""
self._state = self._player.pause()
self.schedule_update_ha_state()
@property
def media_title(self):
@ -155,33 +168,21 @@ class DuneHDPlayerEntity(MediaPlayerEntity):
self.__update_title()
if self._media_title:
return self._media_title
return self._state.get("playback_url", "Not playing")
def __update_title(self):
if self._state.get("player_state") == "bluray_playback":
self._media_title = "Blu-Ray"
elif "playback_url" in self._state:
sources = self._sources
sval = sources.values()
skey = sources.keys()
pburl = self._state["playback_url"]
if pburl in sval:
self._media_title = list(skey)[list(sval).index(pburl)]
else:
self._media_title = pburl
def select_source(self, source):
"""Select input source."""
self._media_title = source
self._state = self._player.launch_media_url(self._sources.get(source))
self.schedule_update_ha_state()
elif self._state.get("player_state") == "photo_viewer":
self._media_title = "Photo Viewer"
elif self._state.get("playback_url"):
self._media_title = self._state["playback_url"].split("/")[-1]
else:
self._media_title = None
def media_previous_track(self):
"""Send previous track command."""
self._state = self._player.previous_track()
self.schedule_update_ha_state()
def media_next_track(self):
"""Send next track command."""
self._state = self._player.next_track()
self.schedule_update_ha_state()

View File

@ -0,0 +1,21 @@
{
"config": {
"step": {
"user": {
"title": "Dune HD",
"description": "Set up Dune HD integration. If you have problems with configuration go to: https://www.home-assistant.io/integrations/dunehd \n\nEnsure that your player is turned on.",
"data": {
"host": "[%key:common::config_flow::data::host%]"
}
}
},
"error": {
"invalid_host": "Invalid hostname or IP address.",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
}
}

View File

@ -33,6 +33,7 @@ FLOWS = [
"dialogflow",
"directv",
"doorbird",
"dunehd",
"dynalite",
"ecobee",
"elgato",

View File

@ -435,6 +435,9 @@ paho-mqtt==1.5.0
# homeassistant.components.panasonic_viera
panasonic_viera==0.3.5
# homeassistant.components.dunehd
pdunehd==1.3.1
# homeassistant.components.aruba
# homeassistant.components.cisco_ios
# homeassistant.components.pandora

View File

@ -0,0 +1 @@
"""Tests for Dune HD."""

View File

@ -0,0 +1,98 @@
"""Define tests for the Dune HD config flow."""
from homeassistant import data_entry_flow
from homeassistant.components.dunehd.const import DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import CONF_HOST
from tests.async_mock import patch
from tests.common import MockConfigEntry
CONFIG_HOSTNAME = {CONF_HOST: "dunehd-host"}
CONFIG_IP = {CONF_HOST: "10.10.10.12"}
DUNEHD_STATE = {"protocol_version": "4", "player_state": "navigator"}
async def test_import(hass):
"""Test that the import works."""
with patch("pdunehd.DuneHDPlayer.update_state", return_value=DUNEHD_STATE):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=CONFIG_HOSTNAME
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "dunehd-host"
assert result["data"] == {CONF_HOST: "dunehd-host"}
async def test_import_cannot_connect(hass):
"""Test that errors are shown when cannot connect to the host during import."""
with patch("pdunehd.DuneHDPlayer.update_state", return_value={}):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=CONFIG_HOSTNAME
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
async def test_import_duplicate_error(hass):
"""Test that errors are shown when duplicates are added during import."""
config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_HOST: "dunehd-host"}, title="dunehd-host",
)
config_entry.add_to_hass(hass)
with patch("pdunehd.DuneHDPlayer.update_state", return_value=DUNEHD_STATE):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=CONFIG_HOSTNAME
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_user_invalid_host(hass):
"""Test that errors are shown when the host is invalid."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "invalid/host"}
)
assert result["errors"] == {CONF_HOST: "invalid_host"}
async def test_user_cannot_connect(hass):
"""Test that errors are shown when cannot connect to the host."""
with patch("pdunehd.DuneHDPlayer.update_state", return_value={}):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG_IP
)
assert result["errors"] == {CONF_HOST: "cannot_connect"}
async def test_duplicate_error(hass):
"""Test that errors are shown when duplicates are added."""
config_entry = MockConfigEntry(
domain=DOMAIN, data=CONFIG_HOSTNAME, title="dunehd-host",
)
config_entry.add_to_hass(hass)
with patch("pdunehd.DuneHDPlayer.update_state", return_value=DUNEHD_STATE):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG_HOSTNAME
)
assert result["errors"] == {CONF_HOST: "already_configured"}
async def test_create_entry(hass):
"""Test that the user step works."""
with patch("pdunehd.DuneHDPlayer.update_state", return_value=DUNEHD_STATE):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG_HOSTNAME
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "dunehd-host"
assert result["data"] == {CONF_HOST: "dunehd-host"}