Passed
Branch master (13db34)
by Grega
02:16
created

NiaPy.algorithms.basic.fwa   F

Complexity

Total Complexity 66

Size/Duplication

Total Lines 801
Duplicated Lines 4.87 %

Importance

Changes 0
Metric Value
eloc 205
dl 39
loc 801
rs 3.12
c 0
b 0
f 0
wmc 66

37 Methods

Rating   Name   Duplication   Size   Complexity  
A BareBonesFireworksAlgorithm.runIteration() 0 25 2
A FireworksAlgorithm.ExplosionAmplitude() 0 13 1
A BareBonesFireworksAlgorithm.initPopulation() 0 16 1
A FireworksAlgorithm.SparsksNo() 0 13 2
A BareBonesFireworksAlgorithm.typeParameters() 0 5 4
A BareBonesFireworksAlgorithm.setParameters() 0 10 2
A DynamicFireworksAlgorithmGauss.Mapping() 0 15 1
A EnhancedFireworksAlgorithm.setParameters() 0 13 1
A EnhancedFireworksAlgorithm.typeParameters() 0 16 3
A FireworksAlgorithm.setParameters() 0 14 2
A DynamicFireworksAlgorithm.runIteration() 0 30 1
A EnhancedFireworksAlgorithm.initPopulation() 0 22 1
A FireworksAlgorithm.initAmplitude() 0 10 1
A FireworksAlgorithm.initPopulation() 0 20 1
A DynamicFireworksAlgorithmGauss.setParameters() 0 15 1
A EnhancedFireworksAlgorithm.initRanges() 0 14 1
A EnhancedFireworksAlgorithm.ExplosionAmplitude() 0 18 1
A FireworksAlgorithm.Mapping() 15 15 1
A FireworksAlgorithm.p() 0 11 1
A EnhancedFireworksAlgorithm.NextGeneration() 0 21 4
A DynamicFireworksAlgorithmGauss.NextGeneration() 0 21 3
A FireworksAlgorithm.R() 0 11 1
A DynamicFireworksAlgorithmGauss.ExplosionAmplitude() 0 2 1
A DynamicFireworksAlgorithmGauss.initAmplitude() 0 11 1
A FireworksAlgorithm.GaussianSpark() 0 11 1
A DynamicFireworksAlgorithmGauss.initPopulation() 0 17 1
A DynamicFireworksAlgorithmGauss.runIteration() 0 31 2
A EnhancedFireworksAlgorithm.runIteration() 0 33 2
A DynamicFireworksAlgorithmGauss.uCF() 0 23 3
A EnhancedFireworksAlgorithm.uAmin() 0 12 1
A FireworksAlgorithm.runIteration() 0 34 2
A FireworksAlgorithm.ExplodeSpark() 0 12 1
A EnhancedFireworksAlgorithm.GaussianSpark() 0 12 1
B FireworksAlgorithm.typeParameters() 7 7 6
A FireworksAlgorithm.NextGeneration() 0 23 2
A DynamicFireworksAlgorithmGauss.typeParameters() 17 17 5
A DynamicFireworksAlgorithmGauss.repair() 0 14 1

How to fix   Duplicated Code    Complexity   

Duplicated Code

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:

Complexity

 Tip:   Before tackling complexity, make sure that you eliminate any duplication first. This often can reduce the size of classes significantly.

Complex classes like NiaPy.algorithms.basic.fwa 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
# encoding=utf8
2
# pylint: disable=mixed-indentation, trailing-whitespace, multiple-statements, attribute-defined-outside-init, logging-not-lazy, arguments-differ, line-too-long, redefined-builtin, no-self-use, singleton-comparison, unused-argument, bad-continuation
3
import logging
4
from numpy import apply_along_axis, argmin, argmax, sum, sqrt, round, argsort, fabs, asarray, where
5
from NiaPy.algorithms.algorithm import Algorithm
6
from NiaPy.util import fullArray
7
8
logging.basicConfig()
9
logger = logging.getLogger('NiaPy.algorithms.basic')
10
logger.setLevel('INFO')
11
12
__all__ = ['FireworksAlgorithm', 'EnhancedFireworksAlgorithm', 'DynamicFireworksAlgorithm', 'DynamicFireworksAlgorithmGauss', 'BareBonesFireworksAlgorithm']
13
14
class BareBonesFireworksAlgorithm(Algorithm):
15
	r"""Implementation of bare bone fireworks algorithm.
16
17
	Algorithm:
18
		Bare Bones Fireworks Algorithm
19
20
	Date:
21
		2018
22
23
	Authors:
24
		Klemen Berkovič
25
26
	License:
27
		MIT
28
29
	Reference URL:
30
		https://www.sciencedirect.com/science/article/pii/S1568494617306609
31
32
	Reference paper:
33
		Junzhi Li, Ying Tan, The bare bones fireworks algorithm: A minimalist global optimizer, Applied Soft Computing, Volume 62, 2018, Pages 454-462, ISSN 1568-4946, https://doi.org/10.1016/j.asoc.2017.10.046.
34
35
	Attributes:
36
		Name (lsit of str): List of strings representing algorithm names
37
		n (int): Number of spraks
38
		C_a (float): amplification coefficient
39
		C_r (float): reduction coefficient
40
	"""
