Optim_Metaheuristique/mopso.py

import random as rd
from .particle import Particle

class MOPSO():
    def __init__(self, f_weights:list, A_max:float, price_mean:float, price_std:float, capacities:list, n:int, t:int, w:float, c1:float, c2:float, archive_size:int=10, nb_vehicles:int=10, delta_t:int=60, nb_of_ticks:int=72, x_min=-100, x_max=100, v_alpha=0.1, surrogate=False):
        # Constants
        self.n = n # Number of particles
        self.t = t # Number of simulation iterations
        self.w = w # Inertia (for exploration)
        self.c1 = c1 # Individual trust
        self.c2 = c2 # Social trust
        self.archive_size = archive_size # Archive size
        self.f_weights = f_weights # Weigths for aggregation of all function objective

        self.surrogate = surrogate # Using AI calculation

        # Initialisation of particle's global parameters
        self.A_max = A_max # Network's power limit
        self.socs, self.socs_req = self.generate_state_of_charges(nb_vehicles,nb_of_ticks)
        self.times = self.generate_times(nb_vehicles, nb_of_ticks, delta_t)
        self.prices = self.generates_prices(price_mean,price_std) #TODO: Use RTE France prices for random prices generation according to number of ticks
        self.capacities = capacities

        # Particles of the simulation
        self.particles = [Particle(nb_vehicles=nb_vehicles, nb_of_ticks=nb_of_ticks, delta_t=delta_t, x_min=x_min, x_max=x_max, alpha=v_alpha) for _ in range(self.n)]
        self.archive = []
        self.leader = self.particles[0] # it doesnt matter as the first thing done is choosing a new leader
        
        for i in range(self.n):
            self.particles[i].evaluate(self.f_weights, self.prices, self.socs, self.socs_req, self.times)
        self.update_archive()
    
    def iterate(self):
        if not self.surrogate:
            for t in range(self.t):
                self.select_leader() # Selection of a leader
                for i in range(self.n):
                    # Updating velocity and positions
                    self.particles[i].update_velocity(leader_pos=self.leader.x, c1=self.c1, c2=self.c2, w=self.w)
                    self.particles[i].update_position()

                    # Checking boundaries + updating global state
                    self.particles[i].keep_boudaries(self.A_max)
                    self.particles[i].updating_socs(self.socs, self.capacities)

                    # Evaluating particles
                    self.particles[i].evaluate(self.prices, self.socs, self.socs_req, self.times)
                    self.particles[i].update_best()

                # Update the archive
                self.update_archive()
        else:
            for t in range(self.t):
                self.select_leader() # Selection of a leader
                # Updating velocity and positions
                # Checking boundaries
                # Evaluating particles
                # Update the archive
                # Checking for best positions
    
    # Generation of arriving and leaving times for every vehicle
    def generate_times(self, nb_vehicles, nb_of_ticks, delta_t):
        times = []
        for _ in range(nb_vehicles):
            # Minumun, we have one tick of charging/discharging during simulation
            t_arrived = rd.randrange(0, nb_of_ticks-1, 1)
            t_leaving = rd.randrange(t_arrived, nb_of_ticks, 1)
            times.append((t_arrived,t_leaving))
        return times
    
    # Genrates different prices in [mean - std, mean + std] range
    def generates_prices(self,nb_of_ticks:int, mean:float, std:float):
        prices = []
        for _ in range(nb_of_ticks):
            variation = rd.randrange(-(std*10), (std * 10) +1, 1) / 10 # Random float variation
            prices.append(mean + variation)
        return prices
    
    # Genrates the coordinated states of charges requested and initial (duplicated initially for other ticks)
    def generate_state_of_charges(self, nb_vehicles:int, nb_of_ticks:int):
        socs = []
        socs_req = []
        # We ensure soc_req is greater than what the soc_init is (percentage transformed into floats)
        for _ in range(nb_vehicles):
            soc_init = rd.randrange(0,100,1)
            soc_req = rd.randrange(soc_init+1, 101,1)

            # Creating states of charges for each tick in time
            for _ in range(nb_of_ticks):
                socs.append(soc_init/100)
            
            # Adding the requested state of charge
            socs_req.append(soc_req/100)
        return socs, socs_req
    
    # True if a dominates b, else false
    def dominates(a:Particle, b:Particle):
        dominates = (a.f_current[0] >= b.f_current[0]) and (a.f_current[1] >= b.f_current[1]) and (a.f_current[2] >= b.f_current[2])
        if dominates:
            # Not strict superiority yet
            dominates = (a.f_current[0] > b.f_current[0]) or (a.f_current[1] > b.f_current[1]) or (a.f_current[2] > b.f_current[2])
        return dominates


    def update_archive(self):
        candidates = self.archive + self.particles
        length = len(candidates)

        non_dominated = []
        for i in range(length):
            candidate_i = candidates[i]
            dominates = True
            for j in range(length):
                if i!=j:
                    candidate_j = candidates[j]
                    dominates = dominates and self.dominates(candidate_i, candidate_j)
            if dominates:
                non_dominated.append(candidate_i)

        # Keeping only a certain number of solutions depending on archive_size (to avoid overloading the number of potential directions for particles)
        if len(non_dominated) > self.archive_size:
            final_non_dominated = []
            while len(final_non_dominated) < self.archive_size:
                final_non_dominated.append(rd.choice(non_dominated))
            self.archive = final_non_dominated
        else:
            self.archive = non_dominated


    # Random uniform selection for a leader
    def select_leader(self):
        return rd.choice(self.archive)