Optim_Metaheuristique/particle.py

import random as rd
import copy

class Particle():
    def __init__(self, socs:list, times:list, nb_vehicles:int=10, delta_t:int=60, nb_of_ticks:int=72, x_min=-100, x_max=100, alpha=0.1):
        # Problem specific attributes
        self.nb_vehicles = nb_vehicles # Number of vehicles handles for the generations of position x
        self.delta_t = delta_t # delta_t for update purposes
        self.nb_of_ticks = nb_of_ticks # Accounting for time evolution of the solution (multiplied by delta_t)
        self.socs = socs # States of charges for the particle current position (self.x)
        self.times = times

        # Minima and maxima of a position value
        self.x_min = x_min
        self.x_max = x_max

        # Limitation of the velocity
        self.alpha = alpha
        self.r1 = [rd.randrange(0,101,1)/100 for _ in range(self.nb_vehicles)] # Variable trust of oneself
        self.r2 = [rd.randrange(0,101,1)/100 for _ in range(self.nb_vehicles)] # Variable trust of other particles

        # Particle attributes
        self.x = self.generate_position() # Position Vector (correspond to one solution for the problem)
        self.clean_position() # Staying coherent with problem modelisation for a_i,t
        self.v = self.generate_velocity() # Velocity
        self.p_best = self.x # Best known position (starting with initial position x)

        # Evalution attributes
        self.f_best = [(self.nb_of_ticks * self.nb_vehicles * self.x_max * 100) for _ in range(3)]# Hundred times the max grid power should be large enough to be out of scope (equivalent to inf)
        self.f_current = [0,0,0]  # [f1,f2,f3]


    def update_position(self):
        for tick in range(self.nb_of_ticks):
            for i in range(self.nb_vehicles):
                new_pos_i_t = self.x[tick][i] + self.v[tick][i]
                self.x[tick][i] = new_pos_i_t
        self.clean_position()

    def update_velocity(self, leader_pos, c1, c2, w=0.4):
        for tick in range(self.nb_of_ticks):
            for i in range(self.nb_vehicles):
                new_vel_i_t = w * self.v[tick][i] + (self.p_best[tick][i] - self.x[tick][i]) * c1 * self.r1[i] + (leader_pos[tick][i] - self.x[tick][i]) * c2 * self.r2[i]
                self.v[tick][i] = new_vel_i_t
    
    # BELOW: Modifying position values to keep logical states

    def clean_position(self):
        for tick in range(self.nb_of_ticks):
            for i in range(self.nb_vehicles):
                arriving = self.times[i][0]
                leaving = self.times[i][1]
                # x[arriving][i] != 0 and x[leaving][i] == 0
                if not(tick >= arriving and tick < leaving):
                    self.x[tick][i] = 0.0

    # Done after evaluation to correct out of bounds position
    def keep_boudaries(self,max_power):
        for tick in range(self.nb_of_ticks):
            current_power = self.get_current_grid_stress(tick)
            # As long as there is too much power, we cut supplies from charging vehicles (keeping discharging other vehicles at same current rate)
            while current_power > max_power:
                for i in range(self.nb_vehicles):
                    if self.x[tick][i] > 0:
                        self.x[tick][i] = self.x[tick][i] * 0.9
                current_power = self.get_current_grid_stress(tick)

    def generate_position(self):
        pos = []
        for _ in range(self.nb_of_ticks):
            x_tick = []
            for _ in range(self.nb_vehicles):
                x_tick.append(rd.randrange(self.x_min, self.x_max +1, 1))
            pos.append(x_tick)
        return pos
    
    # Randomize a velocity vector for each tick
    def generate_velocity(self):
        vel = []
        vel_coeff = abs(self.x_max - self.x_min) 
        for _ in range(self.nb_of_ticks):
            v_tick = []
            for _ in range(self.nb_vehicles):
                v_tick.append(rd.randrange(-vel_coeff, vel_coeff +1, 1) * self.alpha)
            vel.append(v_tick)
        return vel
    
    # Function objective
    def evaluate(self,elec_prices,socs,socs_req,times):
        f1 = self.f1(elec_prices)
        f2 = self.f2(self.socs,socs_req,times)
        f3 = self.f3()

        # Keeping in memory evaluation of each objective for domination evaluation
        f_current = []
        f_current.append(f1)
        f_current.append(f2)
        f_current.append(f3)
        
        self.f_current = f_current

    def update_best(self):
        current_better = (self.f_current[0] <= self.f_best[0]) and (self.f_current[1] <= self.f_best[1]) and (self.f_current[2] <= self.f_best[2])
        if current_better:
            # Not strict superiority yet
            current_dominates = (self.f_current[0] < self.f_best[0]) or (self.f_current[1] < self.f_best[1]) or (self.f_current[2] < self.f_best[2])
            if current_dominates:
                self.p_best = copy.deepcopy(self.x)
                self.f_best = self.f_current[:]

    # Calculate the price of the electricity consumption in the grid SUM(1_to_T)(Epsilon_t * A_t * delta_t)
    def f1(self,elec_prices):
        result = 0
        for tick in range(self.nb_of_ticks):
            grid_stress_tick = self.get_current_grid_stress(tick)
            result += elec_prices[tick] * grid_stress_tick * self.delta_t
        return result
    
    # User's insatisfaction
    def f2(self,socs,socs_req,times):
        result = 0
        for i in range(self.nb_vehicles):
            leaving = times[i][1]
            stress = socs_req[i] - socs[leaving][i]
            result += max(0, stress)
        return result

    # Network Stress
    def f3(self):
        current_max = self.nb_vehicles * self.x_min
        for tick in range(self.nb_of_ticks):
            current_max = max(current_max, self.get_current_grid_stress(tick))
        return current_max

    def get_current_grid_stress(self, tick:int):
        assert tick < self.nb_of_ticks # Make sure the tick exist in the position x
        current_grid_stress = 0
        for i in range(self.nb_vehicles):
            current_grid_stress += self.x[tick][i]
        return current_grid_stress
    
    def updating_socs(self, initial_socs, capacities):
    
        # Calcul de l'évolution temporelle
        for tick in range(self.nb_of_ticks - 1): # On s'arrête à l'avant-dernier pour calculer le suivant
            for i in range(self.nb_vehicles):
                # SoC(t+1) = SoC(t) + (Puissance(t) * delta_t / Capacité)
                # Attention: x est en kW, delta_t en minutes -> conversion en heures (/60) si capacité en kWh
                energy_added = (self.x[tick][i] * (self.delta_t / 60)) 
                
                # Mise à jour du tick suivant basé sur le tick actuel
                # On utilise initial_socs comme base si c'est une liste de listes [tick][vehicule]
                self.socs[tick+1][i] = self.socs[tick][i] + (energy_added / capacities[i])
                
                # Bornage entre 0 et 1 (0% et 100%)
                self.socs[tick+1][i] = max(0.0, min(1.0, self.socs[tick+1][i]))