HA-EasyComputerManager/custom_components/easy_computer_manager/switch.py

# Some snippets of code are from the official wake_on_lan integration (inspiration for this custom component)

from __future__ import annotations

import asyncio
from typing import Any

import voluptuous as vol
from homeassistant.components.switch import (SwitchEntity)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    CONF_HOST,
    CONF_MAC,
    CONF_NAME,
    CONF_PASSWORD,
    CONF_PORT,
    CONF_USERNAME, )
from homeassistant.core import HomeAssistant, ServiceResponse, SupportsResponse
from homeassistant.helpers import (
    device_registry as dr,
    entity_platform,
)
from homeassistant.helpers.config_validation import make_entity_service_schema
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback

from .computer import OSType, Computer
from .computer.utils import format_debug_information, get_bluetooth_devices_as_str
from .const import SERVICE_RESTART_TO_WINDOWS_FROM_LINUX, SERVICE_PUT_COMPUTER_TO_SLEEP, \
    SERVICE_START_COMPUTER_TO_WINDOWS, SERVICE_RESTART_COMPUTER, SERVICE_RESTART_TO_LINUX_FROM_WINDOWS, \
    SERVICE_CHANGE_MONITORS_CONFIG, SERVICE_STEAM_BIG_PICTURE, SERVICE_CHANGE_AUDIO_CONFIG, SERVICE_DEBUG_INFO, DOMAIN


async def async_setup_entry(
        hass: HomeAssistant,
        config: ConfigEntry,
        async_add_entities: AddEntitiesCallback,
) -> None:
    # Retrieve the data from the config flow
    mac_address: str = config.data.get(CONF_MAC)
    # broadcast_address: str | None = config.data.get(CONF_BROADCAST_ADDRESS)
    # broadcast_port: int | None = config.data.get(CONF_BROADCAST_PORT)
    host: str = config.data.get(CONF_HOST)
    name: str = config.data.get(CONF_NAME)
    dualboot: bool = config.data.get("dualboot")
    username: str = config.data.get(CONF_USERNAME)
    password: str = config.data.get(CONF_PASSWORD)
    port: int | None = config.data.get(CONF_PORT)

    # Register the computer switch
    async_add_entities(
        [
            ComputerSwitch(
                hass,
                name,
                host,
                mac_address,
                dualboot,
                username,
                password,
                port,
            ),
        ],
        host is not None,
    )

    platform = entity_platform.async_get_current_platform()

    # Synthax : (service_name: str, schema: dict, supports_response: SupportsResponse)
    services = [
        (SERVICE_RESTART_TO_WINDOWS_FROM_LINUX, {}, SupportsResponse.NONE),
        (SERVICE_RESTART_TO_WINDOWS_FROM_LINUX, {}, SupportsResponse.NONE),
        (SERVICE_RESTART_TO_LINUX_FROM_WINDOWS, {}, SupportsResponse.NONE),
        (SERVICE_PUT_COMPUTER_TO_SLEEP, {}, SupportsResponse.NONE),
        (SERVICE_START_COMPUTER_TO_WINDOWS, {}, SupportsResponse.NONE),
        (SERVICE_RESTART_COMPUTER, {}, SupportsResponse.NONE),
        (SERVICE_CHANGE_MONITORS_CONFIG, {vol.Required("monitors_config"): dict}, SupportsResponse.NONE),
        (SERVICE_STEAM_BIG_PICTURE, {vol.Required("action"): str}, SupportsResponse.NONE),
        (SERVICE_CHANGE_AUDIO_CONFIG, {
            vol.Optional("volume"): int,
            vol.Optional("mute"): bool,
            vol.Optional("input_device"): str,
            vol.Optional("output_device"): str
        }, SupportsResponse.NONE),
        (SERVICE_DEBUG_INFO, {}, SupportsResponse.ONLY),
    ]

    # Register the services that depends on the switch
    for service in services:
        platform.async_register_entity_service(
            service[0],
            make_entity_service_schema(service[1]),
            service[0],
            supports_response=service[2]
        )


