From b804a5d7e7dab93bf6056e8969a5dd4d4c52f452 Mon Sep 17 00:00:00 2001 From: Evgeny Zinoviev Date: Sat, 8 May 2021 01:53:47 +0300 Subject: migrate from isv to inverterd --- main.py | 187 ---------------------------------------------------------------- 1 file changed, 187 deletions(-) delete mode 100644 main.py (limited to 'main.py') diff --git a/main.py b/main.py deleted file mode 100644 index 1441728..0000000 --- a/main.py +++ /dev/null @@ -1,187 +0,0 @@ -import logging -import re -import datetime -import isv -import configstore - -from html import escape -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 -) - - -# -# 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 = isv.general_status() - - # render response - power_direction = gs['battery_power_direction'].lower() - power_direction = re.sub(r'ge$', '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__': - config = configstore.get_config() - - logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', - level=logging.INFO) - - updater = Updater(configstore.get_token(), request_kwargs={'read_timeout': 6, 'connect_timeout': 7}) - dispatcher = updater.dispatcher - - user_filter = Filters.user(configstore.get_admins()) - - 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() -- cgit v1.2.3