文章复现: guo_integrated_2012

This commit is contained in:
2023-02-25 21:09:33 +08:00
commit 6b031dc486
9 changed files with 1564 additions and 0 deletions

8
.gitignore vendored Normal file
View File

@ -0,0 +1,8 @@
.idea
result/
data/
*.txt
__pycache__
Lib/
Scripts/
*.cfg

34
SMO/benchmarks.py Normal file
View File

@ -0,0 +1,34 @@
# -*- coding: utf-8 -*-
"""
Python code of Spider-Monkey Optimization (SMO)
Coded by: Mukesh Saraswat (emailid: saraswatmukesh@gmail.com), Himanshu Mittal (emailid: himanshu.mittal224@gmail.com) and Raju Pal (emailid: raju3131.pal@gmail.com)
The code template used is similar to code given at link: https://github.com/himanshuRepo/CKGSA-in-Python
and C++ version of the SMO at link: http://smo.scrs.in/
Reference: Jagdish Chand Bansal, Harish Sharma, Shimpi Singh Jadon, and Maurice Clerc. "Spider monkey optimization algorithm for numerical optimization." Memetic computing 6, no. 1, 31-47, 2014.
@link: http://smo.scrs.in/
-- Benchmark.py: Defining the benchmark function along its range lower bound, upper bound and dimensions
Code compatible:
-- Python: 2.* or 3.*
"""
import numpy
import math
# define the function blocks
def F1(x):
s = numpy.sum(x ** 2);
return s
# define the function parameters
def getFunctionDetails():
# [name, lb, ub, dim, acc_err, obj_val]
param = ["F1", -100, 100, 30, 1.0e-5, 0]
return param

79
SMO/main.py Normal file
View File

@ -0,0 +1,79 @@
# -*- coding: utf-8 -*-
"""
Python code of Spider-Monkey Optimization (SMO)
Coded by: Mukesh Saraswat (emailid: saraswatmukesh@gmail.com), Himanshu Mittal (emailid: himanshu.mittal224@gmail.com) and Raju Pal (emailid: raju3131.pal@gmail.com)
The code template used is similar to code given at link: https://github.com/himanshuRepo/CKGSA-in-Python
and C++ version of the SMO at link: http://smo.scrs.in/
Reference: Jagdish Chand Bansal, Harish Sharma, Shimpi Singh Jadon, and Maurice Clerc. "Spider monkey optimization algorithm for numerical optimization." Memetic computing 6, no. 1, 31-47, 2014.
@link: http://smo.scrs.in/
-- Main.py: Calling the Spider-Monkey Optimization (SMO) Algorithm
for minimizing of an objective Function
Code compatible:
-- Python: 2.* or 3.*
"""
import smo
import benchmarks
import csv
import numpy
import time
import math
def selector(func_details, popSize, Iter, succ_rate, mean_feval):
function_name = func_details[0]
lb = func_details[1]
ub = func_details[2]
dim = func_details[3]
acc_err = func_details[4]
obj_val = func_details[5]
x, succ_rate, mean_feval = smo.main(getattr(benchmarks, function_name), lb, ub, dim, popSize, Iter, acc_err,
obj_val, succ_rate, mean_feval)
return x, succ_rate, mean_feval
# Select number of repetitions for each experiment.
# To obtain meaningful statistical results, usually 30 independent runs are executed for each algorithm.
NumOfRuns = 2
# Select general parameters for all optimizers (population size, number of iterations)
PopulationSize = 10
Iterations = 500
mean_error = 0
total_feval = 0
mean1 = 0
var = 0
sd = 0
mean_feval = 0
succ_rate = 0
GlobalMins = numpy.zeros(NumOfRuns)
for k in range(0, NumOfRuns):
func_details = benchmarks.getFunctionDetails()
print("Run: {}".format(k + 1))
x, succ_rate, mean_feval = selector(func_details, PopulationSize, Iterations, succ_rate, mean_feval)
mean_error = mean_error + x.error;
mean1 = mean1 + x.convergence[-1]
total_feval = total_feval + x.feval
GlobalMins[k] = x.convergence[-1]
mean1 = mean1 / NumOfRuns;
mean_error = mean_error / NumOfRuns
if (succ_rate > 0):
mean_feval = mean_feval / succ_rate
total_feval = total_feval / NumOfRuns
for k in range(NumOfRuns):
var = var + math.pow((GlobalMins[k] - mean1), 2)
var = var / NumOfRuns
sd = math.sqrt(var)
print(
"Values after executing SMO: \n Mean Error:{} \n Mean Function eval:{} \n Total Function eval:{} \n Variance:{} \n STD:{}".format(
mean_error, mean_feval, total_feval, var, sd))

348
SMO/smo.py Normal file
View File

