From 54d2303fb90090018a55a9bbf299ed4eeb2e8542 Mon Sep 17 00:00:00 2001 From: KuMiShi Date: Fri, 16 Jan 2026 20:50:06 +0100 Subject: [PATCH] MOPSO refactor (1/2) --- mopso.py | 78 +++++++++++++++++++++++++++++++----------- particle.py | 98 +++++++++++++++++++++++++++++++++++++++-------------- 2 files changed, 132 insertions(+), 44 deletions(-) diff --git a/mopso.py b/mopso.py index 39e672e..d36ca7d 100644 --- a/mopso.py +++ b/mopso.py @@ -2,7 +2,7 @@ import random as rd from .particle import Particle class MOPSO(): - def __init__(self, f_weigths:list, A_max:float, price_mean:float, price_std:float, 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): + 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 @@ -10,7 +10,7 @@ class MOPSO(): self.c1 = c1 # Individual trust self.c2 = c2 # Social trust self.archive_size = archive_size # Archive size - self.f_weigths = f_weigths # Weigths for aggregation of all function objective + self.f_weights = f_weights # Weigths for aggregation of all function objective self.surrogate = surrogate # Using AI calculation @@ -19,26 +19,38 @@ class MOPSO(): 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(times=self.times,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.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): - nb_iter = 0 if not self.surrogate: - while nb_iter < self.t: - nb_iter += 1 - # Selection of a leader - # Updating velocity and positions - # Checking boundaries - # Evaluating particles + 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.f_weights, self.prices, self.socs, self.socs_req, self.times) + # Update the archive - # Checking for best positions + self.update_archive() else: - while nb_iter < self.t: - nb_iter += 1 - # Selection of a leader + for t in range(self.t): + self.select_leader() # Selection of a leader # Updating velocity and positions # Checking boundaries # Evaluating particles @@ -50,8 +62,8 @@ class MOPSO(): times = [] for _ in range(nb_vehicles): # Minumun, we have one tick of charging/discharging during simulation - t_arrived = rd.randrange(0, (nb_of_ticks * delta_t - delta_t) +1, delta_t) - t_leaving = rd.randrange(t_arrived + delta_t, (nb_of_ticks * delta_t) +1, delta_t) + 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 @@ -79,9 +91,37 @@ class MOPSO(): # 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 = False + + + 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): - n = len(self.archive) # Archive length - rd_pos = rd.randrange(0, n, 1) - return self.archive[rd_pos] \ No newline at end of file + return rd.choice(self.archive) \ No newline at end of file diff --git a/particle.py b/particle.py index f44d791..c0fcfdb 100644 --- a/particle.py +++ b/particle.py @@ -1,16 +1,12 @@ import random as rd class Particle(): - def __init__(self, times:list, nb_vehicles:int=10, delta_t:int=60, nb_of_ticks:int=72, x_min=-100, x_max=100, alpha=0.1): + def __init__(self, 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= self.generate_state_of_charges() # States of charge (initial, requested) - - self.times = times # (arrived, leaving) - # Minima and maxima of a position value self.x_min = x_min self.x_max = x_max @@ -22,21 +18,51 @@ class Particle(): # 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) - self.eval = 0 #TODO: self.evaluate() + + # Evalution attributes + self.f_memory = [0,0,0] + self.eval = 0 + # Initial evaluation in MOPSO + def update_position(self): - for i in range(self.nb_vehicles): - new_pos_i = self.x[i] + self.v[i] - self.x[i] = new_pos_i + 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, c1, c2, w=0.4): - for i in range(self.nb_vehicles): - new_vel_i = w * self.v[i] + (self.p_best - self.x[i]) * c1 * self.r1[i] + (leader - self.x[i]) * c2 * self.r2[i] - self.v[i] = new_vel_i + 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 - #TODO: Modify for uses of ticks + # 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): @@ -58,36 +84,58 @@ class Particle(): return vel # Function objective - def evaluate(self, elec_prices, max_power): - pass + def evaluate(self,f_weights,elec_prices,socs,socs_req,times): + f1 = self.f1(elec_prices) + f2 = self.f2(socs,socs_req,times) + f3 = self.f3() + + # Keeping in memory evaluation of each objective for domination evaluation + memory = [] + memory.append(f1) + memory.append(f2) + memory.append(f3) + + # Global weigthed evaluation of the position + f = (f1 * f_weights[0]) + (f2 * f_weights[1]) + (f3 * f_weights[2]) + + # Best position check + if f < self.eval: + self.p_best = self.x + + # Updating the previous evaluation + self.f_memory = memory + self.eval = f # 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): + 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 - #TODO: Modify for uses of ticks # User's insatisfaction - def f2(self): + def f2(self,socs,socs_req,times): result = 0 for i in range(self.nb_vehicles): - soc_req_i = self.socs[i][1] - result += max(0, ) + leaving = times[i][1] + stress = socs_req[i] - socs[leaving][i] + result += max(0, stress) + return result # Network Stress def f3(self): - current_max = 0 + 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 - - #TODO: Modify for uses of ticks + 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 \ No newline at end of file + return current_grid_stress + + def updating_socs(self, socs, capacities): + pass \ No newline at end of file