Files
assembly-line-optimizer/base_optimizer/smopt_twophase.py

508 lines
24 KiB
Python

from base_optimizer.optimizer_common import *
from base_optimizer.smtopt_route import *
from base_optimizer.result_analysis import *
def list_range(start, end=None):
return list(range(start)) if end is None else list(range(start, end))
@timer_wrapper
def gurobi_optimizer(pcb_data, component_data, feeder_data, reduction=True, partition=True, initial=False, hinter=True):
# data preparation: convert data to index
component_list, nozzle_list = defaultdict(int), defaultdict(int)
component_feeder = defaultdict(int)
cpidx_2_part, nzidx_2_nozzle, cpidx_2_nzidx = {}, {}, {}
arg_slot_rng = None if len(feeder_data) == 0 else [feeder_data.iloc[0].slot, feeder_data.iloc[-1].slot]
for idx, data in component_data.iterrows():
part, nozzle = data.part, data.nz
cpidx_2_part[idx] = part
nz_key = [key for key, val in nzidx_2_nozzle.items() if val == nozzle]
nz_idx = len(nzidx_2_nozzle) if len(nz_key) == 0 else nz_key[0]
nzidx_2_nozzle[nz_idx] = nozzle
component_list[part] = 0
component_feeder[part] = data.fdn
cpidx_2_nzidx[idx] = nz_idx
for _, data in pcb_data.iterrows():
idx = component_data[component_data.part == data.part].index.tolist()[0]
nozzle = component_data.loc[idx].nz
nozzle_list[nozzle] += 1
component_list[data.part] += 1
part_feederbase = defaultdict(int)
if feeder_data is not None:
for _, data in feeder_data.iterrows():
idx = -1
for idx, part_ in cpidx_2_part.items():
if data.part == part_:
break
assert idx != -1
part_feederbase[idx] = data.slot # part index - slot
ratio = 1 if reduction else 2
I, J = len(cpidx_2_part.keys()), len(nzidx_2_nozzle.keys())
# === determine the hyper-parameter of L ===
# first phase: calculate the number of heads for each type of nozzle
nozzle_heads = defaultdict(int)
for nozzle in nozzle_list.keys():
nozzle_heads[nozzle] = 1
while sum(nozzle_heads.values()) != max_head_index:
max_cycle_nozzle = None
for nozzle, head_num in nozzle_heads.items():
if max_cycle_nozzle is None or nozzle_list[nozzle] / head_num > nozzle_list[max_cycle_nozzle] / \
nozzle_heads[max_cycle_nozzle]:
max_cycle_nozzle = nozzle
assert max_cycle_nozzle is not None
nozzle_heads[max_cycle_nozzle] += 1
nozzle_comp_points = defaultdict(list)
for part, points in component_list.items():
idx = component_data[component_data.part == part].index.tolist()[0]
nozzle = component_data.loc[idx].nz
nozzle_comp_points[nozzle].append([part, points])
level = 1 if len(component_list) == 1 or len(component_list) % max_head_index == 0 else 2
part_assignment, cycle_assignment = [], []
def aux_func(info):
return max(map(lambda points: max([p[1] for p in points]), info))
pre_objbst, pre_changetime = None, None
def terminate_condition(mdl, where):
if where == GRB.Callback.MIP:
objbst, objbnd = mdl.cbGet(GRB.Callback.MIP_OBJBST), mdl.cbGet(GRB.Callback.MIP_OBJBND)
changetime = mdl.cbGet(GRB.Callback.RUNTIME)
nonlocal pre_objbst, pre_changetime
# condition: value change
if abs(objbst - 1e+100) > 1: # 避免未找到可行解提前退出
if pre_objbst and abs(pre_objbst - objbst) < 1e-3:
if pre_changetime and changetime - pre_changetime > 90 * (1 - objbnd / objbst):
mdl.terminate()
else:
pre_changetime = changetime
pre_objbst = objbst
def recursive_assign(assign_points, nozzle_compo_points, cur_level, total_level) -> int:
def func(points):
return map(lambda points: max([p[1] for p in points]), points)
if cur_level > total_level and sum(func(nozzle_compo_points.values())) == 0:
return 0
elif assign_points <= 0 and cur_level == 1:
return -1 # backtrack
elif assign_points <= 0 or cur_level > total_level:
return 1 # fail
nozzle_compo_points_cpy = copy.deepcopy(nozzle_compo_points)
prev_assign = 0
for part in part_assignment[cur_level - 1]:
if part != -1:
prev_assign += 1
head_idx = 0
for nozzle, head in nozzle_heads.items():
while head:
min_idx = -1
for idx, (part, points) in enumerate(nozzle_compo_points_cpy[nozzle]):
if points >= assign_points and (
min_idx == -1 or points < nozzle_compo_points_cpy[nozzle][min_idx][1]):
min_idx = idx
part_assignment[cur_level - 1][head_idx] = -1 if min_idx == -1 else \
nozzle_compo_points_cpy[nozzle][min_idx][0]
if min_idx != -1:
nozzle_compo_points_cpy[nozzle][min_idx][1] -= assign_points
head -= 1
head_idx += 1
cycle_assignment[cur_level - 1] = assign_points
for part in part_assignment[cur_level - 1]:
if part != -1:
prev_assign -= 1
if prev_assign == 0:
res = 1
else:
points = min(len(pcb_data) // max_head_index + 1, aux_func(nozzle_compo_points_cpy.values()))
res = recursive_assign(points, nozzle_compo_points_cpy, cur_level + 1, total_level)
if res == 0:
return 0
elif res == 1:
# All cycles have been completed, but there are still points left to be allocated
return recursive_assign(assign_points - 1, nozzle_compo_points, cur_level, total_level)
# second phase: (greedy) recursive search to assign points for each cycle set and obtain an initial solution
while True:
part_assignment = [[-1 for _ in range(max_head_index)] for _ in range(level)]
cycle_assignment = [-1 for _ in range(level)]
points = min(len(pcb_data) // max_head_index + 1, max(component_list.values()))
if recursive_assign(points, nozzle_comp_points, 1, level) == 0:
break
level += 1
L = len(cycle_assignment) if partition else len(pcb_data)
S = ratio * sum(component_feeder.values()) if len(feeder_data) == 0 else arg_slot_rng[-1] - arg_slot_rng[0] + 1 # the available feeder num
M = len(pcb_data) # a sufficiently large number (number of placement points)
HC = [[0 for _ in range(J)] for _ in range(I)]
for i in range(I):
for j in range(J):
HC[i][j] = 1 if cpidx_2_nzidx[i] == j else 0
mdl = Model('SMT')
mdl.setParam('Seed', 0)
mdl.setParam('OutputFlag', hinter) # set whether output the debug information
mdl.setParam('TimeLimit', 3600)
mdl.setParam('PoolSearchMode', 2)
mdl.setParam('PoolSolutions', 100)
mdl.setParam('PoolGap', 1e-4)
# mdl.setParam('MIPFocus', 2)
mdl.setParam("Heuristics", 0.5)
# Use only if other methods, including exploring the tree with the default settings, do not yield a viable solution
# mdl.setParam("ZeroObjNodes", 100)
# === Decision Variables ===
x = mdl.addVars(list_range(I), list_range(S), list_range(max_head_index), list_range(L), vtype=GRB.BINARY, name='x')
y = mdl.addVars(list_range(I), list_range(max_head_index), list_range(L), vtype=GRB.BINARY, name='y')
v = mdl.addVars(list_range(S), list_range(max_head_index), list_range(L), vtype=GRB.BINARY, name='v')
c = mdl.addVars(list_range(I), list_range(max_head_index), list_range(L), vtype=GRB.INTEGER, name='c')
mdl.addConstrs(
c[i, h, l] <= component_list[cpidx_2_part[i]] for i in range(I) for h in range(max_head_index) for l in
range(L))
f = {}
for i in range(I):
if i not in part_feederbase.keys():
for s in range(S):
f[s, i] = mdl.addVar(vtype=GRB.BINARY, name='f_' + str(s) + '_' + str(i))
else:
for s in range(S):
f[s, i] = 1 if part_feederbase[i] == s + arg_slot_rng[0] else 0
p = mdl.addVars(list_range(-(max_head_index - 1) * ratio, S), list_range(L), vtype=GRB.BINARY, name='p')
z = mdl.addVars(list_range(J), list_range(max_head_index), list_range(L), vtype=GRB.BINARY)
d = mdl.addVars(list_range(L), list_range(max_head_index), vtype=GRB.INTEGER, name='d')
d_plus = mdl.addVars(list_range(J), list_range(max_head_index), list_range(L), vtype=GRB.INTEGER,
name='d_plus')
d_minus = mdl.addVars(list_range(J), list_range(max_head_index), list_range(L), vtype=GRB.INTEGER,
name='d_minus')
max_cycle = math.ceil(len(pcb_data) / max_head_index)
PU = mdl.addVars(list_range(-(max_head_index - 1) * ratio, S), list_range(L), vtype=GRB.INTEGER, name='PU')
WL = mdl.addVars(list_range(L), vtype=GRB.INTEGER, ub=max_cycle, name='WL')
NC = mdl.addVars(list_range(max_head_index), vtype=GRB.INTEGER, name='NC')
part_2_cpidx = defaultdict(int)
for idx, part in cpidx_2_part.items():
part_2_cpidx[part] = idx
if initial:
# initial some variables to speed up the search process
# ensure the priority of the workload assignment
cycle_index = sorted(range(len(cycle_assignment)), key=lambda k: cycle_assignment[k], reverse=True)
part_list = []
for cycle in cycle_index:
cycle_part = part_assignment[cycle]
for part in cycle_part:
if part != -1 and part not in part_list:
part_list.append(part)
slot = 0
for part in part_list:
if feeder_data is not None:
while slot in feeder_data.keys():
slot += 1 # skip assigned feeder slot
if part_2_cpidx[part] in part_feederbase.keys():
continue
part_feederbase[part_2_cpidx[part]] = slot
f[slot, part_2_cpidx[part]].Start = 1
slot += 1
for idx, cycle in enumerate(cycle_index):
WL[idx].Start = cycle_assignment[cycle]
for h in range(max_head_index):
part = part_assignment[cycle][h]
if part == -1:
continue
i = part_2_cpidx[part]
y[i, h, idx].Start = 1
v[part_feederbase[i], h, idx].Start = 1
# === Objective ===
mdl.setObjective(Fit_cy * quicksum(WL[l] for l in range(L)) + 2 * Fit_nz * quicksum(
NC[h] for h in range(max_head_index)) + Fit_pu * quicksum(
PU[s, l] for s in range(-(max_head_index - 1) * ratio, S) for l in range(L)))
# === Constraint ===
if not partition:
mdl.addConstrs(WL[l] <= 1 for l in range(L))
# work completion
mdl.addConstrs(c[i, h, l] == WL[l] * y[i, h, l] for i in range(I) for h in range(max_head_index) for l in range(L))
# mdl.addConstrs(
# c[i, h, l] <= max_cycle * y[i, h, l] for i in range(I) for h in range(max_head_index) for l in range(L))
# mdl.addConstrs(c[i, h, l] <= WL[l] for i in range(I) for h in range(max_head_index) for l in range(L))
# mdl.addConstrs(
# c[i, h, l] >= WL[l] - max_cycle * (1 - y[i, h, l]) for i in range(I) for h in range(max_head_index) for l in
# range(L))
mdl.addConstrs(
quicksum(c[i, h, l] for h in range(max_head_index) for l in range(L)) == component_list[cpidx_2_part[i]] for i
in range(I))
# variable constraint
mdl.addConstrs(quicksum(y[i, h, l] for i in range(I)) <= 1 for h in range(max_head_index) for l in range(L))
# simultaneous pick
for s in range(-(max_head_index - 1) * ratio, S):
rng = list(range(max(0, -math.floor(s / ratio)), min(max_head_index, math.ceil((S - s) / ratio))))
for l in range(L):
mdl.addConstr(quicksum(v[s + h * ratio, h, l] for h in rng) <= max_head_index * p[s, l])
mdl.addConstr(quicksum(v[s + h * ratio, h, l] for h in rng) >= p[s, l])
mdl.addConstrs(PU[s, l] == p[s, l] * WL[l] for s in range(-(max_head_index - 1) * ratio, S) for l in range(L))
# mdl.addConstrs(PU[s, l] <= max_cycle * p[s, l] for s in range(-(max_head_index - 1) * ratio, S) for l in range(L))
# mdl.addConstrs(PU[s, l] <= WL[l] for s in range(-(max_head_index - 1) * ratio, S) for l in range(L))
# mdl.addConstrs(
# PU[s, l] >= WL[l] - max_cycle * (1 - p[s, l]) for s in range(-(max_head_index - 1) * ratio, S) for l in
# range(L))
# nozzle change
mdl.addConstrs(
z[j, h, l] - z[j, h, l + 1] == d_plus[j, h, l] - d_minus[j, h, l] for l in range(L - 1) for j in range(J) for h
in range(max_head_index))
mdl.addConstrs(z[j, h, 0] - z[j, h, L - 1] == d_plus[j, h, L - 1] - d_minus[j, h, L - 1] for j in range(J) for h
in range(max_head_index))
mdl.addConstrs(
2 * d[l, h] == quicksum(d_plus[j, h, l] for j in range(J)) + quicksum(d_minus[j, h, l] for j in range(J)) for l
in range(L) for h in range(max_head_index))
mdl.addConstrs(NC[h] == quicksum(d[l, h] for l in range(L)) for h in range(max_head_index))
mdl.addConstrs(quicksum(y[i, h, l] for i in range(I) for h in range(max_head_index)) * M >= WL[l] for l in range(L))
# nozzle-component compatibility
mdl.addConstrs(
y[i, h, l] <= quicksum(HC[i][j] * z[j, h, l] for j in range(J)) for i in range(I) for h in range(max_head_index)
for l in range(L))
# available number of feeder
mdl.addConstrs(quicksum(f[s, i] for s in range(S)) <= component_feeder[cpidx_2_part[i]] for i in range(I))
# available number of nozzle
mdl.addConstrs(quicksum(z[j, h, l] for h in range(max_head_index)) <= max_head_index for j in range(J) for l in range(L))
# upper limit for occupation for feeder slot
mdl.addConstrs(quicksum(f[s, i] for i in range(I)) <= 1 for s in range(S))
mdl.addConstrs(
quicksum(v[s, h, l] for s in range(S)) >= quicksum(y[i, h, l] for i in range(I)) for h in range(max_head_index)
for l in range(L))
# others
mdl.addConstrs(quicksum(z[j, h, l] for j in range(J)) <= 1 for h in range(max_head_index) for l in range(L))
mdl.addConstrs(
quicksum(x[i, s, h, l] for h in range(max_head_index) for l in range(L)) >= f[s, i] for i in range(I)
for s in range(S))
mdl.addConstrs(
quicksum(x[i, s, h, l] for h in range(max_head_index) for l in range(L)) <= M * f[s, i] for i in
range(I) for s in range(S))
# mdl.addConstrs(
# f[s, i] >= x[i, s, h, l] for s in range(S) for i in range(I) for h in range(max_head_index) for l in range(L))
#
# mdl.addConstrs(
# quicksum(x[i, s, h, l] for h in range(max_head_index) for l in range(L)) >= f[s, i] for s in
# range(S) for i in range(I))
# the constraints to speed up the search process
mdl.addConstrs(
quicksum(x[i, s, h, l] for i in range(I) for s in range(S)) <= 1 for h in range(max_head_index) for l
in range(L))
if reduction:
mdl.addConstrs(WL[l] >= WL[l + 1] for l in range(L - 1))
mdl.addConstr(quicksum(WL[l] for l in range(L)) <= sum(cycle_assignment))
mdl.addConstr(quicksum(WL[l] for l in range(L)) >= math.ceil(len(pcb_data) / max_head_index))
mdl.addConstrs(quicksum(z[j, h, l] for j in range(J) for h in range(max_head_index)) >= quicksum(
z[j, h, l + 1] for j in range(J) for h in range(max_head_index)) for l in range(L - 1))
mdl.addConstrs(y[i, h, l] <= WL[l] for i in range(I) for h in range(max_head_index) for l in range(L))
mdl.addConstrs(v[s, h, l] <= WL[l] for s in range(S) for h in range(max_head_index) for l in range(L))
mdl.addConstrs(
x[i, s, h, l] >= y[i, h, l] + v[s, h, l] - 1 for i in range(I) for s in range(S) for h in range(max_head_index)
for l in range(L))
mdl.addConstrs(
x[i, s, h, l] <= y[i, h, l] for i in range(I) for s in range(S) for h in range(max_head_index)
for l in range(L))
mdl.addConstrs(
x[i, s, h, l] <= v[s, h, l] for i in range(I) for s in range(S) for h in range(max_head_index)
for l in range(L))
# === search process ===
mdl.update()
# mdl.write('mdl.lp')
if hinter:
print('num of constrs: ', str(len(mdl.getConstrs())), ', num of vars: ', str(len(mdl.getVars())))
mdl.optimize(terminate_condition)
# mdl.optimize()
# === result generation ===
opt_res_list = defaultdict(OptResult)
if mdl.Status == GRB.OPTIMAL or mdl.Status == GRB.INTERRUPTED or mdl.Status == GRB.TIME_LIMIT:
# === selection from solution pool ===
component_pos = defaultdict(list[Point])
for _, data in pcb_data.iterrows():
component_index = component_data[component_data.part == data.part].index.tolist()[0]
component_pos[component_index].append(Point(data.x, data.y))
for part in component_pos.keys():
component_pos[part] = sorted(component_pos[part], key=lambda pos: (pos.x, pos.y))
min_dist, solution_number = None, 0
for sol_counter in range(mdl.SolCount):
mdl.Params.SolutionNumber = sol_counter
opt_res = OptResult()
# == 转换标准的贴装头分配的解 ===
for l in range(L):
if abs(WL[l].Xn) <= 1e-4:
continue
opt_res.cycle_assign.append(round(WL[l].Xn))
opt_res.component_assign.append([-1] * max_head_index)
opt_res.feeder_slot_assign.append([-1] * max_head_index)
for h in range(max_head_index):
for i in range(I):
if abs(y[i, h, l].Xn) <= 1e-4:
continue
opt_res.component_assign[-1][h] = i
for s in range(S):
if abs(v[s, h, l].Xn - 1) < 1e-4 and opt_res.component_assign[-1][h] != -1:
opt_res.feeder_slot_assign[-1][h] = s
# 根据贴装头位置,转换供料器槽位
cp_avg_head, cp_sum_cycle = defaultdict(float), defaultdict(int)
for cycle, component_assign in enumerate(opt_res.component_assign):
for head, part in enumerate(component_assign):
if part == -1:
continue
cp_avg_head[part] += opt_res.cycle_assign[cycle] * head
cp_sum_cycle[part] += opt_res.cycle_assign[cycle]
for part, head in cp_avg_head.items():
cp_avg_head[part] = head / cp_sum_cycle[part]
avg_position = sum([data.x - cp_avg_head[part_2_cpidx[data.part]] * head_interval for _, data in
pcb_data.iterrows()]) / len(pcb_data)
avg_slot = 0
D_PU, D_PL, D_BW, D_FW = 0, 0, 0, 0
for cycle, slots in enumerate(opt_res.feeder_slot_assign):
min_slot, max_slot = max_slot_index, 0
for head, slot in enumerate(slots):
if slot == -1:
continue
min_slot = min(min_slot, slot - head * ratio)
max_slot = max(max_slot, slot - head * ratio)
avg_slot += (max_slot - min_slot) * opt_res.cycle_assign[cycle]
D_PU += (max_slot - min_slot) * slot_interval * opt_res.cycle_assign[cycle] # 拾取路径
avg_slot /= sum(opt_res.cycle_assign)
start_slot = round((avg_position + stopper_pos[0] - slotf1_pos[0]) / slot_interval + avg_slot / 2) + 1
for cycle in range(len(opt_res.feeder_slot_assign)):
for head in range(max_head_index):
if (slot := opt_res.feeder_slot_assign[cycle][head]) == -1:
continue
opt_res.feeder_slot_assign[cycle][head] = start_slot + slot * (2 if ratio == 1 else 1)
component_pos_counter = defaultdict(int)
cycle_place_pos = defaultdict(list[Point])
for head in range(max_head_index):
for cycle in range(len(opt_res.cycle_assign)):
if (part := opt_res.component_assign[cycle][head]) == -1:
continue
avg_place_pos = Point(0, 0, _h=head)
for counter in range(round(opt_res.cycle_assign[cycle])):
avg_place_pos.x = (1 - 1.0 / (counter + 1)) * avg_place_pos.x + (
component_pos[part][component_pos_counter[part]].x - head * head_interval) / (
counter + 1)
avg_place_pos.y = (1 - 1.0 / (counter + 1)) * avg_place_pos.y + component_pos[part][
component_pos_counter[part]].y / (counter + 1)
component_pos_counter[part] += 1
avg_place_pos.x += stopper_pos[0]
avg_place_pos.y += stopper_pos[1]
cycle_place_pos[cycle].append(avg_place_pos)
for cycle in range(len(opt_res.cycle_assign)):
min_slot, max_slot = max_slot_index, 0
for head in range(max_head_index):
if (slot := opt_res.feeder_slot_assign[cycle][head]) == -1:
continue
min_slot = min(min_slot, slot - head * interval_ratio)
max_slot = max(max_slot, slot - head * interval_ratio)
# cycle_place_pos[cycle] = sorted(cycle_place_pos[cycle], key=lambda pt: pt.x)
pick_pos = Point(slotf1_pos[0] + (min_slot + max_slot) / 2 * slot_interval, slotf1_pos[1])
_, seq = dynamic_programming_cycle_route(cycle_place_pos[cycle], pick_pos)
head_position = [Point(0, 0) for _ in range(max_head_index)]
for point in cycle_place_pos[cycle]:
head_position[point.h] = point
for idx in range(len(seq) - 1):
h1, h2 = seq[idx], seq[idx + 1]
D_PL += max(abs(head_position[h1].x - head_position[h2].x),
abs(head_position[h1].y - head_position[h2].y)) * opt_res.cycle_assign[cycle]
# opt_res.placement_assign, opt_res.head_sequence = scan_based_placement_route_generation(component_data,
# pcb_data,
# opt_res.component_assign,
# opt_res.cycle_assign,
# opt_res.feeder_slot_assign,
# hinter=False)
# info = placement_info_evaluation(component_data, pcb_data, opt_res)
# print(f'{info.place_distance + info.pickup_distance: .3f}\t{D_PL + D_PU: .3f}')
opt_res_list[sol_counter] = opt_res
solution_number = 0
mdl.Params.SolutionNumber = 0
if hinter:
print('total cost = {}'.format(mdl.objval))
print('cycle = {}, nozzle change = {}, pick up = {}'.format(quicksum(WL[l].Xn for l in range(L)), quicksum(
NC[h].Xn for h in range(max_head_index)), quicksum(
PU[s, l].Xn for s in range(-(max_head_index - 1) * ratio, S) for l in range(L))))
print('workload: ')
for l in range(L):
print(WL[l].Xn, end=', ')
print('')
print('result')
print('component assignment: ', opt_res_list[solution_number].component_assign)
print('feeder assignment: ', opt_res_list[solution_number].feeder_slot_assign)
print('cycle assignment: ', opt_res_list[solution_number].cycle_assign)
return opt_res_list[solution_number].component_assign, opt_res_list[solution_number].feeder_slot_assign, \
opt_res_list[solution_number].cycle_assign