64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
import time
|
|
import paho.mqtt.client as mqtt
|
|
import serial
|
|
|
|
# Configuration
|
|
SERIAL_PORT = '/dev/ttyAMA0' # Adjust for your system (e.g., 'COM3' on Windows)
|
|
BAUD_RATE = 9600
|
|
MQTT_BROKER = "localhost" # Public test broker
|
|
MQTT_PORT = 1883
|
|
MQTT_TOPIC = "python/mqtt/test"
|
|
|
|
def main():
|
|
# 1. Initialize Serial Connection
|
|
try:
|
|
ser = serial.Serial(
|
|
port=SERIAL_PORT,
|
|
baudrate=BAUD_RATE,
|
|
parity=serial.PARITY_NONE,
|
|
stopbits=serial.STOPBITS_ONE,
|
|
bytesize=serial.EIGHTBITS,
|
|
timeout=1
|
|
)
|
|
except serial.SerialException as e:
|
|
print(f"Could not open serial port: {e}")
|
|
return
|
|
|
|
# 2. Initialize MQTT Client
|
|
client = mqtt.Client()
|
|
|
|
# Optional: Add callbacks for connection status
|
|
def on_connect(client, userdata, flags, reason_code, properties):
|
|
if reason_code == 0:
|
|
print("Connected to MQTT Broker")
|
|
else:
|
|
print(f"Failed to connect, return code {reason_code}")
|
|
|
|
client.on_connect = on_connect
|
|
client.connect(MQTT_BROKER, MQTT_PORT, 60)
|
|
|
|
# 3. Read Serial and Publish to MQTT
|
|
print("Reading from serial port and publishing to MQTT...")
|
|
try:
|
|
for i in range(10): # Read 10 messages as an example
|
|
if ser.in_waiting > 0:
|
|
# Read line from serial port
|
|
data = ser.readline().decode('utf-8').strip()
|
|
if data:
|
|
# Publish to MQTT topic
|
|
client.publish(MQTT_TOPIC, data)
|
|
print(f"Published: {data}")
|
|
else:
|
|
print("No data available on serial port")
|
|
time.sleep(1) # Wait 1 second between reads
|
|
except KeyboardInterrupt:
|
|
print("Interrupted by user")
|
|
finally:
|
|
# Cleanup
|
|
ser.close()
|
|
client.disconnect()
|
|
client.loop_stop()
|
|
print("Disconnected and closed serial port.")
|
|
|
|
if __name__ == "__main__":
|
|
main() |