diff options
| -rw-r--r-- | .gitignore | 2 | ||||
| -rw-r--r-- | README.md | 30 | ||||
| -rw-r--r-- | configstore.py | 51 | ||||
| -rw-r--r-- | isv.py | 42 | ||||
| -rw-r--r-- | main.py | 186 | ||||
| -rw-r--r-- | requirements.txt | 1 | ||||
| -rw-r--r-- | strings.py | 12 |
7 files changed, 324 insertions, 0 deletions
diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e04276f --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +.idea +venv diff --git a/README.md b/README.md new file mode 100644 index 0000000..752e17e --- /dev/null +++ b/README.md @@ -0,0 +1,30 @@ +# inverter-bot + +This is a Telegram bot for querying information from an InfiniSolar V family of hybrid solar inverters, in particular +inverters supported by **isv** utility, which is an older version of **infinisolarctl** from **infinisolar-tools** +package. + +It supports querying general status, such as battery voltage or power usage, printing amounts of energy generated in +the last days, dumping status or rated information and more. + +It requires Python 3.6+ or so. + +## Configuration + +Configuration is stored in `config.ini` file in `~/.config/inverter-bot`. + +Config example: +``` +token=YOUR_TOKEN +admins= + 123456 ; admin id + 000123 ; another admin id +isv_bin=/path/to/isv +use_sudo=0 +``` + +Only users in `admins` are allowed to use the bot. + +## License + +BSD-2c diff --git a/configstore.py b/configstore.py new file mode 100644 index 0000000..fcaa5c8 --- /dev/null +++ b/configstore.py @@ -0,0 +1,51 @@ +import os, re +from configparser import ConfigParser + +CONFIG_DIR = os.environ['HOME'] + '/.config/inverter-bot' +CONFIG_FILE = 'config.ini' + +__config__ = None + + +def _get_config_path() -> str: + return "%s/%s" % (CONFIG_DIR, CONFIG_FILE) + + +def get_config(): + global __config__ + if __config__ is not None: + return __config__['root'] + + if not os.path.exists(CONFIG_DIR): + raise IOError("%s directory not found" % CONFIG_DIR) + + if not os.path.isdir(CONFIG_DIR): + raise IOError("%s is not a directory" % CONFIG_DIR) + + config_path = _get_config_path() + if not os.path.isfile(config_path): + raise IOError("%s file not found" % config_path) + + __config__ = ConfigParser() + with open(config_path) as config_content: + __config__.read_string("[root]\n" + config_content.read()) + + return __config__['root'] + + +def get_token() -> str: + return get_config()['token'] + + +def get_admins() -> tuple: + config = get_config() + return tuple([int(s) for s in re.findall(r'\b\d+\b', config['admins'], flags=re.MULTILINE)]) + + +def get_isv_bin() -> str: + return get_config()['isv_bin'] + + +def use_sudo() -> bool: + config = get_config() + return 'use_sudo' in config and config['use_sudo'] == '1'
\ No newline at end of file @@ -0,0 +1,42 @@ +import subprocess +import configstore +import json + + +def __run(argv: list, fmt='json-w-units'): + argv.insert(0, configstore.get_isv_bin()) + if configstore.use_sudo(): + argv.insert(0, 'sudo') + argv.append('--format') + argv.append(fmt) + + result = subprocess.run(argv, capture_output=True) + if result.returncode != 0: + raise ChildProcessError("isv returned %d: %s" % (result.returncode, result.stderr)) + + return json.loads(result.stdout) if 'json' in fmt else result.stdout.decode('utf-8') + + +def general_status(as_table=False): + kwargs = {} + if as_table: + kwargs['fmt'] = 'table' + return __run(['--get-general-status'], **kwargs) + + +def day_generated(y: int, m: int, d: int): + return __run(['--get-day-generated', str(y), str(m), str(d)]) + + +def rated_information(as_table=False): + kwargs = {} + if as_table: + kwargs['fmt'] = 'table' + return __run(['--get-rated-information'], **kwargs) + + +def faults(as_table=False): + kwargs = {} + if as_table: + kwargs['fmt'] = 'table' + return __run(['--get-faults-warnings'], **kwargs) @@ -0,0 +1,186 @@ +import logging +import re +import datetime +import isv +import configstore + +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 = '<b>Battery:</b> %s %s' % tuple(gs['battery_voltage']) + html += ' (%s%s, ' % tuple(gs['battery_capacity']) + html += '%s%s)' % (power_direction, charging_rate) + + html += '\n<b>Load:</b> %s %s' % tuple(gs['ac_output_active_power']) + html += ' (%s%%)' % (gs['output_load_percent'][0]) + + if gs['pv1_input_power'][0] > 0: + html += '\n<b>Input power:</b> %s%s' % tuple(gs['pv1_input_power']) + + if gs['grid_voltage'][0] > 0 or gs['grid_freq'][0] > 0: + html += '\n<b>Generator:</b> %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 = '<b>Input power:</b> %s %s' % tuple(gs['pv1_input_power']) + html += ' (%s %s)' % tuple(gs['pv1_input_voltage']) + + html += '\n<b>Today:</b> %s Wh' % (gen_today['wh']) + + if gen_yday is not None: + html += '\n<b>Yesterday:</b> %s Wh' % (gen_yday['wh']) + + if gen_yday2 is not None: + html += '\n<b>The day before yesterday:</b> %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() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..ccd058e --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +python-telegram-bot~=13.1
\ No newline at end of file diff --git a/strings.py b/strings.py new file mode 100644 index 0000000..065210a --- /dev/null +++ b/strings.py @@ -0,0 +1,12 @@ +__strings = { + 'status': 'Status', + 'generation': 'Generation', + 'gs': 'GS', + 'ri': 'RI', + 'errors': 'Errors' +} + + +def lang(key): + global __strings + return __strings[key] if key in __strings else f'{{{key}}}' |