41
	Name = ['BareBonesFireworksAlgorithm', 'BBFWA']
42
43
	@staticmethod
44
	def typeParameters(): return {
45
			'n': lambda x: isinstance(x, int) and x > 0,
46
			'C_a': lambda x: isinstance(x, (float, int)) and x > 1,
47
			'C_r': lambda x: isinstance(x, (float, int)) and 0 < x < 1
48
	}
49
50
	def setParameters(self, n=10, C_a=1.5, C_r=0.5, **ukwargs):
51
		r"""Set the arguments of an algorithm.
52
53
		Arguments:
54
			n (int): Number of sparks :math:`\in [1, \infty)`.
55
			C_a (float): Amplification coefficient :math:`\in [1, \infty)`.
56
			C_r (float): Reduction coefficient :math:`\in (0, 1)`.
57
		"""
58
		self.n, self.C_a, self.C_r = n, C_a, C_r
59
		if ukwargs: logger.info('Unused arguments: %s' % (ukwargs))
60
61
	def initPopulation(self, task):
62
		r"""Initialize starting population.
63
64
		Args:
65
			task (Task): Optimization task.
66
67
		Returns:
68
			Tuple[numpy.ndarray, float, Dict[str, Any]]:
69
				1. Initial solution.
70
				2. Initial solution function/fitness value.
71
				3. Additional arguments:
72
					* A (numpy.ndarray): Starting aplitude or search range.
73
		"""
74
		x, A = self.uniform(task.Lower, task.Upper, task.D), task.bRange
75
		x_fit = task.eval(x)
76
		return x, x_fit, {'A': A}
77
78
	def runIteration(self, task, x, x_fit, xb, fxb, A, **dparams):
79
		r"""Core function of Bare Bones Fireworks Algorithm.
80
81
		Args:
82
			task (Task): Optimization task.
83
			x (numpy.ndarray): Current solution.
84
			x_fit (float): Current solution fitness/function value.
85
			xb (numpy.ndarray): Current best solution.
86
			fxb (float): Current best solution fitness/function value.
87
			A (numpy.ndarray): Serach range.
88
			dparams (Dict[str, Any]): Additional parameters.
89
90
		Returns:
91
			Tuple[numpy.ndarray, float, Dict[str, Any]]:
92
				1. New solution.
93
				2. New solution fitness/function value.
94
				3. Additional arguments:
95
					* A (numpy.ndarray): Serach range.
96
		"""
97
		S = apply_along_axis(task.repair, 1, self.uniform(x - A, x + A, [self.n, task.D]), self.Rand)
98
		S_fit = apply_along_axis(task.eval, 1, S)
99
		iS = argmin(S_fit)
100
		if S_fit[iS] < x_fit: x, x_fit, A = S[iS], S_fit[iS], self.C_a * A
101
		else: A = self.C_r * A
102
		return x, x_fit, {'A': A}
103
104
class FireworksAlgorithm(Algorithm):
105
	r"""Implementation of fireworks algorithm.
106
107
	Algorithm:
108
		Fireworks Algorithm
109
110
	Date:
111
		2018
112
113
	Authors:
114
		Klemen Berkovič
115
116
	License:
117
		MIT
118
119
	Reference URL:
120
		https://www.springer.com/gp/book/9783662463529
121
122
	Reference paper:
123
		Tan, Ying. "Firework Algorithm: A Novel Swarm Intelligence Optimization Method." (2015).
124
125
	Attributes:
126
		Name (List[str]): List of stirngs representing algorithm names.
127
	"""
128
	Name = ['FireworksAlgorithm', 'FWA']
129
130 View Code Duplication
	@staticmethod
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
131
	def typeParameters(): return {
132
			'N': lambda x: isinstance(x, int) and x > 0,
133
			'm': lambda x: isinstance(x, int) and x > 0,
134
			'a': lambda x: isinstance(x, (int, float)) and x > 0,
135
			'b': lambda x: isinstance(x, (int, float)) and x > 0,
136
			'epsilon': lambda x: isinstance(x, float) and 0 < x < 1
137
	}
138
139
	def setParameters(self, N=40, m=40, a=1, b=2, A=40, epsilon=1e-31, **ukwargs):
140
		r"""Set the arguments of an algorithm.
141
142
		Arguments:
143
			N (int): Number of Fireworks
144
			m (int): Number of sparks
145
			a (int): Limitation of sparks
146
			b (int): Limitation of sparks
147
			A (float): --
148
			epsilon (float): Small number for non 0 devision
149
		"""
