Control OH via Python Telegram Bot

der_optimist · October 19, 2018, 10:30am

Hi OH-community,
I would like to share my current status of my telegram bot. It communicates with OH via REST api - both directions. One “speciality” is, that I save most of the “set OH item to a value” funcions into a xml file. This makes it easy to change the behavier of the bot without chaning the python code.
But please be warned:

This is only my 2nd python project - so it might not be the best python code of this world
It’s not a “public release for everyone to use” - you will need to edit python code if you want to use it

I use 2 files, in addition to the python script: bot.ini saves the api keys, allowed user-IDs and coordinates of home and work (at the moment). The structure should be easy to use I think.
Second file is bot.xml - I use a small conversation for setting some values in OH, and all the information for these conversation is stored in bot.xml
It works like that:

You write to the bot: Light
Bot: OK, which room? (Showing you buttons for all available rooms)
You: (click on or write) Kitchen
Bot: Great. Which light there? (again keyboard with available lights)
You: Wall-Light
Bot: Fine. What value? (showing predefined values)
You: 50%
Bot: OK, should be done
(and it hopefully really is done)

Sounds very long now - but when I tested it, it takes below 5 seconds (because it’s only 1 word and 2 or 3 clicks). I prepared conversations with 2 and 3 steps, you can define the questions for each step in the xml - and of course all the predefined values and the name of the items (the user doesn’t have to know the name of the item! You only need it when you write the xml file)

The bot also tells me some temperatures when I ask him/her to do so, it gives me some infos when I write “Good Morning” and I can reset my Garbage-Reminder. And it tells me the driving time to work or home from work.

Why I did that? Because I was looking for a python project to learn some python. Will I really use that? No idea, I think sometimes, yes. But it will not replace all my switches in the house I think

You should create a telegram bot and add the token and your user ID to the bot.ini - if you don’t know your user ID, you can run the script via “python3 telegram-bot.py” and start a chat with your bot. You should see a logging info like “access denied to user 1234”. That’s your user ID.

I’m glad if someone likes to use that script, or tells me how I can do things better, or has some ideas for updates…

And finally, here are the files:
bot.xml (5.0 KB)

bot.ini:

[KEYS]
bot_api = 123:XYZ
maps_api = 123XYZ
[USERS]
me = 123
wife = 456
[PLACES]
home = 1.123,1.456
work1 = 2.789,2.123

and the python script (you will need some python modules like python-telegram-bot, python-openhab, configparser and requests):

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
"""
Jochens Telegram-OH-Bot
v01: Basis
v02: Bug fixes
v03: Bug fixes
v04: added try statement for openhab connection
v05: changed emojis (very important...)
v06: added google maps drive time and good morning
v07: Bug fixes
v08: Clean Strings from xml file (remove '%' and translate to ON/OFF...)
"""

from telegram import (ReplyKeyboardMarkup, ReplyKeyboardRemove, KeyboardButton)
from telegram.ext import (Updater, CommandHandler, MessageHandler, Filters, RegexHandler,
                          ConversationHandler, BaseFilter)
import configparser
import xml.etree.ElementTree as ET
from functools import wraps
import logging
from openhab import openHAB
import requests

# Input
url_oh = 'http://openhabianpi:8080/rest'

# Enable logging
logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s',
                    level=logging.INFO)

logger = logging.getLogger(__name__)

# Config
config = configparser.ConfigParser()
config.read('bot.ini')
allowed_users = [int(config['USERS']['me']),int(config['USERS']['wife'])]
google_maps_api_key = config['KEYS']['maps_api']
coord_home = config['PLACES']['home']
coord_work1 = config['PLACES']['work1']
print('allowed user ids: ' + str(allowed_users))

# Read xml data
xml_tree = ET.parse('bot.xml')
xml_root = xml_tree.getroot()
ThreeSteps_Keywords = []
for elem in xml_root.find('ThreeSteps'):
    ThreeSteps_Keywords.append(elem.tag)
TwoSteps_Keywords = []
for elem in xml_root.find('TwoSteps'):
    TwoSteps_Keywords.append(elem.tag)