class ComputerSwitch(SwitchEntity):
    """Representation of a computer switch."""

    def __init__(
            self,
            hass: HomeAssistant,
            name: str,
            host: str | None,
            mac_address: str,
            dualboot: bool | False,
            username: str,
            password: str,
            port: int | None,
    ) -> None:
        """Initialize the WOL switch."""

        self._hass = hass
        self._attr_name = name

        self.computer = Computer(host, mac_address, username, password, port, dualboot)

        self._state = False
        self._attr_assumed_state = host is None
        self._attr_should_poll = bool(not self._attr_assumed_state)
        self._attr_unique_id = dr.format_mac(mac_address)
        self._attr_extra_state_attributes = {}

    @property
    def device_info(self) -> DeviceInfo | None:
        """Return the device information."""
        # TODO: remove this?
        # if self._host is None:
        #     return None

        return DeviceInfo(
            identifiers={(DOMAIN, self.computer.mac)},
            name=self._attr_name,
            manufacturer="Generic",
            model="Computer",
            sw_version=self.computer.operating_system_version,
            connections={(dr.CONNECTION_NETWORK_MAC, self.computer.mac)},
        )

    @property
    def icon(self) -> str:
        return "mdi:monitor" if self._state else "mdi:monitor-off"

    @property
    def is_on(self) -> bool:
        return self._state

    async def turn_on(self, **kwargs: Any) -> None:
        await self.computer.start()

        if self._attr_assumed_state:
            self._state = True
            self.async_write_ha_state()

    async def turn_off(self, **kwargs: Any) -> None:
        await self.computer.shutdown()

        if self._attr_assumed_state:
            self._state = False
            self.async_write_ha_state()

    async def async_update(self) -> None:
        """Ping the computer to see if it is online and update the state."""
        self._state = await self.computer.is_on()

        await self.computer.update()

        # Update the state attributes and the connection only if the computer is on
        if self._state:
            self._attr_extra_state_attributes = {
                "operating_system": self.computer.operating_system,
                "operating_system_version": self.computer.operating_system_version,
                "mac_address": self.computer.mac,
                "ip_address": self.computer.host,
                "connected_devices": get_bluetooth_devices_as_str(self.computer),
            }

    # Services
    async def restart_to_windows_from_linux(self) -> None:
        """(Service Handler) Restart the computer to Windows from a running Linux by setting grub-reboot and restarting."""
        await self.computer.restart(OSType.LINUX, OSType.WINDOWS)

    async def restart_to_linux_from_windows(self) -> None:
        """(Service Handler) Restart the computer to Linux from a running Windows by setting grub-reboot and restarting."""
        await self.computer.restart()

    async def put_computer_to_sleep(self) -> None:
        """(Service Handler) Put the computer to sleep."""
        await self.computer.put_to_sleep()

    async def start_computer_to_windows(self) -> None:
        """(Service Handler) Start the computer to Windows"""
        async def wait_task():
            while not self.is_on:
                await asyncio.sleep(3)

            await self.computer.restart(OSType.LINUX, OSType.WINDOWS)

        """Start the computer to Linux, wait for it to boot, and then set grub-reboot and restart."""
        await self.computer.start()
        self._hass.loop.create_task(wait_task())

    async def restart_computer(self) -> None:
        """(Service Handler) Restart the computer."""
        await self.computer.restart()

    async def change_monitors_config(self, monitors_config: dict | None = None) -> None:
        """(Service Handler) Change the monitors configuration using a YAML config file."""
        await self.computer.set_monitors_config(monitors_config)

    async def steam_big_picture(self, action: str) -> None:
        """(Service Handler) Control Steam Big Picture mode."""
        await self.computer.steam_big_picture(action)

    async def change_audio_config(self, volume: int | None = None, mute: bool | None = None,
                                  input_device: str | None = None,
                            output_device: str | None = None) -> None:
        """(Service Handler) Change the audio configuration using a YAML config file."""
        await self.computer.set_audio_config(volume, mute, input_device, output_device)

    async def debug_info(self) -> ServiceResponse:
        """(Service Handler) Prints debug info."""
        return await format_debug_information(self.computer)