DogmaticDict._log_blocked_setitem() - Code Metrics - IDSIA/sacred - Measure and Improve Code Quality continuously with Scrutinizer

DogmaticDict._log_blocked_setitem() B
last analyzed 2018-07-01 19:23 UTC

↳ Parent: DogmaticDict

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
c	0
b	0
f	0
dl	0
loc	13
rs	8

#!/usr/bin/env python
# coding=utf-8
from __future__ import division, print_function, unicode_literals

import sacred.optional as opt
from sacred.utils import join_paths


class FallbackDict(dict):
    """Dictionary that defaults to a fallback dict for missing keys."""

    def __init__(self, fallback, **kwargs):
        super(FallbackDict, self).__init__(**kwargs)
        self.fallback = fallback

    def __getitem__(self, item):
        if dict.__contains__(self, item):
            return dict.__getitem__(self, item)
        else:
            return self.fallback[item]

    def __contains__(self, item):
        return dict.__contains__(self, item) or (item in self.fallback)

    def get(self, k, d=None):
        if dict.__contains__(self, k):
            return dict.__getitem__(self, k)
        else:
            return self.fallback.get(k, d)

    def items(self):
        raise NotImplementedError()

    def iteritems(self):
        raise NotImplementedError()

    def iterkeys(self):
        raise NotImplementedError()

    def itervalues(self):
        raise NotImplementedError()

    def keys(self):
        raise NotImplementedError()

    def pop(self, k, d=None):
        raise NotImplementedError()

    def popitem(self):
        raise NotImplementedError()

    def setdefault(self, k, d=None):
        raise NotImplementedError()

    def update(self, e=None, **f):
        raise NotImplementedError()

    def values(self):
        raise NotImplementedError()

    def viewitems(self):
        raise NotImplementedError()

    def viewkeys(self):
        raise NotImplementedError()

    def viewvalues(self):
        raise NotImplementedError()

    def __iter__(self):
        raise NotImplementedError()

    def __len__(self):
        raise NotImplementedError()


class DogmaticDict(dict):
    def __init__(self, fixed=None, fallback=None):
        super(DogmaticDict, self).__init__()
        self.typechanges = {}
        self.fallback_writes = []
        self.modified = set()
        self.fixed = fixed or {}
        self._fallback = {}
        if fallback:
            self.fallback = fallback

    @property
    def fallback(self):
        return self._fallback

    @fallback.setter
    def fallback(self, newval):
        ffkeys = set(self.fixed.keys()).intersection(set(newval.keys()))
        for k in ffkeys:
            if isinstance(self.fixed[k], DogmaticDict):
                self.fixed[k].fallback = newval[k]
            elif isinstance(self.fixed[k], dict):
                self.fixed[k] = DogmaticDict(self.fixed[k])
                self.fixed[k].fallback = newval[k]

        self._fallback = newval

    def _log_blocked_setitem(self, key, value, fixed_value):
        if type_changed(value, fixed_value):
            self.typechanges[key] = (type(value), type(fixed_value))

        if value != fixed_value:
            self.modified.add(key)

        # if both are dicts recursively collect modified and typechanges
        if isinstance(fixed_value, DogmaticDict) and isinstance(value, dict):
            for k, val in fixed_value.typechanges.items():
                self.typechanges[join_paths(key, k)] = val

            self.modified |= {join_paths(key, m) for m in fixed_value.modified}

    def __setitem__(self, key, value):
        if key not in self.fixed:
            if key in self.fallback:
                self.fallback_writes.append(key)
            return dict.__setitem__(self, key, value)

        fixed_value = self.fixed[key]
        dict.__setitem__(self, key, fixed_value)
        # if both are dicts do a recursive update
        if isinstance(fixed_value, DogmaticDict) and isinstance(value, dict):
            for k, val in value.items():
                fixed_value[k] = val

        self._log_blocked_setitem(key, value, fixed_value)

    def __getitem__(self, item):
        if dict.__contains__(self, item):
            return dict.__getitem__(self, item)
        elif item in self.fallback:
            if item in self.fixed:
                return self.fixed[item]
            else:
                return self.fallback[item]
        raise KeyError(item)

    def __contains__(self, item):
        return dict.__contains__(self, item) or (item in self.fallback)

    def get(self, k, d=None):
        if dict.__contains__(self, k):
            return dict.__getitem__(self, k)
        else:
            return self.fallback.get(k, d)

    def has_key(self, item):
        return self.__contains__(item)

    def __delitem__(self, key):
        if key not in self.fixed:
            dict.__delitem__(self, key)

    def update(self, iterable=None, **kwargs):
        if iterable is not None:
            if hasattr(iterable, 'keys'):
                for key in iterable:
                    self[key] = iterable[key]
            else:
                for (key, value) in iterable:
                    self[key] = value
        for key in kwargs:
            self[key] = kwargs[key]

    def revelation(self):
        missing = set()
        for key in self.fixed:
            if not dict.__contains__(self, key):
                self[key] = self.fixed[key]
                missing.add(key)

            if isinstance(self[key], (DogmaticDict, DogmaticList)):
                missing |= {key + "." + k for k in self[key].revelation()}
        return missing


