-
Notifications
You must be signed in to change notification settings - Fork 0
/
clustering_test.py
90 lines (78 loc) · 3.88 KB
/
clustering_test.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
import unittest
import clustering as c
import api as a
import os
import datetime
from clustering import cluster_ts
import matplotlib.pyplot as plt
import pandas as pd
import plotly.express as px
MAX_TEMP = 50
MIN_TEMP = -50
class TestCluster(unittest.TestCase):
def plot_cluster(self, cluster, sensor_list, vector):
n_clusters = len(cluster)
fig, axs = plt.subplots(n_clusters, sharex=True, sharey=True)
for i in range(len(cluster)):
c = cluster[i][1]
for id in c['sensor_ids']:
axs[i].plot(list(sensor_list[id].dataFrame[vector]), linewidth=1)
axs[i].plot(c['centroid'], linewidth=3)
axs[i].grid()
plt.show()
def test_austria_sensors(self):
from_time = datetime.datetime(2020, 3, 1)
to_time = datetime.datetime(2020, 3, 3)
sensors = a.get_sensors("AUT", "Steiermark", "Graz", return_sensors=True)
sensor_list = a.download_sensors(sensors, from_time, to_time)
result = cluster_ts(sensor_list, 'temperature', 7, MIN_TEMP, MAX_TEMP)
self.assertEqual(len(result), 7)
# self.plot_cluster(result, sensor_list, 'temperature')
def test_berlin_sensors(self):
from_time = datetime.datetime(2021, 3, 1)
to_time = datetime.datetime(2021, 3, 3)
sensors = a.get_sensors("DEU", "Berlin", return_sensors=True)
sensor_list = a.download_sensors(sensors, from_time, to_time)
result = cluster_ts(sensor_list, 'temperature', 6, MIN_TEMP, MAX_TEMP)
self.assertEqual(len(result), 6)
# self.plot_cluster(result, sensor_list, 'temperature')
def test_world_sensors(self):
from_time = datetime.datetime(2021, 3, 1)
to_time = datetime.datetime(2021, 3, 3)
sensors = a.get_sensors(return_sensors=True, num_sensors=1)
sensor_list = a.download_sensors(sensors, from_time, to_time)
result = cluster_ts(sensor_list, 'temperature', 6, MIN_TEMP, MAX_TEMP)
self.assertEqual(len(sensors), 80)
# self.plot_cluster(result, sensor_list, 'temperature')
def test_germany_regional_sensors(self):
from_time = datetime.datetime(2021, 3, 1)
to_time = datetime.datetime(2021, 3, 3)
sensors = a.get_sensors("DEU", return_sensors=True, num_sensors=1)
sensor_list = a.download_sensors(sensors, from_time, to_time)
result = cluster_ts(sensor_list, 'temperature', 4, MIN_TEMP, MAX_TEMP)
self.assertEqual(len(sensors), 16)
# self.plot_cluster(result, sensor_list, 'temperature')
def test_all_states(self):
from_time = datetime.datetime(2021, 3, 1)
to_time = datetime.datetime(2021, 3, 3)
sensors = a.get_state_sensors(num_cities=1, type=['bme280', 'dht22', 'bmp280'])
sensor_list = a.download_sensors(sensors, from_time, to_time)
result = cluster_ts(sensor_list, 'temperature', 5, MIN_TEMP, MAX_TEMP)
# self.plot_cluster(result, sensor_list, 'temperature')
def test_all_cities(self):
from_time = datetime.datetime(2021, 3, 1)
to_time = datetime.datetime(2021, 3, 3)
sensors = a.get_sensors(return_sensors=True, sensor_per_city=1, type=['bme280', 'dht22', 'bmp280'])
sensor_list = a.download_sensors(sensors, from_time, to_time)
result = cluster_ts(sensor_list, 'temperature', 5, MIN_TEMP, MAX_TEMP)
# self.plot_cluster(result, sensor_list, 'temperature')
def test_ita_ger_aut(self):
from_time = datetime.datetime(2021, 3, 1)
to_time = datetime.datetime(2021, 3, 3)
sensors = a.get_state_sensors(num_cities=1, countries=["DEU", "AUT", "ITA"])
sensor_list = a.download_sensors(sensors, from_time, to_time)
result = cluster_ts(sensor_list, 'temperature', 5, MIN_TEMP, MAX_TEMP)
self.assertEqual(len(sensors), 43)
# self.plot_cluster(result, sensor_list, 'temperature')
if __name__ == '__main__':
unittest.main()