#!/usr/bin/env python3 import logging, re, datetime, json, inverterd from typing import Optional from argparse import ArgumentParser from html import escape from pprint import pprint from time import sleep from strings import lang as _ from telegram import ( Update, ParseMode, KeyboardButton, ReplyKeyboardMarkup ) from telegram.ext import ( Updater, Filters, CommandHandler, MessageHandler, CallbackContext ) inverter: Optional[inverterd.Client] = None # # helpers # def get_markup() -> ReplyKeyboardMarkup: button = [ [ _('status'), _('generation') ], [ _('gs'), _('ri'), _('errors') ] ] return ReplyKeyboardMarkup(button, one_time_keyboard=False) def reply(update: Update, text: str) -> None: update.message.reply_text(text, reply_markup=get_markup(), parse_mode=ParseMode.HTML) # # command/message handlers # def start(update: Update, context: CallbackContext) -> None: reply(update, 'Select a command on the keyboard.') def msg_status(update: Update, context: CallbackContext) -> None: try: gs = json.loads(inverter.exec('get-status'))['data'] pprint(gs) # render response power_direction = gs['battery_power_direction'].lower() power_direction = re.sub(r'ges$', 'ging', power_direction) charging_rate = '' if power_direction == 'charging': charging_rate = ' @ %s %s' % tuple(gs['battery_charging_current']) elif power_direction == 'discharging': charging_rate = ' @ %s %s' % tuple(gs['battery_discharge_current']) html = 'Battery: %s %s' % tuple(gs['battery_voltage']) html += ' (%s%s, ' % tuple(gs['battery_capacity']) html += '%s%s)' % (power_direction, charging_rate) html += '\nLoad: %s %s' % tuple(gs['ac_output_active_power']) html += ' (%s%%)' % (gs['output_load_percent'][0]) if gs['pv1_input_power'][0] > 0: html += '\nInput power: %s%s' % tuple(gs['pv1_input_power']) if gs['grid_voltage'][0] > 0 or gs['grid_freq'][0] > 0: html += '\nGenerator: %s %s' % tuple(gs['grid_voltage']) html += ', %s %s' % tuple(gs['grid_freq']) # send response reply(update, html) except Exception as e: logging.exception(str(e)) reply(update, 'exception: ' + str(e)) def msg_generation(update: Update, context: CallbackContext) -> None: try: today = datetime.date.today() yday = today - datetime.timedelta(days=1) yday2 = today - datetime.timedelta(days=2) gs = isv.general_status() sleep(0.1) gen_today = isv.day_generated(today.year, today.month, today.day) gen_yday = None gen_yday2 = None if yday.month == today.month: sleep(0.1) gen_yday = isv.day_generated(yday.year, yday.month, yday.day) if yday2.month == today.month: sleep(0.1) gen_yday2 = isv.day_generated(yday2.year, yday2.month, yday2.day) # render response html = 'Input power: %s %s' % tuple(gs['pv1_input_power']) html += ' (%s %s)' % tuple(gs['pv1_input_voltage']) html += '\nToday: %s Wh' % (gen_today['wh']) if gen_yday is not None: html += '\nYesterday: %s Wh' % (gen_yday['wh']) if gen_yday2 is not None: html += '\nThe day before yesterday: %s Wh' % (gen_yday2['wh']) # send response reply(update, html) except Exception as e: logging.exception(str(e)) reply(update, 'exception: ' + str(e)) def msg_gs(update: Update, context: CallbackContext) -> None: try: status = isv.general_status(as_table=True) reply(update, status) except Exception as e: logging.exception(str(e)) reply(update, 'exception: ' + str(e)) def msg_ri(update: Update, context: CallbackContext) -> None: try: rated = isv.rated_information(as_table=True) reply(update, rated) except Exception as e: logging.exception(str(e)) reply(update, 'exception: ' + str(e)) def msg_errors(update: Update, context: CallbackContext) -> None: try: faults = isv.faults(as_table=True) reply(update, faults) except Exception as e: logging.exception(str(e)) reply(update, 'exception: ' + str(e)) def msg_all(update: Update, context: CallbackContext) -> None: reply(update, "Command not recognized. Please try again.") if __name__ == '__main__': # command-line arguments parser = ArgumentParser() parser.add_argument('--token', required=True, type=str, help='Telegram bot token') parser.add_argument('--users-whitelist', nargs='+', help='ID of users allowed to use the bot') parser.add_argument('--inverterd-host', default='127.0.0.1', type=str) parser.add_argument('--inverterd-port', default=8305, type=int) args = parser.parse_args() whitelist = list(map(lambda x: int(x), args.users_whitelist)) # connect to inverterd inverter = inverterd.Client(host=args.inverterd_host, port=args.inverterd_port) # configure logging logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO) # configure bot updater = Updater(args.token, request_kwargs={'read_timeout': 6, 'connect_timeout': 7}) dispatcher = updater.dispatcher user_filter = Filters.user(whitelist) dispatcher.add_handler(CommandHandler('start', start)) dispatcher.add_handler(MessageHandler(Filters.text(_('status')) & user_filter, msg_status)) dispatcher.add_handler(MessageHandler(Filters.text(_('generation')) & user_filter, msg_generation)) dispatcher.add_handler(MessageHandler(Filters.text(_('gs')) & user_filter, msg_gs)) dispatcher.add_handler(MessageHandler(Filters.text(_('ri')) & user_filter, msg_ri)) dispatcher.add_handler(MessageHandler(Filters.text(_('errors')) & user_filter, msg_errors)) dispatcher.add_handler(MessageHandler(Filters.all & user_filter, msg_all)) # start the bot updater.start_polling() # run the bot until the user presses Ctrl-C or the process receives SIGINT, SIGTERM or SIGABRT updater.idle()