1 |
d6ca840d
|
petrh
|
import yaml
|
2 |
|
|
import os
|
3 |
af7609b5
|
Tomáš Ballák
|
from typing import Dict, Set
|
4 |
|
|
from shared_types import StringSetType
|
5 |
d6ca840d
|
petrh
|
from Utilities.Database import database_record_logs
|
6 |
81980e82
|
ballakt
|
from Utilities.helpers import should_skip
|
7 |
d6ca840d
|
petrh
|
|
8 |
|
|
# Path to dataset configuration files
|
9 |
|
|
CONFIG_FILES_PATH = "DatasetConfigs/"
|
10 |
|
|
# Config file type
|
11 |
|
|
CONFIG_FILE_TYPE = ".yaml"
|
12 |
|
|
|
13 |
|
|
|
14 |
af7609b5
|
Tomáš Ballák
|
def load_configuration(dataset_name: str) -> Dict[str, any]:
|
15 |
d6ca840d
|
petrh
|
"""
|
16 |
|
|
Loads yaml configuration file into memory
|
17 |
|
|
|
18 |
|
|
Args:
|
19 |
|
|
dataset_name: name of dataset that has existing configuration file
|
20 |
|
|
|
21 |
|
|
Returns:
|
22 |
|
|
yaml configuration file as dictionary
|
23 |
|
|
"""
|
24 |
|
|
with open(CONFIG_FILES_PATH + dataset_name + CONFIG_FILE_TYPE, "r") as f:
|
25 |
|
|
data = yaml.load(f, Loader=yaml.FullLoader)
|
26 |
|
|
|
27 |
af7609b5
|
Tomáš Ballák
|
devices_dic = {}
|
28 |
d6ca840d
|
petrh
|
|
29 |
|
|
if data["devices"] is not None:
|
30 |
|
|
for item in data["devices"]:
|
31 |
|
|
devices_dic.update(item)
|
32 |
|
|
|
33 |
|
|
data["devices"] = devices_dic
|
34 |
|
|
|
35 |
|
|
return data
|
36 |
|
|
|
37 |
|
|
|
38 |
af7609b5
|
Tomáš Ballák
|
def update_configuration(dataset_name: str,
|
39 |
|
|
new_devices: StringSetType) -> None:
|
40 |
d6ca840d
|
petrh
|
"""
|
41 |
|
|
Open dataset and appends new_devices to the end
|
42 |
|
|
|
43 |
|
|
Args:
|
44 |
|
|
dataset_name: name of dataset that has existing configuration file
|
45 |
|
|
new_devices: list or set of new devices for dataset
|
46 |
|
|
"""
|
47 |
|
|
|
48 |
81980e82
|
ballakt
|
with open(CONFIG_FILES_PATH + dataset_name + CONFIG_FILE_TYPE,
|
49 |
|
|
"a") as file:
|
50 |
d6ca840d
|
petrh
|
for device in new_devices:
|
51 |
81980e82
|
ballakt
|
if device == "":
|
52 |
|
|
continue
|
53 |
|
|
file.write(" - " + device + ":\n")
|
54 |
d6ca840d
|
petrh
|
file.write(" x: UNKNOWN!\n")
|
55 |
|
|
file.write(" y: UNKNOWN!\n")
|
56 |
|
|
file.write("\n")
|
57 |
|
|
|
58 |
|
|
|
59 |
af7609b5
|
Tomáš Ballák
|
def check_if_there_is_a_config_file(dataset_name: str) -> bool:
|
60 |
d6ca840d
|
petrh
|
"""
|
61 |
|
|
Goes trough all config files (represeting valid dataset in database)
|
62 |
|
|
and checks if dataset_name is there
|
63 |
|
|
|
64 |
|
|
Args:
|
65 |
|
|
dataset_name: name of dataset that has existing configuration file
|
66 |
|
|
|
67 |
|
|
Returns:
|
68 |
|
|
True - if contains
|
69 |
|
|
False - if not
|
70 |
|
|
"""
|
71 |
|
|
datasets = os.listdir(CONFIG_FILES_PATH)
|
72 |
|
|
|
73 |
|
|
for dataset in datasets:
|
74 |
|
|
name = dataset.split('.')
|
75 |
81980e82
|
ballakt
|
if name[0] == dataset_name:
|
76 |
d6ca840d
|
petrh
|
return True
|
77 |
|
|
|
78 |
|
|
return False
|
79 |
|
|
|
80 |
81980e82
|
ballakt
|
|
81 |
af7609b5
|
Tomáš Ballák
|
def return_dictionary_of_valid_devices(
|
82 |
|
|
devices: Dict[str, any]) -> Dict[str, Dict[str, any]]:
|
83 |
d6ca840d
|
petrh
|
"""
|
84 |
|
|
Iterates over all devices specified in config file
|
85 |
|
|
|
86 |
|
|
Extracts only valid one (have both specified coordinates no UNKOWN! OR SKIP)
|
87 |
|
|
|
88 |
|
|
Args:
|
89 |
|
|
devices: dictionary of devices contained in config file
|
90 |
|
|
|
91 |
|
|
Returns:
|
92 |
|
|
Dictonary containing only valid devices
|
93 |
|
|
"""
|
94 |
af7609b5
|
Tomáš Ballák
|
valid_devices = {}
|
95 |
d6ca840d
|
petrh
|
|
96 |
|
|
for device in devices.keys():
|
97 |
81980e82
|
ballakt
|
if not should_skip(devices[device]):
|
98 |
|
|
valid_devices[device] = {
|
99 |
|
|
'name': device,
|
100 |
|
|
'x': devices[device]['x'],
|
101 |
|
|
'y': devices[device]['y']
|
102 |
|
|
}
|
103 |
d6ca840d
|
petrh
|
|
104 |
|
|
return valid_devices
|