150
		Algorithm.setParameters(self, NP=N, **ukwargs)
151
		self.m, self.a, self.b, self.A, self.epsilon = m, a, b, A, epsilon
152
		if ukwargs: logger.info('Unused arguments: %s' % (ukwargs))
153
154
	def initAmplitude(self, task):
155
		r"""Initialize amplitudes for dimensions.
156
157
		Args:
158
			task (Task): Optimization task.
159
160
		Returns:
161
			numpy.ndarray[float]: Starting amplitudes.
162
		"""
163
		return fullArray(self.A, task.D)
164
165
	def SparsksNo(self, x_f, xw_f, Ss):
166
		r"""Calculate number of sparks based on function value of individual.
167
168
		Args:
169
			x_f (float): Individuals function/fitness value.
170
			xw_f (float): Worst individual function/fitness value.
171
			Ss (): TODO
172
173
		Returns:
174
			int: Number of sparks that individual will create.
175
		"""
176
		s = self.m * (xw_f - x_f + self.epsilon) / (Ss + self.epsilon)
177
		return round(self.b * self.m) if s > self.b * self.m and self.a < self.b < 1 else round(self.a * self.m)
178
179
	def ExplosionAmplitude(self, x_f, xb_f, A, As):
180
		r"""Calculate explosion amplitude.
181
182
		Args:
183
			x_f (float): Individuals function/fitness value.
184
			xb_f (float): Best individuals function/fitness value.
185
			A (numpy.ndarray): Amplitudes.
186
			As ():
187
188
		Returns:
189
			numpy.ndarray: TODO.
190
		"""
191
		return A * (x_f - xb_f - self.epsilon) / (As + self.epsilon)
192
193
	def ExplodeSpark(self, x, A, task):
194
		r"""Explode a spark.
195
196
		Args:
197
			x (numpy.ndarray): Individuals creating spark.
198
			A (numpy.ndarray): Amplitude of spark.
199
			task (Task): Optimization task.
200
201
		Returns:
202
			numpy.ndarray: Sparks exploded in with specified amplitude.
203
		"""
204
		return self.Mapping(x + self.rand(task.D) * self.uniform(-A, A, task.D), task)
205
206
	def GaussianSpark(self, x, task):
207
		r"""Create gaussian spark.
208
209
		Args:
210
			x (numpy.ndarray): Individual creating a spark.
211
			task (Task): Optimization task.
212
213
		Returns:
214
			numpy.ndarray: Spark exploded based on gaussian amplitude.
215
		"""
216
		return self.Mapping(x + self.rand(task.D) * self.normal(1, 1, task.D), task)
217
218 View Code Duplication
	def Mapping(self, x, task):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
219
		r"""Fix value to bounds..
220
221
		Args:
222
			x (numpy.ndarray): Individual to fix.
223
			task (Task): Optimization task.
224
225
		Returns:
226
			numpy.ndarray: Individual in search range.
227
		"""
228
		ir = where(x > task.Upper)
229
		x[ir] = task.Lower[ir] + x[ir] % task.bRange[ir]
230
		ir = where(x < task.Lower)
231
		x[ir] = task.Lower[ir] + x[ir] % task.bRange[ir]
232
		return x
233
234
	def R(self, x, FW):
235
		r"""Calculate ranges.
236
237
		Args:
238
			x (numpy.ndarray): Individual in population.
239
			FW (numpy.ndarray): Current population.
240
241
		Returns:
242
			numpy,ndarray[float]: Ranges values.
243
		"""
244
		return sqrt(sum(fabs(x - FW)))
245
246
	def p(self, r, Rs):
247
		r"""Calculate p.
248
249
		Args:
250
			r (float): Range of individual.
251
			Rs (float): Sum of ranges.
252
253
		Returns:
254
			float: p value.
255
		"""
256
		return r / Rs
257
258
	def NextGeneration(self, FW, FW_f, FWn, task):
259
		r"""Generate new generation of individuals.
260
261
		Args:
262
			FW (numpy.ndarray): Current population.
263
			FW_f (numpy.ndarray[float]): Currents population fitness/function values.
264
			FWn (numpy.ndarray): New population.
265
			task (Task): Optimization task.
266
267
		Returns:
268
			Tuple[numpy.ndarray, numpy.ndarray[float]]:
269
				1. New population.
270
				2. New populations fitness/function values.
271
		"""
272
		FWn_f = apply_along_axis(task.eval, 1, FWn)
273
		ib = argmin(FWn_f)
274
		if FWn_f[ib] < FW_f[0]: FW[0], FW_f[0] = FWn[ib], FWn_f[ib]
275
		R = asarray([self.R(FWn[i], FWn) for i in range(len(FWn))])
276
		Rs = sum(R)
277
		P = asarray([self.p(R[i], Rs) for i in range(len(FWn))])
