130 řádky
4.2 KiB
Python
130 řádky
4.2 KiB
Python
import paho.mqtt.client as mqtt
|
|
import paho.mqtt.subscribe as subscribe
|
|
import json
|
|
|
|
def create_client(broker: str, port: int, username: str, password: str) -> mqtt.Client:
|
|
"""Create an MQTT client and connect it to the broker
|
|
|
|
Args:
|
|
broker (str): MQTT broker address
|
|
port (int): MQTT broker port
|
|
username (str): Username for MQTT broker
|
|
password (str): Password for MQTT broker
|
|
|
|
Returns:
|
|
mqtt.Client: Connected MQTT client instance
|
|
"""
|
|
# 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.Client) -> None:
|
|
"""Create Home Assistant discovery topics
|
|
|
|
Args:
|
|
client (mqtt.Client): 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 check_current_disc(client: mqtt.Client) -> str:
|
|
"""Check the current loaded disc by subscribing to the retained message on the state topic.
|
|
|
|
Args:
|
|
client (mqtt.Client): MQTT Client
|
|
|
|
Returns:
|
|
str: Current object at current_disc/state
|
|
"""
|
|
def on_message(client, userdata, message):
|
|
"""Callback function to handle received MQTT messages."""
|
|
userdata['message'] = message.payload.decode()
|
|
|
|
# Create a dictionary to store the message
|
|
userdata = {'message': None}
|
|
|
|
# Set the user data and the on_message callback
|
|
client.user_data_set(userdata)
|
|
client.on_message = on_message
|
|
|
|
# Subscribe to the topic
|
|
client.subscribe("homeassistant/sensor/floppy_player/current_disc/state")
|
|
|
|
# Run the loop manually until we receive a message
|
|
timeout = 5 # Timeout after 5 seconds
|
|
while userdata['message'] is None and timeout > 0:
|
|
client.loop(timeout=0.1) # Process network events for a short time
|
|
timeout -= 0.1
|
|
|
|
# Check if we received the message
|
|
if userdata['message'] is None:
|
|
raise TimeoutError("No retained message was found.")
|
|
|
|
return userdata['message']
|
|
|
|
|
|
def update_disc(client: mqtt.Client, disc_message: dict) -> None:
|
|
"""Update current disc.
|
|
|
|
Args:
|
|
client (mqtt.Client): MQTT Client
|
|
disc_message (dict): Current disc information
|
|
"""
|
|
client.publish("homeassistant/sensor/floppy_player/current_disc/state", json.dumps(disc_message), retain=True)
|
|
print(f"Published current disc: {disc_message}")
|
|
|
|
def control_player(client: mqtt.Client, state: str) -> None:
|
|
"""Control the player
|
|
|
|
Args:
|
|
client (mqtt.Client): MQTT Client
|
|
state (str): Player State
|
|
"""
|
|
client.publish("homeassistant/sensor/floppy_player/status/state", state, retain=True)
|
|
print(f"Published status: {state}")
|