145 lines
6.0 KiB
Python
145 lines
6.0 KiB
Python
import random as rd
|
|
from particle import Particle
|
|
import copy
|
|
|
|
class MOPSO():
|
|
def __init__(self, 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.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)
|
|
self.prices = self.generates_prices(nb_of_ticks,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(
|
|
socs=copy.deepcopy(self.socs),
|
|
times=self.times, # Ajouté ici
|
|
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.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):
|
|
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.uniform(-std, std) # 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):
|
|
# Structure souhaitée : socs[tick][vehicle] pour être cohérent avec self.x[tick][vehicle]
|
|
socs = [[0.0 for _ in range(nb_vehicles)] for _ in range(nb_of_ticks)]
|
|
socs_req = []
|
|
|
|
for i in range(nb_vehicles):
|
|
soc_init = rd.randrange(0, 100, 1)
|
|
soc_req = rd.randrange(soc_init + 1, 101, 1)
|
|
|
|
# Remplissage de la matrice 2D
|
|
for tick in range(nb_of_ticks):
|
|
socs[tick][i] = soc_init / 100.0
|
|
|
|
socs_req.append(soc_req / 100.0)
|
|
|
|
return socs, socs_req
|
|
|
|
# True if a dominates b, else false
|
|
def dominates(self, 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]
|
|
is_dominated = False
|
|
|
|
for j in range(length):
|
|
if i != j:
|
|
candidate_j = candidates[j]
|
|
if self.dominates(candidate_j, candidate_i):
|
|
is_dominated = True
|
|
break
|
|
if not is_dominated:
|
|
non_dominated.append(candidate_i)
|
|
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) |