278
		isort = argsort(P)[-(self.NP - 1):]
279
		FW[1:], FW_f[1:] = asarray(FWn)[isort], FWn_f[isort]
280
		return FW, FW_f
281
282
	def initPopulation(self, task):
283
		r"""Initialize starting population.
284
285
		Args:
286
			task (Task): Optimization task.
287
288
		Returns:
289
			Tuple[numpy.ndarray, numpy.ndarray[float], Dict[str, Any]]:
290
				1. Initialized population.
291
				2. Initialized populations function/fitness values.
292
				3. Additional arguments:
293
					* Ah (numpy.ndarray): Initialized amplitudes.
294
295
		See Also:
296
			* :func:`NiaPy.algorithms.algorithm.Algorithm.initPopulation`
297
		"""
298
		FW, FW_f, d = Algorithm.initPopulation(self, task)
299
		Ah = self.initAmplitude(task)
300
		d.update({'Ah': Ah})
301
		return FW, FW_f, d
302
303
	def runIteration(self, task, FW, FW_f, xb, fxb, Ah, **dparams):
304
		r"""Core function of Fireworks algorithm.
305
306
		Args:
307
			task (Task): Optimization task.
308
			FW (numpy.ndarray): Current population.
309
			FW_f (numpy.ndarray[float]): Current populations function/fitness values.
310
			xb (numpy.ndarray): Global best individual.
311
			fxb (float): Global best individuals fitness/function value.
312
			Ah (numpy.ndarray): Current amplitudes.
313
			**dparams (Dict[str, Any)]: Additional arguments
314
315
		Returns:
316
			Tuple[numpy.ndarray, numpy.ndarray[float], Dict[str, Any]]:
317
				1. Initialized population.
318
				2. Initialized populations function/fitness values.
319
				3. Additional arguments:
320
					* Ah (numpy.ndarray): Initialized amplitudes.
321
322
		See Also:
323
			* :func:`FireworksAlgorithm.SparsksNo`.
324
			* :func:`FireworksAlgorithm.ExplosionAmplitude`
325
			* :func:`FireworksAlgorithm.ExplodeSpark`
326
			* :func:`FireworksAlgorithm.GaussianSpark`
327
			* :func:`FireworksAlgorithm.NextGeneration`
328
		"""
329
		iw, ib = argmax(FW_f), 0
330
		Ss, As = sum(FW_f[iw] - FW_f), sum(FW_f - FW_f[ib])
331
		S = [self.SparsksNo(FW_f[i], FW_f[iw], Ss) for i in range(self.NP)]
332
		A = [self.ExplosionAmplitude(FW_f[i], FW_f[ib], Ah, As) for i in range(self.NP)]
333
		FWn = [self.ExplodeSpark(FW[i], A[i], task) for i in range(self.NP) for _ in range(S[i])]
334
		for i in range(self.m): FWn.append(self.GaussianSpark(self.randint(self.NP), task))
335
		FW, FW_f = self.NextGeneration(FW, FW_f, FWn, task)
336
		return FW, FW_f, {'Ah': Ah}
337
338
class EnhancedFireworksAlgorithm(FireworksAlgorithm):
339
	r"""Implementation of enganced fireworks algorithm.
340
341
	Algorithm:
342
		Enhanced Fireworks Algorithm
343
344
	Date:
345
		2018
346
347
	Authors:
348
		Klemen Berkovič
349
350
	License:
351
		MIT
352
353
	Reference URL:
354
		https://ieeexplore.ieee.org/document/6557813/
355
356
	Reference paper:
357
		S. Zheng, A. Janecek and Y. Tan, "Enhanced Fireworks Algorithm," 2013 IEEE Congress on Evolutionary Computation, Cancun, 2013, pp. 2069-2077. doi: 10.1109/CEC.2013.6557813
358
359
	Attributes:
360
		Name (List[str]): List of strings representing algorithm names.
361
		Ainit (float): TODO
362
		Afinal (float): TODO
363
	"""
364
	Name = ['EnhancedFireworksAlgorithm', 'EFWA']
365
366
	@staticmethod
367
	def typeParameters():
368
		r"""Get dictionary with functions for checking values of parameters.
369
370
		Returns:
371
			Dict[str, Callable]:
372
				* Ainit (Callable[[Union[int, float]], bool]): TODO
373
				* Afinal (Callable[[Union[int, float]], bool]): TODO
374
375
		See Also:
376
			* :func:`FireworksAlgorithm.typeParameters`
377
		"""
378
		d = FireworksAlgorithm.typeParameters()
379
		d['Ainit'] = lambda x: isinstance(x, (float, int)) and x > 0
380
		d['Afinal'] = lambda x: isinstance(x, (float, int)) and x > 0
381
		return d
382
383
	def setParameters(self, Ainit=20, Afinal=5, **ukwargs):
