eswyft/inverter-bot
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

migrate from isv to inverterd

Browse Source
This commit is contained in:
Evgeny Zinoviev 2021-05-08 01:53:47 +03:00
parent a9f26cf874
commit b804a5d7e7
6 changed files with 144 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
README.md
View File

@ -1,33 +1,40 @@
# 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.
This is a Telegram bot for querying information from an InfiniSolar V family of hybrid solar inverters.
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.
## Requirements
- **`inverterd`** from [inverter-tools](https://github.com/gch1p/inverter-tools)
- **[`inverterd`](https://pypi.org/project/inverterd/)** python library
- Python 3.6+ or so
## Configuration
Configuration is stored in `config.ini` file in `~/.config/inverter-bot`.
The bot accepts following parameters:
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.
* ``--token`` — your telegram bot token (required)
* ``--users-whitelist`` — space-separated list of IDs of users who are allowed
to use the bot (required)
* ``--inverterd-host`` (default is `127.0.0.1`)
* ``--inverterd-port`` (default is `8305`)
## Launching with systemd
Create a service file `/etc/systemd/system/inverter-bot.service` with the following content (changing stuff like paths):
This is tested on Debian 10. Something might differ on other systems.
Create environment configuration file `/etc/default/inverter-bot`:
```
TOKEN="YOUR_TOKEN"
USERS="ID ID ID ..."
OPTS="" # here you can pass other options such as --inverterd-host
```
Create systemd service file `/etc/systemd/system/inverter-bot.service` with the following content (changing stuff like paths):
```systemd
[Unit]
@ -35,10 +42,11 @@ Description=inverter bot
After=network.target
[Service]
EnvironmentFile=/etc/default/inverter-bot
User=user
Group=user
Restart=on-failure
ExecStart=python3 /home/user/inverter-bot/main.py
ExecStart=python3 /home/user/inverter-bot/inverter-bot --token $TOKEN --users-whitelist $USERS $PARAMS
WorkingDirectory=/home/user/inverter-bot
[Install]
@ -47,7 +55,6 @@ WantedBy=multi-user.target
Then enable and start the service:
```
systemctl daemon-reload
systemctl enable inverter-bot
systemctl start inverter-bot
```

51
configstore.py
View File

@ -1,51 +0,0 @@
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'

115
main.py → inverter-bot Normal file → Executable file
View File

@ -1,10 +1,11 @@
import logging
import re
import datetime
import isv
import configstore
#!/usr/bin/env python3
import logging, re, datetime, json
from inverterd import Format, Client as InverterClient, InverterError
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 (
@ -22,10 +23,41 @@ from telegram.ext import (
)
class InverterClientWrapper:
def __init__(self, host: str, port: str):
self._host = host
self._port = port
self._inverter = None
self.create()
def create(self):
self._inverter = InverterClient(host=self._host, port=self._port)
self._inverter.connect()
def exec(self, command: str, arguments: tuple = (), format=Format.JSON):
try:
self._inverter.format(format)
return self._inverter.exec(command, arguments)
except InverterError as e:
raise e
except Exception as e:
# silently try to reconnect
try:
self.create()
except Exception:
pass
raise e
inverter: Optional[InverterClientWrapper] = None
#
# helpers
#
def get_markup() -> ReplyKeyboardMarkup:
button = [
[
@ -57,7 +89,7 @@ def start(update: Update, context: CallbackContext) -> None:
def msg_status(update: Update, context: CallbackContext) -> None:
try:
gs = isv.general_status()
gs = json.loads(inverter.exec('get-status'))['data']
# render response
power_direction = gs['battery_power_direction'].lower()
@ -65,23 +97,25 @@ def msg_status(update: Update, context: CallbackContext) -> None:
charging_rate = ''
if power_direction == 'charging':
charging_rate = ' @ %s %s' % tuple(gs['battery_charging_current'])
charging_rate = ' @ %s %s' % (
gs['battery_charging_current']['value'], gs['battery_charging_current']['unit'])
elif power_direction == 'discharging':
charging_rate = ' @ %s %s' % tuple(gs['battery_discharge_current'])
charging_rate = ' @ %s %s' % (
gs['battery_discharging_current']['value'], gs['battery_discharging_current']['unit'])
html = '<b>Battery:</b> %s %s' % tuple(gs['battery_voltage'])
html += ' (%s%s, ' % tuple(gs['battery_capacity'])
html = '<b>Battery:</b> %s %s' % (gs['battery_voltage']['value'], gs['battery_voltage']['unit'])
html += ' (%s%s, ' % (gs['battery_capacity']['value'], gs['battery_capacity']['unit'])
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])
html += '\n<b>Load:</b> %s %s' % (gs['ac_output_active_power']['value'], gs['ac_output_active_power']['unit'])
html += ' (%s%%)' % (gs['output_load_percent']['value'])
if gs['pv1_input_power'][0] > 0:
html += '\n<b>Input power:</b> %s%s' % tuple(gs['pv1_input_power'])
if gs['pv1_input_power']['value'] > 0:
html += '\n<b>Input power:</b> %s%s' % (gs['pv1_input_power']['value'], gs['pv1_input_power']['unit'])
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'])
if gs['grid_voltage']['value'] > 0 or gs['grid_freq']['value'] > 0:
html += '\n<b>Generator:</b> %s %s' % (gs['grid_voltage']['unit'], gs['grid_voltage']['value'])
html += ', %s %s' % (gs['grid_freq']['value'], gs['grid_freq']['unit'])
# send response
reply(update, html)
@ -96,24 +130,24 @@ def msg_generation(update: Update, context: CallbackContext) -> None:
yday = today - datetime.timedelta(days=1)
yday2 = today - datetime.timedelta(days=2)
gs = isv.general_status()
sleep(0.1)
gs = json.loads(inverter.exec('get-status'))['data']
# sleep(0.1)
gen_today = isv.day_generated(today.year, today.month, today.day)
gen_today = json.loads(inverter.exec('get-day-generated', (today.year, today.month, today.day)))['data']
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)
# sleep(0.1)
gen_yday = json.loads(inverter.exec('get-day-generated', (yday.year, yday.month, yday.day)))['data']
if yday2.month == today.month:
sleep(0.1)
gen_yday2 = isv.day_generated(yday2.year, yday2.month, yday2.day)
# sleep(0.1)
gen_yday2 = json.loads(inverter.exec('get-day-generated', (yday2.year, yday2.month, yday2.day)))['data']
# render response
html = '<b>Input power:</b> %s %s' % tuple(gs['pv1_input_power'])
html += ' (%s %s)' % tuple(gs['pv1_input_voltage'])
html = '<b>Input power:</b> %s %s' % (gs['pv1_input_power']['value'], gs['pv1_input_power']['unit'])
html += ' (%s %s)' % (gs['pv1_input_voltage']['value'], gs['pv1_input_voltage']['unit'])
html += '\n<b>Today:</b> %s Wh' % (gen_today['wh'])
@ -132,7 +166,7 @@ def msg_generation(update: Update, context: CallbackContext) -> None:
def msg_gs(update: Update, context: CallbackContext) -> None:
try:
status = isv.general_status(as_table=True)
status = inverter.exec('get-status', format=Format.TABLE)
reply(update, status)
except Exception as e:
logging.exception(str(e))
@ -141,7 +175,7 @@ def msg_gs(update: Update, context: CallbackContext) -> None:
def msg_ri(update: Update, context: CallbackContext) -> None:
try:
rated = isv.rated_information(as_table=True)
rated = inverter.exec('get-rated', format=Format.TABLE)
reply(update, rated)
except Exception as e:
logging.exception(str(e))
@ -150,8 +184,8 @@ def msg_ri(update: Update, context: CallbackContext) -> None:
def msg_errors(update: Update, context: CallbackContext) -> None:
try:
faults = isv.faults(as_table=True)
reply(update, faults)
errors = inverter.exec('get-errors', format=Format.TABLE)
reply(update, errors)
except Exception as e:
logging.exception(str(e))
reply(update, 'exception: ' + str(e))
@ -162,15 +196,30 @@ def msg_all(update: Update, context: CallbackContext) -> None:
if __name__ == '__main__':
config = configstore.get_config()
# 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 = InverterClientWrapper(host=args.inverterd_host, port=args.inverterd_port)
# configure logging
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})
# configure bot
updater = Updater(args.token, request_kwargs={'read_timeout': 6, 'connect_timeout': 7})
dispatcher = updater.dispatcher
user_filter = Filters.user(configstore.get_admins())
user_filter = Filters.user(whitelist)
dispatcher.add_handler(CommandHandler('start', start))
dispatcher.add_handler(MessageHandler(Filters.text(_('status')) & user_filter, msg_status))

42
isv.py
View File

@ -1,42 +0,0 @@
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)

3
requirements.txt
View File

@ -1 +1,2 @@
python-telegram-bot~=13.1
python-telegram-bot~=13.1
inverterd~=1.0.2

35
test.py Normal file
View File

@ -0,0 +1,35 @@
import json, re
from pprint import pprint
s = '{"result":"ok","data":{"grid_voltage":{"unit":"V","value":0.0},"grid_freq":{"unit":"Hz","value":0.0},"ac_output_voltage":{"unit":"V","value":230.0},"ac_output_freq":{"unit":"Hz","value":50.0},"ac_output_apparent_power":{"unit":"VA","value":115},"ac_output_active_power":{"unit":"Wh","value":18},"output_load_percent":{"unit":"%","value":2},"battery_voltage":{"unit":"V","value":50.0},"battery_voltage_scc":{"unit":"V","value":0.0},"battery_voltage_scc2":{"unit":"V","value":0.0},"battery_discharging_current":{"unit":"A","value":0},"battery_charging_current":{"unit":"A","value":0},"battery_capacity":{"unit":"%","value":78},"inverter_heat_sink_temp":{"unit":"°C","value":19},"mppt1_charger_temp":{"unit":"°C","value":0},"mppt2_charger_temp":{"unit":"°C","value":0},"pv1_input_power":{"unit":"Wh","value":1000},"pv2_input_power":{"unit":"Wh","value":0},"pv1_input_voltage":{"unit":"V","value":0.0},"pv2_input_voltage":{"unit":"V","value":0.0},"settings_values_changed":"Custom","mppt1_charger_status":"Abnormal","mppt2_charger_status":"Abnormal","load_connected":"Connected","battery_power_direction":"Discharge","dc_ac_power_direction":"DC/AC","line_power_direction":"Do nothing","local_parallel_id":0}}'
if __name__ == '__main__':
gs = json.loads(s)['data']
# pprint(gs)
# 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' % (gs['battery_charging_current']['value'], gs['battery_charging_current']['unit'])
elif power_direction == 'discharging':
charging_rate = ' @ %s %s' % (gs['battery_discharging_current']['value'], gs['battery_discharging_current']['unit'])
html = '<b>Battery:</b> %s %s' % (gs['battery_voltage']['value'], gs['battery_voltage']['unit'])
html += ' (%s%s, ' % (gs['battery_capacity']['value'], gs['battery_capacity']['unit'])
html += '%s%s)' % (power_direction, charging_rate)
html += '\n<b>Load:</b> %s %s' % (gs['ac_output_active_power']['value'], gs['ac_output_active_power']['unit'])
html += ' (%s%%)' % (gs['output_load_percent']['value'])
if gs['pv1_input_power']['value'] > 0:
html += '\n<b>Input power:</b> %s%s' % (gs['pv1_input_power']['value'], gs['pv1_input_power']['unit'])
if gs['grid_voltage']['value'] > 0 or gs['grid_freq']['value'] > 0:
html += '\n<b>Generator:</b> %s %s' % (gs['grid_voltage']['unit'], gs['grid_voltage']['value'])
html += ', %s %s' % (gs['grid_freq']['value'], gs['grid_freq']['unit'])
print(html)