forked from home-assistant/core
/
media_player.py
245 lines (207 loc) · 8.94 KB
/
media_player.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
"""Support for Enigma2 media players."""
from __future__ import annotations
import contextlib
from logging import getLogger
from aiohttp.client_exceptions import ClientConnectorError, ServerDisconnectedError
from openwebif.api import OpenWebIfDevice
from openwebif.enums import PowerState, RemoteControlCodes, SetVolumeOption
import voluptuous as vol
from yarl import URL
from homeassistant.components.media_player import (
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaPlayerState,
MediaType,
)
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_SSL,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.aiohttp_client import async_create_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.config_validation import PLATFORM_SCHEMA
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import (
CONF_DEEP_STANDBY,
CONF_MAC_ADDRESS,
CONF_SOURCE_BOUQUET,
CONF_USE_CHANNEL_ICON,
DEFAULT_DEEP_STANDBY,
DEFAULT_MAC_ADDRESS,
DEFAULT_NAME,
DEFAULT_PASSWORD,
DEFAULT_PORT,
DEFAULT_SOURCE_BOUQUET,
DEFAULT_SSL,
DEFAULT_USE_CHANNEL_ICON,
DEFAULT_USERNAME,
)
ATTR_MEDIA_CURRENTLY_RECORDING = "media_currently_recording"
ATTR_MEDIA_DESCRIPTION = "media_description"
ATTR_MEDIA_END_TIME = "media_end_time"
ATTR_MEDIA_START_TIME = "media_start_time"
_LOGGER = getLogger(__name__)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_USERNAME, default=DEFAULT_USERNAME): cv.string,
vol.Optional(CONF_PASSWORD, default=DEFAULT_PASSWORD): cv.string,
vol.Optional(CONF_SSL, default=DEFAULT_SSL): cv.boolean,
vol.Optional(
CONF_USE_CHANNEL_ICON, default=DEFAULT_USE_CHANNEL_ICON
): cv.boolean,
vol.Optional(CONF_DEEP_STANDBY, default=DEFAULT_DEEP_STANDBY): cv.boolean,
vol.Optional(CONF_MAC_ADDRESS, default=DEFAULT_MAC_ADDRESS): cv.string,
vol.Optional(CONF_SOURCE_BOUQUET, default=DEFAULT_SOURCE_BOUQUET): cv.string,
}
)
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up of an enigma2 media player."""
if discovery_info:
# Discovery gives us the streaming service port (8001)
# which is not useful as OpenWebif never runs on that port.
# So use the default port instead.
config[CONF_PORT] = DEFAULT_PORT
config[CONF_NAME] = discovery_info["hostname"]
config[CONF_HOST] = discovery_info["host"]
config[CONF_USERNAME] = DEFAULT_USERNAME
config[CONF_PASSWORD] = DEFAULT_PASSWORD
config[CONF_SSL] = DEFAULT_SSL
config[CONF_USE_CHANNEL_ICON] = DEFAULT_USE_CHANNEL_ICON
config[CONF_MAC_ADDRESS] = DEFAULT_MAC_ADDRESS
config[CONF_DEEP_STANDBY] = DEFAULT_DEEP_STANDBY
config[CONF_SOURCE_BOUQUET] = DEFAULT_SOURCE_BOUQUET
base_url = URL.build(
scheme="https" if config[CONF_SSL] else "http",
host=config[CONF_HOST],
port=config.get(CONF_PORT),
user=config.get(CONF_USERNAME),
password=config.get(CONF_PASSWORD),
)
session = async_create_clientsession(hass, verify_ssl=False, base_url=base_url)
device = OpenWebIfDevice(
host=session,
turn_off_to_deep=config.get(CONF_DEEP_STANDBY, False),
source_bouquet=config.get(CONF_SOURCE_BOUQUET),
)
try:
about = await device.get_about()
except ClientConnectorError as err:
raise PlatformNotReady from err
async_add_entities([Enigma2Device(config[CONF_NAME], device, about)])
class Enigma2Device(MediaPlayerEntity):
"""Representation of an Enigma2 box."""
_attr_has_entity_name = True
_attr_media_content_type = MediaType.TVSHOW
_attr_supported_features = (
MediaPlayerEntityFeature.VOLUME_SET
| MediaPlayerEntityFeature.VOLUME_MUTE
| MediaPlayerEntityFeature.TURN_OFF
| MediaPlayerEntityFeature.NEXT_TRACK
| MediaPlayerEntityFeature.STOP
| MediaPlayerEntityFeature.PREVIOUS_TRACK
| MediaPlayerEntityFeature.VOLUME_STEP
| MediaPlayerEntityFeature.TURN_ON
| MediaPlayerEntityFeature.PAUSE
| MediaPlayerEntityFeature.SELECT_SOURCE
)
def __init__(self, name: str, device: OpenWebIfDevice, about: dict) -> None:
"""Initialize the Enigma2 device."""
self._device: OpenWebIfDevice = device
self._device.mac_address = about["info"]["ifaces"][0]["mac"]
self._attr_name = name
self._attr_unique_id = device.mac_address
async def async_turn_off(self) -> None:
"""Turn off media player."""
if self._device.turn_off_to_deep:
with contextlib.suppress(ServerDisconnectedError):
await self._device.set_powerstate(PowerState.DEEP_STANDBY)
self._attr_available = False
else:
await self._device.set_powerstate(PowerState.STANDBY)
async def async_turn_on(self) -> None:
"""Turn the media player on."""
await self._device.turn_on()
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
await self._device.set_volume(int(volume * 100))
async def async_volume_up(self) -> None:
"""Volume up the media player."""
await self._device.set_volume(SetVolumeOption.UP)
async def async_volume_down(self) -> None:
"""Volume down media player."""
await self._device.set_volume(SetVolumeOption.DOWN)
async def async_media_stop(self) -> None:
"""Send stop command."""
await self._device.send_remote_control_action(RemoteControlCodes.STOP)
async def async_media_play(self) -> None:
"""Play media."""
await self._device.send_remote_control_action(RemoteControlCodes.PLAY)
async def async_media_pause(self) -> None:
"""Pause the media player."""
await self._device.send_remote_control_action(RemoteControlCodes.PAUSE)
async def async_media_next_track(self) -> None:
"""Send next track command."""
await self._device.send_remote_control_action(RemoteControlCodes.CHANNEL_UP)
async def async_media_previous_track(self) -> None:
"""Send previous track command."""
await self._device.send_remote_control_action(RemoteControlCodes.CHANNEL_DOWN)
async def async_mute_volume(self, mute: bool) -> None:
"""Mute or unmute."""
await self._device.toggle_mute()
async def async_select_source(self, source: str) -> None:
"""Select input source."""
await self._device.zap(self._device.sources[source])
async def async_update(self) -> None:
"""Update state of the media_player."""
try:
await self._device.update()
except ClientConnectorError as err:
if self._attr_available:
_LOGGER.warning(
"%s is unavailable. Error: %s", self._device.base.host, err
)
self._attr_available = False
return
if not self._attr_available:
_LOGGER.debug("%s is available", self._device.base.host)
self._attr_available = True
if not self._device.status.in_standby:
self._attr_extra_state_attributes = {
ATTR_MEDIA_CURRENTLY_RECORDING: self._device.status.is_recording,
ATTR_MEDIA_DESCRIPTION: self._device.status.currservice.fulldescription,
ATTR_MEDIA_START_TIME: self._device.status.currservice.begin,
ATTR_MEDIA_END_TIME: self._device.status.currservice.end,
}
else:
self._attr_extra_state_attributes = {}
self._attr_media_title = self._device.status.currservice.station
self._attr_media_series_title = self._device.status.currservice.name
self._attr_media_channel = self._device.status.currservice.station
self._attr_is_volume_muted = self._device.status.muted
self._attr_media_content_id = self._device.status.currservice.serviceref
self._attr_media_image_url = self._device.picon_url
self._attr_source = self._device.status.currservice.station
self._attr_source_list = self._device.source_list
if self._device.status.in_standby:
self._attr_state = MediaPlayerState.OFF
else:
self._attr_state = MediaPlayerState.ON
if (volume_level := self._device.status.volume) is not None:
self._attr_volume_level = volume_level / 100
else:
self._attr_volume_level = None