384
		r"""Set EnhancedFireworksAlgorithm algorithms core parameters.
385
386
		Args:
387
			Ainit (float): TODO
388
			Afinal (float): TODO
389
			**ukwargs (Dict[str, Any]): Additional arguments.
390
391
      See Also:
392
      	* :func:`FireworksAlgorithm.setParameters`
393
		"""
394
		FireworksAlgorithm.setParameters(self, **ukwargs)
395
		self.Ainit, self.Afinal = Ainit, Afinal
396
397
	def initRanges(self, task):
398
		r"""Initialize ranges.
399
400
		Args:
401
			task (Task): Optimization task.
402
403
		Returns:
404
			Tuple[numpy.ndarray[float], numpy.ndarray[float], numpy.ndarray[float]]:
405
				1. Initial amplitude values over dimensions.
406
				2. Final amplitude values over dimensions.
407
				3. uAmin.
408
		"""
409
		Ainit, Afinal = fullArray(self.Ainit, task.D), fullArray(self.Afinal, task.D)
410
		return Ainit, Afinal, self.uAmin(Ainit, Afinal, task)
411
412
	def uAmin(self, Ainit, Afinal, task):
413
		r"""Calculate the value of `uAmin`.
414
415
		Args:
416
			Ainit (numpy.ndarray[float]): Initial amplitude values over dimensions.
417
			Afinal (numpy.ndarray[float]): Final amplitude values over dimensions.
418
			task (Task): Optimization task.
419
420
		Returns:
421
			numpy.ndarray[float]: uAmin.
422
		"""
423
		return Ainit - sqrt(task.Evals * (2 * task.nFES - task.Evals)) * (Ainit - Afinal) / task.nFES
424
425
	def ExplosionAmplitude(self, x_f, xb_f, Ah, As, A_min=None):
426
		r"""Calculate explosion amplitude.
427
428
		Args:
429
			x_f (float): Individuals function/fitness value.
430
			xb_f (float): Best individual function/fitness value.
431
			Ah (numpy.ndarray):
432
			As (): TODO.
433
			A_min (Optional[numpy.ndarray]): Minimal amplitude values.
434
			task (Task): Optimization task.
435
436
		Returns:
437
			numpy.ndarray: New amplitude.
438
		"""
439
		A = FireworksAlgorithm.ExplosionAmplitude(self, x_f, xb_f, Ah, As)
440
		ifix = where(A < A_min)
441
		A[ifix] = A_min[ifix]
442
		return A
443
444
	def GaussianSpark(self, x, xb, task):
445
		r"""Create new individual.
446
447
		Args:
448
			x (numpy.ndarray):
449
			xb (numpy.ndarray):
450
			task (Task): Optimization task.
451
452
		Returns:
453
			numpy.ndarray: New individual generated by gaussian noise.
454
		"""
455
		return self.Mapping(x + self.rand(task.D) * (xb - x) * self.normal(1, 1, task.D), task)
456
457
	def NextGeneration(self, FW, FW_f, FWn, task):
458
		r"""Generate new population.
459
460
		Args:
461
			FW (numpy.ndarray): Current population.
462
			FW_f (numpy.ndarray[float]): Current populations fitness/function values.
463
			FWn (numpy.ndarray): New population.
464
			task (Task): Optimization task.
465
466
		Returns:
467
			Tuple[numpy.ndarray, numpy.ndarray[float]]:
468
				1. New population.
469
				2. New populations fitness/function values.
470
		"""
471
		FWn_f = apply_along_axis(task.eval, 1, FWn)
472
		ib = argmin(FWn_f)
473
		if FWn_f[ib] < FW_f[0]: FW[0], FW_f[0] = FWn[ib], FWn_f[ib]
474
		for i in range(1, self.NP):
475
			r = self.randint(len(FWn))
476
			if FWn_f[r] < FW_f[i]: FW[i], FW_f[i] = FWn[r], FWn_f[r]
477
		return FW, FW_f
478
479
	def initPopulation(self, task):
480
		r"""Initialize population.
481
482
		Args:
483
			task (Task): Optimization task.
484
485
		Returns:
486
			Tuple[numpy.ndarray, numpy.ndarray[float], Dict[str, Any]]:
487
				1. Initial population.
488
				2. Initial populations fitness/function values.
489
				3. Additional arguments:
490
					* Ainit (numpy.ndarray): Initial amplitude values.
491
					* Afinal (numpy.ndarray): Final amplitude values.
492
					* A_min (numpy.ndarray): Minimal amplitude values.
493
494
		See Also:
495
			* :func:`FireworksAlgorithm.initPopulation`
496
		"""
497
		FW, FW_f, d = FireworksAlgorithm.initPopulation(self, task)
498
		Ainit, Afinal, A_min = self.initRanges(task)
499
		d.update({'Ainit': Ainit, 'Afinal': Afinal, 'A_min': A_min})