@ -0,0 +1,348 @@
# -*- coding: utf-8 -*-
"""
Python code of Spider-Monkey Optimization (SMO)
Coded by: Mukesh Saraswat (emailid: saraswatmukesh@gmail.com), Himanshu Mittal (emailid: himanshu.mittal224@gmail.com) and Raju Pal (emailid: raju3131.pal@gmail.com)
The code template used is similar to code given at link: https://github.com/himanshuRepo/CKGSA-in-Python
and C++ version of the SMO at link: http://smo.scrs.in/
Reference: Jagdish Chand Bansal, Harish Sharma, Shimpi Singh Jadon, and Maurice Clerc. "Spider monkey optimization algorithm for numerical optimization." Memetic computing 6, no. 1, 31-47, 2014.
@link: http://smo.scrs.in/
-- smo.py: Performing the Spider-Monkey Optimization (SMO) Algorithm
Code compatible:
-- Python: 2.* or 3.*
"""
from __future__ import division
import time
import random
import numpy
import math
from solution import solution
class SMO():
def __init__(self, objf1, lb1, ub1, dim1, PopSize1, acc_err1, iters1):
self.PopSize = PopSize1
self.dim = dim1
self.acc_err = acc_err1
self.lb = lb1
self.ub = ub1
self.objf = objf1
self.pos = numpy.zeros((PopSize1, dim1))
self.fun_val = numpy.zeros(PopSize1)
self.fitness = numpy.zeros(PopSize1)
self.gpoint = numpy.zeros((PopSize1, 2))
self.prob = numpy.zeros(PopSize1)
self.LocalLimit = dim1 * PopSize1;
self.GlobalLimit = PopSize1;
self.fit = numpy.zeros(PopSize1)
self.MinCost = numpy.zeros(iters1)
self.Bestpos = numpy.zeros(dim1)
self.group = 0
self.func_eval = 0
self.part = 1
self.max_part = 5
self.cr = 0.1
# ====== Function: CalculateFitness() ========= #
def CalculateFitness(self, fun1):
if fun1 >= 0:
result = (1 / (fun1 + 1))
else:
result = (1 + math.fabs(fun1))
return result
# ================ X X X ===================== #
# ==================================== Function: Initialization() ============================================ #
def initialize(self):
global GlobalMin, GlobalLeaderPosition, GlobalLimitCount, LocalMin, LocalLimitCount, LocalLeaderPosition
S_max = int(self.PopSize / 2)
LocalMin = numpy.zeros(S_max)
LocalLeaderPosition = numpy.zeros((S_max, self.dim))
LocalLimitCount = numpy.zeros(S_max)
for i in range(self.PopSize):
for j in range(self.dim):
if type(self.ub) == int:
self.pos[i, j] = random.random() * (self.ub - self.lb) + self.lb
else:
self.pos[i, j] = random.random() * (self.ub[j] - self.lb[j]) + self.lb[j]
# Calculate objective function for each particle
for i in range(self.PopSize):
# Performing the bound checking
self.pos[i, :] = numpy.clip(self.pos[i, :], self.lb, self.ub)
self.fun_val[i] = self.objf(self.pos[i, :])
self.func_eval += 1
self.fitness[i] = self.CalculateFitness(self.fun_val[i])
# Initialize Global Leader Learning
GlobalMin = self.fun_val[0]
GlobalLeaderPosition = self.pos[0, :]
GlobalLimitCount = 0
# Initialize Local Leader Learning
for k in range(self.group):
LocalMin[k] = self.fun_val[int(self.gpoint[k, 0])]
LocalLimitCount[k] = 0
LocalLeaderPosition[k, :] = self.pos[int(self.gpoint[k, 0]), :]
# ============================================ X X X ======================================================= #
# =========== Function: CalculateProbabilities() ============ #
def CalculateProbabilities(self):
maxfit = self.fitness[0];
i = 1
while (i < self.PopSize):
if (self.fitness[i] > maxfit):
maxfit = self.fitness[i];
i += 1
for i in range(self.PopSize):
self.prob[i] = (0.9 * (self.fitness[i] / maxfit)) + 0.1;
# ========================== X X X ======================== #
# ================= Function: create_group() ================ #
def create_group(self):
g = 0
lo = 0
while (lo < self.PopSize):
hi = lo + int(self.PopSize / self.part)
self.gpoint[g, 0] = lo
self.gpoint[g, 1] = hi
if ((self.PopSize - hi) < (int(self.PopSize / self.part))):
self.gpoint[g, 1] = (self.PopSize - 1)
g = g + 1
lo = hi + 1
self.group = g
# ========================== X X X ======================== #
# ================= Function: LocalLearning() ================ #
def LocalLearning(self):
global LocalMin, LocalLimitCount, LocalLeaderPosition
S_max = int(self.PopSize / 2)
OldMin = numpy.zeros(S_max)
for k in range(self.group):
OldMin[k] = LocalMin[k]
for k in range(self.group):
i = int(self.gpoint[k, 0])
while (i <= int(self.gpoint[k, 1])):
if (self.fun_val[i] < LocalMin[k]):
LocalMin[k] = self.fun_val[i]
LocalLeaderPosition[k, :] = self.pos[i, :]
i = i + 1
for k in range(self.group):
if (math.fabs(OldMin[k] - LocalMin[k]) < self.acc_err):
LocalLimitCount[k] = LocalLimitCount[k] + 1
else:
LocalLimitCount[k] = 0
# ========================== X X X ======================== #
# ================= Function: GlobalLearning() ================ #
def GlobalLearning(self):
global GlobalMin, GlobalLeaderPosition, GlobalLimitCount
G_trial = GlobalMin
for i in range(self.PopSize):
if (self.fun_val[i] < GlobalMin):
GlobalMin = self.fun_val[i]
GlobalLeaderPosition = self.pos[i, :]
if (math.fabs(G_trial - GlobalMin) < self.acc_err):
GlobalLimitCount = GlobalLimitCount + 1
else:
GlobalLimitCount = 0
# ========================== X X X ======================== #
# ================= Function: LocalLeaderPhase() ================ #
def LocalLeaderPhase(self, k):
global LocalLeaderPosition
new_position = numpy.zeros((1, self.dim))
lo = int(self.gpoint[k, 0])
hi = int(self.gpoint[k, 1])
i = lo
while (i <= hi):
while True:
PopRand = int((random.random() * (hi - lo) + lo))
if (PopRand != i):
break
for j in range(self.dim):
if (random.random() >= self.cr):
new_position[0, j] = self.pos[i, j] + (LocalLeaderPosition[k, j] - self.pos[i, j]) * (
random.random()) + (self.pos[PopRand, j] - self.pos[i, j]) * (random.random() - 0.5) * 2
else:
new_position[0, j] = self.pos[i, j]
new_position = numpy.clip(new_position, self.lb, self.ub)
ObjValSol = self.objf(new_position)
self.func_eval += 1
FitnessSol = self.CalculateFitness(ObjValSol)
if (FitnessSol > self.fitness[i]):
self.pos[i, :] = new_position
self.fun_val[i] = ObjValSol
self.fitness[i] = FitnessSol
i += 1
# ========================== X X X ======================== #
# ================= Function: GlobalLeaderPhase() ================ #
def GlobalLeaderPhase(self, k):
global GlobalLeaderPosition
new_position = numpy.zeros((1, self.dim))
lo = int(self.gpoint[k, 0])
hi = int(self.gpoint[k, 1])
i = lo;
l = lo;
while (l < hi):
if (random.random() < self.prob[i]):
l += 1
while True:
PopRand = int(random.random() * (hi - lo) + lo)
if (PopRand != i):
break
param2change = int(random.random() * self.dim)
new_position = self.pos[i, :]
new_position[param2change] = self.pos[i, param2change] + (
GlobalLeaderPosition[param2change] - self.pos[i, param2change]) * (random.random()) + (
self.pos[PopRand, param2change] - self.pos[
i, param2change]) * (random.random() - 0.5) * 2
new_position = numpy.clip(new_position, self.lb, self.ub)
ObjValSol = self.objf(new_position)
self.func_eval += 1
FitnessSol = self.CalculateFitness(ObjValSol)
if (FitnessSol > self.fitness[i]):
self.pos[i, :] = new_position
self.fun_val[i] = ObjValSol
self.fitness[i] = FitnessSol
i += 1;
if (i == hi):
i = lo;
# ========================== X X X ======================== #
# ================= Function: GlobalLeaderDecision() ================ #
def GlobalLeaderDecision(self):
global GlobalLimitCount
if (GlobalLimitCount > self.GlobalLimit):
GlobalLimitCount = 0
if (self.part < self.max_part):
self.part = self.part + 1
self.create_group()
self.LocalLearning()
else:
self.part = 1
self.create_group()
self.LocalLearning()
# ========================== X X X ======================== #
# ================= Function: LocalLeaderDecision() ================ #
def LocalLeaderDecision(self):
global GlobalLeaderPosition, LocalLimitCount, LocalLeaderPosition
for k in range(self.group):
if (LocalLimitCount[k] > self.LocalLimit):
i = self.gpoint[k, 0]
while (i <= int(self.gpoint[k, 1])):
for j in range(self.dim):
if (random.random() >= self.cr):
if type(self.ub) == int:
self.pos[i, j] = random.random() * (self.ub - self.lb) + self.lb
else:
self.pos[i, j] = random.random() * (self.ub[j] - self.lb[j]) + self.lb[j]
else:
self.pos[i, j] = self.pos[i, j] + (
GlobalLeaderPosition[j] - self.pos[i, j]) * random.random() + (
self.pos[i, j] - LocalLeaderPosition[k, j]) * random.random()
self.pos[i, :] = numpy.clip(self.pos[i, :], self.lb, self.ub)
self.fun_val[i] = self.objf(self.pos[i, :])
self.func_eval += 1
self.fitness[i] = self.CalculateFitness(self.fun_val[i])
i += 1
LocalLimitCount[k] = 0
# ========================== X X X ======================== #
# ==================================== Main() ===================================== #
def main(objf1, lb1, ub1, dim1, PopSize1, iters, acc_err1, obj_val, succ_rate, mean_feval):
smo = SMO(objf1, lb1, ub1, dim1, PopSize1, acc_err1, iters)
s = solution()
print("SMO is optimizing \"" + smo.objf.__name__ + "\"")
timerStart = time.time()
s.startTime = time.strftime("%Y-%m-%d-%H-%M-%S")
# =========================== Calling: initialize() =========================== #
smo.initialize()
# ========================== Calling: GlobalLearning() ======================== #
smo.GlobalLearning()
# ========================= Calling: LocalLearning() ========================== #
smo.LocalLearning()
# ========================== Calling: create_group() ========================== #
smo.create_group()
# ================================= Looping ================================== #
for l in range(iters):
for k in range(smo.group):
# ==================== Calling: LocalLeaderPhase() =================== #
smo.LocalLeaderPhase(k)
# =================== Calling: CalculateProbabilities() ================== #
smo.CalculateProbabilities()
for k in range(smo.group):
# ==================== Calling: GlobalLeaderPhase() ================== #
smo.GlobalLeaderPhase(k)
# ======================= Calling: GlobalLearning() ====================== #
smo.GlobalLearning()
# ======================= Calling: LocalLearning() ======================= #
smo.LocalLearning()
# ================== Calling: LocalLeaderDecision() ====================== #
smo.LocalLeaderDecision()
# ===================== Calling: GlobalLeaderDecision() ================== #
smo.GlobalLeaderDecision()
# ======================= Updating: 'cr' parameter ======================= #
smo.cr = smo.cr + (0.4 / iters)
# ====================== Saving the best individual ====================== #
smo.MinCost[l] = GlobalMin
gBestScore = GlobalMin
# ================ Displaying the fitness of each iteration ============== #
if (l % 1 == 0):
print(['At iteration ' + str(l + 1) + ' the best fitness is ' + str(gBestScore)]);
# ====================== Checking: acc_error ============================ #
if (math.fabs(GlobalMin - obj_val) <= smo.acc_err):
succ_rate += 1
mean_feval = mean_feval + smo.func_eval
break
# ========================= XXX Ending of Loop XXX ========================== #
# =========================== XX Result saving XX =========================== #
error1 = math.fabs(GlobalMin - obj_val)
timerEnd = time.time()
s.endTime = time.strftime("%Y-%m-%d-%H-%M-%S")
s.executionTime = timerEnd - timerStart
s.convergence = smo.MinCost
s.optimizer = "SMO"
s.error = error1
s.feval = smo.func_eval
s.objfname = smo.objf.__name__
return s, succ_rate, mean_feval
# ================================ X X X =================================== #

