-
Notifications
You must be signed in to change notification settings - Fork 2
/
delete_telemetry.py
77 lines (69 loc) · 2.85 KB
/
delete_telemetry.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
"""Delete telemetry from Thingsboard for the specified devices, keys and time.
"""
#import io
#import json
import sys
import json
from queue import Queue
import download_new as dwn
import tb_rest as tb
import datetime as dt
TIME_FORMAT = "%d.%m.%Y %H:%M:%S"
MAX_TRY = 50
def delete_all_data(tb_connection, devices, keys, start_time, end_time):
"""
devices - {'name':'id',...}
"""
device_queue = Queue()
for name, _id in devices.items():
elem = {'name': name, 'id': _id, 'num_try': 0}
device_queue.put(elem)
done = False
while not done:
if not device_queue.empty():
elem = device_queue.get()
print(f"Delete telemetry for {elem['name']}. \
From {start_time.strftime(TIME_FORMAT)} to {end_time.strftime(TIME_FORMAT)}...", end = " ")
resp = tb.delete_telemetry(tb_connection.url, tb_connection.get_token(), elem['id'], keys, start_time, end_time)
if tb.request_success(resp):
print("Success")
elif resp.status_code == 204:
print("No data")
else:
print("ERROR")
print(f"Code={resp.status_code}")
try :
print(resp.json())
except:
print(resp.content)
if elem['num_try'] < MAX_TRY:
elem['num_try'] += 1
device_queue.put(elem)
else:
print(f"Max number of tries={elem['num_try']} reached for {elem['name']}")
else:
done = True
def get_config_params(config_file):
with open(config_file, 'r') as f:
params = json.load(f)
params['start_time'] = dt.datetime.strptime(params['start_time'], TIME_FORMAT)
if params['end_time']:
params['end_time'] = dt.datetime.strptime(params['end_time'], TIME_FORMAT)
else:
params['end_time'] = dt.datetime.now()
return params
if __name__ == "__main__":
params = get_config_params(sys.argv[1])
if "password" not in params.keys():
params['password'] = input("Enter password: ")
tb_connection = tb.TbConnection(params['url'], params['user'], params['password'])
if 'device_type' in params.keys():#if device type is specified, ignore the specified devices
devices, resp = tb.get_tenant_devices(tb_connection.url, tb_connection.get_token(), params['device_type'])
if not tb.request_success(resp):
raise tb.ConnectionError(resp)
params['devices'] = dwn.name_id_dict(devices)
print("*** Purge started ***")
print(f"Total interval: {params['start_time'].strftime(TIME_FORMAT)}-{params['end_time'].strftime(TIME_FORMAT)}")
delete_all_data(tb_connection, params['devices'], params['keys'],
params['start_time'], params['end_time'])
print("*** Purge finished ***")