def initialize_vars():
    global Keyword
    Keyword = ""
    global Values_1
    Values_1 = []
    global StepOne
    StepOne = ""
    global Values_2
    Values_2 = []
    global StepTwo
    StepTwo = ""
    global Values_3
    Values_3 = []
    global StepThree
    StepThree = ""

def restricted(func):
    @wraps(func)
    def wrapped(bot, update, *args, **kwargs):
        user_id = update.effective_user.id
        user_name = update.effective_user.full_name
        if user_id not in allowed_users:
            logger.info("access denied to user id %d, %s", user_id, user_name)
            return
        return func(bot, update, *args, **kwargs)
    return wrapped

def build_menu(values,n_cols):
    while len(values) < 5*n_cols:
        values.append(" ")
    button_list = [KeyboardButton(s) for s in values]
    menu = [button_list[i:i + n_cols] for i in range(0, len(button_list), n_cols)]
    return menu

def isfloat(value):
    try:
        float(value)
        return True
    except ValueError:
        return False

def cleanup(value):
    value = value.replace(" %","")
    value = value.replace("%","")
    value = value.replace(" °C","")
    value = value.replace("°C","")
    if (value == "An") or (value == "AN"):
        value = "ON"
    if (value == "Aus") or (value == "AUS"):
        value = "OFF"
    return value

def send_oh(item, value):
    value = cleanup(value)
    try: # an error should not lead to a stop of the script
#        print(item + ": " + str(value))
        if isfloat(value):
            Items.get(item).command(float(value))
        else:
            Items.get(item).command(value)
        return True
    except Exception as e:
        logger.info("Error sending to openHAB: %s", e)
        return False

def get_oh(item):
    try:
        return Items.get(item).state
    except Exception as e:
        logger.info("Error loading from openHAB: %s", e)
        return "Fehler, sorry..."