33
SMO/solution.py Normal file
View File

@ -0,0 +1,33 @@
# -*- coding: utf-8 -*-
"""
Python code of Spider-Monkey Optimization (SMO)
Coded by: Mukesh Saraswat (emailid: saraswatmukesh@gmail.com), Himanshu Mittal (emailid: himanshu.mittal224@gmail.com) and Raju Pal (emailid: raju3131.pal@gmail.com)
The code template used is similar to code given at link: https://github.com/himanshuRepo/CKGSA-in-Python
and C++ version of the SMO at link: http://smo.scrs.in/
Reference: Jagdish Chand Bansal, Harish Sharma, Shimpi Singh Jadon, and Maurice Clerc. "Spider monkey optimization algorithm for numerical optimization." Memetic computing 6, no. 1, 31-47, 2014.
@link: http://smo.scrs.in/
-- solution.py: Defining the solution variable for saving the output variables
Code compatible:
-- Python: 2.* or 3.*
"""
class solution:
def __init__(self):
self.best = 0
self.bestIndividual=[]
self.convergence = []
self.optimizer=""
self.objfname=""
self.startTime=0
self.endTime=0
self.executionTime=0
self.lb=0
self.ub=0
self.dim=0
self.popnum=0
self.error =0
self.feval=0
self.maxiers=0

89
dataloader.py Normal file
View File

