-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathresponse.py
More file actions
107 lines (82 loc) · 3 KB
/
Copy pathresponse.py
File metadata and controls
107 lines (82 loc) · 3 KB
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
import pandas as pd
from operator import itemgetter
import json
class Response:
"""
A class, that handles the responses done by the route. This is the base
class that returns a generic json
"""
def __init__(self, return_value=None, *args, **kwargs):
self.raw_out = return_value
self.df = None
def to_json(self, filename=None, path='./', safe=False):
if safe:
if filename is None or path is None:
raise ValueError("If you want to save to .json please provide "
"a filename")
else:
with open(path + filename + '.json', 'w') as out_json:
json.dump(self.to_json(), out_json)
return self.raw_out
def to_dataframe(self):
raise NotImplementedError
def to_csv(self, filename, path='./'):
raise NotImplementedError
def to_pickle(self, filename: str, path: str = './'):
raise NotImplementedError
def __repr__(self):
return str(self.raw_out)
def __str__(self):
return str(self.raw_out)
def __iter__(self):
return ResponseIterator(self.raw_out)
def __getitem__(self, item):
return self.raw_out[item]
class ValueResponse(Response):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def to_csv(self, filename: str, path: str = './'):
if self.df is None:
self.df = self.to_dataframe()
file = path + filename + '.csv'
self.df.to_csv(file)
def to_pickle(self, filename: str, path: str = './'):
if self.df is None:
self.df = self.to_dataframe()
file = path + filename + '.pickle'
self.df.to_pickle(file)
def to_dataframe(self):
# for ind in range(len(self.raw_out)):
# col_name = self.raw_out[ind]['objectId']
rows = self.get_timestamp_set()
columns = self.get_columns()
self.df = pd.DataFrame(index=rows, columns=columns)
for obj in self.raw_out:
self.add_object_data(obj)
return self.df
def add_object_data(self, obj):
obj_name = obj['objectId']
values = sorted(obj['values'], key=itemgetter('timestamp'))
for value in values:
self.df.at[value['timestamp'], obj_name] = value['value']
def get_columns(self):
return [obj['objectId'] for obj in self.raw_out]
def get_timestamp_set(self):
# [x for b in a for x in b]
timestamps = sorted(
list(set([d['timestamp'] for obj in self.raw_out
for d in obj['values'] if 'timestamp' in d])))
return timestamps
class ResponseIterator:
"""
Iterator for the Response-object
"""
def __init__(self, raw_out):
self._raw_out = raw_out
self._index = 0
def __next__(self):
if self._index < len(self._raw_out):
result = self._raw_out[self._index]
self._index += 1
return result
raise StopIteration