diff options
| -rwxr-xr-x | src/electricity_calc.py | 167 | ||||
| -rw-r--r-- | src/home/config/__init__.py | 2 | ||||
| -rw-r--r-- | src/home/config/config.py | 4 | ||||
| -rw-r--r-- | src/home/database/inverter.py | 133 |
4 files changed, 302 insertions, 4 deletions
diff --git a/src/electricity_calc.py b/src/electricity_calc.py new file mode 100755 index 0000000..8ad2957 --- /dev/null +++ b/src/electricity_calc.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +import logging +import os +import sys +import inspect +import zoneinfo + +from home.config import config # do not remove this import! +from datetime import datetime, timedelta +from logging import Logger +from home.database import InverterDatabase +from argparse import ArgumentParser, ArgumentError +from typing import Optional + +_logger: Optional[Logger] = None +_progname = os.path.basename(__file__) +_is_verbose = False + +fmt_time = '%Y-%m-%d %H:%M:%S' +fmt_date = '%Y-%m-%d' +tz = zoneinfo.ZoneInfo('Europe/Moscow') + + +def method_usage() -> str: + # https://stackoverflow.com/questions/2654113/how-to-get-the-callers-method-name-in-the-called-method + curframe = inspect.currentframe() + calframe = inspect.getouterframes(curframe, 2) + return f'{_progname} {calframe[1][3]} [ARGS]' + + +def fmt_escape(s: str): + return s.replace('%', '%%') + + +def setup_logging(verbose: bool): + global _is_verbose + + logging_level = logging.INFO if not verbose else logging.DEBUG + logging.basicConfig(level=logging_level) + + _is_verbose = verbose + + +class SubParser: + def __init__(self, description: str, usage: str): + self.parser = ArgumentParser( + description=description, + usage=usage + ) + + def add_argument(self, *args, **kwargs): + self.parser.add_argument(*args, **kwargs) + + def parse_args(self): + self.add_argument('--verbose', '-V', action='store_true', + help='enable debug logs') + + args = self.parser.parse_args(sys.argv[2:]) + setup_logging(args.verbose) + + return args + + +def strptime_auto(s: str) -> datetime: + e = None + for fmt in (fmt_time, fmt_date): + try: + return datetime.strptime(s, fmt) + except ValueError as _e: + e = _e + raise e + + +def get_dt_from_to_arguments(parser): + parser.add_argument('--from', type=str, dest='date_from', required=True, + help=f'From date, format: {fmt_escape(fmt_time)} or {fmt_escape(fmt_date)}') + parser.add_argument('--to', type=str, dest='date_to', default='now', + help=f'To date, format: {fmt_escape(fmt_time)}, {fmt_escape(fmt_date)}, \'now\' or \'24h\'') + arg = parser.parse_args() + + dt_from = strptime_auto(arg.date_from) + + if arg.date_to == 'now': + dt_to = datetime.now() + elif arg.date_to == '24h': + dt_to = dt_from + timedelta(days=1) + else: + dt_to = strptime_auto(arg.date_to) + + return dt_from, dt_to + + +def print_intervals(intervals): + for interval in intervals: + start, end = interval + buf = f'{start.strftime(fmt_time)} .. ' + if end: + buf += f'{end.strftime(fmt_time)}' + else: + buf += 'now' + + print(buf) + + +class Electricity(): + def __init__(self): + global _logger + + methods = [func.replace('_', '-') + for func in dir(Electricity) + if callable(getattr(Electricity, func)) and not func.startswith('_') and func != 'query'] + + parser = ArgumentParser( + usage=f'{_progname} METHOD [ARGS]' + ) + parser.add_argument('method', choices=methods, + help='Method to run') + parser.add_argument('--verbose', '-V', action='store_true', + help='enable debug logs') + + argv = sys.argv[1:2] + for arg in ('-V', '--verbose'): + if arg in sys.argv: + argv.append(arg) + args = parser.parse_args(argv) + + setup_logging(args.verbose) + self.db = InverterDatabase() + + method = args.method.replace('-', '_') + getattr(self, method)() + + def get_grid_connected_intervals(self): + parser = SubParser('Returns datetime intervals when grid was connected', method_usage()) + dt_from, dt_to = get_dt_from_to_arguments(parser) + + intervals = self.db.get_grid_connected_intervals(dt_from, dt_to) + print_intervals(intervals) + + def get_grid_used_intervals(self): + parser = SubParser('Returns datetime intervals when power grid was actually used', method_usage()) + dt_from, dt_to = get_dt_from_to_arguments(parser) + + intervals = self.db.get_grid_used_intervals(dt_from, dt_to) + print_intervals(intervals) + + def get_grid_consumed_energy(self): + parser = SubParser('Returns sum of energy consumed from util grid', method_usage()) + dt_from, dt_to = get_dt_from_to_arguments(parser) + + wh = self.db.get_grid_consumed_energy(dt_from, dt_to) + print('%.2f' % wh,) + + def get_consumed_energy(self): + parser = SubParser('Returns total consumed energy', method_usage()) + dt_from, dt_to = get_dt_from_to_arguments(parser) + + wh = self.db.get_consumed_energy(dt_from, dt_to) + print('%.2f' % wh,) + + +if __name__ == '__main__': + try: + Electricity() + except Exception as e: + _logger.exception(e) + sys.exit(1) diff --git a/src/home/config/__init__.py b/src/home/config/__init__.py index d4b1c27..cc9c091 100644 --- a/src/home/config/__init__.py +++ b/src/home/config/__init__.py @@ -1 +1 @@ -from .config import ConfigStore, config, is_development_mode +from .config import ConfigStore, config, is_development_mode, setup_logging diff --git a/src/home/config/config.py b/src/home/config/config.py index f953c43..9882bfa 100644 --- a/src/home/config/config.py +++ b/src/home/config/config.py @@ -34,7 +34,7 @@ class ConfigStore: data: MutableMapping[str, Any] app_name: Optional[str] - def __int__(self): + def __init__(self): self.data = {} self.app_name = None @@ -57,7 +57,7 @@ class ConfigStore: parser = ArgumentParser() if not no_config: parser.add_argument('-c', '--config', type=str, required=name is None, - help='Path to the config in TOML format') + help='Path to the config in TOML or YAML format') parser.add_argument('-V', '--verbose', action='store_true') parser.add_argument('--log-file', type=str) parser.add_argument('--log-default-fmt', action='store_true') diff --git a/src/home/database/inverter.py b/src/home/database/inverter.py index 6fdf9d7..8ebfae2 100644 --- a/src/home/database/inverter.py +++ b/src/home/database/inverter.py @@ -1,10 +1,40 @@ -from .clickhouse import get_clickhouse +import logging + +from zoneinfo import ZoneInfo from time import time +from datetime import datetime, timedelta +from typing import Optional +from collections import namedtuple + +from ..config import is_development_mode +from .clickhouse import get_clickhouse + + +IntervalList = list[list[Optional[datetime]]] class InverterDatabase: def __init__(self): self.db = get_clickhouse('solarmon') + self.server_timezone = self.query('SELECT timezone()')[0][0] + + self.logger = logging.getLogger(self.__class__.__name__) + + def query(self, *args, **kwargs): + settings = {'use_client_time_zone': True} + kwargs['settings'] = settings + + if 'no_tz_fix' not in kwargs and len(args) > 1 and isinstance(args[1], dict): + for k, v in args[1].items(): + if isinstance(v, datetime): + args[1][k] = v.astimezone(tz=ZoneInfo(self.server_timezone)) + + result = self.db.execute(*args, **kwargs) + + if is_development_mode(): + self.logger.debug(args[0] if len(args) == 1 else args[0] % args[1]) + + return result def add_generation(self, home_id: int, client_time: int, watts: int) -> None: self.db.execute( @@ -100,3 +130,104 @@ class InverterDatabase: line_power_direction, load_connected ]]) + + def get_consumed_energy(self, dt_from: datetime, dt_to: datetime) -> float: + rows = self.query('SELECT ClientTime, ACOutputActivePower FROM status' + ' WHERE ClientTime >= %(from)s AND ClientTime <= %(to)s' + ' ORDER BY ClientTime', {'from': dt_from, 'to': dt_to}) + prev_time = None + prev_wh = 0 + + ws = 0 # watt-seconds + for t, wh in rows: + if prev_time is not None: + n = (t - prev_time).total_seconds() + ws += prev_wh * n + + prev_time = t + prev_wh = wh + + return ws / 3600 # convert to watt-hours + + def get_intervals_by_condition(self, + dt_from: datetime, + dt_to: datetime, + cond_start: str, + cond_end: str) -> IntervalList: + rows = None + ranges = [[None, None]] + + while rows is None or len(rows) > 0: + if ranges[len(ranges) - 1][0] is None: + condition = cond_start + range_idx = 0 + else: + condition = cond_end + range_idx = 1 + + rows = self.query('SELECT ClientTime FROM status ' + f'WHERE ClientTime > %(from)s AND ClientTime <= %(to)s AND {condition}' + ' ORDER BY ClientTime LIMIT 1', + {'from': dt_from, 'to': dt_to}) + if not rows: + break + + row = rows[0] + + ranges[len(ranges) - 1][range_idx] = row[0] + if range_idx == 1: + ranges.append([None, None]) + + dt_from = row[0] + + if ranges[len(ranges)-1][0] is None: + ranges.pop() + + return ranges + + def get_grid_connected_intervals(self, dt_from: datetime, dt_to: datetime) -> IntervalList: + return self.get_intervals_by_condition(dt_from, dt_to, 'GridFrequency > 0', 'GridFrequency = 0') + + def get_grid_used_intervals(self, dt_from: datetime, dt_to: datetime) -> IntervalList: + return self.get_intervals_by_condition(dt_from, + dt_to, + "LinePowerDirection = 'Input'", + "LinePowerDirection != 'Input'") + + def get_grid_consumed_energy(self, dt_from: datetime, dt_to: datetime) -> float: + PrevData = namedtuple('PrevData', 'time, pd, bat_chg, bat_dis, wh') + + ws = 0 # watt-seconds + amps = 0 # amper-seconds + + intervals = self.get_grid_used_intervals(dt_from, dt_to) + for dt_start, dt_end in intervals: + fields = ', '.join([ + 'ClientTime', + 'DCACPowerDirection', + 'BatteryChargingCurrent', + 'BatteryDischargingCurrent', + 'ACOutputActivePower' + ]) + rows = self.query(f'SELECT {fields} FROM status' + ' WHERE ClientTime >= %(from)s AND ClientTime < %(to)s ORDER BY ClientTime', + {'from': dt_start, 'to': dt_end}) + + prev = PrevData(time=None, pd=None, bat_chg=None, bat_dis=None, wh=None) + for ct, pd, bat_chg, bat_dis, wh in rows: + if prev.time is not None: + n = (ct-prev.time).total_seconds() + ws += prev.wh * n + + if pd == 'DC/AC': + amps -= prev.bat_dis * n + elif pd == 'AC/DC': + amps += prev.bat_chg * n + + prev = PrevData(time=ct, pd=pd, bat_chg=bat_chg, bat_dis=bat_dis, wh=wh) + + amps /= 3600 + wh = ws / 3600 + wh += amps*48 + + return wh |