Total Complexity | 50 |
Total Lines | 208 |
Duplicated Lines | 10.58 % |
Changes | 0 |
Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.
Common duplication problems, and corresponding solutions are:
Complex classes like reporting.fitness often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
1 | import sys |
||
2 | from abc import ABCMeta |
||
3 | from collections import defaultdict |
||
4 | from functools import reduce |
||
5 | |||
6 | |||
7 | class FitnessValue(object): |
||
8 | def __init__(self, value): |
||
9 | self.value = value |
||
10 | def __abs__(self): |
||
11 | return abs(self.value) |
||
12 | def __eq__(self, other): |
||
13 | return self.value == other.value |
||
14 | def __ne__(self, other): |
||
15 | return not self.__ne__(other) |
||
16 | def __str__(self): |
||
17 | return "{:.2f}".format(self.value) |
||
18 | def __repr__(self): |
||
19 | return "{:.2f}".format(self.value) |
||
20 | View Code Duplication | class NaturalFitnessValue(FitnessValue): |
|
|
|||
21 | def __init__(self, value): |
||
22 | super(NaturalFitnessValue, self).__init__(value) |
||
23 | def __lt__(self, other): |
||
24 | return self.value < other.value |
||
25 | def __le__(self, other): |
||
26 | return self.value <= other.value |
||
27 | def __gt__(self, other): |
||
28 | return self.value > other.value |
||
29 | def __ge__(self, other): |
||
30 | return self.value >= other.value |
||
31 | View Code Duplication | class ReversedFitnessValue(FitnessValue): |
|
32 | def __init__(self, value): |
||
33 | super(ReversedFitnessValue, self).__init__(value) |
||
34 | def __lt__(self, other): |
||
35 | return self.value > other.value |
||
36 | def __le__(self, other): |
||
37 | return self.value >= other.value |
||
38 | def __gt__(self, other): |
||
39 | return self.value < other.value |
||
40 | def __ge__(self, other): |
||
41 | return self.value <= other.value |
||
42 | |||
43 | |||
44 | _ORDERING_HASH = defaultdict(lambda: 'natural', perplexity='reversed') |
||
45 | _INITIAL_BEST = defaultdict(lambda: 0, perplexity=float('inf')) |
||
46 | _FITNESS_VALUE_CONSTRUCTORS_HASH = {'natural': NaturalFitnessValue, 'reversed': ReversedFitnessValue} |
||
47 | |||
48 | |||
49 | class FitnessFunctionBuilder(object): |
||
50 | def __init__(self): |
||
51 | self._column_definitions = [] |
||
52 | self._extractors = [] |
||
53 | self._coeff_values = [] |
||
54 | self._order = '' |
||
55 | self._names = [] |
||
56 | |||
57 | def _create_extractor(self, column_definition): |
||
58 | return lambda x: x[self._column_definitions.index(column_definition)] |
||
59 | |||
60 | def start(self, column_definitions, ordering='natural'): |
||
61 | self._column_definitions = column_definitions |
||
62 | self._extractors = [] |
||
63 | self._coeff_values = [] |
||
64 | self._order = ordering |
||
65 | self._names = [] |
||
66 | return self |
||
67 | |||
68 | def coefficient(self, name, value): |
||
69 | assert name in self._column_definitions |
||
70 | self._extractors.append(self._create_extractor(name)) |
||
71 | self._coeff_values.append(value) |
||
72 | self._names.append(name) |
||
73 | return self |
||
74 | |||
75 | def build(self): |
||
76 | return FitnessFunction(self._extractors, self._coeff_values, self._names, ordering=self._order) |
||
77 | |||
78 | function_builder = FitnessFunctionBuilder() |
||
79 | |||
80 | |||
81 | class FitnessFunction(object): |
||
82 | def __init__(self, extractors, coefficients, names, ordering='natural'): |
||
83 | self._extr = extractors |
||
84 | self._coeff = coefficients |
||
85 | self._names = names |
||
86 | self._order = ordering |
||
87 | assert ordering in ('natural', 'reversed') |
||
88 | assert abs(sum(map(abs, self._coeff))) - 1 < 1e-6 |
||
89 | |||
90 | @classmethod |
||
91 | def single_metric(cls, metric_definition): |
||
92 | return function_builder.start([metric_definition], ordering=_ORDERING_HASH[metric_definition]).coefficient(metric_definition, 1).build() |
||
93 | |||
94 | @property |
||
95 | def ordering(self): |
||
96 | return self._order |
||
97 | |||
98 | def compute(self, individual): |
||
99 | """ |
||
100 | :param list individual: list of values for metrics [prpl, top-tokens-coh-10, ..] |
||
101 | :return: |
||
102 | """ |
||
103 | try: |
||
104 | c = [x(individual) for x in self._extr] |
||
105 | except IndexError as e: |
||
106 | raise IndexError("Error: {}. Current builder column definitions: [{}], Input: [{}]".format(e, ', '.join(str(_) for _ in sorted(function_builder._column_definitions)), |
||
107 | ', '.join(str(_) for _ in sorted(individual)))) |
||
108 | c1 = [self._wrap(x) for x in c] |
||
109 | c2 = [self._coeff[i] * x for i,x in enumerate(c1)] |
||
110 | c3 = reduce(lambda i,j: i+j, c2) |
||
111 | c4 = _FITNESS_VALUE_CONSTRUCTORS_HASH[self._order] |
||
112 | r = c4(c3) |
||
113 | return r |
||
114 | # return _FITNESS_VALUE_CONSTRUCTORS_HASH[self._order]( |
||
115 | # reduce(lambda i, j: i + j, [x[0] * self._wrap(x[1](individual)) for x in zip(self._coeff, self._extr)])) |
||
116 | |||
117 | def _wrap(self, value): |
||
118 | if value is None: |
||
119 | return {'natural': 0, 'reversed': float('inf')}[self._order] |
||
120 | return value |
||
121 | |||
122 | def __str__(self): |
||
123 | return ' + '.join(['{}*{}'.format(x[0], x[1]) for x in zip(self._names, self._coeff)]) |
||
124 | |||
125 | |||
126 | class FitnessCalculator(object): |
||
127 | def __new__(cls, *args, **kwargs): |
||
128 | x = super(FitnessCalculator, cls).__new__(cls) |
||
129 | x._func = None |
||
130 | x._column_defs, x._best = [], [] |
||
131 | x._highlightable_columns = [] |
||
132 | return x |
||
133 | |||
134 | def __init__(self, single_metric=None, column_definitions=None): |
||
135 | if type(single_metric) == str and type(column_definitions) == list: |
||
136 | # FitnessFunctionBuilder.start(column_definitions).coefficient(single_metric, 1).build() |
||
137 | self._func = function_builder.start(column_definitions, ordering=_ORDERING_HASH[single_metric]).coefficient(single_metric, 1).build() |
||
138 | assert isinstance(self._func, FitnessFunction) |
||
139 | self._column_defs = column_definitions |
||
140 | |||
141 | # def initialize(self, fitness_function, column_definitions): |
||
142 | # """ |
||
143 | # :param FitnessFunction fitness_function: |
||
144 | # :param list column_definitions: i.e. ['perplexity', 'kernel-coherence-0.8', 'kernel-contrast-0.8', 'top-tokens-coherence-10', 'top-tokens-coherence-100'] |
||
145 | # """ |
||
146 | # assert isinstance(fitness_function, FitnessFunction) |
||
147 | # self.function = fitness_function |
||
148 | # self._column_defs = column_definitions |
||
149 | |||
150 | @property |
||
151 | def highlightable_columns(self): |
||
152 | return self._highlightable_columns |
||
153 | |||
154 | @highlightable_columns.setter |
||
155 | def highlightable_columns(self, column_definitions): |
||
156 | self._highlightable_columns = column_definitions |
||
157 | self._best = dict([(x, _INITIAL_BEST[FitnessCalculator._get_column_key(x)]) for x in column_definitions]) |
||
158 | |||
159 | @property |
||
160 | def best(self): |
||
161 | return self._best |
||
162 | |||
163 | @property |
||
164 | def function(self): |
||
165 | return self._func |
||
166 | |||
167 | # @function.setter |
||
168 | # def function(self, a_fitness_function): |
||
169 | # self._func = a_fitness_function |
||
170 | |||
171 | def pass_vector(self, values_vector): |
||
172 | self._update_best(values_vector) |
||
173 | return values_vector |
||
174 | |||
175 | def compute_fitness(self, values_vector): |
||
176 | self._update_best(values_vector) |
||
177 | return self._func.compute(values_vector) |
||
178 | |||
179 | def _update_best(self, values_vector): |
||
180 | self._best.update([(column_def, value) for column_key, column_def, value in |
||
181 | [(FitnessCalculator._get_column_key(x[0]), x[0], x[1]) for x in zip(self._column_defs, values_vector)] |
||
182 | if column_def in self._best and |
||
183 | FitnessCalculator._fitness(column_key, value) > FitnessCalculator._fitness(column_key, self._best[column_def])]) |
||
184 | |||
185 | def __call__(self, *args, **kwargs): |
||
186 | return self.compute_fitness(args[0]) |
||
187 | |||
188 | @staticmethod |
||
189 | def _fitness(column_key, value): |
||
190 | if value is None: |
||
191 | return _FITNESS_VALUE_CONSTRUCTORS_HASH[_ORDERING_HASH[column_key]](_INITIAL_BEST[column_key]) |
||
192 | return _FITNESS_VALUE_CONSTRUCTORS_HASH[_ORDERING_HASH[column_key]](value) |
||
193 | |||
194 | @staticmethod |
||
195 | def _get_column_key(column_definition): |
||
196 | return '-'.join([_f for _f in map(FitnessCalculator._get_token, column_definition.split('-')) if _f]) |
||
197 | # return '-'.join([_f for _f in [FitnessCalculator._get_token(x) for x in column_definition.split('-')] if _f]) |
||
198 | |||
199 | @staticmethod |
||
200 | def _get_token(definition_element): |
||
201 | try: |
||
202 | _ = float(definition_element) |
||
203 | return None |
||
204 | except ValueError: |
||
205 | if definition_element[0] == '@' or len(definition_element) == 1: |
||
206 | return None |
||
207 | return definition_element |
||
208 | |||
212 |