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
168
169
170
171
|
from dataclasses import dataclass
import pymysql
import os
@dataclass
class PowerDatabase:
host: str = None
user: str = "root"
passwd: str = None
db: str = "power"
port: int = 3306
def __enter__(self):
if self.passwd is None:
self.passwd = os.environ["MYSQL_ROOT_PASSWORD"]
if self.host is None:
self.host = os.environ["MYSQL_HOST"]
try:
self.__connection = self.__get_connection()
except Exception as e:
print(e)
if e.args[0] == 1049:
self.__connection = self.__build_db()
elif e.args[0] == 2003:
raise ConnectionError(e.args[1])
with self.__connection.cursor() as cursor:
if "TASMOTA_DEVICES" in os.environ.keys():
for host, username, password, description in self.get_tasmota_devices():
cursor.execute("""
INSERT INTO tasmota_devices (host, username, password)
VALUES (%s, %s, %s)
ON DUPLICATE KEY UPDATE username = %s, password = %s;
""", (host, username, password, username, password))
return self
def __exit__(self, type, value, traceback):
self.__connection.commit()
self.__connection.close()
def __get_connection(self):
return pymysql.connect(
host = self.host,
port = self.port,
user = self.user,
passwd = self.passwd,
charset = "utf8mb4",
database = self.db
)
def __build_db(self):
print("Building database...")
self.__connection = pymysql.connect(
host = self.host,
port = self.port,
user = self.user,
passwd = self.passwd,
charset = "utf8mb4",
)
with self.__connection.cursor() as cursor:
# unsafe:
cursor.execute("CREATE DATABASE %s" % self.db)
cursor.execute("USE %s" % self.db)
cursor.execute("""
CREATE TABLE tasmota_devices (
host VARCHAR(25) NOT NULL PRIMARY KEY,
username VARCHAR(50) NOT NULL,
password VARCHAR(50) NOT NULL
);
""")
cursor.execute("""
CREATE TABLE watt_readings (
host VARCHAR(25) NOT NULL,
`datetime` DATETIME DEFAULT NOW(),
reading FLOAT(24) NOT NULL,
FOREIGN KEY (host) REFERENCES tasmota_devices (host),
PRIMARY KEY (host, `datetime`)
);
""")
cursor.execute("""
CREATE TABLE kwh_readings (
host VARCHAR(25) NOT NULL,
`datetime` DATETIME DEFAULT NOW(),
reading FLOAT(24) NOT NULL,
FOREIGN KEY (host) REFERENCES tasmota_devices (host),
PRIMARY KEY (host, `datetime`)
);
""")
self.__connection.commit()
return self.__connection
def get_tasmota_devices(self):
o = []
for d in os.environ["TASMOTA_DEVICES"].split(";"):
line = d.split(",")
o.append(line)
return o
def append_watt_readings(self, host, reading):
with self.__connection.cursor() as cursor:
cursor.execute("DELETE FROM watt_readings WHERE `datetime` < DATE_SUB(NOW(), INTERVAL 1 DAY);")
cursor.execute("INSERT INTO watt_readings (host, reading) VALUES (%s, %s);", (host, reading))
def append_kwh_readings(self, host, reading):
with self.__connection.cursor() as cursor:
cursor.execute("INSERT INTO kwh_readings (host, reading) VALUES (%s, %s);", (host, reading))
def get_last_plug_readings(self):
plugs = [i[0] for i in self.get_tasmota_devices()]
with self.__connection.cursor() as cursor:
cursor.execute("SELECT host, MAX(datetime) FROM watt_readings WHERE host IN %s GROUP BY host;", (plugs, ))
wattplugtimes = cursor.fetchall()
cursor.execute("SELECT host, MAX(datetime) FROM kwh_readings WHERE host IN %s GROUP BY host;", (plugs, ))
kwhplugtimes = {i[0]: i[1] for i in cursor.fetchall()}
readings = {}
for host, datetime in wattplugtimes:
cursor.execute("SELECT host, datetime, reading FROM watt_readings WHERE host = %s AND datetime = %s;", (host, datetime))
o1 = cursor.fetchone()
readings[host] = {"watts": (o1[1], o1[2])}
cursor.execute("SELECT host, datetime, reading FROM kwh_readings WHERE host = %s AND datetime = %s;", (host, kwhplugtimes[host]))
o2 = cursor.fetchone()
readings[host]["kWh"] = (o2[1], o2[2])
return readings
def get_watt_chart(self):
with self.__connection.cursor() as cursor:
cursor.execute("SELECT DISTINCT host FROM watt_readings;")
hosts = [i[0] for i in cursor.fetchall()]
out = {}
for host in hosts:
cursor.execute("SELECT datetime, reading FROM watt_readings WHERE host = %s ORDER BY datetime;", (host, ))
out[host] = cursor.fetchall()
return out
def get_kwh_chart(self):
with self.__connection.cursor() as cursor:
cursor.execute("SELECT DISTINCT host FROM kwh_readings;")
hosts = [i[0] for i in cursor.fetchall()]
out = {}
for host in hosts:
cursor.execute("SELECT datetime, reading FROM kwh_readings WHERE host = %s ORDER BY datetime;", (host, ))
out[host] = cursor.fetchall()
return out
def to_series(timeseriesforeach):
print(timeseriesforeach)
if __name__ == "__main__":
if not os.path.exists(".docker"):
import dotenv
dotenv.load_dotenv(dotenv_path = "power.env")
host = "srv.athome"
else:
host = None
with PowerDatabase(host = host) as db:
print(db.get_last_plug_readings())
|
