|
|
@@ -1,33 +1,131 @@
|
|
1
|
|
-from util import getQuery, pickleQuery, getkwh
|
|
|
1
|
+from argparse import ArgumentParser
|
|
|
2
|
+# from psycopg2 import sql
|
|
|
3
|
+import gc
|
|
|
4
|
+from util import getQuery, datevalid
|
|
2
|
5
|
import pandas as p
|
|
3
|
6
|
import gc
|
|
4
|
7
|
from datetime import datetime
|
|
5
|
|
-from tqdm import tqdm
|
|
6
|
|
-
|
|
7
|
|
-months = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']
|
|
8
|
|
-mstarts = list(range(1, 13))
|
|
9
|
|
-mends = mstarts[1:13]
|
|
10
|
|
-mends.append(1)
|
|
11
|
|
-yends = [2017] * 11
|
|
12
|
|
-yends.append(2018)
|
|
13
|
|
-
|
|
14
|
|
-for i, m in tqdm(enumerate(months)):
|
|
15
|
|
- # if i < 11:
|
|
16
|
|
- # continue
|
|
17
|
|
- print(m)
|
|
18
|
|
- print(datetime.now().time())
|
|
19
|
|
- kwhdata = getkwh('2017-{:02d}-01'.format(mstarts[i]),
|
|
20
|
|
- '{}-{:02d}-01'.format(yends[i], mends[i]),
|
|
21
|
|
- '2017-{:02d}-01 00:30:00'.format(mstarts[i]),
|
|
22
|
|
- '{}-{:02d}-01 00:00:00'.format(yends[i], mends[i]),
|
|
23
|
|
- '%%1')
|
|
24
|
|
- print("Pivoting")
|
|
25
|
|
- kwhpiv = kwhdata.pivot(index = 'read_time', columns = 'icp_id', values = 'kwh_tot')
|
|
26
|
|
- print("Pickling")
|
|
27
|
|
- kwhpiv.to_pickle('../data/2017-{}-5k.pkl'.format(m))
|
|
28
|
|
- del kwhdata
|
|
29
|
|
- del kwhpiv
|
|
30
|
|
- gc.collect()
|
|
31
|
|
-
|
|
32
|
|
-print('Done')
|
|
|
8
|
+from tqdm import tqdm, trange
|
|
|
9
|
+from pprint import pprint
|
|
|
10
|
+from tempfile import TemporaryDirectory
|
|
|
11
|
+import numpy as np
|
|
|
12
|
+
|
|
|
13
|
+tables = [
|
|
|
14
|
+ 'public.best_icp', # All icps with at least 360 days of data in 2017
|
|
|
15
|
+ 'public.best_icp_1618', # All icps with at least 720 days of data in 2 years from 1 April 2016
|
|
|
16
|
+ 'public.best_icp_18m', # All icps with at least 540 days of data from July 2016 to end of 2017
|
|
|
17
|
+ 'public.icp_sample', # A pre-generated 1k sample from best_icp
|
|
|
18
|
+ 'public.icp_sample_5k', # A pre-generated 5k sample from best_icp
|
|
|
19
|
+ 'public.icp_sample_1618', # A pre-generated 1k sample from best_icp_1618
|
|
|
20
|
+ 'public.icp_sample_18m' # A pre-generated 1k sample from best_icp_18m
|
|
|
21
|
+]
|
|
|
22
|
+
|
|
|
23
|
+
|
|
|
24
|
+def getkwh(datestart, dateend, timestart, timeend, icp_tab, verbose = True):
|
|
|
25
|
+ """Get kwh data from database
|
|
|
26
|
+ """
|
|
|
27
|
+ query = """
|
|
|
28
|
+ SELECT SUBSTRING(comb.icp_id FROM 2 FOR 6)::int AS icp_id, comb.read_time, COALESCE(kwh_tot, 0) AS kwh_tot
|
|
|
29
|
+ FROM
|
|
|
30
|
+ (
|
|
|
31
|
+ SELECT read_time, icp_id
|
|
|
32
|
+ FROM
|
|
|
33
|
+ (
|
|
|
34
|
+ SELECT read_time
|
|
|
35
|
+ FROM GENERATE_SERIES(%(tsstart)s::timestamp, %(tsend)s::timestamp,
|
|
|
36
|
+ '30 minutes'::interval) read_time
|
|
|
37
|
+ ) AS tsdata CROSS JOIN {}
|
|
|
38
|
+ ) AS comb
|
|
|
39
|
+ LEFT JOIN
|
|
|
40
|
+ (
|
|
|
41
|
+ SELECT *, read_date + CONCAT(period / 2, ':', period %% 2 * 30, ':00')::time AS read_time
|
|
|
42
|
+ FROM (
|
|
|
43
|
+ SELECT a.icp_id
|
|
|
44
|
+ , a.read_date
|
|
|
45
|
+ , c.period
|
|
|
46
|
+ , sum(c.read_kwh) as kwh_tot
|
|
|
47
|
+ , sum(case when a.content_code = 'UN' then c.read_kwh else 0 end) as kwh_un
|
|
|
48
|
+ , sum(case when a.content_code in ('CN','EG') then c.read_kwh else 0 end) as kwh_cn
|
|
|
49
|
+ FROM coup_prd.coupdatamaster a,
|
|
|
50
|
+ unnest(a.read_array) WITH ORDINALITY c(read_kwh, period)
|
|
|
51
|
+ WHERE a.read_date >= to_date(%(datestart)s,'yyyy-mm-dd')
|
|
|
52
|
+ and a.read_date < to_date(%(dateend)s,'yyyy-mm-dd')
|
|
|
53
|
+ and a.content_code ~ ('UN|CN|EG')
|
|
|
54
|
+ AND a.icp_id IN (
|
|
|
55
|
+ SELECT icp_id FROM {}
|
|
|
56
|
+ )
|
|
|
57
|
+ GROUP BY 1, 2, 3
|
|
|
58
|
+ ) AS coup_tall
|
|
|
59
|
+ ) AS tall_timestamp
|
|
|
60
|
+ ON comb.read_time = tall_timestamp.read_time AND comb.icp_id = tall_timestamp.icp_id;
|
|
|
61
|
+ """
|
|
|
62
|
+ query = query.format(icp_tab, icp_tab)
|
|
|
63
|
+ pdict = {
|
|
|
64
|
+ 'datestart': datestart,
|
|
|
65
|
+ 'dateend': dateend,
|
|
|
66
|
+ 'tsstart': timestart,
|
|
|
67
|
+ 'tsend': timeend
|
|
|
68
|
+ # 'subset': subset
|
|
|
69
|
+ }
|
|
|
70
|
+
|
|
|
71
|
+ if verbose:
|
|
|
72
|
+ print("Getting data with parameters:")
|
|
|
73
|
+ pprint(pdict)
|
|
|
74
|
+ qdf = getQuery(query, pdict, verbose)
|
|
|
75
|
+ if verbose:
|
|
|
76
|
+ print("Optimising")
|
|
|
77
|
+ qdf['icp_id'] = qdf['icp_id'].astype(np.int32)
|
|
|
78
|
+ qdf['kwh_tot'] = qdf['kwh_tot'].astype(np.float16)
|
|
|
79
|
+ # print("Done")
|
|
|
80
|
+ return(qdf)
|
|
|
81
|
+
|
|
|
82
|
+
|
|
|
83
|
+def collateddownload(startd, endd, numdivis, icp_tab, pivot, verbose):
|
|
|
84
|
+ """
|
|
|
85
|
+ Download dataset in pieces, then combine
|
|
|
86
|
+ """
|
|
|
87
|
+ with TemporaryDirectory() as tempdir:
|
|
|
88
|
+ divset = p.date_range(startd, endd, periods = numdivis + 1).strftime("%Y-%m-%d")
|
|
|
89
|
+ divlow = divset[:-1]
|
|
|
90
|
+ divhig = divset[1:]
|
|
|
91
|
+ for i in trange(numdivis):
|
|
|
92
|
+ datestart = divlow[i]
|
|
|
93
|
+ dateend = divhig[i]
|
|
|
94
|
+ datetimeend = dateend + " 00:00:00"
|
|
|
95
|
+ datetimestart = datestart + " 00:30:00"
|
|
|
96
|
+ filename = "{}/{}temp.pkl".format(tempdir, i)
|
|
|
97
|
+ tdf = getkwh(datestart, dateend, datetimestart, datetimeend, icp_tab, verbose)
|
|
|
98
|
+ if pivot:
|
|
|
99
|
+ if verbose:
|
|
|
100
|
+ print("Pivoting")
|
|
|
101
|
+ tdf = tdf.pivot(index = 'read_time', columns = 'icp_id', values = 'kwh_tot')
|
|
|
102
|
+ tdf.to_pickle(filename)
|
|
|
103
|
+ del tdf
|
|
|
104
|
+ coldf = p.read_pickle("{}/{}temp.pkl".format(tempdir, 0))
|
|
|
105
|
+ for i in trange(1, numdivis):
|
|
|
106
|
+ filename = "{}/{}temp.pkl".format(tempdir, i)
|
|
|
107
|
+ tdf = p.read_pickle(filename)
|
|
|
108
|
+ coldf = p.concat([coldf, tdf])
|
|
|
109
|
+ del tdf
|
|
|
110
|
+ gc.collect()
|
|
|
111
|
+ return coldf
|
|
|
112
|
+
|
|
|
113
|
+
|
|
|
114
|
+def main():
|
|
|
115
|
+ parser = ArgumentParser(description='Download kwh data from dataframe')
|
|
|
116
|
+ parser.add_argument("-o", "--output", dest="output", help = "output pickle path; default: ../data/2017-5k-wide.pkl", metavar="[PATH]", default = "../data/2017-5k-wide.pkl")
|
|
|
117
|
+ parser.add_argument("-s", "--start-date", dest = "startdate", help = "start date for download; format: YYYY-MM-DD; default: 2017-01-01", metavar="[DATE]", default = "2017-01-01", type = datevalid)
|
|
|
118
|
+ parser.add_argument("-e", "--end-date", dest = "enddate", help = "end date for download; format: YYYY-MM-DD; default: 2018-01-01", metavar="[DATE]", default = "2018-01-01", type = datevalid)
|
|
|
119
|
+ parser.add_argument("-t", "--table", dest = "table", help = "table for download (constrained to specific values in source); default: public.icp_sample", metavar="[TABLE]", default = "public.icp_sample", choices = tables)
|
|
|
120
|
+ parser.add_argument("-n", "--num-div", dest="numdiv", help = "number of segments to divide download into", metavar = "[NUM]", default = 12, type = int)
|
|
|
121
|
+ parser.add_argument("--no-pivot", dest = "pivot", help = "output dataframe in tall (non-pivoted) format", action ="store_false")
|
|
|
122
|
+ parser.add_argument("-v", "--verbose", dest = "verbose", action ="store_true")
|
|
|
123
|
+ args = parser.parse_args()
|
|
|
124
|
+ cdata = collateddownload(args.startdate, args.enddate, args.numdiv, args.table, args.pivot, args.verbose)
|
|
|
125
|
+ cdata.to_pickle(args.output)
|
|
|
126
|
+
|
|
|
127
|
+
|
|
|
128
|
+
|
|
33
|
129
|
|
|
|
130
|
+if __name__ == "__main__":
|
|
|
131
|
+ main()
|
