I couldn’t figure out why it doesn’t work so I merged the original small piece of code with the full home assistant samsung media player and now all works fine.
If I have time I’ll try to figure it out later. Here it is anyway. It updates status properly and you also have the send_key service.
"""Support for interface with an Samsung TV."""
import asyncio
from datetime import timedelta
import logging
import socket
import voluptuous as vol
# added DOMAIN
from homeassistant.components.media_player import (
MediaPlayerDevice, PLATFORM_SCHEMA, DOMAIN)
from homeassistant.components.media_player.const import (
MEDIA_TYPE_CHANNEL, SUPPORT_NEXT_TRACK, SUPPORT_PAUSE,
SUPPORT_PLAY, SUPPORT_PLAY_MEDIA, SUPPORT_PREVIOUS_TRACK, SUPPORT_TURN_OFF,
SUPPORT_TURN_ON, SUPPORT_VOLUME_MUTE, SUPPORT_VOLUME_STEP)
# added ATTR_ENTITY_ID
from homeassistant.const import (
CONF_HOST, CONF_MAC, CONF_NAME, CONF_PORT, CONF_TIMEOUT, STATE_OFF,
STATE_ON, ATTR_ENTITY_ID)
import homeassistant.helpers.config_validation as cv
from homeassistant.util import dt as dt_util
_LOGGER = logging.getLogger(__name__)
# added
SAMSUNG_TV_CUSTOM_DATA = 'samsungtv_custom'
# added
SERVICE_KEY = 'send_key'
# added
# Service call validation schemas
ATTR_KEY = 'key_code'
# added
SAMSUNG_TV_CUSTOM_KEY_SCHEMA = vol.Schema({
vol.Required(ATTR_ENTITY_ID): cv.entity_ids,
vol.Required(ATTR_KEY): cv.string,
})
DEFAULT_NAME = 'Samsung TV Remote'
DEFAULT_PORT = 55000
DEFAULT_TIMEOUT = 1
KEY_PRESS_TIMEOUT = 1.2
KNOWN_DEVICES_KEY = 'samsungtv_known_devices'
SUPPORT_SAMSUNGTV = SUPPORT_PAUSE | SUPPORT_VOLUME_STEP | \
SUPPORT_VOLUME_MUTE | SUPPORT_PREVIOUS_TRACK | \
SUPPORT_NEXT_TRACK | SUPPORT_TURN_OFF | SUPPORT_PLAY | SUPPORT_PLAY_MEDIA
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_MAC): cv.string,
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
})
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Samsung TV platform."""
known_devices = hass.data.get(KNOWN_DEVICES_KEY)
if known_devices is None:
known_devices = set()
hass.data[KNOWN_DEVICES_KEY] = known_devices
if SAMSUNG_TV_CUSTOM_DATA not in hass.data:
hass.data[SAMSUNG_TV_CUSTOM_DATA] = []
uuid = None
# Is this a manual configuration?
if config.get(CONF_HOST) is not None:
host = config.get(CONF_HOST)
port = config.get(CONF_PORT)
name = config.get(CONF_NAME)
mac = config.get(CONF_MAC)
timeout = config.get(CONF_TIMEOUT)
elif discovery_info is not None:
tv_name = discovery_info.get('name')
model = discovery_info.get('model_name')
host = discovery_info.get('host')
name = "{} ({})".format(tv_name, model)
port = DEFAULT_PORT
timeout = DEFAULT_TIMEOUT
mac = None
udn = discovery_info.get('udn')
if udn and udn.startswith('uuid:'):
uuid = udn[len('uuid:'):]
else:
_LOGGER.warning("Cannot determine device")
return
# Only add a device once, so discovered devices do not override manual
# config.
ip_addr = socket.gethostbyname(host)
if ip_addr not in known_devices:
known_devices.add(ip_addr)
# modified
device = [SamsungTVDevice(host, port, name, timeout, mac, uuid)]
# modified
add_entities(device)
# added
hass.data[SAMSUNG_TV_CUSTOM_DATA].append(device[0])
_LOGGER.info("Samsung TV %s:%d added as '%s'", host, port, name)
else:
_LOGGER.info("Ignoring duplicate Samsung TV %s:%d", host, port)
def service_handle(service):
_LOGGER.debug("service_handle called for %s with %s", service.service, service.data)
entity_ids = service.data.get('entity_id')
devices = hass.data[SAMSUNG_TV_CUSTOM_DATA]
for device in devices:
if device.entity_id in entity_ids:
if service.service == SERVICE_KEY:
device.send_key(service.data.get(ATTR_KEY))
device.schedule_update_ha_state(True)
hass.services.register(
DOMAIN, SERVICE_KEY, service_handle,
schema=SAMSUNG_TV_CUSTOM_KEY_SCHEMA)
class SamsungTVDevice(MediaPlayerDevice):
"""Representation of a Samsung TV."""
def __init__(self, host, port, name, timeout, mac, uuid):
"""Initialize the Samsung device."""
from samsungctl import exceptions
from samsungctl import Remote
import wakeonlan
# Save a reference to the imported classes
self._exceptions_class = exceptions
self._remote_class = Remote
self._name = name
self._mac = mac
self._uuid = uuid
self._wol = wakeonlan
# Assume that the TV is not muted
self._muted = False
# Assume that the TV is in Play mode
self._playing = True
self._state = None
self._remote = None
# Mark the end of a shutdown command (need to wait 15 seconds before
# sending the next command to avoid turning the TV back ON).
self._end_of_power_off = None
# Generate a configuration for the Samsung library
self._config = {
'name': 'HomeAssistant',
'description': name,
'id': 'ha.component.samsung',
'port': port,
'host': host,
'timeout': timeout,
}
if self._config['port'] == 8001:
self._config['method'] = 'websocket'
else:
self._config['method'] = 'legacy'
def update(self):
"""Update state of device."""
self.send_key("KEY")
def get_remote(self):
"""Create or return a remote control instance."""
if self._remote is None:
# We need to create a new instance to reconnect.
self._remote = self._remote_class(self._config)
return self._remote
def send_key(self, key):
"""Send a key to the tv and handles exceptions."""
if self._power_off_in_progress() \
and key not in ('KEY_POWER', 'KEY_POWEROFF'):
_LOGGER.info("TV is powering off, not sending command: %s", key)
return
try:
# recreate connection if connection was dead
retry_count = 1
for _ in range(retry_count + 1):
try:
self.get_remote().control(key)
break
except (self._exceptions_class.ConnectionClosed,
BrokenPipeError):
# BrokenPipe can occur when the commands is sent to fast
self._remote = None
self._state = STATE_ON
except (self._exceptions_class.UnhandledResponse,
self._exceptions_class.AccessDenied):
# We got a response so it's on.
self._state = STATE_ON
self._remote = None
_LOGGER.debug("Failed sending command %s", key, exc_info=True)
return
except OSError:
self._state = STATE_OFF
self._remote = None
if self._power_off_in_progress():
self._state = STATE_OFF
def _power_off_in_progress(self):
return self._end_of_power_off is not None and \
self._end_of_power_off > dt_util.utcnow()
@property
def unique_id(self) -> str:
"""Return the unique ID of the device."""
return self._uuid
@property
def name(self):
"""Return the name of the device."""
return self._name
@property
def state(self):
"""Return the state of the device."""
return self._state
@property
def is_volume_muted(self):
"""Boolean if volume is currently muted."""
return self._muted
@property
def supported_features(self):
"""Flag media player features that are supported."""
if self._mac:
return SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON
return SUPPORT_SAMSUNGTV
def turn_off(self):
"""Turn off media player."""
self._end_of_power_off = dt_util.utcnow() + timedelta(seconds=15)
if self._config['method'] == 'websocket':
self.send_key('KEY_POWER')
else:
self.send_key('KEY_POWEROFF')
# Force closing of remote session to provide instant UI feedback
try:
self.get_remote().close()
self._remote = None
except OSError:
_LOGGER.debug("Could not establish connection.")
def volume_up(self):
"""Volume up the media player."""
self.send_key('KEY_VOLUP')
def volume_down(self):
"""Volume down media player."""
self.send_key('KEY_VOLDOWN')
def mute_volume(self, mute):
"""Send mute command."""
self.send_key('KEY_MUTE')
def media_play_pause(self):
"""Simulate play pause media player."""
if self._playing:
self.media_pause()
else:
self.media_play()
def media_play(self):
"""Send play command."""
self._playing = True
self.send_key('KEY_PLAY')
def media_pause(self):
"""Send media pause command to media player."""
self._playing = False
self.send_key('KEY_PAUSE')
def media_next_track(self):
"""Send next track command."""
self.send_key('KEY_FF')
def media_previous_track(self):
"""Send the previous track command."""
self.send_key('KEY_REWIND')
async def async_play_media(self, media_type, media_id, **kwargs):
"""Support changing a channel."""
if media_type != MEDIA_TYPE_CHANNEL:
_LOGGER.error('Unsupported media type')
return
# media_id should only be a channel number
try:
cv.positive_int(media_id)
except vol.Invalid:
_LOGGER.error('Media ID must be positive integer')
return
for digit in media_id:
await self.hass.async_add_job(self.send_key, 'KEY_' + digit)
await asyncio.sleep(KEY_PRESS_TIMEOUT, self.hass.loop)
def turn_on(self):
"""Turn the media player on."""
if self._mac:
self._wol.send_magic_packet(self._mac)
else:
self.send_key('KEY_POWERON')