500
		return FW, FW_f, d
501
502
	def runIteration(self, task, FW, FW_f, xb, fxb, Ah, Ainit, Afinal, A_min, **dparams):
503
		r"""Core function of EnhancedFireworksAlgorithm algorithm.
504
505
		Args:
506
			task (Task): Optimization task.
507
			FW (numpy.ndarray): Current population.
508
			FW_f (numpy.ndarray[float]): Current populations fitness/function values.
509
			xb (numpy.ndarray): Global best individual.
510
			fxb (float): Global best individuals function/fitness value.
511
			Ah (numpy.ndarray[float]): Current amplitude.
512
			Ainit (numpy.ndarray[float]): Initial amplitude.
513
			Afinal (numpy.ndarray[float]): Final amplitude values.
514
			A_min (numpy.ndarray[float]): Minial amplitude values.
515
			**dparams (Dict[str, Any]): Additional arguments.
516
517
		Returns:
518
			Tuple[numpy.ndarray, numpy.ndarray[float], Dict[str, Any]]:
519
				1. Initial population.
520
				2. Initial populations fitness/function values.
521
				3. Additional arguments:
522
					* Ainit (numpy.ndarray): Initial amplitude values.
523
					* Afinal (numpy.ndarray): Final amplitude values.
524
					* A_min (numpy.ndarray): Minimal amplitude values.
525
		"""
526
		iw, ib = argmax(FW_f), 0
527
		Ss, As = sum(FW_f[iw] - FW_f), sum(FW_f - FW_f[ib])
528
		S = [self.SparsksNo(FW_f[i], FW_f[iw], Ss) for i in range(self.NP)]
529
		A = [self.ExplosionAmplitude(FW_f[i], FW_f[ib], Ah, As, A_min) for i in range(self.NP)]
530
		A_min = self.uAmin(Ainit, Afinal, task)
531
		FWn = [self.ExplodeSpark(FW[i], A[i], task) for i in range(self.NP) for _ in range(S[i])]
532
		for i in range(self.m): FWn.append(self.GaussianSpark(self.randint(self.NP), FW[ib], task))
533
		FW, FW_f = self.NextGeneration(FW, FW_f, FWn, task)
534
		return FW, FW_f, {'Ah': Ah, 'Ainit': Ainit, 'Afinal': Afinal, 'A_min': A_min}
535
536
class DynamicFireworksAlgorithmGauss(EnhancedFireworksAlgorithm):
537
	r"""Implementation of dynamic fireworks algorithm.
538
539
	Algorithm:
540
		Dynamic Fireworks Algorithm
541
542
	Date:
543
		2018
544
545
	Authors:
546
		Klemen Berkovič
547
548
	License:
549
		MIT
550
551
	Reference URL:
552
		http://ieeexplore.ieee.org/stamp/stamp.jsp?tp=&arnumber=6900485&isnumber=6900223
553
554
	Reference paper:
555
		S. Zheng, A. Janecek, J. Li and Y. Tan, "Dynamic search in fireworks algorithm," 2014 IEEE Congress on Evolutionary Computation (CEC), Beijing, 2014, pp. 3222-3229. doi: 10.1109/CEC.2014.6900485
556
557
	Attributes:
558
		Name (List[str]): List of strings representing algorithm names.
559
		A_cf (Union[float, int]): TODO
560
		C_a (Union[float, int]): TODO
561
		C_r (Union[float, int]): TODO
562
		epsilon (Union[float, int]): TODO
563
	"""
564
	Name = ['DynamicFireworksAlgorithmGauss', 'dynFWAG']
565
566 View Code Duplication
	@staticmethod
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
567
	def typeParameters():
568
		r"""Get dictionary with functions for checking values of parameters.
569
570
		Returns:
571
			Dict[str, Callable]:
572
				* A_cr (Callable[[Union[float, int], bool]): TODo
573
574
		See Also:
575
			* :func:`FireworksAlgorithm.typeParameters`
576
		"""
577
		d = FireworksAlgorithm.typeParameters()
578
		d['A_cf'] = lambda x: isinstance(x, (float, int)) and x > 0
579
		d['C_a'] = lambda x: isinstance(x, (float, int)) and x > 1
580
		d['C_r'] = lambda x: isinstance(x, (float, int)) and 0 < x < 1
581
		d['epsilon'] = lambda x: isinstance(x, (float, int)) and 0 < x < 1
582
		return d
583
584
	def setParameters(self, A_cf=20, C_a=1.2, C_r=0.9, epsilon=1e-8, **ukwargs):
585
		r"""Set core arguments of DynamicFireworksAlgorithmGauss.
586
587
		Args:
588
			A_cf (Union[int, float]):
589
			C_a (Union[int, float]):
590
			C_r (Union[int, float]):
591
			epsilon (Union[int, float]):
592
			**ukwargs (Dict[str, Any]): Additional arguments.
593
594
		See Also:
595
			* :func:`FireworksAlgorithm.setParameters`
596
		"""
