floppy-ytube-player/lib/mqtt.py

import paho.mqtt.client as mqtt
import json

def create_client(broker:str, port:int, username:str, password:str)-> mqtt:
    """_summary_

    Args:
        broker (str): _description_
        port (int): _description_
        username (str): _description_
        password (str): _description_

    Returns:
        mqtt: _description_
    """    
    # Create a new MQTT client instance
    client = mqtt.Client()

    # Set username and password
    client.username_pw_set(username, password)

    # Connect to the MQTT broker
    client.connect(broker, port, 60)
    
    return client

def create_config(client:mqtt)->None:
    """Create Home Assistant discovery topics

    Args:
        client (mqtt): MQTT Client
    """    
    # Device-specific information for multiple sensors
    node_id = "floppy_player"  # Unique device ID

    # Define discovery and state topics for each sensor
    discovery_topic_disc = f"homeassistant/sensor/floppy_player/current_disc/config"
    state_topic_disc = f"homeassistant/sensor/floppy_player/current_disc/state"

    discovery_topic_status = f"homeassistant/sensor/floppy_player/status/config"
    state_topic_status = f"homeassistant/sensor/floppy_player/status/state"

    # Sensor 1: current_disc (a text-based sensor)
    disc_config = {
        "name": "Current Disc",
        "state_topic": state_topic_disc,
        "value_template": "{{ value }}",  # Textual value
        "unique_id": f"{node_id}_current_disc",
        "device": {
            "identifiers": [node_id],
            "name": "Floppy Player",
            "model": "v1",
            "manufacturer": "Karl"
        }
    }

    # Sensor 2: status (another text-based sensor)
    status_config = {
        "name": "Device Status",
        "state_topic": state_topic_status,
        "value_template": "{{ value }}",  # Textual value
        "unique_id": f"{node_id}_status",
        "device": {
            "identifiers": [node_id],
            "name": "Floppy Player",
            "model": "v1",
            "manufacturer": "Karl"
        }
    }
    client.publish(discovery_topic_disc, json.dumps(disc_config), retain=True)
    client.publish(discovery_topic_status, json.dumps(status_config), retain=True)

def update_disc(client:mqtt, disc_message:dict)->None:
    """Update current disc.

    Args:
        client (mqtt): MQTT Client
        disc_message (dict): Current disc information
    """
    client.publish("homeassistant/sensor/floppy_player/current_disc/state", json.dumps(disc_message))
    print(f"Published current disc: {disc_message}")


def control_player(client:mqtt, state:str)->None:
    """Control the player

    Args:
        client (mqtt): MQTT Client
        state (str): Player State
    """    
    client.publish("homeassistant/sensor/floppy_player/status/state", state)
    print(f"Published status: {state}")