ha-pjlink2/custom_components/pjlink2/media_player.py

"""PJLink2 media_player platform."""

from __future__ import annotations

from collections.abc import Callable
from datetime import timedelta
import logging
from typing import Any

from aiopjlink import (
    PJLink,
    PJLinkException,
    PJLinkProjectorError,
    Power,
    Sources,
    Lamp,
    Information,
)

from homeassistant import config_entries, core
from homeassistant.components.media_player import (
    MediaPlayerEntity,
    MediaPlayerEntityFeature,
    MediaPlayerState,
    PLATFORM_SCHEMA,
)
from homeassistant.const import (
    CONF_HOST,
    CONF_PORT,
    CONF_NAME,
    CONF_PASSWORD,
    CONF_TIMEOUT,
)
from homeassistant.core import HomeAssistant as HomeAssistantType

import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType

import voluptuous as vol

from .const import (
    DOMAIN,
    CONF_ENCODING,
    DEFAULT_ENCODING,
    DEFAULT_PORT,
    DEFAULT_TIMEOUT,
    ATTR_PRODUCT_NAME,
    ATTR_MANUFACTURER_NAME,
    ATTR_PROJECTOR_NAME,
    ATTR_RESOLUTION_X,
    ATTR_RESOLUTION_Y,
    ATTR_LAMP_HOURS,
    ProjectorState,
)

_LOGGER = logging.getLogger(__name__)
# Time between updating data from projector
SCAN_INTERVAL = timedelta(seconds=3)

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_HOST): cv.string,
        vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
        vol.Optional(CONF_NAME): cv.string,
        vol.Optional(CONF_ENCODING, default=DEFAULT_ENCODING): cv.string,
        vol.Optional(CONF_PASSWORD): cv.string,
        vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_float,
    }
)


async def async_setup_platform(
    hass: HomeAssistantType,
    config: ConfigType,
    async_add_entities: Callable,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the media_player platform."""
    host = config.get(CONF_HOST)
    port = config.get(CONF_PORT)
    password = config.get(CONF_PASSWORD)
    timeout = config.get(CONF_TIMEOUT)
    name = config.get(CONF_NAME)
    pjl = PJLink(host, port, password, timeout)
    devices = [PJLink2MediaPlayer(pjl, name)]
    async_add_entities(devices, update_before_add=False)


class PJLink2MediaPlayer(MediaPlayerEntity):
    """Representation of a PJLink2 media player."""

    _attr_supported_features = (
        MediaPlayerEntityFeature.TURN_ON
        | MediaPlayerEntityFeature.TURN_OFF
        | MediaPlayerEntityFeature.SELECT_SOURCE
    )

    def __init__(self, pjl, name):
        super().__init__()
        self._projector = pjl
        self.attrs: dict[str, Any] = {}
        self._name = name
        self._state = MediaPlayerState.OFF
        self._available = False
        self._connectionErrorLogged = False
        self._current_source = None

        # PJLink standard inputs. You can modify these to friendly names later.
        # Format is typically "31" for HDMI1, "32" for HDMI2, etc.
        self._source_list = ["11", "12", "21", "22", "31", "32", "33"]

    async def async_will_remove_from_hass(self) -> None:
        """Close connection."""
        await super().async_will_remove_from_hass()
        if self._available:
            try:
                await self._projector.__aexit__(0, 0, 0)
            except (PJLinkException, OSError) as err:
                _LOGGER.error(
                    "PJLink2 ERROR when closing connection: %s", repr(err)
                )

    @property
    def name(self) -> str:
        return self._name

    @property
    def unique_id(self) -> str:
        return self._projector._address

    @property
    def available(self) -> bool:
        return self._available

    @property
    def state(self) -> MediaPlayerState:
        return self._state

    @property
    def source(self) -> str | None:
        """Name of the current input source."""
        return self._current_source

    @property
    def source_list(self) -> list[str]:
        """List of available input sources."""
        return self._source_list

    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        """Return the custom PJLink2 attributes."""
        return self.attrs

    async def async_turn_on(self) -> None:
        """Turn the projector on."""
        await Power(self._projector).set(Power.ON)
        self._state = MediaPlayerState.ON

    async def async_turn_off(self) -> None:
        """Turn the projector off."""
        await Power(self._projector).set(Power.OFF)
        self._state = MediaPlayerState.OFF

    async def async_select_source(self, source: str) -> None:
        """Select input source."""
        source_type = source[0]
        source_index = source[1]

        await Sources(self._projector).set(source_type, source_index)
        self._current_source = source

    async def async_update(self) -> None:
        """Update data from projector."""
        try:
            if not self._available:
                await self._projector.__aenter__()
                self._available = True
                info = await Information(self._projector).table()
                self.attrs[ATTR_PRODUCT_NAME] = info["product_name"]
                self.attrs[ATTR_MANUFACTURER_NAME] = info["manufacturer_name"]
                self.attrs[ATTR_PROJECTOR_NAME] = info["projector_name"]
                if self._name == None:
                    self._name = info["projector_name"]

            pwr = await Power(self._projector).get()
            if pwr == Power.State.OFF:
                self._state = MediaPlayerState.OFF
            elif pwr == Power.State.ON:
                self._state = MediaPlayerState.ON
            elif pwr in (Power.State.COOLING, Power.State.WARMING):
                self._state = (
                    MediaPlayerState.ON
                )  # Keeps UI active during transition

            if pwr == Power.ON:
                res = await Sources(self._projector).resolution()
                self.attrs[ATTR_RESOLUTION_X] = res[0]
                self.attrs[ATTR_RESOLUTION_Y] = res[1]
                self.attrs[ATTR_LAMP_HOURS] = await Lamp(
                    self._projector
                ).hours()

                # Fetch current source
                current = await Sources(self._projector).get()
                if isinstance(current, (tuple, list)):
                    self._current_source = "".join(map(str, current))
                else:
                    self._current_source = str(current)
            else:
                self.attrs.pop(ATTR_RESOLUTION_X, None)
                self.attrs.pop(ATTR_RESOLUTION_Y, None)
                self._current_source = None

            self._connectionErrorLogged = False

        except PJLinkProjectorError:
            self.attrs.pop(ATTR_RESOLUTION_X, None)
            self.attrs.pop(ATTR_RESOLUTION_Y, None)
        except (PJLinkException, OSError) as err:
            if not self._connectionErrorLogged:
                _LOGGER.error("PJLink2 ERROR for %s: %s", self._name, repr(err))
                self._connectionErrorLogged = True
            self._state = MediaPlayerState.OFF
            if self._available:
                self._available = False
                try:
                    await self._projector.__aexit__(0, 0, 0)
                except Exception:
                    pass