fuzz-scene-sim-test/FuzzScene/code/GA/ga.py

# ga.py
import numpy as np
import random
import fit_funcs
from sampling import *


class Individual:
    def __init__(self, dna, pop_seed_name):
        self.dna = dna
        self.dna_len = len(dna)
        self.pop_seed_name = pop_seed_name
        self.f = []
        self.choose_time = 0
        self.error_number = None
        self.list_er = []

    def __lt__(self, other):
        if self.f[0] > other.f[0]:
            return True

        return False

    def get_fitness(self, F, domain_flags, data_collection_para):
        # len_F = len(F)
        # fitness = np.zeros(shape=len_F)
        # for i in range(len_F):
        fitness, error_number, list_er = F(self.dna, self.pop_seed_name, self.error_number, self.choose_time,
                                           self.list_er, domain_flags, data_collection_para)
        print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", type(fitness))
        return fitness, error_number, list_er

    # def cross_newObj(self, p1, p2,iterate_time):
    #     # 不能修改之前的，需要生成一个新的对象
    #     p1_dna = p1.dna.copy()
    #     p2_dna = p2.dna.copy()
    #     cut_i = np.random.randint(1, self.dna_len - 1)
    #     temp = p1_dna[cut_i:].copy()
    #     p1_dna[cut_i:] = p2_dna[cut_i:]
    #     p2_dna[cut_i:] = temp

    #     cross_name=p1.pop_seed_name.split("_")
    #     p1_name="seed_"+iterate_time+"_1_"+cross_name[3]
    #     cross_name=p2.pop_seed_name.split("_")
    #     p2_name="seed_"+iterate_time+"_2_"+cross_name[3]
    #     q1 = individual(p1_dna,p1_name)
    #     q2 = individual(p2_dna,p2_name)
    #     return q1, q2

    def cross(self, p1, p2):
        cut_i = np.random.randint(1, self.dna_len - 1)
        temp = p1[cut_i:].copy()
        p1[cut_i:] = p2[cut_i:]
        p2[cut_i:] = temp
        return p1, p2

    # def mutation(self, domain):
    #     point = np.random.randint(0, self.dna_len)
    #     low = domain[point][0]
    #     high = domain[point][1]
    #     self.dna[point] = random.randint(low,high)
    #     return self.dna

    # 实数编码,SBX交叉
    def cross_newObj(self, domain, p1, p2, iterate_time, m, n):
        eta_c = 10
        pcross_real = 1.0  # ori : 0.9
        # 如果随机概率大于交叉概率则不进行交叉操作
        p1_dna = p1.dna.copy()
        p2_dna = p2.dna.copy()

        if random.random() <= pcross_real:
            # 对两个个体执行SBX交叉操作
            for j in range(self.dna_len):
                # 判断是否对某自变量交叉
                if random.random() > 0.5:
                    continue
                # 如果两个体某自变量相等则不操作
                if p1_dna[j] == p2_dna[j]:
                    continue
                # 对某自变量交叉
                y1 = min(p1_dna[j], p2_dna[j])
                y2 = max(p1_dna[j], p2_dna[j])
                ylow = domain[j][0]
                yup = domain[j][1]
                r = random.random()
                beta = 1.0 + (2.0 * (y1 - ylow) / (y2 - y1))
                alpha = 2.0 - beta ** (-(eta_c + 1.0))
                if r <= (1.0 / alpha):
                    betaq = (r * alpha) ** (1.0 / (eta_c + 1.0))
                else:
                    betaq = (1.0 / (2.0 - r * alpha)) ** (1.0 / (eta_c + 1.0))
                child1 = 0.5 * ((y1 + y2) - betaq * (y2 - y1))

                beta = 1.0 + (2.0 * (yup - y2) / (y2 - y1))
                alpha = 2.0 - beta ** (-(eta_c + 1.0))
                if r <= (1.0 / alpha):
                    betaq = (r * alpha) ** (1.0 / (eta_c + 1.0))
                else:
                    betaq = (1.0 / (2.0 - r * alpha)) ** (1.0 / (eta_c + 1.0))
                child2 = 0.5 * ((y1 + y2) - betaq * (y2 - y1))

                if str(child1) == 'nan':
                    child1 = 0
                if str(child2) == 'nan':
                    child2 = 0
                child1 = min(max(child1, ylow), yup)
                child2 = min(max(child2, ylow), yup)
                p1_dna[j] = child1
                p2_dna[j] = child2

        cross_name = p1.pop_seed_name.split("_")
        p1_name = "seed_" + iterate_time + "_" + str(m) + "_" + cross_name[3]
        cross_name = p2.pop_seed_name.split("_")
        p2_name = "seed_" + iterate_time + "_" + str(n) + "_" + cross_name[3]
        q1 = Individual(p1_dna, p1_name)
        q2 = Individual(p2_dna, p2_name)

        return q1, q2

    def mutation(self, domain, pm):
        eta_m = 5
        for j in range(self.dna_len):
            r = random.random()
            # 对个体某变量进行变异
            if r <= pm:
                y = self.dna[j]
                ylow = domain[j][0]
                yup = domain[j][1]
                delta1 = 1.0 * (y - ylow) / (yup - ylow)
                delta2 = 1.0 * (yup - y) / (yup - ylow)
                # delta=min(delta1, delta2)
                r = random.random()
                mut_pow = 1.0 / (eta_m + 1.0)
                if r <= 0.5:
                    xy = 1.0 - delta1
                    val = 2.0 * r + (1.0 - 2.0 * r) * (xy ** (eta_m + 1.0))
                    deltaq = val ** mut_pow - 1.0
                else:
                    xy = 1.0 - delta2
                    val = 2.0 * (1.0 - r) + 2.0 * (r - 0.5) * (xy ** (eta_m + 1.0))
                    deltaq = 1.0 - val ** mut_pow
                y = y + deltaq * (yup - ylow)
                y = min(yup, max(y, ylow))
                self.dna[j] = y
        print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", self.dna)
        print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", "77777777777777777777777777777")


