LU/smt-optimizer
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

启发式产线分配

Browse Source
This commit is contained in:
hit_lu
2023-07-16 23:19:34 +08:00
parent 6e56f796f0
commit 96430a4b6c
11 changed files with 439 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
base_optimizer/optimizer_aggregation.py
View File

@ -1,15 +1,18 @@
from base_optimizer.optimizer_common import *
from ortools.sat.python import cp_model
from gurobipy import *
from collections import defaultdict
def list_range(start, end=None):
return list(range(start)) if end is None else list(range(start, end))
@timer_wrapper
def optimizer_aggregation(component_data, pcb_data):
# === phase 0: data preparation ===
M = 1000 # a sufficient large number
a, b = 1, 6 # coefficient
K, I, J, L = max_head_index, 0, 0, 0 # the maximum number of heads, component types, nozzle types and batch level
component_list, nozzle_list = defaultdict(int), defaultdict(int)
cpidx_2_part, nzidx_2_nozzle = {}, {}
@ -26,10 +29,11 @@ def optimizer_aggregation(component_data, pcb_data):
nzidx_2_nozzle[len(nzidx_2_nozzle)] = nozzle
nozzle_list[nozzle] += 1
I, J = len(component_list.keys()), len(nozzle_list.keys())
L = I + 1
HC = [[M for _ in range(J)] for _ in range(I)] # the handing class when component i is handled by nozzle type j
# represent the nozzle-component compatibility
I, J = len(component_list.keys()), len(nozzle_list.keys()) # the maximum number of component types and nozzle types
L = I + 1 # the maximum number of batch level
K = max_head_index # the maximum number of heads
HC = [[M for _ in range(J)] for _ in range(I)] # represent the nozzle-component compatibility
for i in range(I):
for _, item in enumerate(cpidx_2_part.items()):
index, part = item
@ -41,105 +45,71 @@ def optimizer_aggregation(component_data, pcb_data):
HC[index][j] = 0
# === phase 1: mathematical model solver ===
model = cp_model.CpModel()
solver = cp_model.CpSolver()
mdl = Model('SMT')
mdl.setParam('OutputFlag', 0)
# === Decision Variables ===
# the number of components of type i that are placed by nozzle type j on placement head k
X = {}
for i in range(I):
for j in range(J):
for k in range(K):
X[i, j, k] = model.NewIntVar(0, component_list[cpidx_2_part[i]], 'X_{}_{}_{}'.format(i, j, k))
X = mdl.addVars(list_range(I), list_range(J), list_range(K), vtype=GRB.INTEGER, ub=max(component_list.values()))
# the total number of nozzle changes on placement head k
N = {}
for k in range(K):
N[k] = model.NewIntVar(0, J, 'N_{}'.format(k))
N = mdl.addVars(list_range(K), vtype=GRB.INTEGER)
# the largest workload of all placement heads
WL = model.NewIntVar(0, len(pcb_data), 'WL')
WL = mdl.addVar(vtype=GRB.INTEGER, lb=0, ub=len(pcb_data))
# whether batch Xijk is placed on level l
Z = {}
for i in range(I):
for j in range(J):
for l in range(L):
for k in range(K):
Z[i, j, l, k] = model.NewBoolVar('Z_{}_{}_{}_{}'.format(i, j, l, k))
Z = mdl.addVars(list_range(I), list_range(J), list_range(L), list_range(K), vtype=GRB.BINARY)
# Dlk := 2 if a change of nozzles in the level l + 1 on placement head k
# Dlk := 1 if there are no batches placed on levels higher than l
D = {}
for l in range(L):
for k in range(K):
D[l, k] = model.NewIntVar(0, 2, 'D_{}_{}'.format(l, k))
D_abs = {}
for l in range(L):
for j in range(J):
for k in range(K):
D_abs[l, j, k] = model.NewIntVar(0, M, 'D_abs_{}_{}_{}'.format(l, j, k))
# Dlk := 0 otherwise
D = mdl.addVars(list_range(L), list_range(K), vtype=GRB.BINARY, ub=2)
D_plus = mdl.addVars(list_range(L), list_range(J), list_range(K), vtype=GRB.INTEGER)
D_minus = mdl.addVars(list_range(L), list_range(J), list_range(K), vtype=GRB.INTEGER)
# == Objective function ===
model.Minimize(a * WL + b * sum(N[k] for k in range(K)))
mdl.modelSense = GRB.MINIMIZE
mdl.setObjective(a * WL + b * quicksum(N[k] for k in range(K)))
# === Constraint ===
for i in range(I):
model.Add(sum(X[i, j, k] for j in range(J) for k in range(K)) == component_list[cpidx_2_part[i]])
mdl.addConstrs(
quicksum(X[i, j, k] for j in range(J) for k in range(K)) == component_list[cpidx_2_part[i]] for i in range(I))
for k in range(K):
model.Add(sum(X[i, j, k] for i in range(I) for j in range(J)) <= WL)
mdl.addConstrs(quicksum(X[i, j, k] for i in range(I) for j in range(J)) <= WL for k in range(K))
for i in range(I):
for j in range(J):
for k in range(K):
model.Add(X[i, j, k] <= M * sum(Z[i, j, l, k] for l in range(L)))
mdl.addConstrs(
X[i, j, k] <= M * quicksum(Z[i, j, l, k] for l in range(L)) for i in range(I) for j in range(J) for k in
range(K))
for i in range(I):
for j in range(J):
for k in range(K):
model.Add(sum(Z[i, j, l, k] for l in range(L)) <= 1)
mdl.addConstrs(quicksum(Z[i, j, l, k] for l in range(L)) <= 1 for i in range(I) for j in range(J) for k in range(K))
mdl.addConstrs(
quicksum(Z[i, j, l, k] for l in range(L)) <= X[i, j, k] for i in range(I) for j in range(J) for k in range(K))
for i in range(I):
for j in range(J):
for k in range(K):
model.Add(sum(Z[i, j, l, k] for l in range(L)) <= X[i, j, k])
mdl.addConstrs(quicksum(Z[i, j, l, k] for j in range(J) for i in range(I)) >= quicksum(
Z[i, j, l + 1, k] for j in range(J) for i in range(I)) for k in range(K) for l in range(L - 1))
for k in range(K):
for l in range(L - 1):
model.Add(sum(Z[i, j, l, k] for j in range(J) for i in range(I)) >= sum(
Z[i, j, l + 1, k] for j in range(J) for i in range(I)))
mdl.addConstrs(quicksum(Z[i, j, l, k] for i in range(I) for j in range(J)) <= 1 for k in range(K) for l in range(L))
mdl.addConstrs(D_plus[l, j, k] - D_minus[l, j, k] == quicksum(Z[i, j, l, k] for i in range(I)) - quicksum(
Z[i, j, l + 1, k] for i in range(I)) for l in range(L - 1) for j in range(J) for k in range(K))
for l in range(I):
for k in range(K):
model.Add(sum(Z[i, j, l, k] for i in range(I) for j in range(J)) <= 1)
mdl.addConstrs(
D[l, k] == quicksum((D_plus[l, j, k] + D_minus[l, j, k]) for j in range(J)) for k in range(K) for l in
range(L))
for l in range(L - 1):
for j in range(J):
for k in range(K):
model.AddAbsEquality(D_abs[l, j, k],
sum(Z[i, j, l, k] for i in range(I)) - sum(Z[i, j, l + 1, k] for i in range(I)))
for k in range(K):
for l in range(L):
model.Add(D[l, k] == sum(D_abs[l, j, k] for j in range(J)))
for k in range(K):
model.Add(N[k] == sum(D[l, k] for l in range(L)) - 1)
for l in range(L):
for k in range(K):
model.Add(0 >= sum(HC[i][j] * Z[i, j, l, k] for i in range(I) for j in range(J)))
mdl.addConstrs(2 * N[k] == quicksum(D[l, k] for l in range(L)) - 1 for k in range(K))
mdl.addConstrs(
0 >= quicksum(HC[i][j] * Z[i, j, l, k] for i in range(I) for j in range(J)) for l in range(L) for k in range(K))
# === Main Process ===
component_result, cycle_result = [], []
feeder_slot_result, placement_result, head_sequence = [], [], []
solver.parameters.max_time_in_seconds = 20.0
mdl.setParam("TimeLimit", 100)
status = solver.Solve(model)
if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE:
print('total cost = {}'.format(solver.ObjectiveValue()))
mdl.optimize()
if mdl.Status == GRB.OPTIMAL:
print('total cost = {}'.format(mdl.objval))
# convert cp model solution to standard output
model_cycle_result, model_component_result = [], []
@ -149,9 +119,9 @@ def optimizer_aggregation(component_data, pcb_data):
for k in range(K):
for i in range(I):
for j in range(J):
if solver.BooleanValue(Z[i, j, l, k]) != 0:
if abs(Z[i, j, l, k].x - 1) <= 1e-3:
model_component_result[-1][k] = cpidx_2_part[i]
model_cycle_result[-1][k] = solver.Value(X[i, j, k])
model_cycle_result[-1][k] = round(X[i, j, k].x)
# remove redundant term
if sum(model_cycle_result[-1]) == 0:
@ -209,7 +179,6 @@ def optimizer_aggregation(component_data, pcb_data):
if component_result[cycle_idx][head] == -1:
continue
index_ = component_result[cycle_idx][head]
placement_result[-1][head] = mount_point_pos[index_][-1][2]
mount_point_pos[index_].pop()
head_sequence.append(dynamic_programming_cycle_path(pcb_data, placement_result[-1], feeder_slot_result[cycle_idx]))