class DogmaticList(list):
    def append(self, p_object):
        pass

    def extend(self, iterable):
        pass

    def insert(self, index, p_object):
        pass

    def reverse(self):
        pass

    def sort(self, compare=None, key=None, reverse=False):
        pass

    def __iadd__(self, other):
        return self

    def __imul__(self, other):
        return self

    def __setitem__(self, key, value):
        pass

    def __setslice__(self, i, j, sequence):
        pass

    def __delitem__(self, key):
        pass

    def __delslice__(self, i, j):
        pass

    def pop(self, index=None):
        raise TypeError('Cannot pop from DogmaticList')

    def remove(self, value):
        pass

    def revelation(self):
        for obj in self:
            if isinstance(obj, (DogmaticDict, DogmaticList)):
                obj.revelation()
        return set()


SIMPLIFY_TYPE = {
    type(None): type(None),
    bool: bool,
    float: float,
    int: int,
    str: str,
    list: list,
    tuple: list,
    dict: dict,
    DogmaticDict: dict,
    DogmaticList: list,
}

# if in python 2 we want to ignore unicode/str and int/long typechanges
try:
    SIMPLIFY_TYPE[unicode] = str
    SIMPLIFY_TYPE[long] = int
except NameError:
    pass

# if numpy is available we also want to ignore typechanges from numpy
# datatypes to the corresponding python datatype
if opt.has_numpy:
    from sacred.optional import np
    NP_FLOATS = ['float', 'float16', 'float32', 'float64', 'float128']
    for npf in NP_FLOATS:
        if hasattr(np, npf):
            SIMPLIFY_TYPE[getattr(np, npf)] = float

    NP_INTS = ['int', 'int8', 'int16', 'int32', 'int64',
               'uint', 'uint8', 'uint16', 'uint32', 'uint64']
    for npi in NP_INTS:
        if hasattr(np, npi):
            SIMPLIFY_TYPE[getattr(np, npi)] = int

    SIMPLIFY_TYPE[np.bool_] = bool


def type_changed(old_value, new_value):
    sot = SIMPLIFY_TYPE.get(type(old_value), type(old_value))
    snt = SIMPLIFY_TYPE.get(type(new_value), type(new_value))
    return sot != snt and old_value is not None  # ignore typechanges from None