class ga:
    def __init__(self, pri_queue_size, pop_size, dna_size, pc, pm, f_funcs, domain, domain_flags):
        self.pri_queue_size = pri_queue_size
        self.pop_size = pop_size
        self.dna_size = dna_size
        # self.f_size = len(f_funcs)
        self.pc = pc
        self.pm = pm
        self.f_funcs = f_funcs
        self.domain = domain
        self.domain_flags = domain_flags

    def ini_pop(self, if_sampling):
        rseed = 8.35
        random.seed(rseed)
        if not if_sampling:
            # 24x8 随机参数矩阵  初始种群24x可变参数8
            P_dna = np.random.random_integers(0, 0, size=(self.pop_size, self.dna_size))
            for i in range(self.pop_size):
                for j in range(len(self.domain)):
                    item = self.domain[j]
                    P_dna[i, j] = random.randint(item[0], item[1])
            print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", P_dna)
            P = []  # 初始种群P\
            Seed_ini = []
            for j in range(self.pop_size):
                seed = "seed_0_" + str(j + 1) + "_" + str(int(j / 4) + 1) + ".xosc"
                Seed_ini.append(seed)
            for m in range(self.pop_size):
                P.append(Individual(P_dna[m], Seed_ini[m]))
        else:
            # 72x8
            P_dna = np.random.random_integers(0, 0, size=(3 * self.pop_size, self.dna_size))
            # print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", P_dna)
            for i in range(3 * self.pop_size):
                for j in range(len(self.domain)):
                    item = self.domain[j]
                    P_dna[i, j] = random.randint(item[0], item[1])
            print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", P_dna)
            P = []  # 初始种群P\
            Seed_ini = []
            Sample_predict = []
            choose = [0, 0, 0, 0, 0, 0]
            choose_result = []
            for j in range(3 * self.pop_size):
                print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", "Sampling # ", j)
                # 前25个 前12：0_j+1_4  后12：0_j+1_5  0_25_6
                if int(j / 12) + 4 <= 6:
                    num = int(j / 12) + 4
                    seed = "seed_0_" + str(j + 1) + "_" + str(num) + ".xosc"
                # 26-72个 0_26_1 0_27_1 ... 0_49_2 0_50_2... 0_61_
                else:                                                                                               
                    num = int((j-36)/ 12 + 1)
                    seed = "seed_0_" + str(j + 1) + "_" + str(num) + ".xosc"
                print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", "Generate seed_name: ", seed)
                #seed = "seed_0_" + str(j + 1) + "_" + str(int(j / 12) + 1) + ".xosc"
                Seed_ini.append(seed)
                cal_number = sample_simulate(P_dna[j, 0], P_dna[j, 1], P_dna[j, 2], P_dna[j, 3], P_dna[j, 4],
                                             P_dna[j, 5], P_dna[j, 6], P_dna[j, 7], seed, self.domain_flags)
                Sample_predict.append(cal_number)
                print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", "*******************")
                print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", "VGG error num:"+str(cal_number))
                with open('r_list.csv', 'a+', encoding='utf-8') as l:  # save entropy
                    cw = csv.writer(l)
                    cw.writerow([seed, P_dna[j], cal_number])
                print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", Sample_predict)

            Sample_predict = np.array(Sample_predict)
            # 加负号后按从大到小排序
            result = np.argsort(-Sample_predict)
            # result为错误数排序后的下标索引数组
            print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", "Sample_predict result:", result)

            for m in range(3 * self.pop_size):  # select top 4 seed of every scene( 4 * 6 ) from 72 seeds
                # 12应该为pop_size * 3 / seed_num
                loc = int(result[m] / 12)
                print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", "loc:", loc)
                # 4应该为pop_size / seed_num
                if (choose[loc] < 4):
                    choose[loc] = choose[loc] + 1
                    choose_result.append(result[m])

            print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", "Sample choose_result:", choose_result)

            for m in range(self.pop_size):
                # print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", P_dna[m])
                # print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", Seed_ini[m])
                P.append(Individual(P_dna[choose_result[m]], Seed_ini[choose_result[m]]))

        return P

    # def calculate_pop_fitness(self, pop, total):
    #     # 计算Pop适应度，存储到individual中
    #     normalization = []
    #     #for p in pop:
    #     #    if p.f == None:
    #     #        p.f, p.error_number, p.list_er = p.get_fitness(self.f_funcs, total)
    #     for p in pop:
    #         p.f, p.error_number, p.list_er = p.get_fitness(self.f_funcs, total)
    #         print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", p.f)
    #     return pop

    def run_sim(self, R,
                data_collection_para):  # data_collection_para : [is_new_seed(for entropy), is_err_collection(collect err or not), err_type(collect normal(1)/sampling(2)/random data(3))]
        for p in R:
            print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", "Run sim index: # ", R.index(p))
            p.f, p.error_number, p.list_er = p.get_fitness(self.f_funcs, self.domain_flags, data_collection_para)
            print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", p.f, '\n')

    def select(self, R):
        R.sort()
        choose_number = [0, 0]
        # print("[" + os.path.basename(__file__) + ", Line " + str(sys._getframe().f_lineno) + ", " + sys._getframe().f_code.co_name + "] ", len(R))
        pop = []
        for i in range(2):
            randomnumber1 = random.randint(0, self.pop_size - 1)
            randomnumber2 = random.randint(0, self.pop_size - 1)
            randomnumber3 = random.randint(0, self.pop_size - 1)
            randomnumber4 = random.randint(0, self.pop_size - 1)
            if R[randomnumber1].f > R[randomnumber2].f:
                pop.append(R[randomnumber1])
                choose_number[0] = randomnumber1
            else:
                pop.append(R[randomnumber2])
                choose_number[0] = randomnumber2
            if R[randomnumber3].f > R[randomnumber4].f:
                pop.append(R[randomnumber3])
                choose_number[1] = randomnumber3
            else:
                pop.append(R[randomnumber4])
                choose_number[1] = randomnumber4

            R[choose_number[0]].choose_time = R[choose_number[0]].choose_time + 1
            R[choose_number[1]].choose_time = R[choose_number[1]].choose_time + 1
        return pop, R

    def pop_cross(self, P, iterate_time):
        new_Q = []
        i = 0
        P_len = len(P)
        while i < P_len:
            # q1, q2 = P[i].cross_newObj(P[i], P[i + 1],str(iterate_time))
            q1, q2 = P[i].cross_newObj(self.domain, P[i], P[i + 1], str(iterate_time), (i + 1), (i + 2))
            new_Q.append(q1)
            new_Q.append(q2)
            i += 2
        return new_Q

    def pop_mutation(self, Q):
        for q in Q:
            q.mutation(self.domain, self.pm)
        return Q

    def calulate_pop(self, R, pri_queue_size):
        R.sort()
        return R[:pri_queue_size]

    def calculate_pop_fitness(self, R):
        fit_funcs.cal_fitness(R)
        normalization = []
        for r in R:
            normalization.append(r.f[2])
        max_val = max(normalization)
        min_val = min(normalization)
        minus = max_val - min_val if max_val - min_val != 0 else 1
        for r in R:
            r.f[0] += r.f[1] + (r.f[2] - min_val) / minus