Particle swarm algorithm with multiple threads

Question

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
    return total


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)]

    condition_wait = threading.Condition()


    producer = ParticleSwarmProducer(initial_particle_position=dict_shared_new_position,
                                     bounds=bounds,
                                     dict_shared_errors=dict_shared_errors,
                                     dict_shared_is_ready=dict_shared_is_ready,
                                     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):
        p = Particle(thread_id=i,
                     name='Thread ' + str(i+1),
                     dict_shared_errors=dict_shared_errors,
                     dict_shared_is_ready=dict_shared_is_ready,
                     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

threadLock = threading.Lock()


class Particle(threading.Thread):
    """ Represents one particle object with specific position/velocity"""
    def __init__(self,
                 thread_id,
                 name,
                 dict_shared_errors,
                 dict_shared_is_ready,
                 dict_shared_new_position,
                 dict_shared_best_positions,
                 fnc,
                 condition_wait
                 ):
        threading.Thread.__init__(self)

        self.thread_id = thread_id
        self.name = name

        self.dict_shared_errors = dict_shared_errors
        self.dict_shared_is_ready = dict_shared_is_ready
        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...')
            threadLock.acquire(blocking=True)
            self.dict_shared_is_ready[self.thread_id] = True
            threadLock.release()

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

            threadLock.acquire(blocking=True)
            position = self.dict_shared_new_positions[self.thread_id]
            error = self.fnc(position)

            if error < best_particle_error:
                self.dict_shared_best_positions[self.thread_id] = position
                best_particle_error = error

            self.dict_shared_errors[self.thread_id] = self.fnc(position)

            # set to not ready
            self.dict_shared_is_ready[self.thread_id] = False
            threadLock.release()
            # 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
    
    threadLock = threading.Lock()
    
    
    class ParticleSwarmProducer(threading.Thread):
        """ Synchronizes all particles and computes their next position/velocity"""
        def __init__(self,
                     initial_particle_position,
                     bounds,
                     dict_shared_errors,
                     dict_shared_is_ready,
                     dict_shared_new_position,
                     dict_shared_best_positions,
                     dict_velocity,
                     condition_wait):
    
            threading.Thread.__init__(self)
    
            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_is_ready = dict_shared_is_ready
            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:
                threadLock.acquire(blocking=True)
                ready = list(self.dict_shared_is_ready.values())
    
                # If all particles have finished their current jobs...
                if all(ready):
                    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.add_pos_to_out()
    
                    self.evaluate_all_particles()
                    self.update_all_particles()
    
                    print('All particles go!')
    
                    with self.condition_wait:
                        self.condition_wait.notifyAll()
                    i += 1
                threadLock.release()
                # 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)
    
        def add_pos_to_out(self):
            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]

Answer 1

Some suggestions:

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.
4. Code should ideally be grouped according to what it belongs with, rather than what it is. For example, you wouldn't group all classes in one file, so why group all the constants?
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).

[cat overflow. please hold.]