8
base_optimizer/optimizer_common.py
View File

@ -49,6 +49,12 @@ feeder_width = {'SM8': (7.25, 7.25), 'SM12': (7.00, 20.00), 'SM16': (7.00, 22.00
# 可用吸嘴数量限制
nozzle_limit = {'CN065': 6, 'CN040': 6, 'CN220': 6, 'CN400': 6, 'CN140': 6}
# 时间参数
t_cycle = 0.3
t_pick, t_place = .078, .051 # 贴装/拾取用时
t_nozzle_put, t_nozzle_pick = 0.9, 0.75 # 装卸吸嘴用时
t_nozzle_change = t_nozzle_put + t_nozzle_pick
t_fix_camera_check = 0.12 # 固定相机检测时间
def axis_moving_time(distance, axis=0):
distance = abs(distance) * 1e-3
@ -880,7 +886,7 @@ def constraint_swap_mutation(component_points, individual):
offspring = individual.copy()
idx, component_index = 0, random.randint(0, len(component_points) - 1)
for points in component_points.values():
for _, points in component_points:
if component_index == 0:
while True:
index1, index2 = random.sample(range(points + max_machine_index - 2), 2)

2
base_optimizer/optimizer_feederpriority.py
View File

@ -2,7 +2,7 @@ from base_optimizer.optimizer_common import *
@timer_wrapper
def feeder_allocate(component_data, pcb_data, feeder_data, nozzle_pattern, figure=False):
def feeder_allocate(component_data, pcb_data, feeder_data, figure=False):
feeder_points, feeder_division_points = defaultdict(int), defaultdict(int) # 供料器贴装点数
mount_center_pos = defaultdict(int)

59
base_optimizer/optimizer_hybridgenetic.py
View File

@ -234,8 +234,8 @@ def cal_individual_val(component_nozzle, component_point_pos, designated_nozzle,
return V[-1], pickup_result, pickup_cycle_result
def convert_individual_2_result(component_data, component_point_pos, designated_nozzle, pickup_group, pickup_group_cycle,
pair_group, feeder_lane, individual):
def convert_individual_2_result(component_data, component_point_pos, designated_nozzle, pickup_group,
pickup_group_cycle, pair_group, feeder_lane, individual):
component_result, cycle_result, feeder_slot_result = [], [], []
placement_result, head_sequence_result = [], []
@ -418,19 +418,19 @@ def optimizer_hybrid_genetic(pcb_data, component_data, hinter=True):
pick_part = pickup[pickup_index]
# 检查槽位占用情况
if feeder_lane[slot] is not None and pick_part is not None:
if feeder_lane[slot] and pick_part:
assign_available = False
break
# 检查机械限位冲突
if pick_part is not None and (slot - CT_Head[pick_part][0] * interval_ratio <= 0 or
slot + (max_head_index - CT_Head[pick_part][1] - 1) * interval_ratio > max_slot_index // 2):
if pick_part and (slot - CT_Head[pick_part][0] * interval_ratio <= 0 or slot + (
max_head_index - CT_Head[pick_part][1] - 1) * interval_ratio > max_slot_index // 2):
assign_available = False
break
if assign_available:
for idx, component in enumerate(pickup):
if component is not None:
if component:
feeder_lane[assign_slot + idx * interval_ratio] = component
CT_Group_slot[CTIdx] = assign_slot
break
@ -509,32 +509,31 @@ def optimizer_hybrid_genetic(pcb_data, component_data, hinter=True):
with tqdm(total=n_generations) as pbar:
pbar.set_description('hybrid genetic process')
# calculate fitness value
pop_val = []
for pop_idx, individual in enumerate(population):
val, _, _ = cal_individual_val(component_nozzle, component_point_pos, designated_nozzle, pickup_group,
pickup_group_cycle, pair_group, feeder_part_arrange, individual)
pop_val.append(val) # val is related to assembly time
for _ in range(n_generations):
# calculate fitness value
pop_val = []
for pop_idx, individual in enumerate(population):
val, _, _ = cal_individual_val(component_nozzle, component_point_pos, designated_nozzle, pickup_group,
pickup_group_cycle, pair_group, feeder_part_arrange, individual)
pop_val.append(val)
idx = np.argmin(pop_val)
if len(best_pop_val) == 0 or pop_val[idx] < best_pop_val[-1]:
best_individual = copy.deepcopy(population[idx])
best_pop_val.append(pop_val[idx])
# idx = np.argmin(pop_val)
# if len(best_pop_val) == 0 or pop_val[idx] < best_pop_val[-1]:
# best_individual = copy.deepcopy(population[idx])
# best_pop_val.append(pop_val[idx])
# min-max convert
max_val = 1.5 * max(pop_val)
pop_val = list(map(lambda v: max_val - v, pop_val))
convert_pop_val = list(map(lambda v: max_val - v, pop_val))
# crossover and mutation
c = 0
new_population = []
new_population, new_pop_val = [], []
for pop in range(population_size):
if pop % 2 == 0 and np.random.random() < crossover_rate:
index1, index2 = roulette_wheel_selection(pop_val), -1
index1, index2 = roulette_wheel_selection(convert_pop_val), -1
while True:
index2 = roulette_wheel_selection(pop_val)
index2 = roulette_wheel_selection(convert_pop_val)
if index1 != index2:
break
# 两点交叉算子
@ -552,13 +551,27 @@ def optimizer_hybrid_genetic(pcb_data, component_data, hinter=True):
new_population.append(offspring1)
new_population.append(offspring2)
# selection
top_k_index = get_top_k_value(pop_val, population_size - len(new_population))
val, _, _ = cal_individual_val(component_nozzle, component_point_pos, designated_nozzle,
pickup_group,
pickup_group_cycle, pair_group, feeder_part_arrange, offspring1)
new_pop_val.append(val)
val, _, _ = cal_individual_val(component_nozzle, component_point_pos, designated_nozzle,
pickup_group,
pickup_group_cycle, pair_group, feeder_part_arrange, offspring2)
new_pop_val.append(val)
# generate next generation
top_k_index = get_top_k_value(pop_val, population_size - len(new_population), reverse=False)
for index in top_k_index:
new_population.append(population[index])
new_pop_val.append(pop_val[index])
population = new_population
pop_val = new_pop_val
pbar.update(1)
best_individual = population[np.argmin(pop_val)]
return convert_individual_2_result(component_data, component_point_pos, designated_nozzle, pickup_group,
pickup_group_cycle, pair_group, feeder_lane, best_individual)

57
base_optimizer/optimizer_scanbased.py
View File

@ -3,11 +3,11 @@ from base_optimizer.optimizer_common import *
@timer_wrapper
def optimizer_scanbased(component_data, pcb_data, hinter):
def optimizer_genetic_scanning(component_data, pcb_data, hinter):
population_size = 200 # 种群规模
crossover_rate, mutation_rate = .4, .02
n_generation = 5
n_generation = 500
component_points = [0] * len(component_data)
for i in range(len(pcb_data)):
@ -31,49 +31,51 @@ def optimizer_scanbased(component_data, pcb_data, hinter):
pop_val.append(feeder_arrange_evaluate(feeder_slot_result, cycle_result))
# todo: 过程写的有问题，暂时不想改
sigma_scaling(pop_val, 1)
with tqdm(total=n_generation) as pbar:
pbar.set_description('hybrid genetic process')
new_pop_val, new_pop_individual = [], []
# min-max convert
max_val = 1.5 * max(pop_val)
convert_pop_val = list(map(lambda v: max_val - v, pop_val))
for _ in range(n_generation):
# 交叉
for pop in range(population_size):
if pop % 2 == 0 and np.random.random() < crossover_rate:
index1, index2 = roulette_wheel_selection(pop_val), -1
index1, index2 = roulette_wheel_selection(convert_pop_val), -1
while True:
index2 = roulette_wheel_selection(pop_val)
index2 = roulette_wheel_selection(convert_pop_val)
if index1 != index2:
break
# 两点交叉算子
offspring1, offspring2 = cycle_crossover(pop_individual[index1], pop_individual[index2])
# 变异
if np.random.random() < mutation_rate:
offspring1 = swap_mutation(offspring1)
if np.random.random() < mutation_rate:
offspring2 = swap_mutation(offspring2)
_, cycle_result, feeder_slot_result = convert_individual_2_result(component_points, offspring1)
pop_val.append(feeder_arrange_evaluate(feeder_slot_result, cycle_result))
pop_individual.append(offspring1)
new_pop_val.append(feeder_arrange_evaluate(feeder_slot_result, cycle_result))
new_pop_individual.append(offspring1)
_, cycle_result, feeder_slot_result = convert_individual_2_result(component_points, offspring2)
pop_val.append(feeder_arrange_evaluate(feeder_slot_result, cycle_result))
pop_individual.append(offspring2)
new_pop_val.append(feeder_arrange_evaluate(feeder_slot_result, cycle_result))
new_pop_individual.append(offspring2)
sigma_scaling(pop_val, 1)
# generate next generation
top_k_index = get_top_k_value(pop_val, population_size - len(new_pop_individual), reverse=False)
for index in top_k_index:
new_pop_individual.append(pop_individual[index])
new_pop_val.append(pop_val[index])
# 变异
if np.random.random() < mutation_rate:
index_ = roulette_wheel_selection(pop_val)
offspring = swap_mutation(pop_individual[index_])
_, cycle_result, feeder_slot_result = convert_individual_2_result(component_points, offspring)
pop_val.append(feeder_arrange_evaluate(feeder_slot_result, cycle_result))
pop_individual.append(offspring)
sigma_scaling(pop_val, 1)
new_population, new_popval = [], []
for index in get_top_k_value(pop_val, population_size):
new_population.append(pop_individual[index])
new_popval.append(pop_val[index])
pop_individual, pop_val = new_population, new_popval
pop_individual, pop_val = new_pop_individual, new_pop_val
sigma_scaling(pop_val, 1)
# select the best individual
pop = np.argmin(pop_val)
@ -98,7 +100,6 @@ def convert_individual_2_result(component_points, pop):
feeder_part[gene], feeder_base_points[gene] = idx, component_points[idx]
# TODO: 暂时未考虑可用吸嘴数的限制
# for _ in range(math.ceil(sum(component_points) / max_head_index)):
while True:
# === 周期内循环 ===
assigned_part = [-1 for _ in range(max_head_index)] # 当前扫描到的头分配元件信息