597
		FireworksAlgorithm.setParameters(self, **ukwargs)
598
		self.A_cf, self.C_a, self.C_r, self.epsilon = A_cf, C_a, C_r, epsilon
599
600
	def initAmplitude(self, task):
601
		r"""Initialize amplitude.
602
603
		Args:
604
			task (Task): Optimization task.
605
606
		Returns:
607
			Tuple[numpy.ndarray, numpy.ndarray]:
608
				1. TODO
609
		"""
610
		return FireworksAlgorithm.initAmplitude(self, task), task.bRange
611
612
	def Mapping(self, x, task):
613
		r"""Fix out of bound solution/individual.
614
615
		Args:
616
			x (numpy.ndarray): Individual.
617
			task (Task): Optimization task.
618
619
		Returns:
620
			numpy.ndarray: Fixed individual.
621
		"""
622
		ir = where(x > task.Upper)
623
		x[ir] = self.uniform(task.Lower[ir], task.Upper[ir])
624
		ir = where(x < task.Lower)
625
		x[ir] = self.uniform(task.Lower[ir], task.Upper[ir])
626
		return x
627
628
	def repair(self, x, d, epsilon):
629
		r"""Repair solution.
630
631
		Args:
632
			x (numpy.ndarray): Individual.
633
			d (numpy.ndarray): Default value.
634
			epsilon (float): Limiting value.
635
636
		Returns:
637
			numpy.ndarray: Fixed solution.
638
		"""
639
		ir = where(x <= epsilon)
640
		x[ir] = d[ir]
641
		return x
642
643
	def NextGeneration(self, FW, FW_f, FWn, task):
644
		r"""TODO.
645
646
		Args:
647
			FW (numpy.ndarray): Current population.
648
			FW_f (numpy.ndarray[float]): Current populations function/fitness values.
649
			FWn (numpy.ndarray): New population.
650
			task (Task): Optimization task.
651
652
		Returns:
653
			Tuple[numpy.ndarray, numpy.ndarray[float]]:
654
				1. New population.
655
				2. New populations function/fitness values.
656
		"""
657
		FWn_f = apply_along_axis(task.eval, 1, FWn)
658
		ib = argmin(FWn_f)
659
		for i, f in enumerate(FW_f):
660
			r = self.randint(len(FWn))
661
			if FWn_f[r] < f: FW[i], FW_f[i] = FWn[r], FWn_f[r]
662
		FW[0], FW_f[0] = FWn[ib], FWn_f[ib]
663
		return FW, FW_f
664
665
	def uCF(self, xnb, xcb, xcb_f, xb, xb_f, Acf, task):
666
		r"""TODO.
667
668
		Args:
669
			xnb:
670
			xcb:
671
			xcb_f:
672
			xb:
673
			xb_f:
674
			Acf:
675
			task (Task): Optimization task.
676
677
		Returns:
678
			Tuple[]:
679
				1. TODO
680
		"""
681
		xnb_f = apply_along_axis(task.eval, 1, xnb)
682
		ib_f = argmin(xnb_f)
683
		if xnb_f[ib_f] <= xb_f: xb, xb_f = xnb[ib_f], xnb_f[ib_f]
684
		Acf = self.repair(Acf, task.bRange, self.epsilon)
685
		if xb_f >= xcb_f: xb, xb_f, Acf = xcb, xcb_f, Acf * self.C_a
686
		else: Acf = Acf * self.C_r
687
		return xb, xb_f, Acf
688
689
	def ExplosionAmplitude(self, x_f, xb_f, Ah, As, A_min=None):
690
		return FireworksAlgorithm.ExplosionAmplitude(self, x_f, xb_f, Ah, As)
691
692
	def initPopulation(self, task):
693
		r"""Initialize population.
694
695
		Args:
696
			task (Task): Optimization task.
697
698
		Returns:
699
			Tuple[numpy.ndarray, numpy.ndarray[float], Dict[str, Any]]:
700
				1. Initialized population.
701
				2. Initialized population function/fitness values.
702
				3. Additional arguments:
703
					* Ah (): TODO
704
					* Ab (): TODO
705
		"""
706
		FW, FW_f, _ = Algorithm.initPopulation(self, task)
707
		Ah, Ab = self.initAmplitude(task)
708
		return FW, FW_f, {'Ah': Ah, 'Ab': Ab}
709
710
	def runIteration(self, task, FW, FW_f, xb, fxb, Ah, Ab, **dparams):
