initial

.gitignore

@@ -0,0 +1,2 @@
+.idea
+venv

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

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'

isv.py

@@ -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)

main.py

@@ -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()

requirements.txt

@@ -0,0 +1 @@
+python-telegram-bot~=13.1

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}}}'