146 lines
4.6 KiB
Python
146 lines
4.6 KiB
Python
import paho.mqtt.client as mqtt
|
|
import json
|
|
import time
|
|
|
|
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_
|
|
"""
|
|
# 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": node_id,
|
|
"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": node_id,
|
|
"model": "v1",
|
|
"manufacturer": "Karl"
|
|
}
|
|
}
|
|
|
|
# # Create a function to handle connection
|
|
# def on_connect(client, userdata, flags, rc):
|
|
# if rc == 0:
|
|
# print("Connected to broker")
|
|
# # Publish device configuration for both sensors
|
|
# client.publish(discovery_topic_disc, json.dumps(disc_config), retain=True)
|
|
# client.publish(discovery_topic_status, json.dumps(status_config), retain=True)
|
|
# else:
|
|
# print(f"Connection failed with code {rc}")
|
|
|
|
# Create a new MQTT client instance
|
|
client = mqtt.Client()
|
|
|
|
# Set username and password
|
|
client.username_pw_set(username, password)
|
|
|
|
# # Assign the on_connect function
|
|
# client.on_connect = on_connect
|
|
|
|
# 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}")
|
|
|