219 lines
7.0 KiB
Python
219 lines
7.0 KiB
Python
"""Media Player platform for BenQ Smart Board integration."""
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from homeassistant.components.media_player import (
|
|
MediaPlayerEntity,
|
|
MediaPlayerEntityFeature,
|
|
)
|
|
from homeassistant.components.media_player.const import (
|
|
MEDIA_TYPE_TVSHOW,
|
|
SUPPORT_VOLUME_MUTE,
|
|
SUPPORT_VOLUME_SET,
|
|
SUPPORT_TURN_ON,
|
|
SUPPORT_TURN_OFF,
|
|
)
|
|
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE
|
|
from homeassistant.helpers.entity import DeviceInfo
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
|
|
from . import DOMAIN
|
|
from .benq_smartboard_lib import (
|
|
BenQSmartBoard,
|
|
BenQSmartBoardError,
|
|
ConnectionError,
|
|
PowerState,
|
|
)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
SUPPORT_BENQ_SMARTBOARD = (
|
|
SUPPORT_VOLUME_SET
|
|
| SUPPORT_VOLUME_MUTE
|
|
| SUPPORT_TURN_OFF
|
|
| SUPPORT_TURN_ON
|
|
)
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
|
):
|
|
"""Set up the BenQ Smart Board media player from a config entry."""
|
|
data = hass.data[DOMAIN][entry.entry_id]
|
|
host = data["host"]
|
|
port = data["port"]
|
|
|
|
name = f"BenQ Smart Board ({host})"
|
|
|
|
entity = BenQSmartBoardMediaPlayer(
|
|
name=name,
|
|
host=host,
|
|
port=port,
|
|
unique_id=f"benq_smartboard_{host}_{port}",
|
|
)
|
|
async_add_entities([entity], update_before_add=True)
|
|
|
|
|
|
class BenQSmartBoardMediaPlayer(MediaPlayerEntity):
|
|
"""Representation of the BenQ Smart Board as a Media Player."""
|
|
|
|
_attr_should_poll = True
|
|
_attr_supported_features = SUPPORT_BENQ_SMARTBOARD
|
|
_attr_media_content_type = MEDIA_TYPE_TVSHOW
|
|
|
|
def __init__(self, name: str, host: str, port: int, unique_id: str):
|
|
"""Initialize."""
|
|
self._attr_name = name
|
|
self._attr_unique_id = unique_id
|
|
self._host = host
|
|
self._port = port
|
|
|
|
# Internal states
|
|
self._state = STATE_UNAVAILABLE
|
|
self._volume_level = 0.0
|
|
self._is_muted = False
|
|
|
|
self._board: Optional[BenQSmartBoard] = None
|
|
|
|
# Setup device info so it appears as a device in HA
|
|
self._attr_device_info = DeviceInfo(
|
|
identifiers={(DOMAIN, unique_id)},
|
|
name=name,
|
|
manufacturer="BenQ",
|
|
model="Smart Board",
|
|
configuration_url=f"http://{host}", # or other relevant link
|
|
)
|
|
|
|
@property
|
|
def state(self):
|
|
"""Return the current power state: ON, OFF, or if unreachable -> UNAVAILABLE."""
|
|
return self._state
|
|
|
|
@property
|
|
def volume_level(self):
|
|
"""Volume level (0..1)."""
|
|
return self._volume_level
|
|
|
|
@property
|
|
def is_volume_muted(self):
|
|
"""Return True if muted."""
|
|
return self._is_muted
|
|
|
|
def _ensure_board(self):
|
|
"""Ensure we have a connected BenQSmartBoard instance."""
|
|
if self._board is None:
|
|
self._board = BenQSmartBoard(self._host, self._port, timeout=3.0)
|
|
return self._board
|
|
|
|
def _refresh_from_device(self):
|
|
"""
|
|
Attempt to query the current state from the device (power, volume, mute).
|
|
Called in update().
|
|
"""
|
|
board = self._ensure_board()
|
|
try:
|
|
# Connect if not connected
|
|
board.connect()
|
|
|
|
# Get power
|
|
pwr = board.get_power() # string e.g. "001" -> ON
|
|
if pwr == PowerState.ON.value:
|
|
self._state = STATE_ON
|
|
elif pwr == PowerState.STANDBY.value:
|
|
# Some prefer modeling STANDBY as OFF or custom
|
|
self._state = STATE_OFF
|
|
elif pwr == PowerState.OFF.value:
|
|
self._state = STATE_OFF
|
|
else:
|
|
# If unknown power code, consider it OFF or UNAVAILABLE
|
|
self._state = STATE_OFF
|
|
|
|
# Volume: e.g. "030"
|
|
vol_str = board.get_volume()
|
|
volume_int = int(vol_str)
|
|
self._volume_level = volume_int / 100.0
|
|
|
|
# Mute: "000" => Unmuted, "001" => Muted
|
|
mute_str = board.get_mute()
|
|
self._is_muted = (mute_str == "001")
|
|
|
|
except BenQSmartBoardError as err:
|
|
_LOGGER.warning("Error talking to BenQ Smart Board: %s", err)
|
|
self._state = STATE_UNAVAILABLE
|
|
except ConnectionError as err:
|
|
_LOGGER.warning("Connection to BenQ Smart Board lost: %s", err)
|
|
self._state = STATE_UNAVAILABLE
|
|
|
|
async def async_update(self):
|
|
"""Update the entity state by polling the device."""
|
|
await self.hass.async_add_executor_job(self._refresh_from_device)
|
|
|
|
async def async_turn_on(self):
|
|
"""Turn on the Smart Board (if off or in standby)."""
|
|
board = self._ensure_board()
|
|
try:
|
|
board.connect()
|
|
success = board.set_power(PowerState.ON)
|
|
if success:
|
|
self._state = STATE_ON
|
|
else:
|
|
_LOGGER.warning("Smart Board refused 'power on' command.")
|
|
except BenQSmartBoardError as e:
|
|
_LOGGER.error("Failed to turn on: %s", e)
|
|
self._state = STATE_UNAVAILABLE
|
|
self.async_write_ha_state()
|
|
|
|
async def async_turn_off(self):
|
|
"""Turn off the Smart Board."""
|
|
board = self._ensure_board()
|
|
try:
|
|
board.connect()
|
|
success = board.set_power(PowerState.OFF)
|
|
if success:
|
|
self._state = STATE_OFF
|
|
else:
|
|
_LOGGER.warning("Smart Board refused 'power off' command.")
|
|
except BenQSmartBoardError as e:
|
|
_LOGGER.error("Failed to turn off: %s", e)
|
|
self._state = STATE_UNAVAILABLE
|
|
self.async_write_ha_state()
|
|
|
|
async def async_set_volume_level(self, volume: float):
|
|
"""Set volume level (0..1)."""
|
|
board = self._ensure_board()
|
|
vol_int = int(volume * 100)
|
|
try:
|
|
board.connect()
|
|
success = board.set_volume(vol_int)
|
|
if success:
|
|
self._volume_level = volume
|
|
else:
|
|
_LOGGER.warning("Smart Board refused 'set volume' command.")
|
|
except BenQSmartBoardError as e:
|
|
_LOGGER.error("Failed to set volume: %s", e)
|
|
self._state = STATE_UNAVAILABLE
|
|
self.async_write_ha_state()
|
|
|
|
async def async_mute_volume(self, mute: bool):
|
|
"""Mute or unmute the volume."""
|
|
board = self._ensure_board()
|
|
try:
|
|
board.connect()
|
|
success = board.set_mute(mute)
|
|
if success:
|
|
self._is_muted = mute
|
|
else:
|
|
_LOGGER.warning("Smart Board refused 'mute' command.")
|
|
except BenQSmartBoardError as e:
|
|
_LOGGER.error("Failed to mute: %s", e)
|
|
self._state = STATE_UNAVAILABLE
|
|
self.async_write_ha_state()
|
|
|
|
async def async_will_remove_from_hass(self):
|
|
"""Disconnect from device on removal."""
|
|
if self._board:
|
|
await self.hass.async_add_executor_job(self._board.disconnect)
|