1			#!/usr/bin/env python
2			# coding=utf-8
3			from __future__ import division, print_function, unicode_literals
4
5			import sacred.optional as opt
6			from sacred.utils import join_paths
7
8
9			class FallbackDict(dict):
10			"""Dictionary that defaults to a fallback dict for missing keys."""
11
12			def __init__(self, fallback, **kwargs):
13			super(FallbackDict, self).__init__(**kwargs)
14			self.fallback = fallback
15
16			def __getitem__(self, item):
17			if dict.__contains__(self, item):
18			return dict.__getitem__(self, item)
19			else:
20			return self.fallback[item]
21
22			def __contains__(self, item):
23			return dict.__contains__(self, item) or (item in self.fallback)
24
25			def get(self, k, d=None):
26			if dict.__contains__(self, k):
27			return dict.__getitem__(self, k)
28			else:
29			return self.fallback.get(k, d)
30
31			def items(self):
32			raise NotImplementedError()
33
34			def iteritems(self):
35			raise NotImplementedError()
36
37			def iterkeys(self):
38			raise NotImplementedError()
39
40			def itervalues(self):
41			raise NotImplementedError()
42
43			def keys(self):
44			raise NotImplementedError()
45
46			def pop(self, k, d=None):
47			raise NotImplementedError()
48
49			def popitem(self):
50			raise NotImplementedError()
51
52			def setdefault(self, k, d=None):
53			raise NotImplementedError()
54
55			def update(self, e=None, **f):
56			raise NotImplementedError()
57
58			def values(self):
59			raise NotImplementedError()
60
61			def viewitems(self):
62			raise NotImplementedError()
63
64			def viewkeys(self):
65			raise NotImplementedError()
66
67			def viewvalues(self):
68			raise NotImplementedError()
69
70			def __iter__(self):
71			raise NotImplementedError()
72
73			def __len__(self):
74			raise NotImplementedError()
75
76
77			class DogmaticDict(dict):
78			def __init__(self, fixed=None, fallback=None):
79			super(DogmaticDict, self).__init__()
80			self.typechanges = {}
81			self.fallback_writes = []
82			self.modified = set()
83			self.fixed = fixed or {}
84			self._fallback = {}
85			if fallback:
86			self.fallback = fallback
87
88			@property
89			def fallback(self):
90			return self._fallback
91
92			@fallback.setter
93			def fallback(self, newval):
94			ffkeys = set(self.fixed.keys()).intersection(set(newval.keys()))
95			for k in ffkeys:
96			if isinstance(self.fixed[k], DogmaticDict):
97			self.fixed[k].fallback = newval[k]
98			elif isinstance(self.fixed[k], dict):
99			self.fixed[k] = DogmaticDict(self.fixed[k])
100			self.fixed[k].fallback = newval[k]
101
102			self._fallback = newval
103
104			def _log_blocked_setitem(self, key, value, fixed_value):
105			if type_changed(value, fixed_value):
106			self.typechanges[key] = (type(value), type(fixed_value))
107
108			if value != fixed_value:
109			self.modified.add(key)
110
111			# if both are dicts recursively collect modified and typechanges
112			if isinstance(fixed_value, DogmaticDict) and isinstance(value, dict):
113			for k, val in fixed_value.typechanges.items():
114			self.typechanges[join_paths(key, k)] = val
115
116			self.modified \|= {join_paths(key, m) for m in fixed_value.modified}
117
118			def __setitem__(self, key, value):
119			if key not in self.fixed:
120			if key in self.fallback:
121			self.fallback_writes.append(key)
122			return dict.__setitem__(self, key, value)
123
124			fixed_value = self.fixed[key]
125			dict.__setitem__(self, key, fixed_value)
126			# if both are dicts do a recursive update
127			if isinstance(fixed_value, DogmaticDict) and isinstance(value, dict):
128			for k, val in value.items():
129			fixed_value[k] = val
130
131			self._log_blocked_setitem(key, value, fixed_value)
132
133			def __getitem__(self, item):
134			if dict.__contains__(self, item):
135			return dict.__getitem__(self, item)
136			elif item in self.fallback:
137			if item in self.fixed:
138			return self.fixed[item]
139			else:
140			return self.fallback[item]
141			raise KeyError(item)
142
143			def __contains__(self, item):
144			return dict.__contains__(self, item) or (item in self.fallback)
145
146			def get(self, k, d=None):
147			if dict.__contains__(self, k):
148			return dict.__getitem__(self, k)
149			else:
150			return self.fallback.get(k, d)
151
152			def has_key(self, item):
153			return self.__contains__(item)
154
155			def __delitem__(self, key):
156			if key not in self.fixed:
157			dict.__delitem__(self, key)
158
159			def update(self, iterable=None, **kwargs):
160			if iterable is not None:
161			if hasattr(iterable, 'keys'):
162			for key in iterable:
163			self[key] = iterable[key]
164			else:
165			for (key, value) in iterable:
166			self[key] = value
167			for key in kwargs:
168			self[key] = kwargs[key]
169
170			def revelation(self):
171			missing = set()
172			for key in self.fixed:
173			if not dict.__contains__(self, key):
174			self[key] = self.fixed[key]
175			missing.add(key)
176
177			if isinstance(self[key], (DogmaticDict, DogmaticList)):
178			missing \|= {key + "." + k for k in self[key].revelation()}
179			return missing
180
181
182			class DogmaticList(list):
183			def append(self, p_object):
184			pass
185
186			def extend(self, iterable):
187			pass
188
189			def insert(self, index, p_object):
190			pass
191
192			def reverse(self):
193			pass
194
195			def sort(self, compare=None, key=None, reverse=False):
196			pass
197
198			def __iadd__(self, other):
199			return self
200
201			def __imul__(self, other):
202			return self
203
204			def __setitem__(self, key, value):
205			pass
206
207			def __setslice__(self, i, j, sequence):
208			pass
209
210			def __delitem__(self, key):
211			pass
212
213			def __delslice__(self, i, j):
214			pass
215
216			def pop(self, index=None):
217			raise TypeError('Cannot pop from DogmaticList')
218
219			def remove(self, value):
220			pass
221
222			def revelation(self):
223			for obj in self:
224			if isinstance(obj, (DogmaticDict, DogmaticList)):
225			obj.revelation()
226			return set()
227
228
229			SIMPLIFY_TYPE = {
230			type(None): type(None),
231			bool: bool,
232			float: float,
233			int: int,
234			str: str,
235			list: list,
236			tuple: list,
237			dict: dict,
238			DogmaticDict: dict,
239			DogmaticList: list,
240			}
241
242			# if in python 2 we want to ignore unicode/str and int/long typechanges
243			try:
244			SIMPLIFY_TYPE[unicode] = str
245			SIMPLIFY_TYPE[long] = int
246			except NameError:
247			pass
248
249			# if numpy is available we also want to ignore typechanges from numpy
250			# datatypes to the corresponding python datatype
251			if opt.has_numpy:
252			from sacred.optional import np
253			NP_FLOATS = ['float', 'float16', 'float32', 'float64', 'float128']
254			for npf in NP_FLOATS:
255			if hasattr(np, npf):
256			SIMPLIFY_TYPE[getattr(np, npf)] = float
257
258			NP_INTS = ['int', 'int8', 'int16', 'int32', 'int64',
259			'uint', 'uint8', 'uint16', 'uint32', 'uint64']
260			for npi in NP_INTS:
261			if hasattr(np, npi):
262			SIMPLIFY_TYPE[getattr(np, npi)] = int
263
264			SIMPLIFY_TYPE[np.bool_] = bool
265
266
267			def type_changed(old_value, new_value):
268			sot = SIMPLIFY_TYPE.get(type(old_value), type(old_value))
269			snt = SIMPLIFY_TYPE.get(type(new_value), type(new_value))
270			return sot != snt and old_value is not None # ignore typechanges from None
271

IDSIA / sacred

DogmaticDict._log_blocked_setitem() B last analyzed 2018-07-01 19:23 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

DogmaticDict._log_blocked_setitem() B
last analyzed 2018-07-01 19:23 UTC