-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathDataExport.py
198 lines (177 loc) · 12.3 KB
/
DataExport.py
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
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
#!python3
"""
Project: Voltcraft Data Analyzer
Author: Valer Bocan, PhD <[email protected]>
Last updated: September 14th, 2014
Module
description: The VoltcraftDataFile module processes data files containing history of voltage, current and power factor,
as generated by the Voltcraft Energy-Logger 4000.
License: This project is placed in the public domain, hoping that it will be useful to people tinkering with Voltcraft products.
Reference: Voltcraft File Format: http://www2.produktinfo.conrad.com/datenblaetter/125000-149999/125323-da-01-en-Datenprotokoll_SD_card_file_Formatv1_2.pdf
"""
import csv
from datetime import timedelta
from datetime import datetime
def WriteInfoData(filename, info, powerdata, blackoutdata):
"""
Write informational data to a text file
"""
try:
with open(filename, "wt") as fout:
fout.write("Voltcraft Data Analyzer v1.2\n")
fout.write("Valer Bocan, PhD <[email protected]>\n")
fout.write("\n")
fout.write("Initial time on device: {0}\n".format(info["InitialDateTime"]))
fout.write("Unit number: {0}\n".format(info["UnitNumber"]))
fout.write("\n")
fout.write("--- DEVICE BASED STATISTICS (may not be accurate)\n")
fout.write("Total power consumed: {0:.3f} kWh\n".format(info["TotalPowerConsumed"]))
fout.write("History:\n")
fout.write(" Today: {0:.3f} kWh\n".format(info["ConsumptionHistory"][0]))
fout.write(" Yesterday: {0:.3f} kWh\n".format(info["ConsumptionHistory"][1]))
fout.write(" 2 days ago: {0:.3f} kWh\n".format(info["ConsumptionHistory"][2]))
fout.write(" 3 days ago: {0:.3f} kWh\n".format(info["ConsumptionHistory"][3]))
fout.write(" 4 days ago: {0:.3f} kWh\n".format(info["ConsumptionHistory"][4]))
fout.write(" 5 days ago: {0:.3f} kWh\n".format(info["ConsumptionHistory"][5]))
fout.write(" 6 days ago: {0:.3f} kWh\n".format(info["ConsumptionHistory"][6]))
fout.write(" 7 days ago: {0:.3f} kWh\n".format(info["ConsumptionHistory"][7]))
fout.write(" 8 days ago: {0:.3f} kWh\n".format(info["ConsumptionHistory"][8]))
fout.write(" 9 days ago: {0:.3f} kWh\n".format(info["ConsumptionHistory"][9]))
fout.write("\n")
fout.write("Total recorded time: {0}\n".format(GetDurationStringFromMinutes(info["TotalRecordedTime"])))
fout.write("History:\n")
fout.write(" Today: {0}\n".format(GetDurationStringFromHours(info["RecordedTimeHistory"][0])))
fout.write(" Yesterday: {0}\n".format(GetDurationStringFromHours(info["RecordedTimeHistory"][1])))
fout.write(" 2 days ago: {0}\n".format(GetDurationStringFromHours(info["RecordedTimeHistory"][2])))
fout.write(" 3 days ago: {0}\n".format(GetDurationStringFromHours(info["RecordedTimeHistory"][3])))
fout.write(" 4 days ago: {0}\n".format(GetDurationStringFromHours(info["RecordedTimeHistory"][4])))
fout.write(" 5 days ago: {0}\n".format(GetDurationStringFromHours(info["RecordedTimeHistory"][5])))
fout.write(" 6 days ago: {0}\n".format(GetDurationStringFromHours(info["RecordedTimeHistory"][6])))
fout.write(" 7 days ago: {0}\n".format(GetDurationStringFromHours(info["RecordedTimeHistory"][7])))
fout.write(" 8 days ago: {0}\n".format(GetDurationStringFromHours(info["RecordedTimeHistory"][8])))
fout.write(" 9 days ago: {0}\n".format(GetDurationStringFromHours(info["RecordedTimeHistory"][9])))
fout.write("\n")
fout.write("Total time with power consumption: {0}\n".format(GetDurationStringFromMinutes(info["TotalOnTime"])))
fout.write("History:\n")
fout.write(" Today: {0}\n".format(GetDurationStringFromHours(info["OnTimeHistory"][0])))
fout.write(" Yesterday: {0}\n".format(GetDurationStringFromHours(info["OnTimeHistory"][1])))
fout.write(" 2 days ago: {0}\n".format(GetDurationStringFromHours(info["OnTimeHistory"][2])))
fout.write(" 3 days ago: {0}\n".format(GetDurationStringFromHours(info["OnTimeHistory"][3])))
fout.write(" 4 days ago: {0}\n".format(GetDurationStringFromHours(info["OnTimeHistory"][4])))
fout.write(" 5 days ago: {0}\n".format(GetDurationStringFromHours(info["OnTimeHistory"][5])))
fout.write(" 6 days ago: {0}\n".format(GetDurationStringFromHours(info["OnTimeHistory"][6])))
fout.write(" 7 days ago: {0}\n".format(GetDurationStringFromHours(info["OnTimeHistory"][7])))
fout.write(" 8 days ago: {0}\n".format(GetDurationStringFromHours(info["OnTimeHistory"][8])))
fout.write(" 9 days ago: {0}\n".format(GetDurationStringFromHours(info["OnTimeHistory"][9])))
fout.write("\n")
fout.write("Tariff 1: {0}\n".format(info["Tariff1"]))
fout.write("Tariff 2: {0}\n".format(info["Tariff2"]))
fout.write("\n")
fout.write("--- PARAMETER HISTORY\n")
for d in powerdata:
fout.write("[{0}] U={1:02}V I={2:.3f}A cosPHI={3:.2f} P={4:.3f}kW S={4:.3f}kVA\n".format(d["Timestamp"].strftime("%Y-%m-%d %H:%M"), d["Voltage"], d["Current"], d["PowerFactor"], d["Power"], d["ApparentPower"]))
stats1 = GetDataStatistics(powerdata)
fout.write("\n")
fout.write("--- VOLTAGE AND POWER\n")
fout.write("Minimum voltage : {0:.1f}V ({1} occurences, first on {2})\n".format(stats1["MinVoltage"], len(stats1["MinVoltageTimestamps"]), stats1["MinVoltageTimestamps"][0].strftime("%Y-%m-%d %H:%M")))
fout.write("Maximum voltage : {0:.1f}V ({1} occurences, first on {2})\n".format(stats1["MaxVoltage"], len(stats1["MaxVoltageTimestamps"]), stats1["MaxVoltageTimestamps"][0].strftime("%Y-%m-%d %H:%M")))
fout.write("Average voltage : {0:.1f}V\n".format(stats1["AvgVoltage"]))
fout.write("Maximum power : {0:.3f}kW ({1} occurences, first on {2})\n".format(stats1["MaxPower"], len(stats1["MaxPowerTimestamps"]), stats1["MaxPowerTimestamps"][0].strftime("%Y-%m-%d %H:%M")))
fout.write("Maximum apparent power : {0:.3f}kVA ({1} occurences, first on {2})\n".format(stats1["MaxApparentPower"], len(stats1["MaxApparentPowerTimestamps"]), stats1["MaxApparentPowerTimestamps"][0].strftime("%Y-%m-%d %H:%M")))
stats2 = GetBlackoutStatistics(blackoutdata)
fout.write("\n")
fout.write("--- BLACKOUTS\n")
fout.write("{0} blackout(s) for a total of {1}\n".format(stats2["Count"], GetDurationString(stats2["TotalDuration"])))
for b in blackoutdata:
fout.write("[{0}] Blackout for {1}\n".format(b["Timestamp"].strftime("%Y-%m-%d %H:%M"), GetDurationString(b["Duration"])))
fout.write("\n")
stats3 = list(GetPowerStatistics(powerdata))
fout.write("--- POWER CONSUMPTION\n")
for c in stats3:
fout.write("[{0}] - {1:.3f}kWh\n".format(c["Day"].strftime("%Y-%m-%d"), c["Consumption"]))
fout.write(" Recorded: {0}\n".format(GetDurationStringFromMinutes(c["TotalMinutes"])))
fout.write(" Power on: {0} ({1:.1f}%)\n".format(GetDurationStringFromMinutes(c["TotalMinutesWithPowerConsumption"]), c["TotalMinutesWithPowerConsumption"] / c["TotalMinutes"] * 100))
TotalPowerConsumption = sum(item['Consumption'] for item in stats3)
fout.write("\nTOTAL CONSUMPTION : {0:.3f}kWh (avg. {1:.3f}kWh/day)\n".format(TotalPowerConsumption, TotalPowerConsumption / len(stats3)))
TotalRecordedTime = len(powerdata) # minutes
TotalTimeWithPowerConsumption = len(tuple(item for item in powerdata if item['Power'] > 0))
fout.write("Total recorded time : {0}\n".format(GetDurationStringFromMinutes(TotalRecordedTime)))
fout.write("Total time with power consumption : {0} ({1:.1f}%)\n".format(GetDurationStringFromMinutes(TotalTimeWithPowerConsumption), TotalTimeWithPowerConsumption / TotalRecordedTime * 100 ))
fout.write("\nFile generated on: {0}\n".format(datetime.now().strftime("%Y-%m-%d %H:%M:%S")))
except IOError:
raise Exception('Could not write out information file.')
def WriteHistoricData(filename, data):
"""
Write historic data to a CSV file
"""
with open(filename, 'w', newline='') as fp:
wr = csv.writer(fp, delimiter=';')
header = [['Timestamp', 'Voltage (V)', 'Current (A)', 'Power (kW)', 'Apparent power (kVA)']]
wr.writerows(header) # Write header
for d in data:
str = [[d["Timestamp"], d["Voltage"], d["Current"], round(d["Power"], 3), round(d["ApparentPower"], 3)]]
wr.writerows(str)
def WriteBlackoutData(filename, data):
"""
Write blackout data to a CSV file
"""
with open(filename, 'w', newline='') as fp:
wr = csv.writer(fp, delimiter=';')
header = [['Timestamp', 'Duration']]
wr.writerows(header) # Write header
for d in data:
str = [[d["Timestamp"], d["Duration"]]]
wr.writerows(str)
def GetDurationString(duration):
"""
Convert the duration timedelta in a day:hour:min:sec string representation
"""
total_days, total_hours, total_minutes = duration.days, duration.seconds // 3600, duration.seconds // 60 % 60
if total_days >= 1:
return "{0:02}d {1:02}h {2:02}m".format(total_days, total_hours, total_minutes)
elif total_hours >= 1:
return "{0:02}h {1:02}m".format(total_hours, total_minutes)
else:
return "{0:02}m".format(total_minutes)
def GetDurationStringFromMinutes(duration):
return GetDurationString(timedelta(minutes=duration))
def GetDurationStringFromHours(duration):
return GetDurationString(timedelta(hours=duration))
def GetDataStatistics(data):
# Compute minimum voltage and its occurence times
MinVoltage = min(item['Voltage'] for item in data)
MinVoltageTimestamps = tuple(item['Timestamp'] for item in data if item["Voltage"] == MinVoltage)
# Compute maximum voltage and its occurence times
MaxVoltage = max(item['Voltage'] for item in data)
MaxVoltageTimestamps = tuple(item['Timestamp'] for item in data if item["Voltage"] == MaxVoltage)
# Compute average voltage
AvgVoltage = sum(item['Voltage'] for item in data) / len(data)
# Compute maximum power and its occurence times
MaxPower = max(item['Power'] for item in data)
MaxPowerTimestamps = tuple(item['Timestamp'] for item in data if item["Power"] == MaxPower)
MaxApparentPower = max(item['ApparentPower'] for item in data)
MaxApparentPowerTimestamps = tuple(item['Timestamp'] for item in data if item["ApparentPower"] == MaxApparentPower)
return {
"AvgVoltage": AvgVoltage,
"MinVoltage":MinVoltage,
"MinVoltageTimestamps":MinVoltageTimestamps,
"MaxVoltage":MaxVoltage,
"MaxVoltageTimestamps":MaxVoltageTimestamps,
"MaxPower":MaxPower,
"MaxPowerTimestamps":MaxPowerTimestamps,
"MaxApparentPower":MaxApparentPower,
"MaxApparentPowerTimestamps":MaxApparentPowerTimestamps
}
def GetBlackoutStatistics(data):
Count = len(data)
TotalDuration = sum((item['Duration'] for item in data), timedelta())
return {"Count":Count, "TotalDuration":TotalDuration}
def GetPowerStatistics(data):
# Determine the unique days in the data log
UniqueOrdinalDays = set(item['Timestamp'].toordinal() for item in data)
for day in sorted(UniqueOrdinalDays):
ConsumptionPerDay = sum(item['Power'] * 1 / 60 for item in data if item['Timestamp'].toordinal() == day)
TotalMinutes = len(tuple(item for item in data if item['Timestamp'].toordinal() == day))
TotalMinutesWithPowerConsumption = len(tuple(item for item in data if item['Timestamp'].toordinal() == day and item['Power'] > 0))
Day = datetime.fromordinal(day)
yield { "Day" : Day, "Consumption" : ConsumptionPerDay, "TotalMinutes": TotalMinutes, "TotalMinutesWithPowerConsumption": TotalMinutesWithPowerConsumption }