KuMiShi/Optim_Metaheuristique
1
0
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MOPSO refactor (1/2)

Browse Source
This commit is contained in:
KuMiShi
2026-01-16 20:50:06 +01:00
parent 18c9fa43c3
commit 54d2303fb9
2 changed files with 132 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
mopso.py
View File

@@ -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]
return rd.choice(self.archive)

98
particle.py
View File

@@ -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
return current_grid_stress
def updating_socs(self, socs, capacities):
pass