1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
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)
|