711
		r"""Core function of DynamicFireworksAlgorithmGauss algorithm.
712
713
		Args:
714
			task (Task): Optimization task.
715
			FW (numpy.ndarray): Current population.
716
			FW_f (numpy.ndarray[float]): Current populations function/fitness values.
717
			xb (numpy.ndarray): Global best individual.
718
			fxb (float): Global best fitness/function value.
719
			Ah (): TODO
720
			Ab (): TODO
721
			**dparams (Dict[str, Any]): Additional arguments.
722
723
		Returns:
724
			Tuple[numpy.ndarray, numpy.ndarray[float], Dict[str, Any]]:
725
				1. New population.
726
				2. New populations fitness/function values.
727
				3. Additional arguments:
728
					* Ah (): TODO
729
					* Ab (): TODO
730
		"""
731
		iw, ib = argmax(FW_f), argmin(FW_f)
732
		Ss, As = sum(FW_f[iw] - FW_f), sum(FW_f - FW_f[ib])
733
		S, sb = [self.SparsksNo(FW_f[i], FW_f[iw], Ss) for i in range(len(FW))], self.SparsksNo(fxb, FW_f[iw], Ss)
734
		A = [self.ExplosionAmplitude(FW_f[i], FW_f[ib], Ah, As) for i in range(len(FW))]
735
		FWn, xnb = [self.ExplodeSpark(FW[i], A[i], task) for i in range(self.NP) for _ in range(S[i])], [self.ExplodeSpark(xb, Ab, task) for _ in range(sb)]
736
		for i in range(self.m): FWn.append(self.GaussianSpark(self.randint(self.NP), FW[ib], task))
737
		FW, FW_f = self.NextGeneration(FW, FW_f, FWn, task)
738
		iw, ib = argmax(FW_f), 0
739
		_, _, Ab = self.uCF(xnb, FW[ib], FW_f[ib], xb, fxb, Ab, task)
740
		return FW, FW_f, {'Ah': Ah, 'Ab': Ab}
741
742
class DynamicFireworksAlgorithm(DynamicFireworksAlgorithmGauss):
743
	r"""Implementation of dynamic fireworks algorithm.
744
745
	Algorithm:
746
		Dynamic Fireworks Algorithm
747
748
	Date:
749
		2018
750
751
	Authors:
752
		Klemen Berkovič
753
754
	License:
755
		MIT
756
757
	Reference URL:
758
		http://ieeexplore.ieee.org/stamp/stamp.jsp?tp=&arnumber=6900485&isnumber=6900223
759
760
	Reference paper:
761
	 	S. Zheng, A. Janecek, J. Li and Y. Tan, "Dynamic search in fireworks algorithm," 2014 IEEE Congress on Evolutionary Computation (CEC), Beijing, 2014, pp. 3222-3229. doi: 10.1109/CEC.2014.6900485
762
763
	Attributes:
764
		Name (List[str]): List of strings representing algorithm name.
765
766
	See Also:
767
		* :class:`NiaPy.algorithms.basic.DynamicFireworksAlgorithmGauss`
768
	"""
769
	Name = ['DynamicFireworksAlgorithm', 'dynFWA']
770
771
	def runIteration(self, task, FW, FW_f, xb, fxb, Ah, Ab, **dparams):
772
		r"""Core function of Dynamic Fireworks Algorithm.
773
774
		Args:
775
			task (Task): Optimization task
776
			FW (numpy.ndarray): Current population
777
			FW_f (numpy.ndarray[float]): Current population fitness/function values
778
			xb (numpy.ndarray): Current best solution
779
			fxb (float): Current best solution's fitness/function value
780
			Ah (): TODO
781
			Ab (): TODO
782
			**dparams:
783
784
		Returns:
785
			Tuple[numpy.ndarray, numpy.ndarray[float], Dict[str, Any]]:
786
				1. New population.
787
				2. New population function/fitness values.
788
				3. Additional arguments:
789
					* Ah (): TODO
790
					* Ab (): TODO
791
		"""
792
		iw, ib = argmax(FW_f), argmin(FW_f)
793
		Ss, As = sum(FW_f[iw] - FW_f), sum(FW_f - FW_f[ib])
794
		S, sb = [self.SparsksNo(FW_f[i], FW_f[iw], Ss) for i in range(len(FW))], self.SparsksNo(fxb, FW_f[iw], Ss)
795
		A = [self.ExplosionAmplitude(FW_f[i], FW_f[ib], Ah, As) for i in range(len(FW))]
796
		FWn, xnb = [self.ExplodeSpark(FW[i], A[i], task) for i in range(self.NP) for _ in range(S[i])], [self.ExplodeSpark(xb, Ab, task) for _ in range(sb)]
797
		FW, FW_f = self.NextGeneration(FW, FW_f, FWn, task)
798
		iw, ib = argmax(FW_f), 0
799
		_, _, Ab = self.uCF(xnb, FW[ib], FW_f[ib], xb, fxb, Ab, task)
800
		return FW, FW_f, {'Ah': Ah, 'Ab': Ab}
801
802
# vim: tabstop=3 noexpandtab shiftwidth=3 softtabstop=3
803