HASS Integration - Githubissues

I’ve been exploring the creation of a Home Assistant integration, and so far, my approach involved hosting a solution on the HASS instance, using the API to display dashboards. However, this method introduces unnecessary overhead since it requires additional resources for relatively minor functionality. Additionally, it's limited to dashboards and user interactions via clicks, which prevents using voice assistants to start games.

Recently, I’ve come up with a new idea: building a custom Home Assistant integration that generates virtual switches for each game in the library, essentially replicating part of the Playnite Web frontend/API functionality. The concept is to subscribe to the Playnite Web MQTT topic and create switches based on game installation, uninstallation, and synchronization events.

My primary concern is ensuring this remains aligned with your work, @andrew-codes. Would you be open to collaborating or coming to an agreement on how the messages should be structured for consistency?

A cleaner solution that could avoid the need for a custom integration altogether would be to integrate MQTT device discovery directly into the Playnite Web plugin. However, I realize this might be beyond the current focus and scope. Here’s a link to Home Assistant’s MQTT device discovery documentation.

I’d love to hear your thoughts. I already have a proof of concept in place, although it’s not yet fully functional. Here's a glimpse of it:

switch.py

import logging
import json
from homeassistant.components.switch import SwitchEntity
from homeassistant.components.mqtt import async_subscribe

LOGGER = logging.getLogger(__name__)

async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
    if discovery_info is None:
        return

    switch = PlayniteGameSwitch(discovery_info, hass)
    async_add_entities([switch])

class PlayniteGameSwitch(SwitchEntity):
    def __init__(self, game_data, hass):
        self._name = game_data.get('name')
        self._state = False
        self._game_id = game_data.get('id')
        self._game_data = game_data
        self.hass = hass

        # Subscribe to MQTT events for game state changes
        self.topic = f"playnite/response/game/state"
        async_subscribe(self.hass, self.topic, self.mqtt_message_received)

    @property
    def name(self):
        return self._name

    @property
    def is_on(self):
        return self._state  # Reflect the running state of the game

    async def turn_on(self, **kwargs):
        """Start the game via MQTT."""
        self._state = True
        self.schedule_update_ha_state()

        # Publish MQTT message to start the game
        payload = {
            "game": {
                "id": self._game_id,
                "gameId": self._game_data.get('gameId'),
                "name": self._game_data.get('name'),
                "platform": {
                    "id": self._game_data['platform']['id'],
                    "name": self._game_data['platform']['name']
                },
                "source": self._game_data.get('source'),
                "install": not self._game_data.get('isInstalled', False)
            }
        }

        topic = f"playnite/request/game/start"
        await self.hass.components.mqtt.async_publish(topic, json.dumps(payload))

    async def turn_off(self, **kwargs):
        """Stop the game via MQTT."""
        self._state = False
        self.schedule_update_ha_state()

        # Publish MQTT message to stop the game
        payload = {
            "game": {
                "id": self._game_id,
                "gameId": self._game_data.get('gameId'),
                "name": self._game_data.get('name'),
                "platform": {
                    "id": self._game_data['platform']['id'],
                    "name": self._game_data['platform']['name']
                },
                "source": self._game_data.get('source')
            }
        }

        topic = f"playnite/request/game/stop"
        await self.hass.components.mqtt.async_publish(topic, json.dumps(payload))

    async def mqtt_message_received(self, msg):
        """Handle messages for game run state."""
        payload = json.loads(msg.payload)

        if payload.get("id") == self._game_id:
            run_state = payload.get("runState")
            if run_state == "Running":
                self._state = True
            else:
                self._state = False

            self.schedule_update_ha_state()  # Update switch state in Home Assistant

mqtt_handler.py


import json
import logging
from homeassistant.components import mqtt

_LOGGER = logging.getLogger(__name__)

async def setup_mqtt_subscription(hass, topic_base, handle_message):
    """Set up an MQTT subscription to listen for game discovery messages."""
    async def message_received(msg):
        try:
            # Parse the MQTT message payload
            data = json.loads(msg.payload)
            # Call the message handler with the parsed data
            await handle_message(hass, data)
        except Exception as e:
            _LOGGER.error(f"Failed to process MQTT message: {e}")

    # Subscribe to the base topic
    await mqtt.async_subscribe(hass, f"{topic_base}/#", message_received)
    _LOGGER.info(f"Subscribed to MQTT topic: {topic_base}/#")

init.py

import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.discovery import async_load_platform  # Import directly
from .mqtt_handler import setup_mqtt_subscription

_LOGGER = logging.getLogger(__name__)

DOMAIN = "playnite_web_mqtt"

async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
    """Set up the Playnite Web MQTT from a config entry."""
    topic_base = entry.data.get("topic_base")

    if not topic_base:
        _LOGGER.error("No topic base provided in the config entry")
        return False

    # Set up the MQTT subscription
    await setup_mqtt_subscription(hass, topic_base, handle_game_discovery)

    return True

async def handle_game_discovery(hass: HomeAssistant, message: dict):
    """Handle the discovery of a game and create a switch for each game."""
    game_id = message.get('id')
    game_name = message.get('name')

    if not game_id or not game_name:
        _LOGGER.error("Invalid game data received, skipping.")
        return

    switch_data = {
        'id': game_id,
        'name': game_name
    }

    # Dynamically load the switch platform for the discovered game
    hass.async_create_task(
        async_load_platform(hass, 'switch', DOMAIN, switch_data, {})  # Use direct import of async_load_platform
    )

andrew-codes / playnite-web

HASS Integration #566