# Particle swarm algorithm with multiple threads

I wrote a simple multitthreading program which does particle swarm optimization. Optimization itself and some corresponding functions were taken from https://nathanrooy.github.io/posts/2016-08-17/simple-particle-swarm-optimization-with-python/. This is my first attempt to write multithread Python code :)

Code has 4 files:

1. const.py - just some constants
2. main.py - entry point
3. particle.py particle class as a thread
4. particle_producer - class which synchronizes all particles and computes next position/velocity

Thank you for your help :)

const.py

MAX_ITERATIONS = 300
NUMBER_OF_PARTICLES = 10
NUM_DIMENSIONS = 2


main.py

import threading
import random

from consts import NUM_DIMENSIONS, NUMBER_OF_PARTICLES
from particle import Particle
from particle_producer import ParticleSwarmProducer

import numpy as np

# Function to optimize
def fnc(x):
total=0
for i in range(len(x)):
total+=x[i]**2

if __name__ == '__main__':
x_down, x_up = (-100, 100)

# All shared datastrucures between particles and particle producer
dict_shared_new_position = {i: list(np.random.uniform(x_down, x_up, NUM_DIMENSIONS)) for i in range(0, NUMBER_OF_PARTICLES)}
dict_shared_best_position = dict_shared_new_position.copy()
dict_velocity = {i: [random.uniform(-1,1)] * NUM_DIMENSIONS for i in range(0, NUMBER_OF_PARTICLES)}
dict_shared_errors = {i: -1 for i in range(0, NUMBER_OF_PARTICLES)}
dict_shared_is_ready = {i: False for i in range(0, NUMBER_OF_PARTICLES)}

bounds = [(x_down, x_up) for i in range(NUMBER_OF_PARTICLES)]

producer = ParticleSwarmProducer(initial_particle_position=dict_shared_new_position,
bounds=bounds,
dict_shared_errors=dict_shared_errors,
dict_shared_new_position=dict_shared_new_position,
dict_shared_best_positions=dict_shared_best_position,
dict_velocity=dict_velocity,
condition_wait=condition_wait
)

particles = []

for i in range(NUMBER_OF_PARTICLES):
name='Thread ' + str(i+1),
dict_shared_errors=dict_shared_errors,
dict_shared_new_position=dict_shared_new_position,
dict_shared_best_positions=dict_shared_best_position,
fnc=fnc,
condition_wait=condition_wait)
particles.append(p)

producer.start()

for p in particles:
p.start()

producer.join()

for p in particles:
p.join()


particle.py

import threading

from consts import MAX_ITERATIONS

""" Represents one particle object with specific position/velocity"""
def __init__(self,
name,
dict_shared_errors,
dict_shared_new_position,
dict_shared_best_positions,
fnc,
condition_wait
):

self.name = name

self.dict_shared_errors = dict_shared_errors
self.dict_shared_new_positions = dict_shared_new_position
self.dict_shared_best_positions = dict_shared_best_positions

self.fnc = fnc
self.condition_wait = condition_wait

def run(self) -> None:

it = 1
best_particle_error = 9999

while it < MAX_ITERATIONS+1:
# print(f'{self.name} waiting for a new job...')

with self.condition_wait:
self.condition_wait.wait()

error = self.fnc(position)

if error < best_particle_error:
best_particle_error = error

# set to not ready
# print(f'{self.name} working on task')
it +=1
# print(f'{self.name}, {it}')


particle_producer.py

    import threading
import time
import random
import numpy as np

from consts import NUM_DIMENSIONS, NUMBER_OF_PARTICLES, MAX_ITERATIONS

""" Synchronizes all particles and computes their next position/velocity"""
def __init__(self,
initial_particle_position,
bounds,
dict_shared_errors,
dict_shared_new_position,
dict_shared_best_positions,
dict_velocity,
condition_wait):

self.initial_particle_positions = initial_particle_position
self.bounds = bounds

self.dict_velocity = dict_velocity

self.dict_best_positions = dict_shared_best_positions
self.dict_shared_errors = dict_shared_errors
self.dict_shared_new_position = dict_shared_new_position

self.condition_wait = condition_wait

self.current_iteration = 0
self.err_best_g = -1  # best error for group
self.pos_best_g = []  # best position for group

# Used for plotting
self.output_pos = {i: np.empty((0, NUM_DIMENSIONS)) for i in range(NUMBER_OF_PARTICLES)}

def run(self) -> None:

i = 1
while i < MAX_ITERATIONS+1:

# If all particles have finished their current jobs...
print(f'Iteration {i}')
print(f'Current positions: {self.dict_shared_new_position}')
print(f'Current errors: {self.dict_shared_errors}')
print(f'Current velocities: {self.dict_velocity}')

self.evaluate_all_particles()
self.update_all_particles()

print('All particles go!')

with self.condition_wait:
self.condition_wait.notifyAll()
i += 1
# time.sleep(0.2)

time.sleep(0.02)
with self.condition_wait:
self.condition_wait.notifyAll()
print(f'Current positions: {self.dict_shared_new_position}')
print(f'Error: {self.err_best_g}')

def evaluate_all_particles(self):
for i in range(NUMBER_OF_PARTICLES):
if self.dict_shared_errors[i] < self.err_best_g or self.err_best_g == -1:
self.pos_best_g = list(self.dict_shared_new_position[i])
self.err_best_g = float(self.dict_shared_errors[i])

def update_all_particles(self):
for i in range(NUMBER_OF_PARTICLES):
self.update_velocity(i)
self.update_position(i)

for i in range(NUMBER_OF_PARTICLES):
self.output_pos[i] = np.vstack((self.output_pos[i], self.dict_shared_new_position[i]))

def update_velocity(self, i):
w = 0.5  # constant inertia weight (how much to weigh the previous velocity)
c1 = 1  # cognative constant
c2 = 2  # social constant

for j in range(0, NUM_DIMENSIONS):
r1 = random.random()
r2 = random.random()

vel_cognitive = c1 * r1 * (self.dict_best_positions[i][j] - self.dict_shared_new_position[i][j])
vel_social = c2 * r2 * (self.pos_best_g[j] - self.dict_shared_new_position[i][j])

self.dict_velocity[i][j] = w * self.dict_velocity[i][j] + vel_cognitive + vel_social

def update_position(self, i):
for j in range(0, NUM_DIMENSIONS):
self.dict_shared_new_position[i][j] = self.dict_shared_new_position[i][j] + self.dict_velocity[i][j]

# adjust maximum position if necessary
if self.dict_shared_new_position[i][j] > self.bounds[i][1]:
self.dict_shared_new_position[i][j] = self.bounds[i][1]

# adjust minimum position if neseccary
if self.dict_shared_new_position[i][j] < self.bounds[i][0]:
self.dict_shared_new_position[i][j] = self.bounds[i][0]



1. Run the code through black and isort --profile=black to format it more idiomatically without any manual work.
2. Run the code through flake8 or pylint to check for other maintainability issues.
3. Run the code through mypy, ideally with a strict configuration, and annotate accordingly. It'll help readers make sense of the arguments and return values, and makes type prefixes like dict_ redundant.
5. i = 1; while i < MAX_ITERATIONS+1: can be simplified as for iteration in range(1, MAX_ITERATIONS + 1), or more idiomatically, for index in range(MAX_ITERATIONS).
6. range(0, limit) can be simplified to range(limit).