def maps_driving_time(orig,dest):
    try:
        url = "https://maps.googleapis.com/maps/api/distancematrix/json?origins={0}&destinations={1}&mode=driving&language=de-DE&key={2}".format(orig,dest,google_maps_api_key)
        response = requests.get(url)
        result = response.json()
        driving_time = result['rows'][0]['elements'][0]['duration_in_traffic']['value']
        return (driving_time // 60)
    except Exception as e:
        logger.info("Error loading Maps Driving Time: %s", e)
        return "(?)"

class Filter3Keywords(BaseFilter):
    def filter(self, message):
        return message.text in ThreeSteps_Keywords
filter_3_keywords = Filter3Keywords()

class Filter31(BaseFilter):
    def filter(self, message):
        return message.text in Values_1
filter_3_1 = Filter31()

class Filter32(BaseFilter):
    def filter(self, message):
        return message.text in Values_2
filter_3_2 = Filter32()

class Filter33(BaseFilter):
    def filter(self, message):
        return message.text in Values_3
filter_3_3 = Filter33()

class Filter2Keywords(BaseFilter):
    def filter(self, message):
        return message.text in TwoSteps_Keywords
filter_2_keywords = Filter2Keywords()

class Filter21(BaseFilter):
    def filter(self, message):
        return message.text in Values_1
filter_2_1 = Filter21()

class Filter22(BaseFilter):
    def filter(self, message):
        return message.text in Values_2
filter_2_2 = Filter22()

@restricted
def step_one_of_three(bot, update):
    global Keyword
    Keyword = update.message.text
    logger.info("User %s, ID %s: Keyword gesendet: %s", update.effective_user.full_name, update.effective_user.id, Keyword)
    global Values_1
    for elem in xml_root.find('ThreeSteps').find(Keyword):
        Values_1.append(elem.tag)
    reply_markup = ReplyKeyboardMarkup(build_menu(Values_1,3))
    update.message.reply_text(xml_root.find('ThreeSteps').find(Keyword).attrib['q1'],reply_markup=reply_markup)
    return 1

@restricted
def step_two_of_three(bot, update):
    global StepOne
    StepOne = update.message.text
    logger.info("User %s, ID %s: StepOne gewählt: %s", update.effective_user.full_name, update.effective_user.id, StepOne)
    global Values_2
    for elem in xml_root.find('ThreeSteps').find(Keyword).find(StepOne):
        Values_2.append(elem.tag)
    reply_markup = ReplyKeyboardMarkup(build_menu(Values_2,2))
    update.message.reply_text(xml_root.find('ThreeSteps').find(Keyword).attrib['q2'],reply_markup=reply_markup)
    return 2


@restricted
def step_three_of_three(bot, update):
    global StepTwo
    StepTwo = update.message.text
    logger.info("User %s, ID %s: StepTwo gewählt: %s", update.effective_user.full_name, update.effective_user.id, StepTwo)
    global Values_3
    for elem in xml_root.find('ThreeSteps').find(Keyword).find(StepOne).find(StepTwo):
        Values_3.append(elem.text)
    reply_markup = ReplyKeyboardMarkup(build_menu(Values_3,2))
    update.message.reply_text(xml_root.find('ThreeSteps').find(Keyword).attrib['q3'],reply_markup=reply_markup)
    return 3

@restricted
def action_of_three_steps(bot, update):
    global StepThree
    StepThree = update.message.text
    logger.info("User %s, ID %s: StepThree gewählt: %s", update.effective_user.full_name, update.effective_user.id, StepThree)
    item_name = xml_root.find('ThreeSteps').find(Keyword).find(StepOne).find(StepTwo).attrib['name']
    if send_oh(item_name,StepThree):
        reply = 'OK 👍🏻 sollte erledigt sein.'
        logger.info("User %s, ID %s: Aktion: Setze Item %s auf %s", update.effective_user.full_name, update.effective_user.id, item_name, StepThree) 
    else:
        reply = 'Oh 😳 da ist leider was schief gegangen, sorry!'
        logger.info("User %s, ID %s: Aktion: Setze Item %s auf %s - FEHLER!", update.effective_user.full_name, update.effective_user.id, item_name, StepThree) 
    update.message.reply_text(reply,reply_markup=ReplyKeyboardRemove())
    initialize_vars()
    return ConversationHandler.END

@restricted
def step_one_of_two(bot, update):
    global Keyword
    Keyword = update.message.text
    logger.info("User %s, ID %s: Keyword gesendet: %s", update.effective_user.full_name, update.effective_user.id, Keyword)
    global Values_1
    for elem in xml_root.find('TwoSteps').find(Keyword):
        Values_1.append(elem.tag)
    reply_markup = ReplyKeyboardMarkup(build_menu(Values_1,3))
    update.message.reply_text(xml_root.find('TwoSteps').find(Keyword).attrib['q1'],reply_markup=reply_markup)
    return 1

@restricted
def step_two_of_two(bot, update):
    global StepOne
    StepOne = update.message.text
    logger.info("User %s, ID %s: StepOne gewählt: %s", update.effective_user.full_name, update.effective_user.id, StepOne)
    global Values_2
    for elem in xml_root.find('TwoSteps').find(Keyword).find(StepOne):
        Values_2.append(elem.text)
    reply_markup = ReplyKeyboardMarkup(build_menu(Values_2,2))
    update.message.reply_text(xml_root.find('TwoSteps').find(Keyword).attrib['q2'],reply_markup=reply_markup)
    return 2

@restricted
def action_of_two_steps(bot, update):
    global StepTwo
    StepTwo = update.message.text
    logger.info("User %s, ID %s: StepTwo gewählt: %s", update.effective_user.full_name, update.effective_user.id, StepTwo)
    item_name = xml_root.find('TwoSteps').find(Keyword).find(StepOne).attrib['name']
    if send_oh(item_name,StepTwo):
        reply = 'OK 👍🏻 sollte erledigt sein.'
        logger.info("User %s, ID %s: Aktion: Setze Item %s auf %s", update.effective_user.full_name, update.effective_user.id, item_name, StepTwo)
    else:
        reply = 'Oh 😳 da ist leider was schief gegangen, sorry!'
        logger.info("User %s, ID %s: Aktion: Setze Item %s auf %s - FEHLER!", update.effective_user.full_name, update.effective_user.id, item_name, StepTwo)
    update.message.reply_text(reply,reply_markup=ReplyKeyboardRemove())
    initialize_vars()
    return ConversationHandler.END

@restricted
def cancel(bot, update):
    logger.info("User %s, ID %s: Abbruch", update.effective_user.full_name, update.effective_user.id) 
    update.message.reply_text('abgebrochen',reply_markup=ReplyKeyboardRemove())
    initialize_vars()
    return ConversationHandler.END

@restricted
def help_me(bot, update):
    all_keywords = ThreeSteps_Keywords + TwoSteps_Keywords + ['Temperaturen (anzeigen)','Müll (erledigt)']
    reply = "Hallo, ich bin's, euer Haus 😊\nSchick mir einfach eines der folgenden Worte, ich sag dir dann schon wie's weiter geht...\n"
    for s in all_keywords:
        reply = reply + "\n" + s
    reply_markup = ReplyKeyboardMarkup(build_menu(all_keywords,2))
    update.message.reply_text(reply,reply_markup=reply_markup)

@restricted
def show_temps(bot, update):
    reply = "=== 🔥 Temperaturen ❄️ ==="
    reply += "\n" + "Draußen: " + get_oh('TempAktuellFIO')
    reply += "\n" + "Wohnzimmer: " + get_oh('T_WZ_ist')
    update.message.reply_text(reply,reply_markup=ReplyKeyboardRemove())
    logger.info("User %s, ID %s: Temperaturen angefragt", update.effective_user.full_name, update.effective_user.id)

@restricted
def set_garbage(bot, update):
    if send_oh('Abfall_Steht_An','OFF'):
        reply = "OK, Müll ist also erledigt 🚚 <= 🗑️\nDanke ❤️"
        logger.info("User %s, ID %s: Aktion: Setze Item Abfall_Steht_An auf OFF", update.effective_user.full_name, update.effective_user.id)
    else:
        reply = "OK, Müll ist also erledigt 🚚 <= 🗑️\nDanke ❤️\nLeider konnte ich es aber dem System nicht sagen, da ein Fehler aufgetreten ist. Tut mir leid!"
        logger.info("User %s, ID %s: Aktion: Setze Item Abfall_Steht_An auf OFF - FEHLER", update.effective_user.full_name, update.effective_user.id) 
    update.message.reply_text(reply,reply_markup=ReplyKeyboardRemove())

@restricted
def good_morning(bot, update):
    reply = "Guten Morgen!"
    reply += "\n" + "Draußen hat es " + get_oh('TempAktuellFIO') + "°C, im Wohnzimmer " + get_oh('T_WZ_ist') + "."
    reply += "\nZur Arbeit brauchst du aktuell " + str(maps_driving_time(coord_home,coord_work1)) + " Minuten."
    update.message.reply_text(reply,reply_markup=ReplyKeyboardRemove())
    logger.info("User %s, ID %s: Guten Morgen gesagt", update.effective_user.full_name, update.effective_user.id)

@restricted
def time_to_work(bot, update):
    reply = "\nZur Arbeit brauchst du aktuell " + str(maps_driving_time(coord_home,coord_work1)) + " Minuten."
    update.message.reply_text(reply,reply_markup=ReplyKeyboardRemove())
    logger.info("User %s, ID %s: Zeit zur Arbeit angefragt", update.effective_user.full_name, update.effective_user.id)

@restricted
def time_home(bot, update):
    reply = "\nVon der Arbeit nach Hause brauchst du aktuell " + str(maps_driving_time(coord_work1,coord_home)) + " Minuten."
    update.message.reply_text(reply,reply_markup=ReplyKeyboardRemove())
    logger.info("User %s, ID %s: Zeit Arbeit nach Hause angefragt", update.effective_user.full_name, update.effective_user.id)

def error(bot, update, error):
    """Log Errors caused by Updates."""
    logger.warning('Update "%s" caused error "%s"', update, error)


def main():
    initialize_vars()
    try: # for my test system that does not have a connection to openhab
        openhab = openHAB(url_oh)
        Items = openhab.fetch_all_items()
    except Exception as e:
        logger.info("Error connecting to openHAB: %s", e)
        pass

    # Create the EventHandler and pass it your bot's token.
    updater = Updater(token=config['KEYS']['bot_api'])

    # Get the dispatcher to register handlers
    dp = updater.dispatcher

    # Add conversation handler
    conv_handler_3 = ConversationHandler(
        entry_points=[MessageHandler(filter_3_keywords, step_one_of_three)],

        states={
            1: [MessageHandler(filter_3_1, step_two_of_three)],

            2: [MessageHandler(filter_3_2, step_three_of_three)],

            3: [MessageHandler(filter_3_3, action_of_three_steps)],

        },

        #fallbacks=[RegexHandler('^(Abbruch|abbruch|cancel|stop|Stop|Stopp|stopp)$', cancel)]
        fallbacks=[MessageHandler(Filters.all, cancel)]
    )
    
    conv_handler_2 = ConversationHandler(
        entry_points=[MessageHandler(filter_2_keywords, step_one_of_two)],

        states={
            1: [MessageHandler(filter_2_1, step_two_of_two)],

            2: [MessageHandler(filter_2_2, action_of_two_steps)],

        },

        #fallbacks=[RegexHandler('^(Abbruch|abbruch|cancel|stop|Stop|Stopp|stopp)$', cancel)]
        fallbacks=[MessageHandler(Filters.all, cancel)]
    )

    help_handler = RegexHandler('^(Hilfe|hilfe|Start|start|Hallo|hallo)$',help_me)
    start_handler = CommandHandler('start',help_me)
    temp_handler = RegexHandler('^(Temp|temp)',show_temps)
    garbage_handler = RegexHandler('^(Müll)',set_garbage)
    good_morning_handler = RegexHandler('^((Guten|guten).(Morgen|morgen))',good_morning)
    time_to_work_handler = RegexHandler('^(Arbeit|arbeit)',time_to_work)
    time_home_handler = RegexHandler('^(Feierabend|feierabend|nach Hause|Nach Hause|heim|Heim)',time_home)

    dp.add_handler(conv_handler_3)
    dp.add_handler(conv_handler_2)
    dp.add_handler(help_handler)
    dp.add_handler(start_handler)
    dp.add_handler(temp_handler)
    dp.add_handler(garbage_handler)
    dp.add_handler(good_morning_handler)
    dp.add_handler(time_to_work_handler)
    dp.add_handler(time_home_handler)

    # log all errors
    dp.add_error_handler(error)

    # Start the Bot
    updater.start_polling()

    # Run the bot until you press Ctrl-C or the process receives SIGINT,
    # SIGTERM or SIGABRT. This should be used most of the time, since
    # start_polling() is non-blocking and will stop the bot gracefully.
    updater.idle()


if __name__ == '__main__':
    main()

Have fun!
Jochen

rlkoshak · October 19, 2018, 3:50pm

Thanks for posting! Projects like this are great.

I would suggest creating a Github/Gitlab/Bitbucket or your repository of choice to host this code. That will allow those users who find your script useful the opportunity to post any changes or improvements back to the script.

I’ve been very pleasantly surprised by the number of improvements others have submitted to my own second Python project ever.

Can it support (eventually) putting it all on one line? Something like “Turn on the light over the kitchen sink”? That will involve some more natural language processing but there are lots of examples of how to manage that in the voice control threads on this forum and elsewhere.

I can see this being useful for two way activities. For example, an event occurs in OH and OH sends a Telegram message to you. You can respond with one or more predefined actions to take.

illnesse · October 19, 2018, 6:03pm

You might wanna check this out: https://github.com/ghys/habot/blob/master/README.md

I think it’s being merged by @Kai and @ysc currently. Can’t wait to play with it!

Belgadon · October 20, 2018, 3:41pm

As mentioned in another thread, I started to edit the telegram action.

I want the bot to start the conversation.
e.g.: A rules triggers when I have to get the trash out. The bot then writes me a message like:

You have to bring the Trash out.
[Done] [Remember me again in 10]

It was pretty easy to give the user the possibility to enter a json parameter to configure the buttons. This already works fine with the current action.
But I don’t know how get the info back to openhab, that someone pressed a button.
Constantly pulling for “getUpdates” doesn’t seem a very good idea. This is where I kind of stuck, but haven’t invested much time in it anyway.

The "problem" that I see with the python scripts, is that they are relative hart to use and maintain especially for users with little to no experience in programming/python/linux/scripting etc.
In my opinion a binding/IO/actions … would always be better.

Nevertheless, I will definitely have a look at what you did. Thank you! Really!

illnesse · October 24, 2018, 1:53pm