#!/usr/bin/env python3 import logging import re import datetime import json import itertools import sys import asyncio import __py_include from inverterd import Format, InverterError from html import escape from typing import Optional, Tuple, Union from homekit.util import chunks from homekit.config import config, AppConfigUnit from homekit.telegram import bot from homekit.telegram.config import TelegramBotConfig, TelegramUserListType from homekit.inverter import ( wrapper_instance as inverter, beautify_table, InverterMonitor, ) from homekit.inverter.types import ( ChargingEvent, ACPresentEvent, BatteryState, ACMode, OutputSourcePriority ) from homekit.database.inverter_time_formats import FormatDate from homekit.api import WebApiClient from telegram import ReplyKeyboardMarkup, InlineKeyboardMarkup, InlineKeyboardButton if __name__ != '__main__': print(f'this script can not be imported as module', file=sys.stderr) sys.exit(1) db = None LT = escape('<=') flags_map = { 'buzzer': 'BUZZ', 'overload_bypass': 'OLBP', 'escape_to_default_screen_after_1min_timeout': 'LCDE', 'overload_restart': 'OLRS', 'over_temp_restart': 'OTRS', 'backlight_on': 'BLON', 'alarm_on_on_primary_source_interrupt': 'ALRM', 'fault_code_record': 'FTCR', } logger = logging.getLogger(__name__) class InverterBotConfig(AppConfigUnit, TelegramBotConfig): NAME = 'inverter_bot' @classmethod def schema(cls) -> Optional[dict]: acmode_item_schema = { 'thresholds': { 'type': 'list', 'required': True, 'schema': { 'type': 'list', 'min': 40, 'max': 60 }, }, 'initial_current': {'type': 'integer'} } return { **super(TelegramBotConfig).schema(), 'ac_mode': { 'type': 'dict', 'required': True, 'schema': { 'generator': acmode_item_schema, 'utilities': acmode_item_schema } }, 'monitor': { 'type': 'dict', 'required': True, 'schema': { 'vlow': {'type': 'integer', 'required': True}, 'vcrit': {'type': 'integer', 'required': True}, 'gen_currents': {'type': 'list', 'schema': {'type': 'integer'}, 'required': True}, 'gen_raise_intervals': {'type': 'list', 'schema': {'type': 'integer'}, 'required': True}, 'gen_cur30_v_limit': {'type': 'float', 'required': True}, 'gen_cur20_v_limit': {'type': 'float', 'required': True}, 'gen_cur10_v_limit': {'type': 'float', 'required': True}, 'gen_floating_v': {'type': 'integer', 'required': True}, 'gen_floating_time_max': {'type': 'integer', 'required': True} } } } config.load_app(InverterBotConfig) bot.initialize() bot.lang.ru( socket="В розетке", status='Статус', generation='Генерация', priority='Приоритет', battery="АКБ", load="Нагрузка", generator="Генератор", utilities="Столб", consumption="Статистика потребления", settings="Настройки", done="Готово", unexpected_callback_data="Ошибка: неверные данные", invalid_input="Неверное значение", invalid_mode="Invalid mode", flags_press_button='Нажмите кнопку для переключения настройки', flags_fail='Не удалось установить настройку', flags_invalid='Неизвестная настройка', # generation gen_input_power='Зарядная мощность', # settings settings_msg="Что вы хотите настроить?", settings_osp='Приоритет питания нагрузки', settings_ac_preset="Применить шаблон режима AC", settings_bat_thresholds="Пороги заряда АКБ от AC", settings_bat_cut_off_voltage="Порог отключения АКБ", settings_ac_max_charging_current="Максимальный ток заряда от AC", settings_osp_msg="Установите приоритет:", settings_osp_sub='Solar-Utility-Battery', settings_osp_sbu='Solar-Battery-Utility', settings_select_bottom_threshold="Выберите нижний порог:", settings_select_upper_threshold="Выберите верхний порог:", settings_select_max_current='Выберите максимальный ток:', settings_enter_cutoff_voltage=f'Введите напряжение V, где 40.0 {LT} V {LT} 48.0', # time and date today='Сегодня', yday1='Вчера', yday2='Позавчера', for_7days='За 7 дней', for_30days='За 30 дней', # to_select_interval='Выбрать интервал', # consumption consumption_msg="Выберите тип:", consumption_total="Домашние приборы", consumption_grid="Со столба", consumption_select_interval='Выберите период:', consumption_request_sent="⏳ Запрос отправлен...", # status charging_at=', ', pd_charging='заряжается', pd_discharging='разряжается', pd_nothing='не используется', # flags flag_buzzer='Звуковой сигнал', flag_overload_bypass='Разрешить перегрузку', flag_escape_to_default_screen_after_1min_timeout='Возврат на главный экран через 1 минуту', flag_overload_restart='Перезапуск при перегрузке', flag_over_temp_restart='Перезапуск при перегреве', flag_backlight_on='Подсветка экрана', flag_alarm_on_on_primary_source_interrupt='Сигнал при разрыве основного источника питания', flag_fault_code_record='Запись кодов ошибок', # commands setbatuv_v=f'напряжение, 40.0 {LT} V {LT} 48.0', setgenct_cv=f'напряжение включения заряда, 44 {LT} CV {LT} 51', setgenct_dv=f'напряжение отключения заряда, 48 {LT} DV {LT} 58', setgencc_a='максимальный ток заряда, допустимые значения: %s', # monitor chrg_evt_started='✅ Начали заряжать от генератора.', chrg_evt_finished='✅ Зарядили. Генератор пора выключать.', chrg_evt_disconnected='ℹ️ Генератор отключен.', chrg_evt_current_changed='ℹ️ Ток заряда от генератора установлен в %d A.', chrg_evt_not_charging='ℹ️ Генератор подключен, но не заряжает.', chrg_evt_na_solar='⛔️ Генератор подключен, но аккумуляторы не заряжаются из-за подключенных панелей.', chrg_evt_mostly_charged='✅ Аккумуляторы более-менее заряжены, генератор пора выключать.', battery_level_changed='Уровень заряда АКБ: %s %s (%0.1f V при нагрузке %d W)', error_message='Ошибка: %s.', util_chrg_evt_started='✅ Начали заряжать от столба.', util_chrg_evt_stopped='ℹ️ Перестали заряжать от столба.', util_chrg_evt_stopped_solar='ℹ️ Перестали заряжать от столба из-за подключения панелей.', util_connected='✅️ Столб подключён.', util_disconnected='‼️ Столб отключён.', # other notifications ac_mode_changed_notification='Пользователь %s установил режим AC: %s.', osp_changed_notification='Пользователь %s установил приоритет источника питания нагрузки: %s.', osp_auto_changed_notification='ℹ️ Бот установил приоритет источника питания нагрузки: %s. Причины: напряжение АКБ %.1f V, мощность заряда с панелей %d W.', bat_state_normal='Нормальный', bat_state_low='Низкий', bat_state_critical='Критический', ) bot.lang.en( socket='AC output', status='Status', generation='Generation', priority='Priority', battery="Battery", load="Load", generator="Generator", utilities="Utilities", consumption="Consumption statistics", settings="Settings", done="Done", unexpected_callback_data="Unexpected callback data", select_priortiy="Select priority:", invalid_input="Invalid input", invalid_mode="Invalid mode", flags_press_button='Press a button to toggle a flag.', flags_fail='Failed to toggle flag', flags_invalid='Invalid flag', # settings settings_msg='What do you want to configure?', settings_osp='Output source priority', settings_ac_preset="AC preset", settings_bat_thresholds="Battery charging thresholds", settings_bat_cut_off_voltage="Battery cut-off voltage", settings_ac_max_charging_current="Max AC charging current", settings_osp_msg="Select priority:", settings_osp_sub='Solar-Utility-Battery', settings_osp_sbu='Solar-Battery-Utility', settings_select_bottom_threshold="Select bottom (lower) threshold:", settings_select_upper_threshold="Select top (upper) threshold:", settings_select_max_current='Select max current:', settings_enter_cutoff_voltage=f'Enter voltage V (40.0 {LT} V {LT} 48.0):', # generation gen_input_power='Input power', # time and date today='Today', yday1='Yesterday', yday2='The day before yesterday', for_7days='7 days', for_30days='30 days', # to_select_interval='Select interval', # consumption consumption_msg="Select type:", consumption_total="Home appliances", consumption_grid="Consumed from grid", consumption_select_interval='Select period:', consumption_request_sent="⏳ Request sent...", # status charging_at=' @ ', pd_charging='charging', pd_discharging='discharging', pd_nothing='not used', # flags flag_buzzer='Buzzer', flag_overload_bypass='Overload bypass', flag_escape_to_default_screen_after_1min_timeout='Reset to default LCD page after 1min timeout', flag_overload_restart='Restart on overload', flag_over_temp_restart='Restart on overtemp', flag_backlight_on='LCD backlight', flag_alarm_on_on_primary_source_interrupt='Beep on primary source interruption', flag_fault_code_record='Fault code recording', # commands setbatuv_v=f'floating point number, 40.0 {LT} V {LT} 48.0', setgenct_cv=f'charging voltage, 44 {LT} CV {LT} 51', setgenct_dv=f'discharging voltage, 48 {LT} DV {LT} 58', setgencc_a='max charging current, allowed values: %s', # monitor chrg_evt_started='✅ Started charging from AC.', chrg_evt_finished='✅ Finished charging, it\'s time to stop the generator.', chrg_evt_disconnected='ℹ️ AC disconnected.', chrg_evt_current_changed='ℹ️ AC charging current set to %d A.', chrg_evt_not_charging='ℹ️ AC connected but not charging.', chrg_evt_na_solar='⛔️ AC connected, but battery won\'t be charged due to active solar power line.', chrg_evt_mostly_charged='✅ The battery is mostly charged now. The generator can be turned off.', battery_level_changed='Battery level: %s (%0.1f V under %d W load)', error_message='Error: %s.', util_chrg_evt_started='✅ Started charging from utilities.', util_chrg_evt_stopped='ℹ️ Stopped charging from utilities.', util_chrg_evt_stopped_solar='ℹ️ Stopped charging from utilities because solar panels were connected.', util_connected='✅️ Utilities connected.', util_disconnected='‼️ Utilities disconnected.', # other notifications ac_mode_changed_notification='User %s set AC mode to %s.', osp_changed_notification='User %s set output source priority: %s.', osp_auto_changed_notification='Bot changed output source priority to %s. Reasons: battery voltage is %.1f V, solar input is %d W.', bat_state_normal='Normal', bat_state_low='Low', bat_state_critical='Critical', ) def monitor_charging(event: ChargingEvent, **kwargs) -> None: args = [] is_util = False if event == ChargingEvent.AC_CHARGING_STARTED: key = 'started' elif event == ChargingEvent.AC_CHARGING_FINISHED: key = 'finished' elif event == ChargingEvent.AC_DISCONNECTED: key = 'disconnected' elif event == ChargingEvent.AC_NOT_CHARGING: key = 'not_charging' elif event == ChargingEvent.AC_CURRENT_CHANGED: key = 'current_changed' args.append(kwargs['current']) elif event == ChargingEvent.AC_CHARGING_UNAVAILABLE_BECAUSE_SOLAR: key = 'na_solar' elif event == ChargingEvent.AC_MOSTLY_CHARGED: key = 'mostly_charged' elif event == ChargingEvent.UTIL_CHARGING_STARTED: key = 'started' is_util = True elif event == ChargingEvent.UTIL_CHARGING_STOPPED: key = 'stopped' is_util = True elif event == ChargingEvent.UTIL_CHARGING_STOPPED_SOLAR: key = 'stopped_solar' is_util = True else: logger.error('unknown charging event:', event) return key = f'chrg_evt_{key}' if is_util: key = f'util_{key}' asyncio.ensure_future( bot.notify_all( lambda lang: bot.lang.get(key, lang, *args) ) ) def monitor_battery(state: BatteryState, v: float, load_watts: int) -> None: if state == BatteryState.NORMAL: emoji = '✅' elif state == BatteryState.LOW: emoji = '⚠️' elif state == BatteryState.CRITICAL: emoji = '‼️' else: logger.error('unknown battery state:', state) return asyncio.ensure_future( bot.notify_all( lambda lang: bot.lang.get('battery_level_changed', lang, emoji, bot.lang.get(f'bat_state_{state.name.lower()}', lang), v, load_watts) ) ) def monitor_util(event: ACPresentEvent): if event == ACPresentEvent.CONNECTED: key = 'connected' else: key = 'disconnected' key = f'util_{key}' asyncio.ensure_future( bot.notify_all( lambda lang: bot.lang.get(key, lang) ) ) def monitor_error(error: str) -> None: asyncio.ensure_future( bot.notify_all( lambda lang: bot.lang.get('error_message', lang, error) ) ) def osp_change_cb(new_osp: OutputSourcePriority, solar_input: int, v: float): setosp(new_osp) asyncio.ensure_future( bot.notify_all( lambda lang: bot.lang.get('osp_auto_changed_notification', lang, bot.lang.get(f'settings_osp_{new_osp.value.lower()}', lang), v, solar_input), ) ) @bot.handler(command='status') async def full_status(ctx: bot.Context) -> None: status = inverter.exec('get-status', format=Format.TABLE) await ctx.reply(beautify_table(status)) @bot.handler(command='config') async def full_rated(ctx: bot.Context) -> None: rated = inverter.exec('get-rated', format=Format.TABLE) await ctx.reply(beautify_table(rated)) @bot.handler(command='errors') async def full_errors(ctx: bot.Context) -> None: errors = inverter.exec('get-errors', format=Format.TABLE) await ctx.reply(beautify_table(errors)) @bot.handler(command='flags') async def flags_handler(ctx: bot.Context) -> None: flags = inverter.exec('get-flags')['data'] text, markup = build_flags_keyboard(flags, ctx) await ctx.reply(text, markup=markup) def build_flags_keyboard(flags: dict, ctx: bot.Context) -> Tuple[str, InlineKeyboardMarkup]: keyboard = [] for k, v in flags.items(): label = ('✅' if v else '❌') + ' ' + ctx.lang(f'flag_{k}') proto_flag = flags_map[k] keyboard.append([InlineKeyboardButton(label, callback_data=f'flag_{proto_flag}')]) return ctx.lang('flags_press_button'), InlineKeyboardMarkup(keyboard) def getacmode() -> ACMode: return ACMode(bot.db.get_param('ac_mode', default=ACMode.GENERATOR)) def setacmode(mode: ACMode): monitor.set_ac_mode(mode) cv, dv = config['ac_mode'][str(mode.value)]['thresholds'] a = config['ac_mode'][str(mode.value)]['initial_current'] logger.debug(f'setacmode: mode={mode}, cv={cv}, dv={dv}, a={a}') inverter.exec('set-charge-thresholds', (cv, dv)) inverter.exec('set-max-ac-charge-current', (0, a)) def setosp(sp: OutputSourcePriority): logger.debug(f'setosp: sp={sp}') inverter.exec('set-output-source-priority', (sp.value,)) monitor.notify_osp(sp) class SettingsConversation(bot.conversation): START, OSP, AC_PRESET, BAT_THRESHOLDS_1, BAT_THRESHOLDS_2, BAT_CUT_OFF_VOLTAGE, AC_MAX_CHARGING_CURRENT = range(7) STATE_SEQS = [ [START, OSP], [START, AC_PRESET], [START, BAT_THRESHOLDS_1, BAT_THRESHOLDS_2], [START, BAT_CUT_OFF_VOLTAGE], [START, AC_MAX_CHARGING_CURRENT] ] START_BUTTONS = bot.lang.pfx('settings_', ['ac_preset', 'ac_max_charging_current', 'bat_thresholds', 'bat_cut_off_voltage', 'osp']) OSP_BUTTONS = bot.lang.pfx('settings_osp_', [sp.value.lower() for sp in OutputSourcePriority]) AC_PRESET_BUTTONS = [mode.value for mode in ACMode] RECHARGE_VOLTAGES = [44, 45, 46, 47, 48, 49, 50, 51] REDISCHARGE_VOLTAGES = [48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58] @bot.conventer(START, message='settings') async def start_enter(self, ctx: bot.Context): buttons = list(chunks(list(self.START_BUTTONS), 2)) buttons.reverse() return await self.reply(ctx, self.START, ctx.lang('settings_msg'), buttons, with_cancel=True) @bot.convinput(START, messages={ 'settings_osp': OSP, 'settings_ac_preset': AC_PRESET, 'settings_bat_thresholds': BAT_THRESHOLDS_1, 'settings_bat_cut_off_voltage': BAT_CUT_OFF_VOLTAGE, 'settings_ac_max_charging_current': AC_MAX_CHARGING_CURRENT }) async def start_input(self, ctx: bot.Context): pass @bot.conventer(OSP) async def osp_enter(self, ctx: bot.Context): return await self.reply(ctx, self.OSP, ctx.lang('settings_osp_msg'), self.OSP_BUTTONS, with_back=True) @bot.convinput(OSP, messages=OSP_BUTTONS) async def osp_input(self, ctx: bot.Context): selected_sp = None for sp in OutputSourcePriority: if ctx.text == ctx.lang(f'settings_osp_{sp.value.lower()}'): selected_sp = sp break if selected_sp is None: raise ValueError('invalid sp') # apply the mode setosp(selected_sp) await asyncio.gather( # reply to user ctx.reply(ctx.lang('saved'), markup=bot.IgnoreMarkup()), # notify other users bot.notify_all( lambda lang: bot.lang.get('osp_changed_notification', lang, ctx.user.id, ctx.user.name, bot.lang.get(f'settings_osp_{selected_sp.value.lower()}', lang)), exclude=(ctx.user_id,) ) ) return self.END @bot.conventer(AC_PRESET) async def acpreset_enter(self, ctx: bot.Context): return await self.reply(ctx, self.AC_PRESET, ctx.lang('settings_ac_preset_msg'), self.AC_PRESET_BUTTONS, with_back=True) @bot.convinput(AC_PRESET, messages=AC_PRESET_BUTTONS) async def acpreset_input(self, ctx: bot.Context): if monitor.active_current is not None: raise RuntimeError('generator charging program is active') if ctx.text == ctx.lang('utilities'): newmode = ACMode.UTILITIES elif ctx.text == ctx.lang('generator'): newmode = ACMode.GENERATOR else: raise ValueError('invalid mode') # apply the mode setacmode(newmode) # save bot.db.set_param('ac_mode', str(newmode.value)) await asyncio.gather( # reply to user ctx.reply(ctx.lang('saved'), markup=bot.IgnoreMarkup()), # notify other users bot.notify_all( lambda lang: bot.lang.get('ac_mode_changed_notification', lang, ctx.user.id, ctx.user.name, bot.lang.get(str(newmode.value), lang)), exclude=(ctx.user_id,) ) ) return self.END @bot.conventer(BAT_THRESHOLDS_1) async def thresholds1_enter(self, ctx: bot.Context): buttons = list(map(lambda v: f'{v} V', self.RECHARGE_VOLTAGES)) buttons = chunks(buttons, 4) return await self.reply(ctx, self.BAT_THRESHOLDS_1, ctx.lang('settings_select_bottom_threshold'), buttons, with_back=True, buttons_lang_completed=True) @bot.convinput(BAT_THRESHOLDS_1, messages=list(map(lambda n: f'{n} V', RECHARGE_VOLTAGES)), messages_lang_completed=True) async def thresholds1_input(self, ctx: bot.Context): v = self._parse_voltage(ctx.text) ctx.user_data['bat_thrsh_v1'] = v return await self.invoke(self.BAT_THRESHOLDS_2, ctx) @bot.conventer(BAT_THRESHOLDS_2) async def thresholds2_enter(self, ctx: bot.Context): buttons = list(map(lambda v: f'{v} V', self.REDISCHARGE_VOLTAGES)) buttons = chunks(buttons, 4) return await self.reply(ctx, self.BAT_THRESHOLDS_2, ctx.lang('settings_select_upper_threshold'), buttons, with_back=True, buttons_lang_completed=True) @bot.convinput(BAT_THRESHOLDS_2, messages=list(map(lambda n: f'{n} V', REDISCHARGE_VOLTAGES)), messages_lang_completed=True) async def thresholds2_input(self, ctx: bot.Context): v2 = v = self._parse_voltage(ctx.text) v1 = ctx.user_data['bat_thrsh_v1'] del ctx.user_data['bat_thrsh_v1'] response = inverter.exec('set-charge-thresholds', (v1, v2)) await ctx.reply(ctx.lang('saved') if response['result'] == 'ok' else 'ERROR', markup=bot.IgnoreMarkup()) return self.END @bot.conventer(AC_MAX_CHARGING_CURRENT) async def ac_max_enter(self, ctx: bot.Context): buttons = self._get_allowed_ac_charge_amps() buttons = map(lambda n: f'{n} A', buttons) buttons = [list(buttons)] return await self.reply(ctx, self.AC_MAX_CHARGING_CURRENT, ctx.lang('settings_select_max_current'), buttons, with_back=True, buttons_lang_completed=True) @bot.convinput(AC_MAX_CHARGING_CURRENT, regex=r'^\d+ A$') async def ac_max_input(self, ctx: bot.Context): a = self._parse_amps(ctx.text) allowed = self._get_allowed_ac_charge_amps() if a not in allowed: raise ValueError('input is not allowed') response = inverter.exec('set-max-ac-charge-current', (0, a)) await ctx.reply(ctx.lang('saved') if response['result'] == 'ok' else 'ERROR', markup=bot.IgnoreMarkup()) return self.END @bot.conventer(BAT_CUT_OFF_VOLTAGE) async def cutoff_enter(self, ctx: bot.Context): return await self.reply(ctx, self.BAT_CUT_OFF_VOLTAGE, ctx.lang('settings_enter_cutoff_voltage'), None, with_back=True) @bot.convinput(BAT_CUT_OFF_VOLTAGE, regex=r'^(\d{2}(\.\d{1})?)$') async def cutoff_input(self, ctx: bot.Context): v = float(ctx.text) if 40.0 <= v <= 48.0: response = inverter.exec('set-battery-cutoff-voltage', (v,)) await ctx.reply(ctx.lang('saved') if response['result'] == 'ok' else 'ERROR', markup=bot.IgnoreMarkup()) else: raise ValueError('invalid voltage') return self.END def _get_allowed_ac_charge_amps(self) -> list[int]: l = inverter.exec('get-allowed-ac-charge-currents')['data'] l = filter(lambda n: n <= 40, l) return list(l) def _parse_voltage(self, s: str) -> int: return int(re.match(r'^(\d{2}) V$', s).group(1)) def _parse_amps(self, s: str) -> int: return int(re.match(r'^(\d{1,2}) A$', s).group(1)) class ConsumptionConversation(bot.conversation): START, TOTAL, GRID = range(3) STATE_SEQS = [ [START, TOTAL], [START, GRID] ] START_BUTTONS = bot.lang.pfx('consumption_', ['total', 'grid']) INTERVAL_BUTTONS = [ ['today'], ['yday1'], ['for_7days', 'for_30days'], # ['to_select_interval'] ] INTERVAL_BUTTONS_FLAT = list(itertools.chain.from_iterable(INTERVAL_BUTTONS)) @bot.conventer(START, message='consumption') async def start_enter(self, ctx: bot.Context): return await self.reply(ctx, self.START, ctx.lang('consumption_msg'), [self.START_BUTTONS], with_cancel=True) @bot.convinput(START, messages={ 'consumption_total': TOTAL, 'consumption_grid': GRID }) async def start_input(self, ctx: bot.Context): pass @bot.conventer(TOTAL) async def total_enter(self, ctx: bot.Context): return await self._render_interval_btns(ctx, self.TOTAL) @bot.conventer(GRID) async def grid_enter(self, ctx: bot.Context): return await self._render_interval_btns(ctx, self.GRID) async def _render_interval_btns(self, ctx: bot.Context, state): return await self.reply(ctx, state, ctx.lang('consumption_select_interval'), self.INTERVAL_BUTTONS, with_back=True) @bot.convinput(TOTAL, messages=INTERVAL_BUTTONS_FLAT) async def total_input(self, ctx: bot.Context): return await self._render_interval_results(ctx, self.TOTAL) @bot.convinput(GRID, messages=INTERVAL_BUTTONS_FLAT) async def grid_input(self, ctx: bot.Context): return await self._render_interval_results(ctx, self.GRID) async def _render_interval_results(self, ctx: bot.Context, state): # if ctx.text == ctx.lang('to_select_interval'): # TODO # pass # # else: now = datetime.datetime.now() s_to = now.strftime(FormatDate) if ctx.text == ctx.lang('today'): s_from = now.strftime(FormatDate) s_to = 'now' elif ctx.text == ctx.lang('yday1'): s_from = (now - datetime.timedelta(days=1)).strftime(FormatDate) elif ctx.text == ctx.lang('for_7days'): s_from = (now - datetime.timedelta(days=7)).strftime(FormatDate) elif ctx.text == ctx.lang('for_30days'): s_from = (now - datetime.timedelta(days=30)).strftime(FormatDate) # markup = InlineKeyboardMarkup([ # [InlineKeyboardButton(ctx.lang('please_wait'), callback_data='wait')] # ]) message = await ctx.reply(ctx.lang('consumption_request_sent'), markup=bot.IgnoreMarkup()) api = WebApiClient(timeout=60) method = 'inverter_get_consumed_energy' if state == self.TOTAL else 'inverter_get_grid_consumed_energy' try: wh = getattr(api, method)(s_from, s_to) await bot.delete_message(message.chat_id, message.message_id) await ctx.reply('%.2f Wh' % (wh,), markup=bot.IgnoreMarkup()) return self.END except Exception as e: await asyncio.gather( bot.delete_message(message.chat_id, message.message_id), ctx.reply_exc(e) ) # other # ----- @bot.handler(command='monstatus') async def monstatus_handler(ctx: bot.Context) -> None: msg = '' st = monitor.dump_status() for k, v in st.items(): msg += k + ': ' + str(v) + '\n' await ctx.reply(msg) @bot.handler(command='monsetcur') async def monsetcur_handler(ctx: bot.Context) -> None: await ctx.reply('not implemented yet') @bot.callbackhandler async def button_callback(ctx: bot.Context) -> None: query = ctx.callback_query if query.data.startswith('flag_'): flag = query.data[5:] found = False json_key = None for k, v in flags_map.items(): if v == flag: found = True json_key = k break if not found: await query.answer(ctx.lang('flags_invalid')) return flags = inverter.exec('get-flags')['data'] cur_flag_value = flags[json_key] target_flag_value = '0' if cur_flag_value else '1' # set flag response = inverter.exec('set-flag', (flag, target_flag_value)) # notify user await query.answer(ctx.lang('done') if response['result'] == 'ok' else ctx.lang('flags_fail')) # edit message flags[json_key] = not cur_flag_value text, markup = build_flags_keyboard(flags, ctx) await query.edit_message_text(text, reply_markup=markup) else: await query.answer(ctx.lang('unexpected_callback_data')) @bot.exceptionhandler async def exception_handler(e: Exception, ctx: bot.Context) -> Optional[bool]: if isinstance(e, InverterError): try: err = json.loads(str(e))['message'] except json.decoder.JSONDecodeError: err = str(e) err = re.sub(r'((?:.*)?error:) (.*)', r'\1 \2', err) await ctx.reply(err, markup=bot.IgnoreMarkup()) return True @bot.handler(message='status') async def status_handler(ctx: bot.Context) -> None: gs = inverter.exec('get-status')['data'] rated = inverter.exec('get-rated')['data'] # render response power_direction = gs['battery_power_direction'].lower() power_direction = re.sub(r'ge$', 'ging', power_direction) charging_rate = '' chrg_at = ctx.lang('charging_at') if power_direction == 'charging': charging_rate = f'{chrg_at}%s %s' % ( gs['battery_charge_current']['value'], gs['battery_charge_current']['unit']) pd_label = ctx.lang('pd_charging') elif power_direction == 'discharging': charging_rate = f'{chrg_at}%s %s' % ( gs['battery_discharge_current']['value'], gs['battery_discharge_current']['unit']) pd_label = ctx.lang('pd_discharging') else: pd_label = ctx.lang('pd_nothing') html = f'{ctx.lang("battery")}: %s %s' % (gs['battery_voltage']['value'], gs['battery_voltage']['unit']) html += ' (%s%s)' % (pd_label, charging_rate) html += f'\n{ctx.lang("load")}: %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 += f'\n{ctx.lang("gen_input_power")}: %s %s' % (gs['pv1_input_power']['value'], gs['pv1_input_power']['unit']) if gs['grid_voltage']['value'] > 0 or gs['grid_freq']['value'] > 0: ac_mode = getacmode() html += f'\n{ctx.lang(ac_mode.value)}: %s %s' % (gs['grid_voltage']['value'], gs['grid_voltage']['unit']) html += ', %s %s' % (gs['grid_freq']['value'], gs['grid_freq']['unit']) html += f'\n{ctx.lang("socket")}: %s %s, %s %s' % ( gs['ac_output_voltage']['value'], gs['ac_output_voltage']['unit'], gs['ac_output_freq']['value'], gs['ac_output_freq']['unit'] ) html += f'\n{ctx.lang("priority")}: {rated["output_source_priority"]}' # send response await ctx.reply(html) @bot.handler(message='generation') async def generation_handler(ctx: bot.Context) -> None: today = datetime.date.today() yday = today - datetime.timedelta(days=1) yday2 = today - datetime.timedelta(days=2) gs = inverter.exec('get-status')['data'] gen_today = inverter.exec('get-day-generated', (today.year, today.month, today.day))['data'] gen_yday = None gen_yday2 = None if yday.month == today.month: gen_yday = inverter.exec('get-day-generated', (yday.year, yday.month, yday.day))['data'] if yday2.month == today.month: gen_yday2 = inverter.exec('get-day-generated', (yday2.year, yday2.month, yday2.day))['data'] # render response html = f'{ctx.lang("gen_input_power")}: %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 += f'\n{ctx.lang("today")}: %s Wh' % (gen_today['wh']) if gen_yday is not None: html += f'\n{ctx.lang("yday1")}: %s Wh' % (gen_yday['wh']) if gen_yday2 is not None: html += f'\n{ctx.lang("yday2")}: %s Wh' % (gen_yday2['wh']) # send response await ctx.reply(html) @bot.defaultreplymarkup def markup(ctx: Optional[bot.Context]) -> Optional[ReplyKeyboardMarkup]: button = [ [ctx.lang('status'), ctx.lang('generation')], [ctx.lang('consumption')], [ctx.lang('settings')] ] return ReplyKeyboardMarkup(button, one_time_keyboard=False) class InverterStore(bot.BotDatabase): SCHEMA = 2 def schema_init(self, version: int) -> None: super().schema_init(version) if version < 2: cursor = self.cursor() cursor.execute("""CREATE TABLE IF NOT EXISTS params ( id TEXT NOT NULL PRIMARY KEY, value TEXT NOT NULL )""") cursor.execute("CREATE INDEX param_id_idx ON params (id)") self.commit() def get_param(self, key: str, default=None): cursor = self.cursor() cursor.execute('SELECT value FROM params WHERE id=?', (key,)) row = cursor.fetchone() return default if row is None else row[0] def set_param(self, key: str, value: Union[str, int, float]): cursor = self.cursor() cursor.execute('REPLACE INTO params (id, value) VALUES (?, ?)', (key, str(value))) self.commit() inverter.init(host=config['inverter']['ip'], port=config['inverter']['port']) bot.set_database(InverterStore()) bot.add_conversation(SettingsConversation(enable_back=True)) bot.add_conversation(ConsumptionConversation(enable_back=True)) monitor = InverterMonitor() monitor.set_charging_event_handler(monitor_charging) monitor.set_battery_event_handler(monitor_battery) monitor.set_util_event_handler(monitor_util) monitor.set_error_handler(monitor_error) monitor.set_osp_need_change_callback(osp_change_cb) setacmode(getacmode()) if not config.get('monitor.disabled'): logging.info('starting monitor') monitor.start() bot.run() monitor.stop()