1 | # -*- coding: utf-8 -*- |
||
2 | """ |
||
3 | This file contains the Qudi configuration file module. |
||
4 | |||
5 | A configuration file is saved in YAML format. This module provides a loader |
||
6 | and a dumper using an OrderedDict instead of the regular dict used by PyYAML. |
||
7 | Additionally, it fixes a bug in PyYAML with scientific notation and allows |
||
8 | to dump numpy dtypes and numpy ndarrays. |
||
9 | |||
10 | The fix of the scientific notation is applied globally at module import. |
||
11 | |||
12 | The idea of the implementation of the OrderedDict was taken from |
||
13 | http://stackoverflow.com/questions/5121931/in-python-how-can-you-load-yaml-mappings-as-ordereddicts |
||
14 | |||
15 | |||
16 | |||
17 | Qudi is free software: you can redistribute it and/or modify |
||
18 | it under the terms of the GNU General Public License as published by |
||
19 | the Free Software Foundation, either version 3 of the License, or |
||
20 | (at your option) any later version. |
||
21 | |||
22 | Qudi is distributed in the hope that it will be useful, |
||
23 | but WITHOUT ANY WARRANTY; without even the implied warranty of |
||
24 | MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
||
25 | GNU General Public License for more details. |
||
26 | |||
27 | You should have received a copy of the GNU General Public License |
||
28 | along with Qudi. If not, see <http://www.gnu.org/licenses/>. |
||
29 | |||
30 | Copyright (c) the Qudi Developers. See the COPYRIGHT.txt file at the |
||
31 | top-level directory of this distribution and at <https://github.com/Ulm-IQO/qudi/> |
||
32 | """ |
||
33 | |||
34 | from collections import OrderedDict |
||
35 | import numpy |
||
36 | import re |
||
0 ignored issues
–
show
Unused Code
introduced
by
![]() |
|||
37 | import os |
||
38 | import ruamel.yaml as yaml |
||
39 | from io import BytesIO |
||
40 | |||
41 | |||
42 | def ordered_load(stream, Loader=yaml.Loader): |
||
43 | """ |
||
44 | Loads a YAML formatted data from stream and puts it into an OrderedDict |
||
45 | |||
46 | @param Stream stream: stream the data is read from |
||
47 | @param Loader Loader: Loader base class |
||
48 | |||
49 | Returns OrderedDict with data. If stream is empty then an empty |
||
50 | OrderedDict is returned. |
||
51 | """ |
||
52 | class OrderedLoader(Loader): |
||
53 | """ |
||
54 | Loader using an OrderedDict |
||
55 | """ |
||
56 | pass |
||
57 | |||
58 | def construct_mapping(loader, node): |
||
59 | """ |
||
60 | The OrderedDict constructor. |
||
61 | """ |
||
62 | loader.flatten_mapping(node) |
||
63 | return OrderedDict(loader.construct_pairs(node)) |
||
64 | |||
65 | def construct_ndarray(loader, node): |
||
66 | """ |
||
67 | The ndarray constructor, correctly saves a numpy array |
||
68 | inside the config file as a string. |
||
69 | """ |
||
70 | value = loader.construct_yaml_binary(node) |
||
71 | with BytesIO(bytes(value)) as f: |
||
72 | arrays = numpy.load(f) |
||
73 | return arrays['array'] |
||
74 | |||
75 | def construct_external_ndarray(loader, node): |
||
76 | """ |
||
77 | The constructor for an numoy array that is saved in an external file. |
||
78 | """ |
||
79 | filename = loader.construct_yaml_str(node) |
||
80 | arrays = numpy.load(filename) |
||
81 | return arrays['array'] |
||
82 | |||
83 | def construct_str(loader, node): |
||
84 | """ |
||
85 | construct strings but if the string starts with 'array(' it tries |
||
86 | to evaluate it as numpy array. |
||
87 | |||
88 | TODO: This behaviour should be deprecated at some point. |
||
89 | """ |
||
90 | value = loader.construct_yaml_str(node) |
||
91 | # if a string could be an array, we try to evaluate the string |
||
92 | # to reconstruct a numpy array. If it fails we return the string. |
||
93 | if value.startswith('array('): |
||
94 | try: |
||
95 | local = {"array": numpy.array} |
||
96 | for dtype in ['int8', 'uint8', 'int16', 'uint16', 'float16', |
||
97 | 'int32', 'uint32', 'float32', 'int64', 'uint64', |
||
98 | 'float64']: |
||
99 | local[dtype] = getattr(numpy, dtype) |
||
100 | return eval(value, local) |
||
0 ignored issues
–
show
|
|||
101 | except SyntaxError: |
||
102 | return value |
||
103 | else: |
||
104 | return value |
||
105 | |||
106 | # add constructor |
||
107 | OrderedLoader.add_constructor( |
||
0 ignored issues
–
show
|
|||
108 | yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, |
||
109 | construct_mapping) |
||
110 | OrderedLoader.add_constructor( |
||
0 ignored issues
–
show
|
|||
111 | '!ndarray', |
||
112 | construct_ndarray) |
||
113 | OrderedLoader.add_constructor( |
||
0 ignored issues
–
show
|
|||
114 | '!extndarray', |
||
115 | construct_external_ndarray) |
||
116 | OrderedLoader.add_constructor( |
||
0 ignored issues
–
show
|
|||
117 | yaml.resolver.BaseResolver.DEFAULT_SCALAR_TAG, |
||
118 | construct_str) |
||
119 | |||
120 | # load config file |
||
121 | config = yaml.load(stream, OrderedLoader) |
||
122 | # yaml returns None if the config file was empty |
||
123 | if config is not None: |
||
124 | return config |
||
125 | else: |
||
126 | return OrderedDict() |
||
127 | |||
128 | |||
129 | def ordered_dump(data, stream=None, Dumper=yaml.Dumper, **kwds): |
||
130 | """ |
||
131 | dumps (OrderedDict) data in YAML format |
||
132 | |||
133 | @param OrderedDict data: the data |
||
134 | @param Stream stream: where the data in YAML is dumped |
||
135 | @param Dumper Dumper: The dumper that is used as a base class |
||
136 | """ |
||
137 | class OrderedDumper(Dumper): |
||
138 | """ |
||
139 | A Dumper using an OrderedDict |
||
140 | """ |
||
141 | external_ndarray_counter = 0 |
||
142 | |||
143 | def ignore_aliases(self, data): |
||
0 ignored issues
–
show
This method could be written as a function/class method.
If a method does not access any attributes of the class, it could also be implemented as a function or static method. This can help improve readability. For example class Foo:
def some_method(self, x, y):
return x + y;
could be written as class Foo:
@classmethod
def some_method(cls, x, y):
return x + y;
![]() |
|||
144 | """ |
||
145 | ignore aliases and anchors |
||
146 | """ |
||
147 | return True |
||
148 | |||
149 | def represent_ordereddict(dumper, data): |
||
150 | """ |
||
151 | Representer for OrderedDict |
||
152 | """ |
||
153 | return dumper.represent_mapping( |
||
154 | yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, |
||
155 | data.items()) |
||
156 | |||
157 | def represent_int(dumper, data): |
||
158 | """ |
||
159 | Representer for numpy int dtypes |
||
160 | """ |
||
161 | return dumper.represent_int(numpy.asscalar(data)) |
||
162 | |||
163 | def represent_float(dumper, data): |
||
164 | """ |
||
165 | Representer for numpy float dtypes |
||
166 | """ |
||
167 | return dumper.represent_float(numpy.asscalar(data)) |
||
168 | |||
169 | def represent_ndarray(dumper, data): |
||
170 | """ |
||
171 | Representer for numpy ndarrays |
||
172 | """ |
||
173 | try: |
||
174 | filename = os.path.splitext(os.path.basename(stream.name))[0] |
||
175 | configdir = os.path.dirname(stream.name) |
||
176 | newpath = '{0}-{1:06}.npz'.format( |
||
177 | os.path.join(configdir, filename), |
||
178 | dumper.external_ndarray_counter) |
||
179 | numpy.savez_compressed(newpath, array=data) |
||
180 | node = dumper.represent_str(newpath) |
||
181 | node.tag = '!extndarray' |
||
182 | dumper.external_ndarray_counter += 1 |
||
183 | except: |
||
184 | with BytesIO() as f: |
||
185 | numpy.savez_compressed(f, array=data) |
||
186 | compressed_string = f.getvalue() |
||
187 | node = dumper.represent_binary(compressed_string) |
||
188 | node.tag = '!ndarray' |
||
189 | return node |
||
190 | |||
191 | # add representers |
||
192 | OrderedDumper.add_representer(OrderedDict, represent_ordereddict) |
||
0 ignored issues
–
show
|
|||
193 | OrderedDumper.add_representer(numpy.uint8, represent_int) |
||
0 ignored issues
–
show
|
|||
194 | OrderedDumper.add_representer(numpy.uint16, represent_int) |
||
0 ignored issues
–
show
|
|||
195 | OrderedDumper.add_representer(numpy.uint32, represent_int) |
||
0 ignored issues
–
show
|
|||
196 | OrderedDumper.add_representer(numpy.uint64, represent_int) |
||
0 ignored issues
–
show
|
|||
197 | OrderedDumper.add_representer(numpy.int8, represent_int) |
||
0 ignored issues
–
show
|
|||
198 | OrderedDumper.add_representer(numpy.int16, represent_int) |
||
0 ignored issues
–
show
|
|||
199 | OrderedDumper.add_representer(numpy.int32, represent_int) |
||
0 ignored issues
–
show
|
|||
200 | OrderedDumper.add_representer(numpy.int64, represent_int) |
||
0 ignored issues
–
show
|
|||
201 | OrderedDumper.add_representer(numpy.float16, represent_float) |
||
0 ignored issues
–
show
|
|||
202 | OrderedDumper.add_representer(numpy.float32, represent_float) |
||
0 ignored issues
–
show
|
|||
203 | OrderedDumper.add_representer(numpy.float64, represent_float) |
||
0 ignored issues
–
show
|
|||
204 | # OrderedDumper.add_representer(numpy.float128, represent_float) |
||
205 | OrderedDumper.add_representer(numpy.ndarray, represent_ndarray) |
||
0 ignored issues
–
show
|
|||
206 | |||
207 | # dump data |
||
208 | return yaml.dump(data, stream, OrderedDumper, **kwds) |
||
209 | |||
210 | |||
211 | def load(filename): |
||
212 | """ |
||
213 | Loads a config file |
||
214 | |||
215 | @param filename str: filename of config file |
||
216 | |||
217 | Returns OrderedDict |
||
218 | """ |
||
219 | with open(filename, 'r') as f: |
||
220 | return ordered_load(f, yaml.SafeLoader) |
||
221 | |||
222 | def save(filename, data): |
||
223 | """ |
||
224 | saves data to filename in yaml format. |
||
225 | |||
226 | @param filename str: filename of config file |
||
227 | @param data OrderedDict: config values |
||
228 | """ |
||
229 | with open(filename, 'w') as f: |
||
230 | ordered_dump(data, stream=f, Dumper=yaml.SafeDumper, |
||
231 | default_flow_style=False) |
||
232 |