"""Support for Xiaomi WiFi speakers.""" import logging import requests import hashlib import hmac import time import json import re import voluptuous as vol from datetime import timedelta from functools import partial from urllib.parse import urlencode, urlparse, parse_qsl from homeassistant.const import ( ATTR_FRIENDLY_NAME, CONF_HOST, ) from homeassistant.components import media_source from homeassistant.components.media_player import ( DOMAIN as ENTITY_DOMAIN, MediaPlayerDeviceClass, MediaPlayerEntity, MediaPlayerEntityFeature, # v2022.5 MediaPlayerState, # v2022.10 MediaType, # v2022.10 RepeatMode, # v2022.10 ) from homeassistant.components.media_player.browse_media import ( async_process_play_media_url, SearchMedia, BrowseMedia, ) from homeassistant.components.homekit.const import EVENT_HOMEKIT_TV_REMOTE_KEY_PRESSED from homeassistant.core import HassJob from homeassistant.util.dt import utcnow from homeassistant.helpers.event import async_track_point_in_utc_time import homeassistant.helpers.config_validation as cv from . import ( DOMAIN, CONF_MODEL, XIAOMI_CONFIG_SCHEMA as PLATFORM_SCHEMA, # noqa: F401 XIAOMI_MIIO_SERVICE_SCHEMA, HassEntry, BaseEntity, MiotEntityInterface, MiotEntity, MiirToggleEntity, MiotCloud, async_setup_config_entry, bind_services_to_entries, ) from .core.const import HA_VERSION from .core.miot_spec import ( MiotSpec, MiotService, ) _LOGGER = logging.getLogger(__name__) DATA_KEY = f'{ENTITY_DOMAIN}.{DOMAIN}' SCAN_INTERVAL = timedelta(seconds=30) SERVICE_TO_METHOD = { 'intelligent_speaker': { 'method': 'async_intelligent_speaker', 'schema': XIAOMI_MIIO_SERVICE_SCHEMA.extend( { vol.Required('text'): cv.string, vol.Optional('execute', default=False): cv.boolean, vol.Optional('silent', default=False): cv.boolean, vol.Optional('throw', default=False): cv.boolean, }, ), }, 'xiaoai_wakeup': { 'method': 'async_xiaoai_wakeup', 'schema': XIAOMI_MIIO_SERVICE_SCHEMA.extend( { vol.Optional('text', default=''): vol.Any(cv.string, None), vol.Optional('throw', default=False): cv.boolean, }, ), }, } async def async_setup_entry(hass, config_entry, async_add_entities): HassEntry.init(hass, config_entry).new_adder(ENTITY_DOMAIN, async_add_entities) await async_setup_config_entry(hass, config_entry, async_setup_platform, async_add_entities, ENTITY_DOMAIN) async def async_setup_platform(hass, config, async_add_entities, discovery_info=None): hass.data.setdefault(DATA_KEY, {}) hass.data[DOMAIN]['add_entities'][ENTITY_DOMAIN] = async_add_entities config['hass'] = hass model = str(config.get(CONF_MODEL) or '') spec = hass.data[DOMAIN]['miot_specs'].get(model) entities = [] if isinstance(spec, MiotSpec): for srv in spec.get_services( 'play_control', 'ir_tv_control', 'ir_projector_control', 'ir_box_control', 'ir_stb_control', ): if 'miir.' in model: entities.append(MiirMediaPlayerEntity(config, srv)) continue if model in ['xiaomi.controller.86v1']: pass elif not srv.mapping() and not srv.get_action('play'): continue if spec.get_service('television', 'projector', 'tv_box'): entities.append(MitvMediaPlayerEntity(config, srv)) else: entities.append(MiotMediaPlayerEntity(config, srv)) for entity in entities: hass.data[DOMAIN]['entities'][entity.unique_id] = entity async_add_entities(entities, update_before_add=True) bind_services_to_entries(hass, SERVICE_TO_METHOD) class BaseMediaPlayerEntity(MediaPlayerEntity, MiotEntityInterface, BaseEntity): _attr_state = None def __init__(self, miot_service: MiotService): self._miot_service = miot_service self._prop_state = miot_service.get_property('playing_state') self._prop_volume = miot_service.get_property('volume') self._prop_mute = miot_service.get_property('mute') self._speaker = miot_service.spec.get_service('speaker') if self._speaker: self._prop_volume = self._speaker.get_property('volume') or self._prop_volume self._prop_mute = self._speaker.get_property('mute') or self._prop_mute self._prop_input = None self._act_turn_on = None self._act_turn_off = None for srv in miot_service.spec.services.values(): if p := srv.get_property('input_control'): self._prop_input = p act = srv.get_action('turn_on') if act and not self._act_turn_on: self._act_turn_on = act act = srv.get_action('turn_off') if act and not self._act_turn_off: self._act_turn_off = act if miot_service.get_action('play'): self._supported_features |= MediaPlayerEntityFeature.PLAY if miot_service.get_action('pause'): self._supported_features |= MediaPlayerEntityFeature.PAUSE if miot_service.get_action('previous'): self._supported_features |= MediaPlayerEntityFeature.PREVIOUS_TRACK if miot_service.get_action('next'): self._supported_features |= MediaPlayerEntityFeature.NEXT_TRACK if miot_service.get_action('stop'): self._supported_features |= MediaPlayerEntityFeature.STOP if self._prop_input: self._supported_features |= MediaPlayerEntityFeature.SELECT_SOURCE self._attr_source_list = self._prop_input.list_descriptions() if self._prop_volume: self._supported_features |= MediaPlayerEntityFeature.VOLUME_SET self._supported_features |= MediaPlayerEntityFeature.VOLUME_STEP if self._prop_mute: self._supported_features |= MediaPlayerEntityFeature.VOLUME_MUTE if self._act_turn_on: self._supported_features |= MediaPlayerEntityFeature.TURN_ON if self._act_turn_off: self._supported_features |= MediaPlayerEntityFeature.TURN_OFF @property def supported_features(self): return self._supported_features @property def device_class(self): if cls := self.get_device_class(MediaPlayerDeviceClass): return cls typ = f'{self.model} {self._miot_service.spec.type}' if 'speaker' in typ: return MediaPlayerDeviceClass.SPEAKER if 'receiver' in typ: return MediaPlayerDeviceClass.RECEIVER if 'television' in typ or '.tv.' in typ: return MediaPlayerDeviceClass.TV return None @property def state(self): if self._prop_state and self._prop_state.readable: sta = self._prop_state.from_device(self.device) if sta is not None: if sta in self._prop_state.list_search('Playing', 'Play'): return MediaPlayerState.PLAYING if sta == self._prop_state.list_value('Pause'): return MediaPlayerState.PAUSED if sta == self._prop_state.list_value('Idle'): return MediaPlayerState.IDLE des = self._prop_state.list_description(sta) if des is not None: return des if self._attr_state is not None: return self._attr_state if self.available: return MediaPlayerState.IDLE return None @property def is_volume_muted(self): if self._prop_mute: return self._prop_mute.from_device(self.device) and True return None async def async_mute_volume(self, mute): if self._prop_mute: return await self.async_set_property(self._prop_mute, True if mute else False) return False @property def volume_level(self): if self._prop_volume: val = self._prop_volume.from_device(self.device) if val is not None: try: return round(val or 0) / 100 except (TypeError, ValueError): pass return self._attr_volume_level async def async_set_volume_level(self, volume): if self._prop_volume: val = round(volume * (self._prop_volume.range_max() or 1)) stp = self._prop_volume.range_step() if stp and stp > 1: val = round(val / stp) * stp return await self.async_set_property(self._prop_volume, val) return False async def async_volume_up(self): if self._prop_volume: stp = self._prop_volume.range_step() or 5 val = round(self._prop_volume.from_device(self.device) or 0) + stp return await self.async_set_property(self._prop_volume, val) return False async def async_volume_down(self): if self._prop_volume: stp = self._prop_volume.range_step() or 5 val = round(self._prop_volume.from_device(self.device) or 0) - stp return await self.async_set_property(self._prop_volume, val) return False async def async_media_play(self): act = self._miot_service.get_action('play') if act: if await self.async_call_action(act): if self._prop_state: self.update_attrs({ self._prop_state.full_name: self._prop_state.list_value('Playing'), }) return True return False async def async_media_pause(self): act = self._miot_service.get_action('pause') if act: if await self.async_call_action(act): if self._prop_state: self.update_attrs({ self._prop_state.full_name: self._prop_state.list_value('Pause'), }) return True return False async def async_media_stop(self): act = self._miot_service.get_action('stop') if act: if await self.async_call_action(act): if self._prop_state: self.update_attrs({ self._prop_state.full_name: self._prop_state.list_value('Stopped', 'Stop', 'Idle'), }) return True return self.media_pause() async def async_media_previous_track(self): act = self._miot_service.get_action('previous') if act: return await self.async_call_action(act) return False async def async_media_next_track(self): act = self._miot_service.get_action('next') if act: return await self.async_call_action(act) return False def media_seek(self, position): return False def play_media(self, media_type, media_id, **kwargs): return False @property def source(self): """Name of the current input source.""" if self._prop_input: val = self._prop_input.from_device(self.device) if val is not None: return self._prop_input.list_description(val) return None async def async_select_source(self, source): """Select input source.""" val = self._prop_input.list_value(source) if val is not None: return await self.async_set_property(self._prop_input, val) return False def select_sound_mode(self, sound_mode): return False def clear_playlist(self): return False def set_shuffle(self, shuffle): return False def set_repeat(self, repeat): return False async def async_browse_media(self, media_content_type=None, media_content_id=None): """Return a BrowseMedia instance.""" pass class MiotMediaPlayerEntity(MiotEntity, BaseMediaPlayerEntity): def __init__(self, config: dict, miot_service: MiotService): super().__init__(miot_service, config=config, logger=_LOGGER) BaseMediaPlayerEntity.__init__(self, miot_service) self._intelligent_speaker = miot_service.spec.get_service('intelligent_speaker') self._message_router = miot_service.spec.get_service('message_router') self.xiaoai_cloud = None self.xiaoai_device = None self._supported_features |= MediaPlayerEntityFeature.PLAY_MEDIA | MediaPlayerEntityFeature.BROWSE_MEDIA if HA_VERSION >= '2025.5' and (self._intelligent_speaker or self._message_router): self._supported_features |= MediaPlayerEntityFeature.SEARCH_MEDIA @property def xiaoai_id(self): if not self.xiaoai_device: return None return self.xiaoai_device.get('deviceID') async def async_added_to_hass(self): await super().async_added_to_hass() if self._intelligent_speaker: mic = self.miot_cloud if isinstance(mic, MiotCloud): self.xiaoai_cloud = await mic.async_change_sid('micoapi') async def async_update(self): await super().async_update() if not self._available: return if self._prop_state and not self._prop_state.readable: if self.is_volume_muted is False: self._attr_state = MediaPlayerState.PLAYING else: self._attr_state = MediaPlayerState.IDLE if self.xiaoai_device is None: await self.async_update_xiaoai_device() if self.xiaoai_device: await self.async_update_play_status() from .sensor import XiaoaiConversationSensor add_sensors = self._add_entities.get('sensor') if 'conversation' not in self._subs and add_sensors: self._subs['conversation'] = XiaoaiConversationSensor(self, self.hass) add_sensors([self._subs['conversation']]) async def async_update_xiaoai_device(self): if not isinstance(self.xiaoai_cloud, MiotCloud): return self.xiaoai_device api = 'https://api2.mina.mi.com/admin/v2/device_list' dat = { 'presence': False, 'master': False, } result = await self.xiaoai_cloud.async_request_api(api, data=dat, method='GET') or {} if 'data' in result: self.xiaoai_device = {} for d in result.get('data', []): if not isinstance(d, dict): continue if d.get('miotDID') == self.miot_did or d.get('mac') == self.device.info.mac: self.xiaoai_device = d break return self.xiaoai_device async def async_update_play_status(self, _=None): if not (aid := self.xiaoai_id): return self.update_attrs({'xiaoai_id': aid}) api = 'https://api2.mina.mi.com/remote/ubus' dat = { 'deviceId': aid, 'path': 'mediaplayer', 'method': 'player_get_play_status', 'message': '{}', } try: result = await self.xiaoai_cloud.async_request_api(api, data=dat, method='POST') or {} info = result.get('data', {}).get('info', {}) if not isinstance(info, dict): info = json.loads(info) if info: song = playing = info.get('play_song_detail') or {} mid = song.get('audio_id') or song.get('global_id') if mid and not song.get('title'): song = self._vars.get('latest_song') or {} aid = song.get('audioId') or song.get('songID') if not aid or mid != aid or mid != self._attr_media_content_id: song = await self.async_get_media_detail(song) or {} self._vars['latest_song'] = song song.update(playing) if (sta := info.get('status')) is not None: self._attr_state = { 0: MediaPlayerState.IDLE, 1: MediaPlayerState.PLAYING, 2: MediaPlayerState.PAUSED, }.get(sta) if (typ := info.get('media_type')) is not None: self._attr_media_content_type = {3: MediaType.MUSIC, 13: MediaType.VIDEO}.get(typ) else: self._attr_media_content_type = song.get('audioType') self._attr_volume_level = info.get('volume') if self._attr_volume_level is not None: self._attr_volume_level = self._attr_volume_level / 100 self._attr_repeat = { 0: RepeatMode.ONE, 1: RepeatMode.ALL, 3: RepeatMode.OFF, # random }.get(info.get('loop_type'), RepeatMode.OFF) self._attr_media_content_id = mid self._attr_media_title = song.get('title') or song.get('name') self._attr_media_artist = song.get('artist') or song.get('artistName') self._attr_media_album_name = song.get('album') or song.get('albumName') self._attr_media_image_url = song.get('cover') or song.get('coverURL') self._attr_media_image_remotely_accessible = False self._attr_media_duration = int(song['duration'] / 1000) if 'duration' in song else None self._attr_media_position = int(song['position'] / 1000) if 'position' in song else None if self._attr_state == MediaPlayerState.PLAYING: self._attr_media_position_updated_at = utcnow() if not self._attr_state: self.logger.info('%s: Got empty media info: %s', self.name_model, result) except (TypeError, ValueError, Exception) as exc: self.logger.warning( '%s: Got exception while fetch xiaoai playing status: %s', self.name_model, [aid, exc], ) if unsub := self._vars.pop('unsub_play_status', None): unsub() if self.state not in [MediaPlayerState.PLAYING]: pass elif not self._attr_media_duration or self._attr_media_position is None: pass elif self._attr_media_duration >= self._attr_media_position: rem = timedelta(seconds=self._attr_media_duration - self._attr_media_position + 3) self._vars['unsub_play_status'] = async_track_point_in_utc_time( self.hass, HassJob(self.async_update_play_status), utcnow().replace(microsecond=0) + rem, ) async def async_get_media_detail(self, media: dict): mid = media.get('audio_id') or media.get('global_id') if not mid: return None api = 'https://api2.mina.mi.com/music/song_info' if is3 := self.xiaoai_device.get('capabilities', {}).get('ai_protocol_3_0', 0): api = 'https://api2.mina.mi.com/aivs3/audio/info' dat = { 'audioIdList' if is3 else 'songIdList': json.dumps([mid]), } try: result = await self.xiaoai_cloud.async_request_api(api, data=dat, method='POST') or {} for m in result.get('data') or []: if 'duration' in m: m['duration'] = int(m['duration'] * 1000) return m except (TypeError, ValueError, Exception) as exc: self.logger.info( '%s: Got exception while fetch xiaoai playing media: %s', self.name_model, [mid, exc], ) return None async def async_turn_on(self): if self._act_turn_on: return await self.async_call_action(self._act_turn_on) return False async def async_turn_off(self): if self._act_turn_off: return await self.async_call_action(self._act_turn_off) return False async def async_browse_media(self, media_content_type=None, media_content_id=None): return await media_source.async_browse_media( self.hass, media_content_id, content_filter=lambda item: item.media_content_type.startswith('audio/'), ) async def async_search_media(self, query): return SearchMedia( result=[ BrowseMedia( title=query.search_query, media_class='', media_content_id=query.search_query, media_content_type=query.media_content_type, can_play=True, can_expand=False, ), ], ) async def async_play_media(self, media_type, media_id, **kwargs): if media_source.is_media_source_id(media_id): play_item = await media_source.async_resolve_media(self.hass, media_id, self.entity_id) media_id = async_process_play_media_url(self.hass, play_item.url) self.logger.debug('async_resolve_media(%s) -> %s -> URL(%s)', media_id, play_item, media_id) if not media_id.startswith('http'): cmd = f'播放{media_id}' if media_type in ['music', 'mp3', '音乐', '歌曲']: cmd = f'播放音乐{media_id}' elif media_type in ['video', 'mp4', '视频']: cmd = f'播放视频{media_id}' return await self.async_intelligent_speaker(cmd, execute=True) if not (aid := self.xiaoai_id): return # e.g. audio/mpeg, but also keeps backward compatibility with old enum keys if media_type.startswith('audio/') or media_type in {'audio', 'music', 'voice', 'mp3', 'tts'}: return await self.async_play_music(media_id) api = 'https://api2.mina.mi.com/remote/ubus' dat = { 'deviceId': aid, 'path': 'mediaplayer', 'method': 'player_play_url', 'message': json.dumps({'url': media_id, 'type': media_type, 'media': 'app_ios'}), } rdt = await self.xiaoai_cloud.async_request_api(api, data=dat, method='POST') or {} logger = rdt.get('code') and self.logger.warning or self.logger.info logger('%s: Play media: %s', self.name_model, [dat, rdt]) async def async_play_music(self, media_id, audio_id='1582971365183456177', id='355454500', **kwargs): if not (aid := self.xiaoai_id): return music = { "payload": { "audio_type": "MUSIC", "audio_items": [ { "item_id": { "audio_id": audio_id, "cp": { "album_id": "-1", "episode_index": 0, "id": id, "name": "xiaowei", }, }, "stream": {"url": media_id}, } ], "list_params": { "listId": "-1", "loadmore_offset": 0, "origin": "xiaowei", "type": "MUSIC", }, }, "play_behavior": "REPLACE_ALL", } api = 'https://api2.mina.mi.com/remote/ubus' dat = { 'deviceId': aid, 'path': 'mediaplayer', 'method': 'player_play_music', 'message': json.dumps({"startaudioid": audio_id, "music": json.dumps(music)}), } rdt = await self.xiaoai_cloud.async_request_api(api, data=dat, method='POST') or {} logger = rdt.get('code') and self.logger.warning or self.logger.info logger('%s: Play Music: %s', self.name_model, [dat, rdt]) async def async_intelligent_speaker(self, text, execute=False, silent=False, **kwargs): if srv := self._intelligent_speaker: anm = 'execute_text_directive' if execute else 'play_text' act = srv.get_action(anm) if act: pms = [text] pse = srv.get_property('silent_execution') if execute and pse: sil = silent if pse.value_list: sil = pse.list_value('On' if silent else 'Off') if sil is None: sil = 0 if silent else 1 pms.append(sil) return await self.async_call_action(srv.iid, act.iid, pms, **kwargs) else: self.logger.warning('%s does not have action: %s', self.name_model, anm) elif self._message_router: act = self._message_router.get_action('post') if act: if not execute: text = f'跟我说 {text}' return await self.async_call_action(act, [text], **kwargs) else: self.logger.error('%s does not have service: %s', self.name_model, 'intelligent_speaker/message_router') return False async def async_xiaoai_wakeup(self, text=None, **kwargs): if srv := self._intelligent_speaker: if act := srv.get_action('wake_up'): pms = [text or ''] if act.ins else [] return await self.async_call_action(srv.iid, act.iid, pms, **kwargs) else: self.logger.warning('%s does not have action: %s', self.name_model, 'wake_up') else: self.logger.error('%s does not have service: %s', self.name_model, 'intelligent_speaker') return False class MitvMediaPlayerEntity(MiotMediaPlayerEntity): def __init__(self, config: dict, miot_service: MiotService): super().__init__(config, miot_service) self._host = self._config.get(CONF_HOST) or '' self._api_key = '881fd5a8c94b4945b46527b07eca2431' self._hmac_key = '2840d5f0d078472dbc5fb78e39da123e' self._state_attrs['6095_state'] = True self._keycode_actions = { 'power': 'turn_off', 'home': 'press_home', 'menu': 'press_menu', 'enter': 'press_ok', 'back': 'press_back', 'up': 'press_up', 'down': 'press_down', 'left': 'press_left', 'right': 'press_right', 'volumeup': 'press_volume_up', 'volumedown': 'press_volume_down', } self._speaker_mode = self._miot_service.spec.get_service('speaker_mode') self._speaker_mode_switch = self._speaker_mode.bool_property('is_on') if self._speaker_mode else None self._remote_ctrl = self._miot_service.spec.get_service('remote_control') if self._remote_ctrl: self._keycode_actions.update({ 'settings': 'press_settings', 'play': 'press_play', 'pause': 'press_pause', 'play_pause': 'press_play_pause', }) self._keycodes = list(self._keycode_actions.keys()) self._apps = {} @property def device_class(self): return MediaPlayerDeviceClass.TV @property def mitv_name(self): nam = self.device_info.get('name', '') if not re.match(r'[^x00-xff]', nam): nam = None nam = self.custom_config('television_name') or nam if not nam: sta = self.hass.states.get(self.entity_id) nam = sta.attributes.get(ATTR_FRIENDLY_NAME) if sta else None return nam or self.device_info.get('name', '电视') @property def bind_xiaoai(self): eid = self.custom_config('bind_xiaoai') if not eid or not self.hass: return None return self.hass.states.get(eid) async def async_added_to_hass(self): await super().async_added_to_hass() if lip := self.custom_config('mitv_lan_host'): self._host = lip self._config = {**self._config, CONF_HOST: lip} self._device = None await self.async_update_apps() sva = self.custom_config_list('sources_via_apps') if self.custom_config('sources_via_apps') in [True, 'true', 'all', '*']: sva = list(self._apps.values()) if sva: if not self.custom_config_bool('source_list_append', True): self._attr_source_list = [] self._attr_source_list.extend(sva) self._vars['sources_via_apps'] = sva svk = self.custom_config_list('sources_via_keycodes') if self.custom_config('sources_via_keycodes') in [True, 'true', 'all', '*']: svk = [*self._keycodes] if svk: if not sva: self._attr_source_list = [] self._attr_source_list.extend(svk) self._vars['sources_via_keycodes'] = svk if add_selects := self._add_entities.get('select'): from .select import SelectSubEntity sub = 'keycodes' self._subs[sub] = SelectSubEntity(self, sub, option={ 'options': self._keycodes, 'async_select_option': self.async_press_key, }) add_selects([self._subs[sub]], update_before_add=False) self._vars['homekit_remote_unsub'] = self.hass.bus.async_listen( EVENT_HOMEKIT_TV_REMOTE_KEY_PRESSED, self.async_homekit_remote_event_handler, ) async def async_will_remove_from_hass(self): """Run when entity will be removed from hass. To be extended by integrations. """ await super().async_will_remove_from_hass() if unsub := self._vars.pop('homekit_remote_unsub', None): unsub() async def async_update(self): await super().async_update() if not self._available: return adt = {} pms = self.with_opaque({ 'action': 'capturescreen', 'compressrate': self.custom_config_integer('screenshot_compress') or 50, }) prev_6095 = self._state_attrs.get('6095_state') rdt = await self.async_request_mitv_api('controller', params=pms) if 'url' in rdt: url = rdt.get('url', '') pms = urlparse(url).query url = f'{url}'.replace(pms, '') url = url.replace('//null:', f'//{self._host}:') url = url.replace('//0.0.0.0:', f'//{self._host}:') pms = dict(parse_qsl(pms)) pms = self.with_opaque(pms, token=rdt.get('token')) self._attr_media_image_url = url + urlencode(pms) self._attr_app_id = rdt.get('pkg') self._attr_app_name = rdt.get('label') adt.update({ 'capture': self._attr_media_image_url, 'capture_token': rdt.get('token'), 'app_current': f'{self._attr_app_name} - {self._attr_app_id}', 'app_page': rdt.get('clz'), }) self._state_attrs.update(adt) if prev_6095 != self._state_attrs.get('6095_state'): await self.async_update_apps() async def async_update_apps(self): if not self._state_attrs.get('6095_state', True): return pms = { 'action': 'getinstalledapp', 'count': 999, 'changeIcon': 1, } rdt = await self.async_request_mitv_api('controller', params=pms) if lst := rdt.get('AppInfo', []): self._apps = { a.get('PackageName'): a.get('AppName') for a in lst } als = [ f'{v} - {k}' for k, v in self._apps.items() ] add_selects = self._add_entities.get('select') sub = 'apps' if sub in self._subs: self._subs[sub].update_options(als) self._subs[sub].update_from_parent() elif add_selects: from .select import SelectSubEntity self._subs[sub] = SelectSubEntity(self, 'app_current', option={ 'options': als, 'async_select_option': self.async_start_app, }) add_selects([self._subs[sub]], update_before_add=False) @property def state(self): sta = super().state if not self.cloud_only and not self._local_state and not self._state_attrs.get('6095_state'): sta = MediaPlayerState.OFF if self._speaker_mode_switch and self.custom_config_bool('turn_off_screen'): if self._speaker_mode_switch.from_device(self.device): sta = MediaPlayerState.OFF return sta async def async_turn_on(self): if self._local_state and self._state_attrs.get('6095_state'): # tv is on pass elif self._speaker_mode_switch: await self.async_set_property(self._speaker_mode_switch, False) elif xai := self.bind_xiaoai: nam = self.mitv_name txt = f'{nam}亮屏' if self._local_state else f'打开{nam}' await self.hass.services.async_call(DOMAIN, 'intelligent_speaker', { 'entity_id': xai.entity_id, 'text': txt, 'execute': True, 'silent': self.custom_config_bool('xiaoai_silent', True), }) return await super().async_turn_on() async def async_turn_off(self): if self.custom_config_bool('turn_off_screen'): act = self._message_router.get_action('post') if self._message_router else None if self._speaker_mode_switch: return await self.async_set_property(self._speaker_mode_switch, True) elif xai := self.bind_xiaoai: return await self.hass.services.async_call(DOMAIN, 'intelligent_speaker', { 'entity_id': xai.entity_id, 'text': f'{self.mitv_name}熄屏', 'execute': True, 'silent': self.custom_config_bool('xiaoai_silent', True), }) elif act: return await self.async_call_action(act, ['熄屏']) return await super().async_turn_off() async def async_play_media(self, media_type, media_id, **kwargs): """Play a piece of media.""" if media_source.is_media_source_id(media_id): play_item = await media_source.async_resolve_media(self.hass, media_id, self.entity_id) media_id = async_process_play_media_url(self.hass, play_item.url) tim = str(int(time.time() * 1000)) pms = { 'action': 'play', 'type': media_type, 'url': media_id, 'apikey': self._api_key, 'ts': tim, 'sign': hashlib.md5(f'mitvsignsalt{media_id}{self._api_key}{tim[-5:]}'.encode()).hexdigest(), } rdt = await self.async_request_mitv_api('controller', params=pms) self.logger.info('%s: Play media: %s', self.name_model, [pms, rdt]) @property def source(self): """Name of the current input source.""" if self.app_name in self._vars.get('sources_via_apps', []): return self.app_name return super().source async def async_select_source(self, source): """Select input source.""" if source in self._apps: return await self.async_start_app(self._apps[source]) if source in self._apps.values(): return await self.async_start_app(source) if source in self._keycodes: ret = await self.async_press_key(source) self._attr_app_name = source self.schedule_update_ha_state() return ret if source in self.source_list: return await super().async_select_source(source) return False async def async_start_app(self, app, **kwargs): pkg = f'{app}'.split(' - ').pop(-1).strip() if pkg not in self._apps: pkg = None for k, v in self._apps.items(): if v == app: pkg = k break if pkg is None: return False pms = { 'action': 'startapp', 'type': 'packagename', 'packagename': pkg, } return await self.async_request_mitv_api('controller', params=pms) async def async_press_key(self, key, **kwargs): if self._remote_ctrl: act = self._keycode_actions.get(key) or key if act := self._remote_ctrl.get_action(act): return await self.async_call_action(act) pms = { 'action': 'keyevent', 'keycode': key, } return await self.async_request_mitv_api('controller', params=pms) async def async_homekit_remote_event_handler(self, event): eid = event.data.get('entity_id') if eid != self.entity_id: return dic = { 'arrow_up': 'up', 'arrow_down': 'down', 'arrow_left': 'left', 'arrow_right': 'right', 'back': 'back', 'select': 'enter', 'information': 'menu', } key = dic.get(event.data.get('key_name', '')) if not key: return return await self.async_press_key(key) def with_opaque(self, pms: dict, token=None): if token is None: token = self._api_key pms.update({ 'timestamp': int(time.time() * 1000), 'token': token, }) pms['opaque'] = hmac.new(self._hmac_key.encode(), urlencode(pms).encode(), hashlib.sha1).hexdigest() pms.pop('token', None) return pms def mitv_api_path(self, path=''): return f'http://{self._host}:6095/{path.lstrip("/")}' def request_mitv_api(self, path, **kwargs): kwargs.setdefault('timeout', 5) req = None try: req = requests.get(self.mitv_api_path(path), **kwargs) rdt = json.loads(req.content or '{}') or {} self._state_attrs['6095_state'] = True if 'success' not in rdt.get('msg', ''): self.logger.warning('%s: Request mitv api error: %s', self.name_model, req.text) except requests.exceptions.RequestException as exc: rdt = {} if self._state_attrs.get('6095_state'): log = self.logger.info if 'NewConnectionError' in f'{exc}' else self.logger.warning log('%s: Request mitv api error: %s', self.name_model, exc) self._state_attrs['6095_state'] = False except json.decoder.JSONDecodeError: rdt = {} if req: self.logger.warning('%s: Invalid response data: %s with %s', req.content, kwargs) return rdt.get('data') or {} async def async_request_mitv_api(self, path, **kwargs): return await self.hass.async_add_executor_job( partial(self.request_mitv_api, path, **kwargs) ) class MiirMediaPlayerEntity(MiirToggleEntity, MediaPlayerEntity): def __init__(self, config: dict, miot_service: MiotService): super().__init__(miot_service, config=config, logger=_LOGGER) if self._act_turn_on: self._supported_features |= MediaPlayerEntityFeature.TURN_ON if self._act_turn_off: self._supported_features |= MediaPlayerEntityFeature.TURN_OFF self._attr_is_volume_muted = None self._act_mute_on = miot_service.get_action('mute_on') self._act_mute_off = miot_service.get_action('mute_off') if self._act_mute_on or self._act_mute_off: self._supported_features |= MediaPlayerEntityFeature.VOLUME_MUTE self._attr_volume_level = 0.5 self._act_volume_up = miot_service.get_action('volume_up') self._act_volume_dn = miot_service.get_action('volume_down') if self._act_volume_up or self._act_volume_dn: self._supported_features |= MediaPlayerEntityFeature.VOLUME_SET self._supported_features |= MediaPlayerEntityFeature.VOLUME_STEP if self._miot_actions: self._supported_features |= MediaPlayerEntityFeature.SELECT_SOURCE self._attr_source_list = self._miot_actions async def async_mute_volume(self, mute): """Mute the volume.""" ret = None if not mute and self._act_mute_off: ret = await self.async_call_action(self._act_mute_off) elif mute and self._act_mute_on: ret = await self.async_call_action(self._act_mute_on) if ret: self._attr_is_volume_muted = mute return ret async def async_set_volume_level(self, volume): """Set volume level, range 0..1.""" if volume > self._attr_volume_level and self._act_volume_up: return await self.async_call_action(self._act_volume_up) elif volume < self._attr_volume_level and self._act_volume_dn: return await self.async_call_action(self._act_volume_dn) raise NotImplementedError() async def async_select_source(self, source): """Select input source.""" if act := self._miot_service.get_action(source): return await self.async_call_action(act) raise NotImplementedError()