Mercurial > repos > laurenmarazzi > netisce_test
comparison tools/myTools/bin/sfa/containers.py @ 1:7e5c71b2e71f draft default tip
Uploaded
author | laurenmarazzi |
---|---|
date | Wed, 22 Dec 2021 16:00:34 +0000 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
0:f24d4892aaed | 1:7e5c71b2e71f |
---|---|
1 # -*- coding: utf-8 -*- | |
2 | |
3 import sys | |
4 if sys.version_info <= (2, 8): | |
5 from builtins import super | |
6 | |
7 import os | |
8 import re | |
9 import importlib | |
10 import collections | |
11 | |
12 import abc | |
13 import six | |
14 | |
15 import sfa.base | |
16 import sfa.algorithms | |
17 import sfa.data | |
18 | |
19 from sfa.utils import Singleton | |
20 | |
21 __all__ = ["AlgorithmSet", "DataSet"] | |
22 | |
23 @six.add_metaclass(abc.ABCMeta) | |
24 class Container(collections.MutableMapping): | |
25 """ | |
26 A simple container class, which handles multiple objects with | |
27 its hashable functionality (using dictionary). | |
28 """ | |
29 def __init__(self, *args, **kwargs): | |
30 """ | |
31 _dict: internal data structure, which is hashable. | |
32 _dpath: Path of the directory containing algorithms, data, etc. | |
33 """ | |
34 self._map = dict() | |
35 self._dpath = None | |
36 self._all_keys = None | |
37 self.update(dict(*args, **kwargs)) | |
38 | |
39 def __getitem__(self, key): | |
40 return self._map[key] | |
41 | |
42 def __setitem__(self, key, value): | |
43 if isinstance(value, sfa.base.ContainerItem): | |
44 self._map[key] = value | |
45 else: | |
46 raise TypeError("Container.__setitem__ only accepts " | |
47 "sfa.base.ContainerItem type object.") | |
48 | |
49 def __delitem__(self, key): | |
50 del self._map[key] | |
51 | |
52 def __iter__(self): | |
53 return iter(self._map) | |
54 | |
55 def __len__(self): | |
56 return len(self._map) | |
57 | |
58 def __str__(self): | |
59 return str(self._map) | |
60 | |
61 def keys(self): | |
62 return self._map.keys() | |
63 | |
64 def values(self): | |
65 return self._map.values() | |
66 | |
67 def create(self, keys=None): | |
68 """ | |
69 Create a single or multiple objects according to keys. | |
70 keys: a single string or multiple strings in an iterable object. | |
71 All related objects are created if 'keys' is None. | |
72 """ | |
73 if keys is not None: | |
74 if type(keys) is str: | |
75 return self._create_single(keys) | |
76 elif hasattr(keys, '__iter__'): | |
77 mult = [] # To return multiple elements | |
78 # An iterable object contains multiple keys. | |
79 for elem in keys: | |
80 mult.append(self._create_single(elem)) | |
81 return mult | |
82 else: | |
83 self._create_all() | |
84 return self | |
85 # end of def create | |
86 | |
87 @abc.abstractmethod | |
88 def get_all_keys(self): | |
89 """""" | |
90 # end of def | |
91 | |
92 @abc.abstractmethod | |
93 def _create_single(self, key): | |
94 """Create a single object""" | |
95 # end of def | |
96 | |
97 @abc.abstractmethod | |
98 def _create_all(self): | |
99 """Create all objects""" | |
100 # end of def | |
101 | |
102 | |
103 | |
104 """ | |
105 <References> | |
106 | |
107 [Dictionary functionality] | |
108 http://stackoverflow.com/questions/3387691/python-how-to-perfectly-override-a-dict | |
109 """ | |
110 # end of def class | |
111 | |
112 excluded = ['np.py'] | |
113 | |
114 @Singleton | |
115 class AlgorithmSet(Container): | |
116 _instance = None | |
117 def __new__(cls, *args, **kwargs): | |
118 return super().__new__(cls, *args, **kwargs) | |
119 | |
120 def __init__(self, *args, **kwargs): | |
121 super().__init__(*args, **kwargs) | |
122 """ | |
123 sfa.algorithms.__file__ represents a directory path | |
124 containing sfa.algorithms's init module (__init__.py). | |
125 """ | |
126 self._dpath = os.path.dirname(sfa.algorithms.__file__) | |
127 | |
128 # end of def __init__ | |
129 | |
130 def get_all_keys(self): | |
131 if not self._all_keys: | |
132 self._all_keys = [] | |
133 for entity in os.listdir(self._dpath): | |
134 if re.match(r"[^_]\w+\.py$", entity) \ | |
135 and entity not in excluded: | |
136 mod_name = entity.split('.')[0] # Module name | |
137 print(mod_name, entity) | |
138 key = mod_name.upper() | |
139 self._all_keys.append(key) | |
140 yield key | |
141 # end of for | |
142 else: | |
143 return iter(self._all_keys) | |
144 | |
145 def _create_single(self, key): | |
146 if key in self._map: | |
147 return self._map[key] | |
148 | |
149 | |
150 key_low = key.lower() | |
151 fstr_module_path = "%s.%s" % (sfa.algorithms.__name__, | |
152 key_low) | |
153 | |
154 _key = key.upper() # We use captital characters for the key. | |
155 if _key in self._map: # Avoid redundant importing | |
156 return | |
157 | |
158 mod = importlib.import_module(fstr_module_path) | |
159 # The following is of test purpose (removable in the future). | |
160 if "this_should_be_imported" in mod.__name__: | |
161 return | |
162 | |
163 alg = mod.create_algorithm(_key) | |
164 self._map[_key] = alg | |
165 | |
166 # For testing purpose | |
167 print("%s algorithm has been created." % (_key)) | |
168 return alg | |
169 | |
170 def _create_all(self): | |
171 """ | |
172 Import all algorithms, based on file names | |
173 """ | |
174 for entity in os.listdir(self._dpath): | |
175 if re.match(r"[^_]\w+\.py", entity) \ | |
176 and entity not in excluded: | |
177 mod_name = entity.split('.')[0] # Module name | |
178 self._create_single(mod_name) | |
179 # end of for | |
180 # end of def _create_all | |
181 | |
182 # end of class Algorithms | |
183 | |
184 @Singleton | |
185 class DataSet(Container): | |
186 """ | |
187 The name of this class is similar to that of 'DataSet' in C#. | |
188 The instance of this class handles multiple sfa.base.Data objects. | |
189 """ | |
190 _instance = None | |
191 def __new__(cls, *args, **kwargs): | |
192 return super().__new__(cls, *args, **kwargs) | |
193 | |
194 def __init__(self, *args, **kwargs): | |
195 super().__init__(*args, **kwargs) | |
196 """ | |
197 sfa.data.__file__ represents a directory path | |
198 containing sfa.data's init module (__init__.py). | |
199 """ | |
200 self._dpath = os.path.dirname(sfa.data.__file__) | |
201 | |
202 # end of def __init__ | |
203 | |
204 def get_all_keys(self): | |
205 if not self._all_keys: | |
206 self._all_keys = [] | |
207 for entity in os.listdir(self._dpath): | |
208 dpath = os.path.join(self._dpath, entity) | |
209 if not entity.startswith('_') and os.path.isdir(dpath): | |
210 key = entity.upper() | |
211 self._all_keys.append(key) | |
212 yield key | |
213 else: | |
214 return iter(self._all_keys) | |
215 | |
216 def _create_single(self, key): | |
217 if key in self._map: | |
218 return self._map[key] | |
219 | |
220 key_items = key.split("_") | |
221 key_1st, key_2nd = key_items[:2] | |
222 mod_name = "%s_%s"%(key_1st.lower(), key_2nd.lower()) | |
223 fstr_module_path = "%s.%s" % (sfa.data.__name__, mod_name) | |
224 | |
225 _key = key.upper() | |
226 if _key in self._map: # Avoid redundant importing | |
227 return | |
228 elif mod_name.upper() in self._map: | |
229 # All data object in this directory has been created before. | |
230 return | |
231 | |
232 mod = importlib.import_module(fstr_module_path) | |
233 if len(key_items) > 2: # Create the specified single data object | |
234 data = mod.create_data(key) | |
235 else: # Create all data objects from this directory | |
236 data = mod.create_data() | |
237 | |
238 if isinstance(data, dict): | |
239 self._map[_key] = data | |
240 elif isinstance(data, list): | |
241 self._map[_key] = {obj.abbr.upper(): obj for obj in data} | |
242 elif isinstance(data, sfa.base.Data): | |
243 _key = data.abbr.upper() | |
244 self._map[_key] = data | |
245 else: | |
246 err_msg = "%s.create_data() returns unsupported type."\ | |
247 % (fstr_module_path) | |
248 raise TypeError(err_msg) | |
249 # end of if | |
250 | |
251 # For testing purpose | |
252 print("%s data has been created." % (_key)) | |
253 return self._map[_key] | |
254 # end of def _create_single | |
255 | |
256 def _create_all(self): | |
257 """ | |
258 Import all data, based on the directory names of data modules | |
259 """ | |
260 for entity in os.listdir(self._dpath): | |
261 dpath = os.path.join(self._dpath, entity) | |
262 if not entity.startswith('_') and os.path.isdir(dpath): | |
263 self._create_single(entity) | |
264 # end of for | |
265 # end of def _create_all | |
266 | |
267 # end of def class DataSet |