@ -0,0 +1,89 @@
import random
from optimizer_common import *
def load_data(filename: str, load_cp_data=True, load_feeder_data=True, component_register=False):
# 锁定随机数种子
random.seed(0)
# 读取PCB数据
filename = 'data/' + filename
pcb_data = pd.DataFrame(pd.read_csv(filepath_or_buffer=filename, sep='\t', header=None))
if len(pcb_data.columns) <= 17:
step_col = ["ref", "x", "y", "z", "r", "part", "desc", "fdr", "nz", "hd", "cs", "cy", "sk", "bl", "ar",
"pl", "lv"]
elif len(pcb_data.columns) <= 18:
step_col = ["ref", "x", "y", "z", "r", "part", "desc", "fdr", "nz", "hd", "cs", "cy", "sk", "bl", "ar", "fid",
"pl", "lv"]
else:
step_col = ["ref", "x", "y", "z", "r", "part", "desc", "fdr", "nz", "hd", "cs", "cy", "sk", "bl", "ar", "fid",
"", "pl", "lv"]
pcb_data.columns = step_col
pcb_data = pcb_data.dropna(axis=1)
# 坐标系处理
# pcb_data = pcb_data.sort_values(by = ['x', 'y'], ascending = True)
# pcb_data["x"] = pcb_data["x"].apply(lambda x: -x)
# 注册元件检查
component_data = None
if load_cp_data:
part_feeder_assign = defaultdict(set)
part_col = ["part", "desc", "fdr", "nz", 'camera', 'group', 'feeder-limit']
try:
component_data = pd.DataFrame(pd.read_csv(filepath_or_buffer='component.txt', sep='\t', header=None), columns=part_col)
except:
component_data = pd.DataFrame(columns=part_col)
for _, data in pcb_data.iterrows():
part, nozzle = data.part, data.nz.split(' ')[1]
slot = data['fdr'].split(' ')[0]
if part not in component_data['part'].values:
if not component_register:
raise Exception("unregistered component: " + component_data['part'].values)
else:
component_data = pd.concat([component_data,
pd.DataFrame([part, '', 'SM8', nozzle, '飞行相机1', 'CHIP-Rect', 0],
index=part_col).T], ignore_index=True)
# warning_info = 'register component ' + part + ' with default feeder type'
# warnings.warn(warning_info, UserWarning)
part_index = component_data[component_data['part'] == part].index.tolist()[0]
part_feeder_assign[part].add(slot)
if nozzle != 'A' and component_data.loc[part_index]['nz'] != nozzle:
warning_info = 'the nozzle type of component ' + part + ' is not consistent with the pcb data'
warnings.warn(warning_info, UserWarning)
for idx, data in component_data.iterrows():
if data['fdr'][0:3] == 'SME': # 电动供料器和气动供料器参数一致
component_data.at[idx, 'fdr'] = data['fdr'][0:2] + data['fdr'][3:]
for part, slots in part_feeder_assign.items():
part_index = component_data[component_data['part'] == part].index.tolist()[0]
component_data.at[part_index, 'feeder-limit'] = max(len(slots), component_data.at[part_index, 'feeder-limit'])
# 读取供料器基座数据
feeder_data = pd.DataFrame(columns=range(3))
if load_feeder_data:
for data in pcb_data.iterrows():
fdr = data[1]['fdr']
slot, part = fdr.split(' ')
if slot[0] != 'F' and slot[0] != 'R':
continue
slot = int(slot[1:]) if slot[0] == 'F' else int(slot[1:]) + max_slot_index // 2
feeder_data = pd.concat([feeder_data, pd.DataFrame([slot, part, 1]).T])
feeder_data.columns = ['slot', 'part', 'arg'] # arg表示是否为预分配不表示分配数目
feeder_data.drop_duplicates(subset='slot', inplace=True, ignore_index=True)
# 随机移除部分已安装的供料器
if load_feeder_data == 2:
drop_index = random.sample(list(range(len(feeder_data))), len(feeder_data) // 2)
feeder_data.drop(index=drop_index, inplace=True)
feeder_data.sort_values(by='slot', ascending=True, inplace=True, ignore_index=True)
else:
feeder_data.columns = ['slot', 'part', 'arg'] # 同上
return pcb_data, component_data, feeder_data

316
optimizer.py Normal file
View File

@ -0,0 +1,316 @@
import math
import random
import matplotlib.pyplot as plt
from optimizer_common import *
from dataloader import *
def get_top_k_value(pop_val, k: int):
res = []
pop_val_cpy = copy.deepcopy(pop_val)
pop_val_cpy.sort(reverse=True)
for i in range(min(len(pop_val_cpy), k)):
for j in range(len(pop_val)):
if abs(pop_val_cpy[i] - pop_val[j]) < 1e-9 and j not in res:
res.append(j)
break
return res
def 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():
if component_index == 0:
index1 = random.randint(0, points + max_machine_index - 2)
while True:
index2 = random.randint(0, points + max_machine_index - 2)
if index1 != index2 and offspring[idx + index1] != offspring[idx + index2]:
break
offspring[idx + index1], offspring[idx + index2] = offspring[idx + index2], offspring[idx + index1]
break
component_index -= 1
idx += (points + max_machine_index - 1)
return offspring
def roulette_wheel_selection(pop_eval):
# Roulette wheel
random_val = np.random.random()
for idx, val in enumerate(pop_eval):
random_val -= val
if random_val <= 0:
return idx
return len(pop_eval) - 1
def random_selective(data, possibility): # 依概率选择随机数
assert len(data) == len(possibility) and len(data) > 0
sum_val = sum(possibility)
possibility = [p / sum_val for p in possibility]
random_val = random.random()
for idx, val in enumerate(possibility):
random_val -= val
if random_val <= 0:
break
return data[idx]
def selective_initialization(component_points, population_size):
population = [] # population initialization
for _ in range(population_size):
individual = []
for points in component_points.values():
if points == 0:
continue
avl_machine_num = random.randint(1, min(max_machine_index, points)) # 可用机器数
selective_possibility = []
for p in range(1, avl_machine_num + 1):
selective_possibility.append(pow(2, avl_machine_num - p + 1))
sel_machine_num = random_selective([p + 1 for p in range(avl_machine_num)], selective_possibility) # 选择的机器数
sel_machine_set = random.sample([p for p in range(avl_machine_num)], sel_machine_num)
sel_machine_points = [1 for _ in range(sel_machine_num)]
for p in range(sel_machine_num - 1):
if points == sum(sel_machine_points):
break
assign_points = random.randint(1, points - sum(sel_machine_points))
sel_machine_points[p] += assign_points
if sum(sel_machine_points) < points:
sel_machine_points[-1] += (points - sum(sel_machine_points))
# code component allocation into chromosome
for p in range(max_machine_index):
if p in sel_machine_set:
individual += [0 for _ in range(sel_machine_points[0])]
sel_machine_points.pop(0)
individual.append(1)
individual.pop(-1)
population.append(individual)
return population
def selective_crossover(mother, father, non_decelerating=True):
assert len(mother) == len(father)
offspring1, offspring2 = mother.copy(), father.copy()
one_counter, feasible_cutline = 0, []
for idx in range(len(mother) - 1):
if mother[idx] == 1:
one_counter += 1
if father[idx] == 1:
one_counter -= 1
# first constraint: the total number of “1”s (the number of partitions) in the chromosome is unchanged
if one_counter != 0 or idx == 0 or idx == len(mother) - 2:
continue
# the selected cutline should guarantee there are the same or a larger number unassigned machine
# for each component type
n_bro, n_new = 0, 0
if mother[idx] and mother[idx + 1]:
n_bro += 1
if father[idx] and father[idx + 1]:
n_bro += 1
if mother[idx] and father[idx + 1]:
n_new += 1
if father[idx] and mother[idx + 1]:
n_new += 1
# non_decelerating or accelerating crossover
if (non_decelerating and n_bro <= n_new) or n_bro < n_new:
feasible_cutline.append(idx)
if len(feasible_cutline) == 0:
return offspring1, offspring2
cutline_idx = feasible_cutline[random.randint(0, len(feasible_cutline) - 1)]
offspring1, offspring2 = mother[:cutline_idx + 1] + father[cutline_idx + 1:], father[:cutline_idx + 1] + mother[
cutline_idx + 1:]
return offspring1, offspring2
def cal_individual_val(component_points, component_nozzle, individual):
idx, objective_val = 0, [0]
machine_component_points = [[] for _ in range(max_machine_index)]
# decode the component allocation
for points in component_points.values():
component_gene = individual[idx: idx + points + max_machine_index - 1]
machine_idx, component_counter = 0, 0
for gene in component_gene:
if gene:
machine_component_points[machine_idx].append(component_counter)
machine_idx += 1
component_counter = 0
else:
component_counter += 1
machine_component_points[-1].append(component_counter)
idx += (points + max_machine_index - 1)
for machine_idx in range(max_machine_index):
nozzle_points = defaultdict(int)
for idx, nozzle in component_nozzle.items():
if component_points[idx] == 0:
continue
nozzle_points[nozzle] += machine_component_points[machine_idx][idx]
machine_points = sum(machine_component_points[machine_idx]) # num of placement points
if machine_points == 0:
continue
ul = math.ceil(len(nozzle_points) * 1.0 / max_head_index) - 1 # num of nozzle set
# assignments of nozzles to heads
wl = 0 # num of workload
total_heads = (1 + ul) * max_head_index - len(nozzle_points)
nozzle_heads = defaultdict(int)
for nozzle in nozzle_points.keys():
nozzle_heads[nozzle] = math.floor(nozzle_points[nozzle] * 1.0 / machine_points * total_heads)
nozzle_heads[nozzle] += 1
total_heads = (1 + ul) * max_head_index
for heads in nozzle_heads.values():
total_heads -= heads
for nozzle in nozzle_heads.keys(): # TODO有利于减少周期的方法
if total_heads == 0:
break
nozzle_heads[nozzle] += 1
total_heads -= 1
# averagely assign placements to heads
heads_placement = []
for nozzle in nozzle_heads.keys():
points = math.floor(nozzle_points[nozzle] / nozzle_heads[nozzle])
heads_placement += [[nozzle, points] for _ in range(nozzle_heads[nozzle])]
nozzle_points[nozzle] -= (nozzle_heads[nozzle] * points)
for idx in range(len(heads_placement) - 1, -1, -1):
if nozzle_points[nozzle] <= 0:
break
nozzle_points[nozzle] -= 1
heads_placement[idx][1] += 1
heads_placement = sorted(heads_placement, key=lambda x: x[1], reverse=True)
# every max_head_index heads in the non-decreasing order are grouped together as nozzle set
for idx in range(len(heads_placement) // max_head_index):
wl += heads_placement[idx][1]
objective_val.append(T_pp * machine_points + T_tr * wl + T_nc * ul)
return max(objective_val), machine_component_points
@timer_wrapper
def optimizer(pcb_data, component_data):
# basic parameter
# crossover rate & mutation rate: 80% & 10%
# population size: 200
# the number of generation: 500
crossover_rate, mutation_rate = 0.8, 0.1
population_size, n_generations = 200, 500
# the number of placement points and nozzle type of component
component_points, component_nozzle = defaultdict(int), defaultdict(str)
for data in pcb_data.iterrows():
part_index = component_data[component_data['part'] == data[1]['part']].index.tolist()[0]
nozzle = component_data.loc[part_index]['nz']
component_points[part_index] += 1
component_nozzle[part_index] = nozzle
# population initialization
best_popval = []
population = selective_initialization(component_points, population_size)
with tqdm(total=n_generations) as pbar:
pbar.set_description('genetic process for PCB assembly')
new_population, new_pop_val = [], []
for _ in range(n_generations):
# calculate fitness value
pop_val = []
for individual in population:
val, _ = cal_individual_val(component_points, component_nozzle, individual)
pop_val.append(val)
best_popval.append(min(pop_val))
# min-max convert
max_val = max(pop_val)
pop_val = list(map(lambda v: max_val - v, pop_val))
sum_pop_val = sum(pop_val)
pop_val = [v / sum_pop_val for v in pop_val]
select_index = get_top_k_value(pop_val, population_size - len(new_pop_val))
population = [population[idx] for idx in select_index]
pop_val = [pop_val[idx] for idx in select_index]
population += new_population
for individual in new_population:
val, _ = cal_individual_val(component_points, component_nozzle, individual)
pop_val.append(val)
# crossover and mutation
new_population = []
for pop in range(population_size):
if pop % 2 == 0 and np.random.random() < crossover_rate:
index1 = roulette_wheel_selection(pop_val)
while True:
index2 = roulette_wheel_selection(pop_val)
if index1 != index2:
break
offspring1, offspring2 = selective_crossover(population[index1], population[index2])
if np.random.random() < mutation_rate:
offspring1 = swap_mutation(component_points, offspring1)
if np.random.random() < mutation_rate:
offspring1 = swap_mutation(component_points, offspring1)
new_population.append(offspring1)
new_population.append(offspring2)
pbar.update(1)
best_individual = population[np.argmin(pop_val)]
val, result = cal_individual_val(component_points, component_nozzle, best_individual)
print(result)
plt.plot(best_popval)
plt.show()
# TODO: 计算实际的PCB整线组装时间
if __name__ == '__main__':
# warnings.simplefilter('ignore')
# 参数解析
parser = argparse.ArgumentParser(description='assembly line optimizer implementation')
parser.add_argument('--filename', default='PCB.txt', type=str, help='load pcb data')
parser.add_argument('--auto_register', default=1, type=int, help='register the component according the pcb data')
params = parser.parse_args()
# 结果输出显示所有行和列
pd.set_option('display.max_columns', None)
pd.set_option('display.max_rows', None)
# 加载PCB数据
pcb_data, component_data, _ = load_data(params.filename, component_register=params.auto_register) # 加载PCB数据
optimizer(pcb_data, component_data)

78
optimizer_common.py Normal file
View File

@ -0,0 +1,78 @@
import copy
import time
import math
import argparse
import warnings
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from functools import wraps
from collections import defaultdict
from typing import List, Counter
from tqdm import tqdm
# 机器参数
max_head_index, max_slot_index = 6, 120 # 暂时默认所有机器参数相同
max_machine_index = 3
interval_ratio = 2
slot_interval = 15
head_interval = slot_interval * interval_ratio
head_nozzle = ['' for _ in range(max_head_index)] # 头上已经分配吸嘴
# 位置信息
slotf1_pos, slotr1_pos = [-31.267, 44.], [807., 810.545] # F1(前基座最左侧)、R1(后基座最右侧)位置
fix_camera_pos = [269.531, 694.823] # 固定相机位置
anc_marker_pos = [336.457, 626.230] # ANC基准点位置
stopper_pos = [635.150, 124.738] # 止档块位置
# 时间参数
T_pp, T_tr, T_nc = 2, 5, 25
# 电机参数
head_rotary_velocity = 8e-5 # 贴装头R轴旋转时间
x_max_velocity, y_max_velocity = 1.4, 1.2
x_max_acceleration, y_max_acceleration = x_max_velocity / 0.079, y_max_velocity / 0.079
# 不同种类供料器宽度
feeder_width = {'SM8': (7.25, 7.25), 'SM12': (7.00, 20.00), 'SM16': (7.00, 22.00),
'SM24': (7.00, 29.00), 'SM32': (7.00, 44.00)}
# 可用吸嘴数量限制
nozzle_limit = {'CN065': 6, 'CN040': 6, 'CN220': 6, 'CN400': 6, 'CN140': 6}
def axis_moving_time(distance, axis=0):
distance = abs(distance) * 1e-3
Lamax = x_max_velocity ** 2 / x_max_acceleration if axis == 0 else y_max_velocity ** 2 / y_max_acceleration
Tmax = x_max_velocity / x_max_acceleration if axis == 0 else y_max_velocity / y_max_acceleration
if axis == 0:
return 2 * math.sqrt(distance / x_max_acceleration) if distance < Lamax else 2 * Tmax + (
distance - Lamax) / x_max_velocity
else:
return 2 * math.sqrt(distance / y_max_acceleration) if distance < Lamax else 2 * Tmax + (
distance - Lamax) / y_max_velocity
def head_rotary_time(angle):
while -180 > angle > 180:
if angle > 180:
angle -= 360
else:
angle += 360
return abs(angle) * head_rotary_velocity
def timer_wrapper(func):
@wraps(func)
def measure_time(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
print("function {} running time : {} s".format(func.__name__, time.time() - start_time))
return result
return measure_time

579
result_analysis.py Normal file
View File

@ -0,0 +1,579 @@
from optimizer_common import *
# 绘制各周期从供料器周期拾取的元件位置
def pickup_cycle_schematic(feeder_slot_result, cycle_result):
plt.rcParams['font.sans-serif'] = ['KaiTi'] # 指定默认字体
plt.rcParams['axes.unicode_minus'] = False # 解决保存图像是负号'-'显示为方块的问题
# data
bar_width = .7
feeder_part = np.zeros((int)(max_slot_index / 2), dtype = np.int)
for cycle in range(len(feeder_slot_result)):
label_str = '周期' + str(cycle + 1)
cur_feeder_part = np.zeros((int)(max_slot_index / 2), dtype = np.int)
for slot in feeder_slot_result[cycle]:
if slot > 0:
cur_feeder_part[slot] += cycle_result[cycle]
plt.bar(np.arange(max_slot_index / 2), cur_feeder_part, bar_width, edgecolor='black', bottom=feeder_part,
label=label_str)
for slot in feeder_slot_result[cycle]:
if slot > 0:
feeder_part[slot] += cycle_result[cycle]
plt.legend()
plt.show()
def placement_route_schematic(pcb_data, component_result, cycle_result, feeder_slot_result, placement_result,
head_sequence, cycle=-1):
plt.figure('cycle {}'.format(cycle + 1))
pos_x, pos_y = [], []
for i in range(len(pcb_data)):
pos_x.append(pcb_data.loc[i]['x'] + stopper_pos[0])
pos_y.append(pcb_data.loc[i]['y'] + stopper_pos[1])
# plt.text(pcb_data.loc[i]['x'], pcb_data.loc[i]['y'] + 0.1, '%d' % i, ha='center', va = 'bottom', size = 8)
mount_pos = []
for head in head_sequence[cycle]:
index = placement_result[cycle][head]
plt.text(pos_x[index], pos_y[index] + 0.1, 'HD%d' % (head + 1), ha='center', va = 'bottom', size = 10)
plt.plot([pos_x[index], pos_x[index] - head * head_interval], [pos_y[index], pos_y[index]], linestyle='-.',
color='black', linewidth=1)
mount_pos.append([pos_x[index] - head * head_interval, pos_y[index]])
plt.plot(mount_pos[-1][0], mount_pos[-1][1], marker='^', color='red', markerfacecolor='white')
# plt.text(mount_pos[-1][0], mount_pos[-1][1], '%d' % index, size=8)
# 绘制贴装路径
for i in range(len(mount_pos) - 1):
plt.plot([mount_pos[i][0], mount_pos[i + 1][0]], [mount_pos[i][1], mount_pos[i + 1][1]], color='blue', linewidth=1)
draw_x, draw_y = [], []
for c in range(cycle, len(placement_result)):
for h in range(max_head_index):
i = placement_result[c][h]
if i == -1:
continue
draw_x.append(pcb_data.loc[i]['x'] + stopper_pos[0])
draw_y.append(pcb_data.loc[i]['y'] + stopper_pos[1])
# plt.text(draw_x[-1], draw_y[-1] - 5, '%d' % i, ha='center', va='bottom', size=10)
plt.scatter(draw_x, draw_y, s=8)
# 绘制供料器位置布局
for slot in range(max_slot_index // 2):
plt.scatter(slotf1_pos[0] + slot_interval * slot, slotf1_pos[1], marker = 'x', s = 12, color = 'green')
plt.text(slotf1_pos[0] + slot_interval * slot, slotf1_pos[1] - 50, slot + 1, ha = 'center', va = 'bottom', size = 8)
feeder_part, feeder_counter = {}, {}
placement_cycle = 0
for cycle_, components in enumerate(component_result):
for head, component in enumerate(components):
if component == -1:
continue
placement = placement_result[placement_cycle][head]
slot = feeder_slot_result[cycle_][head]
feeder_part[slot] = pcb_data.loc[placement]['part']
if slot not in feeder_counter.keys():
feeder_counter[slot] = 0
feeder_counter[slot] += cycle_result[cycle_]
placement_cycle += cycle_result[cycle_]
for slot, part in feeder_part.items():
plt.text(slotf1_pos[0] + slot_interval * (slot - 1), slotf1_pos[1] + 15, part + ': ' + str(feeder_counter[slot]), ha = 'center', size = 7, rotation = 90)
plt.plot([slotf1_pos[0] - slot_interval / 2, slotf1_pos[0] + slot_interval * (max_slot_index // 2 - 1 + 0.5)],
[slotf1_pos[1] + 10, slotf1_pos[1] + 10], color = 'black')
plt.plot([slotf1_pos[0] - slot_interval / 2, slotf1_pos[0] + slot_interval * (max_slot_index // 2 - 1 + 0.5)],
[slotf1_pos[1] - 40, slotf1_pos[1] - 40], color = 'black')
for counter in range(max_slot_index // 2 + 1):
pos = slotf1_pos[0] + (counter - 0.5) * slot_interval
plt.plot([pos, pos], [slotf1_pos[1] + 10, slotf1_pos[1] - 40], color='black', linewidth = 1)
# 绘制拾取路径
pick_slot = []
cycle_group = 0
while sum(cycle_result[0: cycle_group + 1]) < cycle:
cycle_group += 1
for head, slot in enumerate(feeder_slot_result[cycle_group]):
if slot == -1:
continue
pick_slot.append(slot - head * interval_ratio)
pick_slot = list(set(pick_slot))
pick_slot = sorted(pick_slot)
next_cycle_group = 0
next_pick_slot = max_slot_index
while sum(cycle_result[0: next_cycle_group + 1]) < cycle + 1:
next_cycle_group += 1
if next_cycle_group < len(feeder_slot_result):
for head, slot in enumerate(feeder_slot_result[cycle_group]):
if slot == -1:
continue
next_pick_slot = min(next_pick_slot, slot - head * interval_ratio)
# 前往PCB贴装
plt.plot([mount_pos[-1][0], slotf1_pos[0] + slot_interval * (pick_slot[-1] - 1)], [mount_pos[-1][1], slotf1_pos[1]],
color='blue', linewidth=1)
# 基座移动路径
plt.plot([slotf1_pos[0] + slot_interval * (pick_slot[0] - 1), slotf1_pos[0] + slot_interval * (pick_slot[-1] - 1)],
[slotf1_pos[1], slotf1_pos[1]], color='blue', linewidth=1)
# 返回基座取料
plt.plot([mount_pos[0][0], slotf1_pos[0] + slot_interval * (next_pick_slot - 1)], [mount_pos[0][1], slotf1_pos[1]],
color='blue', linewidth=1)
plt.show()
def save_placement_route_figure(file_name, pcb_data, component_result, cycle_result, feeder_slot_result, placement_result, head_sequence):
path = 'result/' + file_name[:file_name.find('.')]
if not os.path.exists(path):
os.mkdir(path)
pos_x, pos_y = [], []
for i in range(len(pcb_data)):
pos_x.append(pcb_data.loc[i]['x'] + stopper_pos[0])
pos_y.append(pcb_data.loc[i]['y'] + stopper_pos[1])
# plt.text(pcb_data.loc[i]['x'], pcb_data.loc[i]['y'] + 0.1, '%d' % i, ha='center', va = 'bottom', size = 8)
with tqdm(total=100) as pbar:
pbar.set_description('save figure')
for cycle in range(len(placement_result)):
plt.figure(cycle)
mount_pos = []
for head in head_sequence[cycle]:
index = placement_result[cycle][head]
plt.text(pos_x[index], pos_y[index] + 0.1, 'HD%d' % (head + 1), ha='center', va='bottom', size=10)
plt.plot([pos_x[index], pos_x[index] - head * head_interval], [pos_y[index], pos_y[index]], linestyle='-.',
color='black', linewidth=1)
mount_pos.append([pos_x[index] - head * head_interval, pos_y[index]])
plt.plot(mount_pos[-1][0], mount_pos[-1][1], marker='^', color='red', markerfacecolor='white')
# 绘制贴装路径
for i in range(len(mount_pos) - 1):
plt.plot([mount_pos[i][0], mount_pos[i + 1][0]], [mount_pos[i][1], mount_pos[i + 1][1]], color='blue',
linewidth=1)
draw_x, draw_y = [], []
for c in range(cycle, len(placement_result)):
for h in range(max_head_index):
i = placement_result[c][h]
if i == -1:
continue
draw_x.append(pcb_data.loc[i]['x'] + stopper_pos[0])
draw_y.append(pcb_data.loc[i]['y'] + stopper_pos[1])
# plt.text(draw_x[-1], draw_y[-1] - 5, '%d' % i, ha='center', va='bottom', size=10)
plt.scatter(pos_x, pos_y, s=8)
# 绘制供料器位置布局
for slot in range(max_slot_index // 2):
plt.scatter(slotf1_pos[0] + slot_interval * slot, slotf1_pos[1], marker='x', s=12, color='green')
plt.text(slotf1_pos[0] + slot_interval * slot, slotf1_pos[1] - 50, slot + 1, ha='center', va='bottom', size=8)
feeder_part, feeder_counter = {}, {}
placement_cycle = 0
for cycle_, components in enumerate(component_result):
for head, component in enumerate(components):
if component == -1:
continue
placement = placement_result[placement_cycle][head]
slot = feeder_slot_result[cycle_][head]
feeder_part[slot] = pcb_data.loc[placement]['part']
if slot not in feeder_counter.keys():
feeder_counter[slot] = 0
feeder_counter[slot] += cycle_result[cycle_]
placement_cycle += cycle_result[cycle_]
for slot, part in feeder_part.items():
plt.text(slotf1_pos[0] + slot_interval * (slot - 1), slotf1_pos[1] + 15,
part + ': ' + str(feeder_counter[slot]), ha='center', size=7, rotation=90)
plt.plot([slotf1_pos[0] - slot_interval / 2, slotf1_pos[0] + slot_interval * (max_slot_index // 2 - 1 + 0.5)],
[slotf1_pos[1] + 10, slotf1_pos[1] + 10], color='black')
plt.plot([slotf1_pos[0] - slot_interval / 2, slotf1_pos[0] + slot_interval * (max_slot_index // 2 - 1 + 0.5)],
[slotf1_pos[1] - 40, slotf1_pos[1] - 40], color='black')
for counter in range(max_slot_index // 2 + 1):
pos = slotf1_pos[0] + (counter - 0.5) * slot_interval
plt.plot([pos, pos], [slotf1_pos[1] + 10, slotf1_pos[1] - 40], color='black', linewidth=1)
# 绘制拾取路径
pick_slot = []
cycle_group = 0
while sum(cycle_result[0: cycle_group + 1]) < cycle:
cycle_group += 1
for head, slot in enumerate(feeder_slot_result[cycle_group]):
if slot == -1:
continue
pick_slot.append(slot - head * interval_ratio)
pick_slot = list(set(pick_slot))
pick_slot = sorted(pick_slot)
plt.plot([mount_pos[0][0], slotf1_pos[0] + slot_interval * (pick_slot[0] - 1)], [mount_pos[0][1], slotf1_pos[1]],
color='blue', linewidth=1)
plt.plot([mount_pos[-1][0], slotf1_pos[0] + slot_interval * (pick_slot[-1] - 1)], [mount_pos[-1][1], slotf1_pos[1]],
color='blue', linewidth=1)
plt.plot([slotf1_pos[0] + slot_interval * (pick_slot[0] - 1), slotf1_pos[0] + slot_interval * (pick_slot[-1] - 1)],
[slotf1_pos[1], slotf1_pos[1]], color='blue', linewidth=1)
plt.savefig(path + '/cycle_{}'.format(cycle + 1))
plt.close(cycle)
pbar.update(100 / len(placement_result))
def output_optimize_result(file_name, method, component_data, pcb_data, feeder_data, component_result, cycle_result,
feeder_slot_result, placement_result, head_sequence):
assert len(component_result) == len(feeder_slot_result)
if feeder_data is None:
warning_info = 'file: ' + file_name + ' optimize result is not existed!'
warnings.warn(warning_info, UserWarning)
return
output_data = pcb_data.copy(deep=True)
# 默认ANC参数
anc_list = defaultdict(list)
anc_list['CN065'] = list(range(14, 25, 2))
anc_list['CN220'] = list(range(15, 26, 2))
anc_list['CN140'] = list(range(26, 37, 2))
anc_list['CN400'] = list(range(27, 38, 2))
# 更新供料器组参数
for cycle_set in range(len(cycle_result)):
for head, component in enumerate(component_result[cycle_set]):
if component == -1:
continue
if feeder_data[feeder_data['slot'] == feeder_slot_result[cycle_set][head]].index.empty:
part = component_data.loc[component]['part']
feeder_data.loc[len(feeder_data.index)] = [feeder_slot_result[cycle_set][head], part, 0]
feeder_data.sort_values('slot', inplace=True, ascending=True, ignore_index=True)
placement_index = []
assigned_nozzle, assigned_anc_hole = ['' for _ in range(max_head_index)], [-1 for _ in range(max_head_index)]
for cycle_set in range(len(cycle_result)):
floor_cycle, ceil_cycle = sum(cycle_result[:cycle_set]), sum(cycle_result[:(cycle_set + 1)])
for cycle in range(floor_cycle, ceil_cycle):
cycle_start = True
cycle_nozzle = ['' for _ in range(max_head_index)]
head_indexes = [-1 for _ in range(max_head_index)]
for head in head_sequence[cycle]:
index_ = placement_result[cycle][head]
if index_ == -1:
continue
head_indexes[head] = index_
placement_index.append(index_)
output_data.loc[index_, 'cs'] = 1 if cycle_start else 0
output_data.loc[index_, 'cy'] = cycle + 1
output_data.loc[index_, 'hd'] = head + 1
cycle_start = False
# 供料器信息
slot = feeder_slot_result[cycle_set][head]
fdr = 'F' + str(slot) if slot < max_slot_index // 2 else 'R' + str(slot - max_slot_index // 2)
feeder_index = feeder_data[feeder_data['slot'] == slot].index.tolist()[0]
output_data.loc[index_, 'fdr'] = fdr + ' ' + feeder_data.loc[feeder_index, 'part']
# ANC信息
cycle_nozzle[head] = component_data.loc[component_result[cycle_set][head], 'nz']
for head in range(max_head_index):
nozzle = cycle_nozzle[head]
if nozzle == '':
continue
if nozzle != assigned_nozzle[head]:
# 已分配有吸嘴,卸载原吸嘴
if assigned_nozzle[head] != '':
anc_list[assigned_nozzle[head]].append(assigned_anc_hole[head])
anc_list[assigned_nozzle[head]] = sorted(anc_list[assigned_nozzle[head]])
# 安装新的吸嘴
assigned_nozzle[head] = nozzle
try:
assigned_anc_hole[head] = anc_list[nozzle][0]
except IndexError:
info = 'the number of nozzle for [' + nozzle + '] exceeds the quantity limit'
raise IndexError(info)
anc_list[nozzle].pop(0)
output_data.loc[head_indexes[head], 'nz'] = '1-' + str(assigned_anc_hole[head]) + ' ' + nozzle
output_data = output_data.reindex(placement_index)
output_data = output_data.reset_index(drop=True)
if 'desc' not in output_data.columns:
column_index = int(np.where(output_data.columns.values.reshape(-1) == 'part')[0][0])
output_data.insert(loc=column_index + 1, column='desc', value='')
if not os.path.exists('result/' + method):
os.makedirs('result/' + method)
file_name = method + '/' + file_name.split('.')[0] + '.xlsx'
output_data.to_excel('result/' + file_name, sheet_name='tb1', float_format='%.3f', na_rep='')
def component_assign_evaluate(component_data, component_result, cycle_result, feeder_slot_result) -> float:
nozzle_change_counter = 0
for head in range(max_head_index):
nozzle = ''
for cycle in range(len(component_result)):
component_index = component_result[cycle][head]
if component_index == -1:
continue
if cycle != 0 and nozzle != component_data.loc[component_index, 'nz']:
nozzle_change_counter += 1
nozzle = component_data.loc[component_index, 'nz']
gang_pick_counter = 0
for cycle, feeder_slot in enumerate(feeder_slot_result):
pick_slot = defaultdict(int)
for head, slot in enumerate(feeder_slot):
if slot == -1:
continue
pick_slot[slot - head * interval_ratio] += 1
for _ in pick_slot.values():
gang_pick_counter += cycle_result[cycle]
return sum(cycle_result) + e_nz_change * nozzle_change_counter + e_gang_pick * gang_pick_counter
def optimization_assign_result(component_data, pcb_data, component_result, cycle_result, feeder_slot_result,
nozzle_hinter=False, component_hinter=False, feeder_hinter=False):
if nozzle_hinter:
columns = ['H{}'.format(i + 1) for i in range(max_head_index)] + ['cycle']
nozzle_assign = pd.DataFrame(columns=columns)
for cycle, components in enumerate(component_result):
nozzle_assign.loc[cycle, 'cycle'] = cycle_result[cycle]
for head in range(max_head_index):
index = component_result[cycle][head]
if index == -1:
nozzle_assign.loc[cycle, 'H{}'.format(head + 1)] = ''
else:
nozzle = component_data.loc[index]['nz']
nozzle_assign.loc[cycle, 'H{}'.format(head + 1)] = nozzle
print(nozzle_assign)
print('')
if component_hinter:
columns = ['H{}'.format(i + 1) for i in range(max_head_index)] + ['cycle']
component_assign = pd.DataFrame(columns=columns)
for cycle, components in enumerate(component_result):
component_assign.loc[cycle, 'cycle'] = cycle_result[cycle]
for head in range(max_head_index):
index = component_result[cycle][head]
if index == -1:
component_assign.loc[cycle, 'H{}'.format(head + 1)] = ''
else:
part = component_data.loc[index]['part']
component_assign.loc[cycle, 'H{}'.format(head + 1)] = part
print(component_assign)
print('')
if feeder_hinter:
columns = ['H{}'.format(i + 1) for i in range(max_head_index)] + ['cycle']
feedr_assign = pd.DataFrame(columns=columns)
for cycle, components in enumerate(feeder_slot_result):
feedr_assign.loc[cycle, 'cycle'] = cycle_result[cycle]
for head in range(max_head_index):
slot = feeder_slot_result[cycle][head]
if slot == -1:
feedr_assign.loc[cycle, 'H{}'.format(head + 1)] = 'A'
else:
feedr_assign.loc[cycle, 'H{}'.format(head + 1)] = 'F{}'.format(
slot) if slot <= max_slot_index // 2 else 'R{}'.format(slot - max_head_index)
print(feedr_assign)
print('')
def placement_time_estimate(component_data, pcb_data, component_result, cycle_result, feeder_slot_result,
placement_result, head_sequence, hinter=True) -> float:
# === 校验 ===
total_points = 0
for cycle, components in enumerate(component_result):
for head, component in enumerate(components):
if component == -1:
continue
total_points += cycle_result[cycle]
if total_points != len(pcb_data):
warning_info = 'the number of placement points is not match with the PCB data. '
warnings.warn(warning_info, UserWarning)
return 0.
for placements in placement_result:
for placement in placements:
if placement == -1:
continue
total_points -= 1
if total_points != 0:
warnings.warn(
'the optimization result of component assignment result and placement result are not consistent. ',
UserWarning)
return 0.
feeder_arrangement = defaultdict(set)
for cycle, feeder_slots in enumerate(feeder_slot_result):
for head, slot in enumerate(feeder_slots):
if slot == -1:
continue
feeder_arrangement[component_result[cycle][head]].add(slot)
for part, data in component_data.iterrows():
if part in feeder_arrangement.keys() and data['feeder-limit'] < len(feeder_arrangement[part]):
info = 'the number of arranged feeder of [' + data['part'] + '] exceeds the quantity limit'
warnings.warn(info, UserWarning)
return 0.
t_pick, t_place = .078, .051 # 贴装/拾取用时
t_nozzle_put, t_nozzle_pick = 0.9, 0.75 # 装卸吸嘴用时
t_fix_camera_check = 0.12 # 固定相机检测时间
total_moving_time = .0 # 总移动用时
total_operation_time = .0 # 操作用时
total_nozzle_change_counter = 0 # 总吸嘴更换次数
total_pick_counter = 0 # 总拾取次数
total_mount_distance, total_pick_distance = .0, .0 # 贴装距离、拾取距离
total_distance = 0 # 总移动距离
cur_pos, next_pos = anc_marker_pos, [0, 0] # 贴装头当前位置
# 初始化首个周期的吸嘴装配信息
nozzle_assigned = ['Empty' for _ in range(max_head_index)]
for head in range(max_head_index):
for cycle in range(len(component_result)):
idx = component_result[cycle][head]
if idx == -1:
continue
else:
nozzle_assigned[head] = component_data.loc[idx]['nz']
break
for cycle_set, _ in enumerate(component_result):
floor_cycle, ceil_cycle = sum(cycle_result[:cycle_set]), sum(cycle_result[:(cycle_set + 1)])
for cycle in range(floor_cycle, ceil_cycle):
pick_slot, mount_pos, mount_angle = [], [], []
nozzle_pick_counter, nozzle_put_counter = 0, 0 # 吸嘴更换次数统计(拾取/放置分别算一次)
for head in range(max_head_index):
if feeder_slot_result[cycle_set][head] != -1:
pick_slot.append(feeder_slot_result[cycle_set][head] - interval_ratio * head)
if component_result[cycle_set][head] == -1:
continue
nozzle = component_data.loc[component_result[cycle_set][head]]['nz']
if nozzle != nozzle_assigned[head]:
if nozzle_assigned[head] != 'Empty':
nozzle_put_counter += 1
nozzle_pick_counter += 1
nozzle_assigned[head] = nozzle
# ANC处进行吸嘴更换
if nozzle_pick_counter + nozzle_put_counter > 0:
next_pos = anc_marker_pos
total_moving_time += max(axis_moving_time(cur_pos[0] - next_pos[0], 0),
axis_moving_time(cur_pos[1] - next_pos[1], 1))
total_distance += max(abs(cur_pos[0] - next_pos[0]), abs(cur_pos[1] - next_pos[1]))
cur_pos = next_pos
pick_slot = list(set(pick_slot))
pick_slot = sorted(pick_slot, reverse=True)
# 拾取路径(自右向左)
for slot in pick_slot:
if slot < max_slot_index // 2:
next_pos = [slotf1_pos[0] + slot_interval * (slot - 1), slotf1_pos[1]]
else:
next_pos = [slotr1_pos[0] - slot_interval * (max_slot_index - slot - 1), slotr1_pos[1]]
total_operation_time += t_pick
total_pick_counter += 1
total_moving_time += max(axis_moving_time(cur_pos[0] - next_pos[0], 0),
axis_moving_time(cur_pos[1] - next_pos[1], 1))
total_distance += max(abs(cur_pos[0] - next_pos[0]), abs(cur_pos[1] - next_pos[1]))
if slot != pick_slot[0]:
total_pick_distance += max(abs(cur_pos[0] - next_pos[0]), abs(cur_pos[1] - next_pos[1]))
cur_pos = next_pos
# 固定相机检测
for head in range(max_head_index):
if component_result[cycle_set][head] == -1:
continue
camera = component_data.loc[component_result[cycle_set][head]]['camera']
if camera == '固定相机':
next_pos = [fix_camera_pos[0] - head * head_interval, fix_camera_pos[1]]
total_moving_time += max(axis_moving_time(cur_pos[0] - next_pos[0], 0),
axis_moving_time(cur_pos[1] - next_pos[1], 1))
total_distance += max(abs(cur_pos[0] - next_pos[0]), abs(cur_pos[1] - next_pos[1]))
total_operation_time += t_fix_camera_check
cur_pos = next_pos
# 贴装路径
for head in head_sequence[cycle]:
index = placement_result[cycle][head]
if index == -1:
continue
mount_pos.append([pcb_data.loc[index]['x'] - head * head_interval + stopper_pos[0],
pcb_data.loc[index]['y'] + stopper_pos[1]])
mount_angle.append(pcb_data.loc[index]['r'])
# 单独计算贴装路径
for cntPoints in range(len(mount_pos) - 1):
total_mount_distance += max(abs(mount_pos[cntPoints][0] - mount_pos[cntPoints + 1][0]),
abs(mount_pos[cntPoints][1] - mount_pos[cntPoints + 1][1]))
# 考虑R轴预旋转补偿同轴角度转动带来的额外贴装用时
total_operation_time += head_rotary_time(mount_angle[0]) # 补偿角度转动带来的额外贴装用时
total_operation_time += t_nozzle_put * nozzle_put_counter + t_nozzle_pick * nozzle_pick_counter
for pos in mount_pos:
total_operation_time += t_place
total_moving_time += max(axis_moving_time(cur_pos[0] - pos[0], 0),
axis_moving_time(cur_pos[1] - pos[1], 1))
total_distance += max(abs(cur_pos[0] - pos[0]), abs(cur_pos[1] - pos[1]))
cur_pos = pos
total_nozzle_change_counter += nozzle_put_counter + nozzle_pick_counter
total_time = total_moving_time + total_operation_time
minutes, seconds = int(total_time // 60), int(total_time) % 60
millisecond = int((total_time - minutes * 60 - seconds) * 60)
if hinter:
optimization_assign_result(component_data, pcb_data, component_result, cycle_result, feeder_slot_result,
nozzle_hinter=True, component_hinter=True, feeder_hinter=True)
print('-Cycle counter: {}'.format(sum(cycle_result)))
print('-Nozzle change counter: {}'.format(total_nozzle_change_counter // 2))
print('-Pick operation counter: {}'.format(total_pick_counter))
print('-Expected mounting tour length: {} mm'.format(total_mount_distance))
print('-Expected picking tour length: {} mm'.format(total_pick_distance))
print('-Expected total tour length: {} mm'.format(total_distance))
print('-Expected total moving time: {} s'.format(total_moving_time))
print('-Expected total operation time: {} s'.format(total_operation_time))
if minutes > 0:
print('-Mounting time estimation: {:d} min {} s {:2d} ms ({:.3f}s)'.format(minutes, seconds, millisecond,
total_time))
else:
print('-Mounting time estimation: {} s {:2d} ms ({:.3f}s)'.format(seconds, millisecond, total_time))
return total_time