in nevergrad/functions/powersystems/core.py [0:0]
def _simulate_power_system(self, *arrays: np.ndarray) -> float:
failure_cost = (
self.failure_cost
) # Cost of power demand which is not satisfied (equivalent to a expensive infinite thermal group).
dam_agents = self.dam_agents
for agent, array in zip(dam_agents, arrays):
agent.set_parameters(array)
self.marginal_costs = []
num_dams = int(self.num_dams)
# Assume empty initial stocks.
stocks = np.zeros((num_dams,))
# Nonsense delays.
delay = np.cos(np.arange(num_dams))
cost = 0.0
# Loop on time steps.
num_time_steps = int(365 * 24 * self.number_of_years)
consumption = 0.0
hydro_prod_per_time_step: tp.List[tp.Any] = []
consumption_per_time_step: tp.List[float] = []
for t in range(num_time_steps):
# Rain
stocks += 0.5 * (1.0 + np.cos(2 * pi * t / (24 * 365) + delay)) * np.random.rand(num_dams)
# Consumption model.
base_consumption = (
self.constant_to_year_ratio * self.year_to_day_ratio
+ 0.5 * self.year_to_day_ratio * (1.0 + cos(2 * pi * t / (24 * 365)))
+ 0.5 * (1.0 + cos(2 * pi * t / 24))
)
if t == 0:
consumption = base_consumption
else:
consumption = max(
0.0,
consumption
+ self.consumption_noise * (np.random.rand() - 0.5)
+ self.back_to_normal * (base_consumption - consumption),
)
consumption_per_time_step += [consumption]
# "Needed" stores what we need, and will decrease as we use various power plants for producing.
needed = consumption
# Setting inputs for all agents.
base_x = [
cos(2 * pi * t / 24.0),
sin(2 * pi * t / 24.0),
cos(2 * pi * t / (365 * 24)),
sin(2 * pi * t / (365 * 24)),
needed,
self.average_consumption,
self.year_to_day_ratio,
self.constant_to_year_ratio,
self.back_to_normal,
self.consumption_noise,
]
x = np.concatenate((base_x, self.thermal_power_capacity, self.thermal_power_prices, stocks))
# Prices as a decomposition tool!
price: np.ndarray = np.asarray([a.get_output(x)[0] for a in dam_agents])
dam_index = np.arange(num_dams)
price = np.concatenate((price, self.thermal_power_prices))
capacity = np.concatenate((stocks, self.thermal_power_capacity))
dam_index = np.concatenate((dam_index, -1 * np.ones(len(price), dtype=int)))
assert len(price) == num_dams + self.num_thermal_plants
hydro_prod: np.ndarray = np.zeros(num_dams)
# Let us rank power plants by production cost.
order = sorted(range(len(price)), key=lambda x: price[x]) # pylint: disable=cell-var-from-loop
price = price[order]
capacity = capacity[order]
dam_index = dam_index[order]
# Using power plants in their cost order, so that we use cheap power plants first.
marginal_cost = 0.0
for i, _ in enumerate(price):
if needed <= 0:
break
production = min(capacity[i], needed)
# If this is a dam, producing will reduce the stock.
if dam_index[i] >= 0:
hydro_prod[dam_index[i]] += production # Let us log the hydro prod for this dam.
stocks[dam_index[i]] -= production
assert stocks[dam_index[i]] >= -1e-7
else:
# If this is not a dam, we pay for using thermal plants.
cost += production * price[i]
if production > 1e-7:
marginal_cost = price[i]
needed -= production
# Cost in case of failures -- this is
# harming industries and hospitals, so it can be penalized.
cost += failure_cost * needed
if needed > 1e-7:
marginal_cost = failure_cost
self.marginal_costs += [marginal_cost]
hydro_prod_per_time_step += [hydro_prod]
# Other data of interest: , hydro_prod, hydro_prod_per_time_step, consumption_per_time_step
assert len(hydro_prod_per_time_step) == num_time_steps # Each time steps has 1 value per dam.
assert len(consumption_per_time_step) == num_time_steps
self.hydro_prod_per_time_step = hydro_prod_per_time_step
self.consumption_per_time_step = consumption_per_time_step
self.losses += [cost]
return cost