Grab state from a Vector robot and publish it to a MQTT topic

vector-mqtt.py

#!/usr/bin/env python3

# Copyright (c) 2021 acidzebra
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# for use with Anki's awesome Vector robot: https://www.anki.com/en-us/vector
# 

import anki_vector
import json
import paho.mqtt.client as mqtt

# define variables for MQTT

MQTT_HOST = "YOUR BROKER IP ADDRESS HERE"
MQTT_TOPIC = "YOUR DESIRED TOPIC NAME HERE"
MQTT_PORT = 1883
MQTT_KEEPALIVE_INTERVAL = 20

# define the robot object and variables we'll be reading from it
myrobot = anki_vector.Robot("YOUR SERIAL NUMBER HERE",behavior_control_level=None)
myrobotvoltage = 0
myrobotbatlevel = 0
myrobotcharging = 0
myrobotdocked = 0

# try connecting to the robot and pull the values

try:
    myrobot.connect(timeout=12)
    myrobot_battery_state = myrobot.get_battery_state()
    if myrobot_battery_state:
        myrobotvoltage  = myrobot_battery_state.battery_volts
        myrobotbatlevel = myrobot_battery_state.battery_level
        myrobotcharging = myrobot_battery_state.is_charging
        myrobotdocked   = myrobot_battery_state.is_on_charger_platform
    myrobot.disconnect()
except:
    print("unable to get data from robot")

# format the response as json
data = {}
data['robots'] = []
data['robots'].append({
    'name': 'myrobot',
    'voltage': myrobotvoltage,
    'batlevel': myrobotbatlevel,
    'charging': myrobotcharging,
    'docked': myrobotdocked
})
# ...then convert it to text? Not sure why I did this but it works. Yay, 1am programming.
MQTT_MSG = str(data)
# anyway, in the openHAB channel, use a jsonpath transform to get specific values like this: JSONPATH:$..voltage

# Define on_publish event function
def on_publish(client, userdata, mid):
    print("Message published to broker")
    
# Initiate MQTT Client
mqttc = mqtt.Client()

# Register publish callback function
mqttc.on_publish = on_publish

# Connect with MQTT Broker
mqttc.connect(MQTT_HOST, MQTT_PORT, MQTT_KEEPALIVE_INTERVAL)

# Publish message to MQTT Broker
mqttc.publish(MQTT_TOPIC,MQTT_MSG)

# Disconnect from MQTT_Broker
mqttc.disconnect()