58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
import os
|
|
import json
|
|
import logging
|
|
from typing import Optional
|
|
import paho.mqtt.client as mqtt
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# MQTT Configuration from environment
|
|
MQTT_BROKER_HOST = os.getenv("MQTT_BROKER_HOST", "localhost")
|
|
MQTT_BROKER_PORT = int(os.getenv("MQTT_BROKER_PORT", "1883"))
|
|
MQTT_USERNAME = os.getenv("MQTT_USERNAME", "")
|
|
MQTT_PASSWORD = os.getenv("MQTT_PASSWORD", "")
|
|
MQTT_LABEL_TOPIC = os.getenv("MQTT_LABEL_TOPIC", "vet/labels/print")
|
|
|
|
def publish_label_print(label_data: dict) -> bool:
|
|
"""
|
|
Publish a label print request to MQTT broker
|
|
|
|
Args:
|
|
label_data: Dictionary containing label print information
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
try:
|
|
# Create MQTT client
|
|
client = mqtt.Client()
|
|
|
|
# Set username and password if provided
|
|
if MQTT_USERNAME and MQTT_PASSWORD:
|
|
client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
|
|
|
|
# Connect to broker
|
|
client.connect(MQTT_BROKER_HOST, MQTT_BROKER_PORT, 60)
|
|
|
|
# Start network loop to process connection
|
|
client.loop_start()
|
|
|
|
# Publish message with QoS 0 (fire and forget)
|
|
message = json.dumps(label_data)
|
|
result = client.publish(MQTT_LABEL_TOPIC, message, qos=0)
|
|
|
|
# Stop loop and disconnect
|
|
client.loop_stop()
|
|
client.disconnect()
|
|
|
|
if result.rc == mqtt.MQTT_ERR_SUCCESS:
|
|
logger.info(f"Successfully published label print request to {MQTT_LABEL_TOPIC}")
|
|
return True
|
|
else:
|
|
logger.error(f"Failed to publish label print request: {result.rc}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error publishing MQTT message: {str(e)}")
|
|
return False
|