164 lines
7.1 KiB
Python
164 lines
7.1 KiB
Python
import random
|
||
import numpy as np
|
||
from data_structures import Config # 算法配置参数类
|
||
from chromosome_utils import ChromosomeUtils # 染色体工具类
|
||
|
||
|
||
class GeneticOperator:
|
||
"""遗传操作类：实现交叉和变异操作，用于产生新个体"""
|
||
|
||
def __init__(self, config: Config, utils: ChromosomeUtils):
|
||
"""
|
||
初始化遗传操作器
|
||
:param config: 算法配置参数（如交叉概率、变异概率等）
|
||
:param utils: 染色体工具类实例（提供染色体修复等功能）
|
||
"""
|
||
self.config = config # 配置参数
|
||
self.utils = utils # 染色体工具
|
||
|
||
def two_point_crossover(self, parent1: np.ndarray, parent2: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
|
||
"""
|
||
两点交叉：在染色体上随机选择两个点，交换两点之间的基因片段
|
||
:param parent1: 父代染色体1
|
||
:param parent2: 父代染色体2
|
||
:return: 两个子代染色体（经过修复）
|
||
"""
|
||
length = self.utils.chromosome_length # 染色体总长度
|
||
if length < 2: # 染色体长度不足2时无法交叉，直接返回父代副本
|
||
return parent1.copy(), parent2.copy()
|
||
|
||
# 随机选择两个交叉点（point1 < point2）
|
||
point1 = random.randint(0, length // 2)
|
||
point2 = random.randint(point1 + 1, length - 1)
|
||
|
||
# 复制父代基因并交换片段
|
||
child1 = parent1.copy()
|
||
child2 = parent2.copy()
|
||
child1[point1:point2] = parent2[point1:point2] # 交换point1到point2之间的基因
|
||
child2[point1:point2] = parent1[point1:point2]
|
||
|
||
# 修复染色体（确保满足约束条件）
|
||
child1 = self.utils.repair_chromosome(child1)
|
||
child2 = self.utils.repair_chromosome(child2)
|
||
return child1, child2
|
||
|
||
def uniform_mutation(self, chromosome: np.ndarray) -> np.ndarray:
|
||
"""
|
||
均匀变异：对染色体的三层基因（企业层、能力层、数量层）进行随机变异
|
||
:param chromosome: 待变异的染色体
|
||
:return: 变异后的染色体（经过修复）
|
||
"""
|
||
mutated = chromosome.copy() # 复制原始染色体，避免直接修改
|
||
|
||
# 拆分染色体为三层
|
||
enterprise_layer, capacity_layer, quantity_layer = self.utils._split_chromosome(mutated)
|
||
split_points = np.cumsum(self.utils.material_enterprise_count)
|
||
|
||
# ========== 第一步：记录原始企业层状态 ==========
|
||
original_enterprise_layer = enterprise_layer.copy()
|
||
|
||
# ========== 第二步：企业层变异 ==========
|
||
start = 0
|
||
for i in range(self.utils.I):
|
||
end = split_points[i]
|
||
for idx in range(start, end):
|
||
if random.random() < self.config.mutation_prob:
|
||
enterprise_layer[idx] = 1 - enterprise_layer[idx]
|
||
start = end
|
||
|
||
# 修复企业层
|
||
enterprise_layer = self.utils.repair_enterprise_layer(enterprise_layer)
|
||
|
||
# ========== 第三步：同步更新能力和数量层 ==========
|
||
start = 0
|
||
for i in range(self.utils.I):
|
||
end = split_points[i]
|
||
ents = self.utils.material_optional_enterprises[i]
|
||
|
||
for idx in range(start, end):
|
||
current_ent_selected = enterprise_layer[idx] # 变异后的选择状态
|
||
original_ent_selected = original_enterprise_layer[idx] # 变异前的选择状态
|
||
|
||
# 情况1：企业从选中变为未选中（1→0）
|
||
if original_ent_selected == 1 and current_ent_selected == 0:
|
||
capacity_layer[idx] = 0 # 产能设为0
|
||
quantity_layer[idx] = 0 # 数量设为0
|
||
|
||
# 情况2：企业从未选中变为选中（0→1）
|
||
elif original_ent_selected == 0 and current_ent_selected == 1:
|
||
ent = ents[idx - start]
|
||
|
||
# 初始化产能
|
||
if ent == 0:
|
||
max_cap = self.utils.risk.C0_i_std[i]
|
||
else:
|
||
supplier_id = ent - 1
|
||
max_cap = self.utils.supplier.Cj_i_std[supplier_id][i]
|
||
capacity_layer[idx] = random.uniform(1, max_cap)
|
||
|
||
# 初始化数量
|
||
qi = self.utils.order.Q[i]
|
||
if ent == 0:
|
||
max_q = qi
|
||
else:
|
||
supplier_id = ent - 1
|
||
max_q = self.utils.supplier.MaxOrder[supplier_id][i]
|
||
quantity_layer[idx] = random.uniform(1, max_q)
|
||
|
||
# 情况3：企业保持选中（1→1）或保持未选中（0→0）
|
||
# 保持不变，后续的变异步骤会处理
|
||
|
||
start = end
|
||
|
||
# ========== 第四步：能力层变异（仅对保持选中的企业） ==========
|
||
start = 0
|
||
for i in range(self.utils.I):
|
||
end = split_points[i]
|
||
ents = self.utils.material_optional_enterprises[i]
|
||
e_segment = enterprise_layer[start:end]
|
||
|
||
for idx in range(start, end):
|
||
# 只对保持选中的企业进行变异（1→1的情况）
|
||
if e_segment[idx - start] == 1 and original_enterprise_layer[idx] == 1:
|
||
if random.random() < self.config.mutation_prob:
|
||
ent = ents[idx - start]
|
||
if ent == 0:
|
||
max_cap = self.utils.risk.C0_i_std[i]
|
||
else:
|
||
supplier_id = ent - 1
|
||
max_cap = self.utils.supplier.Cj_i_std[supplier_id][i]
|
||
capacity_layer[idx] = random.uniform(1, max_cap)
|
||
|
||
start = end
|
||
|
||
# 修复能力层（考虑所有选中企业的产能）
|
||
capacity_layer = self.utils.repair_capacity_layer(enterprise_layer, capacity_layer)
|
||
|
||
# ========== 第五步：数量层变异（仅对保持选中的企业） ==========
|
||
start = 0
|
||
for i in range(self.utils.I):
|
||
end = split_points[i]
|
||
ents = self.utils.material_optional_enterprises[i]
|
||
e_segment = enterprise_layer[start:end]
|
||
|
||
for idx in range(start, end):
|
||
# 只对保持选中的企业进行变异（1→1的情况）
|
||
if e_segment[idx - start] == 1 and original_enterprise_layer[idx] == 1:
|
||
if random.random() < self.config.mutation_prob:
|
||
ent = ents[idx - start]
|
||
qi = self.utils.order.Q[i]
|
||
if ent == 0:
|
||
max_q = qi
|
||
else:
|
||
supplier_id = ent - 1
|
||
max_q = self.utils.supplier.MaxOrder[supplier_id][i]
|
||
quantity_layer[idx] = random.uniform(1, max_q)
|
||
|
||
start = end
|
||
|
||
# 修复数量层（考虑所有选中企业的分配）
|
||
quantity_layer = self.utils.repair_quantity_layer(enterprise_layer, quantity_layer)
|
||
|
||
# 合并三层
|
||
mutated = self.utils._merge_chromosome(enterprise_layer, capacity_layer, quantity_layer)
|
||
return mutated |