Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add Radio Browser integration (#66950)

pull/66994/head
Franck Nijhof 2022-02-21 18:13:02 +01:00 committed by GitHub
parent 660fb393f0
commit d839febbe7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 531 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -955,6 +955,8 @@ omit =
homeassistant/components/rachio/switch.py
homeassistant/components/rachio/webhooks.py
homeassistant/components/radarr/sensor.py
homeassistant/components/radio_browser/__init__.py
homeassistant/components/radio_browser/media_source.py
homeassistant/components/radiotherm/climate.py
homeassistant/components/rainbird/*
homeassistant/components/raincloud/*

2
CODEOWNERS
View File

@ -755,6 +755,8 @@ homeassistant/components/qwikswitch/* @kellerza
tests/components/qwikswitch/* @kellerza
homeassistant/components/rachio/* @bdraco
tests/components/rachio/* @bdraco
homeassistant/components/radio_browser/* @frenck
tests/components/radio_browser/* @frenck
homeassistant/components/radiotherm/* @vinnyfuria
homeassistant/components/rainbird/* @konikvranik
homeassistant/components/raincloud/* @vanstinator

17
homeassistant/components/onboarding/views.py
View File

@ -188,9 +188,8 @@ class CoreConfigOnboardingView(_BaseOnboardingView):
await self._async_mark_done(hass)
await hass.config_entries.flow.async_init(
"met", context={"source": "onboarding"}
)
# Integrations to set up when finishing onboarding
onboard_integrations = ["met", "radio_browser"]
# pylint: disable=import-outside-toplevel
from homeassistant.components import hassio
@ -199,9 +198,17 @@ class CoreConfigOnboardingView(_BaseOnboardingView):
hassio.is_hassio(hass)
and "raspberrypi" in hassio.get_core_info(hass)["machine"]
):
await hass.config_entries.flow.async_init(
"rpi_power", context={"source": "onboarding"}
onboard_integrations.append("rpi_power")
# Set up integrations after onboarding
await asyncio.gather(
*(
hass.config_entries.flow.async_init(
domain, context={"source": "onboarding"}
)
for domain in onboard_integrations
)
)
return self.json({})

36
homeassistant/components/radio_browser/__init__.py Normal file
View File

@ -0,0 +1,36 @@
"""The Radio Browser integration."""
from __future__ import annotations
from radios import RadioBrowser, RadioBrowserError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import __version__
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Radio Browser from a config entry.
This integration doesn't set up any enitites, as it provides a media source
only.
"""
session = async_get_clientsession(hass)
radios = RadioBrowser(session=session, user_agent=f"HomeAssistant/{__version__}")
try:
await radios.stats()
except RadioBrowserError as err:
raise ConfigEntryNotReady("Could not connect to Radio Browser API") from err
hass.data[DOMAIN] = radios
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
del hass.data[DOMAIN]
return True

33
homeassistant/components/radio_browser/config_flow.py Normal file
View File

@ -0,0 +1,33 @@
"""Config flow for Radio Browser integration."""
from __future__ import annotations
from typing import Any
from homeassistant.config_entries import ConfigFlow
from homeassistant.data_entry_flow import FlowResult
from .const import DOMAIN
class RadioBrowserConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Radio Browser."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
if user_input is not None:
return self.async_create_entry(title="Radio Browser", data={})
return self.async_show_form(step_id="user")
async def async_step_onboarding(
self, data: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by onboarding."""
return self.async_create_entry(title="Radio Browser", data={})

7
homeassistant/components/radio_browser/const.py Normal file
View File

@ -0,0 +1,7 @@
"""Constants for the Radio Browser integration."""
import logging
from typing import Final
DOMAIN: Final = "radio_browser"
LOGGER = logging.getLogger(__package__)

9
homeassistant/components/radio_browser/manifest.json Normal file
View File

@ -0,0 +1,9 @@
{
"domain": "radio_browser",
"name": "Radio Browser",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/radio",
"requirements": ["radios==0.1.0"],
"codeowners": ["@frenck"],
"iot_class": "cloud_polling"
}

286
homeassistant/components/radio_browser/media_source.py Normal file
View File

@ -0,0 +1,286 @@
"""Expose Radio Browser as a media source."""
from __future__ import annotations
import mimetypes
from radios import FilterBy, Order, RadioBrowser, Station
from homeassistant.components.media_player.const import (
MEDIA_CLASS_CHANNEL,
MEDIA_CLASS_DIRECTORY,
MEDIA_CLASS_MUSIC,
MEDIA_TYPE_MUSIC,
)
from homeassistant.components.media_player.errors import BrowseError
from homeassistant.components.media_source.models import (
BrowseMediaSource,
MediaSource,
MediaSourceItem,
PlayMedia,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from .const import DOMAIN
CODEC_TO_MIMETYPE = {
"MP3": "audio/mpeg",
"AAC": "audio/aac",
"AAC+": "audio/aac",
"OGG": "application/ogg",
}
async def async_get_media_source(hass: HomeAssistant) -> RadioMediaSource:
"""Set up Radio Browser media source."""
# Radio browser support only a single config entry
entry = hass.config_entries.async_entries(DOMAIN)[0]
radios = hass.data[DOMAIN]
return RadioMediaSource(hass, radios, entry)
class RadioMediaSource(MediaSource):
"""Provide Radio stations as media sources."""
def __init__(
self, hass: HomeAssistant, radios: RadioBrowser, entry: ConfigEntry
) -> None:
"""Initialize CameraMediaSource."""
super().__init__(DOMAIN)
self.hass = hass
self.entry = entry
self.radios = radios
async def async_resolve_media(self, item: MediaSourceItem) -> PlayMedia:
"""Resolve selected Radio station to a streaming URL."""
station = await self.radios.station(uuid=item.identifier)
if not station:
raise BrowseError("Radio station is no longer available")
if not (mime_type := self._async_get_station_mime_type(station)):
raise BrowseError("Could not determine stream type of radio station")
# Register "click" with Radio Browser
await self.radios.station_click(uuid=station.uuid)
return PlayMedia(station.url, mime_type)
async def async_browse_media(
self,
item: MediaSourceItem,
) -> BrowseMediaSource:
"""Return media."""
return BrowseMediaSource(
domain=DOMAIN,
identifier=None,
media_class=MEDIA_CLASS_CHANNEL,
media_content_type=MEDIA_TYPE_MUSIC,
title=self.entry.title,
can_play=False,
can_expand=True,
children_media_class=MEDIA_CLASS_DIRECTORY,
children=[
*await self._async_build_popular(item),
*await self._async_build_by_tag(item),
*await self._async_build_by_language(item),
*await self._async_build_by_country(item),
],
)
@callback
@staticmethod
def _async_get_station_mime_type(station: Station) -> str | None:
"""Determine mime type of a radio station."""
mime_type = CODEC_TO_MIMETYPE.get(station.codec)
if not mime_type:
mime_type, _ = mimetypes.guess_type(station.url)
return mime_type
@callback
def _async_build_stations(self, stations: list[Station]) -> list[BrowseMediaSource]:
"""Build list of media sources from radio stations."""
items: list[BrowseMediaSource] = []
for station in stations:
if station.codec == "UNKNOWN" or not (
mime_type := self._async_get_station_mime_type(station)
):
continue
items.append(
BrowseMediaSource(
domain=DOMAIN,
identifier=station.uuid,
media_class=MEDIA_CLASS_MUSIC,
media_content_type=mime_type,
title=station.name,
can_play=True,
can_expand=False,
thumbnail=station.favicon,
)
)
return items
async def _async_build_by_country(
self, item: MediaSourceItem
) -> list[BrowseMediaSource]:
"""Handle browsing radio stations by country."""
category, _, country_code = (item.identifier or "").partition("/")
if country_code:
stations = await self.radios.stations(
filter_by=FilterBy.COUNTRY_CODE_EXACT,
filter_term=country_code,
hide_broken=True,
order=Order.NAME,
reverse=False,
)
return self._async_build_stations(stations)
# We show country in the root additionally, when there is no item
if not item.identifier or category == "country":
countries = await self.radios.countries(order=Order.NAME)
return [
BrowseMediaSource(
domain=DOMAIN,
identifier=f"country/{country.code}",
media_class=MEDIA_CLASS_DIRECTORY,
media_content_type=MEDIA_TYPE_MUSIC,
title=country.name,
can_play=False,
can_expand=True,
thumbnail=country.favicon,
)
for country in countries
]
return []
async def _async_build_by_language(
self, item: MediaSourceItem
) -> list[BrowseMediaSource]:
"""Handle browsing radio stations by language."""
category, _, language = (item.identifier or "").partition("/")
if category == "language" and language:
stations = await self.radios.stations(
filter_by=FilterBy.LANGUAGE_EXACT,
filter_term=language,
hide_broken=True,
order=Order.NAME,
reverse=False,
)
return self._async_build_stations(stations)
if category == "language":
languages = await self.radios.languages(order=Order.NAME, hide_broken=True)
return [
BrowseMediaSource(
domain=DOMAIN,
identifier=f"language/{language.code}",
media_class=MEDIA_CLASS_DIRECTORY,
media_content_type=MEDIA_TYPE_MUSIC,
title=language.name,
can_play=False,
can_expand=True,
thumbnail=language.favicon,
)
for language in languages
]
if not item.identifier:
return [
BrowseMediaSource(
domain=DOMAIN,
identifier="language",
media_class=MEDIA_CLASS_DIRECTORY,
media_content_type=MEDIA_TYPE_MUSIC,
title="By Language",
can_play=False,
can_expand=True,
)
]
return []
async def _async_build_popular(
self, item: MediaSourceItem
) -> list[BrowseMediaSource]:
"""Handle browsing popular radio stations."""
if item.identifier == "popular":
stations = await self.radios.stations(
hide_broken=True,
limit=250,
order=Order.CLICK_COUNT,
reverse=True,
)
return self._async_build_stations(stations)
if not item.identifier:
return [
BrowseMediaSource(
domain=DOMAIN,
identifier="popular",
media_class=MEDIA_CLASS_DIRECTORY,
media_content_type=MEDIA_TYPE_MUSIC,
title="Popular",
can_play=False,
can_expand=True,
)
]
return []
async def _async_build_by_tag(
self, item: MediaSourceItem
) -> list[BrowseMediaSource]:
"""Handle browsing radio stations by tags."""
category, _, tag = (item.identifier or "").partition("/")
if category == "tag" and tag:
stations = await self.radios.stations(
filter_by=FilterBy.TAG_EXACT,
filter_term=tag,
hide_broken=True,
order=Order.NAME,
reverse=False,
)
return self._async_build_stations(stations)
if category == "tag":
tags = await self.radios.tags(
hide_broken=True,
limit=100,
order=Order.STATION_COUNT,
reverse=True,
)
# Now we have the top tags, reorder them by name
tags.sort(key=lambda tag: tag.name)
return [
BrowseMediaSource(
domain=DOMAIN,
identifier=f"tag/{tag.name}",
media_class=MEDIA_CLASS_DIRECTORY,
media_content_type=MEDIA_TYPE_MUSIC,
title=tag.name.title(),
can_play=False,
can_expand=True,
)
for tag in tags
]
if not item.identifier:
return [
BrowseMediaSource(
domain=DOMAIN,
identifier="tag",
media_class=MEDIA_CLASS_DIRECTORY,
media_content_type=MEDIA_TYPE_MUSIC,
title="By Category",
can_play=False,
can_expand=True,
)
]
return []

12
homeassistant/components/radio_browser/strings.json Normal file
View File

@ -0,0 +1,12 @@
{
"config": {
"step": {
"user": {
"description": "Do you want to add Radio Browser to Home Assistant?"
}
},
"abort": {
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]"
}
}
}

12
homeassistant/components/radio_browser/translations/en.json Normal file
View File

@ -0,0 +1,12 @@
{
"config": {
"abort": {
"single_instance_allowed": "Already configured. Only a single configuration possible."
},
"step": {
"user": {
"description": "Do you want to add Radio Browser to Home Assistant?"
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -261,6 +261,7 @@ FLOWS = [
"pvoutput",
"pvpc_hourly_pricing",
"rachio",
"radio_browser",
"rainforest_eagle",
"rainmachine",
"rdw",

3
requirements_all.txt
View File

@ -2077,6 +2077,9 @@ quantum-gateway==0.0.6
# homeassistant.components.rachio
rachiopy==1.0.3
# homeassistant.components.radio_browser
radios==0.1.0
# homeassistant.components.radiotherm
radiotherm==2.1.0

3
requirements_test_all.txt
View File

@ -1296,6 +1296,9 @@ pyzerproc==0.4.8
# homeassistant.components.rachio
rachiopy==1.0.3
# homeassistant.components.radio_browser
radios==0.1.0
# homeassistant.components.rainmachine
regenmaschine==2022.01.0

17
tests/components/onboarding/test_views.py
View File

@ -381,6 +381,23 @@ async def test_onboarding_core_sets_up_met(hass, hass_storage, hass_client):
assert len(hass.states.async_entity_ids("weather")) == 1
async def test_onboarding_core_sets_up_radio_browser(hass, hass_storage, hass_client):
"""Test finishing the core step set up the radio browser."""
mock_storage(hass_storage, {"done": [const.STEP_USER]})
assert await async_setup_component(hass, "onboarding", {})
await hass.async_block_till_done()
client = await hass_client()
resp = await client.post("/api/onboarding/core_config")
assert resp.status == 200
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries("radio_browser")) == 1
async def test_onboarding_core_sets_up_rpi_power(
hass, hass_storage, hass_client, aioclient_mock, rpi
):

1
tests/components/radio_browser/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Radio Browser integration."""

30
tests/components/radio_browser/conftest.py Normal file
View File

@ -0,0 +1,30 @@
"""Fixtures for the Radio Browser integration tests."""
from __future__ import annotations
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant.components.radio_browser.const import DOMAIN
from tests.common import MockConfigEntry
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title="My Radios",
domain=DOMAIN,
data={},
)
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
"""Mock setting up a config entry."""
with patch(
"homeassistant.components.radio_browser.async_setup_entry", return_value=True
) as mock_setup:
yield mock_setup

65
tests/components/radio_browser/test_config_flow.py Normal file
View File

@ -0,0 +1,65 @@
"""Test the Radio Browser config flow."""
from unittest.mock import AsyncMock
from homeassistant.components.radio_browser.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
from tests.common import MockConfigEntry
async def test_full_user_flow(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
"""Test the full user configuration flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result.get("type") == RESULT_TYPE_FORM
assert result.get("errors") is None
assert "flow_id" in result
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={},
)
assert result2.get("type") == RESULT_TYPE_CREATE_ENTRY
assert result2.get("title") == "Radio Browser"
assert result2.get("data") == {}
assert len(mock_setup_entry.mock_calls) == 1
async def test_already_configured(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_setup_entry: AsyncMock,
) -> None:
"""Test we abort if the Radio Browser is already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result.get("type") == RESULT_TYPE_ABORT
assert result.get("reason") == "single_instance_allowed"
async def test_onboarding_flow(
hass: HomeAssistant, mock_setup_entry: AsyncMock
) -> None:
"""Test the onboarding configuration flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "onboarding"}
)
assert result.get("type") == RESULT_TYPE_CREATE_ENTRY
assert result.get("title") == "Radio Browser"
assert result.get("data") == {}
assert len(mock_setup_entry.mock_calls) == 1