QGA.py

import random
import matplotlib.pyplot as plt
import numpy as np
import time
import json 
from diversity_metrics import shannon_from_population

N,M = 50,50
start, goal = (0,0), (49,49)
# obstacles = {(1,1), (1,2), (2,2), (3,1)}

obstacles = {
    (3, 21), (3, 33), (4, 4), (5, 43), (9, 31),
    (10, 13), (10, 41), (11, 2), (11, 25), (13, 33),
    (15, 25), (18, 25), (19, 14), (21, 30), (22, 3),
    (26, 7), (26, 36), (27, 47), (31, 14), (31, 28),
    (31, 39), (32, 17), (32, 33), (34, 24), (35, 24),
    (37, 2), (42, 24), (43, 48), (48, 16), (49, 32), (25, 25)
}

population_size = 70
L = 150 # kromozom uzunlugu, maksimum adım sayisi
GEN = 100 # jenerasyon sayisi
MOVES = ["U", "D", "L", "R"]
DIR = {'U': (-1, 0), 'D': (1, 0), 'L': (0, -1), 'R': (0, 1)}
K = 2 # her Q bireyinden K örnek (gürültüyü azaltir.)
FIT_C = 200 # fitness sabiti, fitness = FIT_C - cost
DELTA_DIST = 3.5 #ulasmazsa manhattan mesafesi
BETA = 5.0 #carpisan engel cezasi
GAMMA = 8.0 #taşma cezasi
ALPHA = 0.2 #adim cezasi
EARLY_GOAL_BONUS = 10 #hedefe erken ulasmanin bonusu


DELTA = 0.035 #olasilik kaydirma adimi (rotation-like)
EPS = 1e-8 # min taban değeri
MUT_RATE = 0.01
MUT_STRENGTH = 0.15
STAGNATION_PATIENCE = 20



def plot_path(moves, start, goal, obstacles, N, M):
    grid = np.zeros((N, M))
    for r, c in obstacles:
        grid[r, c] = -1  # engeller

    r, c = start
    coords = [(r, c)]
    reached = False
    for mv in moves:
        (r, c), _, _ = step((r, c), mv)
        coords.append((r, c))
        if (r, c) == goal:
            reached = True
            break

    ys, xs = zip(*coords)

    plt.figure(figsize=(6, 6))
    plt.imshow(grid, cmap="Greys", origin="upper", alpha=0.3)

    # Engeller
    if obstacles:
        obs_y, obs_x = zip(*obstacles)
        plt.scatter(obs_x, obs_y, marker='s', color='red', label='Obstacle', s=80, edgecolors='black', linewidths=0.5)

    # Yol
    plt.plot(xs, ys, color="blue", linewidth=2, label="Path")
    plt.scatter(xs, ys, color="blue", s=20)

    # Başlangıç
    plt.scatter([start[1]], [start[0]], color="green", marker="o", s=120, label="Start", edgecolors='black', zorder=5)
    plt.text(start[1], start[0], "Start", color="green", fontsize=10, fontweight="bold", ha="right", va="bottom")

    # Hedef
    plt.scatter([goal[1]], [goal[0]], color="gold", marker="*", s=200, label="Goal", edgecolors='black', zorder=5)
    plt.text(goal[1], goal[0], "Goal", color="goldenrod", fontsize=10, fontweight="bold", ha="left", va="top")

    plt.title("Quantum Inspired Genetic Algorithm Path", fontsize=14)
    plt.xlabel("Y coordinate")
    plt.ylabel("X coordinate")
    plt.xlim(-0.5, M-0.5)
    plt.ylim(-0.5, N-0.5)
    plt.gca().set_aspect('equal')
    plt.grid(True, which='both', color='gray', linewidth=0.3, linestyle='--', alpha=0.5)
    plt.legend(loc="upper left", fontsize=10)
    plt.tight_layout()
    # plt.show()

def normalize_with_floor(p, eps=EPS):
    """ min tabanla normalize et """
    q = [max(x, eps) for x in p]
    s = sum(q)
    return [x/s for x in q]

def mutate_q(q_ind, rate=MUT_RATE, strength=MUT_STRENGTH):
    """
    Mikro çeşitlilik: rate olasılıkla, bazı adımların dağılımını
    uniform'a doğru (0.25) strength oranında karıştır.
    """
    for i in range(L):
        if random.random() < rate:
            probs = q_ind[i]
            uniform = [0.25, 0.25, 0.25, 0.25]
            mixed = [(1 - strength)*p + strength*u for p, u in zip(probs, uniform)]
            q_ind[i] = normalize_with_floor(mixed, EPS)


#Quantumdaki rotation_gate'in kucuk bir benzeri 
def update_toward_best(q_ind, best_moves, delta=DELTA):
    """Her adim i için, en iyi çözümün seçtiği hareketin olasiligini +delta arttir,
    diğerlerini -delta/(n-1) azalt; sonra normalize et."""
    
    for i in range(L):
        winner = best_moves[i] if i < len(best_moves) else None
        if winner is None:
            continue
        if winner not in MOVES:
            continue

        idx = MOVES.index(winner)
        probs = q_ind[i][:]

        # artış/azalış
        ncat = len(probs)  #4
        for j in range(ncat):
            if j == idx:
                probs[j] += delta
            else:
                probs[j] -= delta / (ncat - 1)

        # min taban + normalize
        q_ind[i] = normalize_with_floor(probs, EPS)


def step (pos, move):
    r,c = pos
    dr, dc = DIR[move]
    nr, nc = r+dr, c+dc

    if 0 <= nr < N and 0 <= nc < M:
        if (nr, nc) in obstacles:
            return pos, 1, 0  # collision
        else:
            return (nr, nc), 0, 0
    else:
        return pos, 0, 1 # out-of-bounds denemesi

def manhattan(a, b):
    return abs(a[0]-b[0]) + abs(a[1]-b[1])

def simulate_path(moves):
    pos = start 
    used_steps = 0
    collisions = 0
    oob = 0
    reached = False
    path_length = 0

    for mv in moves:
        (nr, nc), col, out = step(pos, mv)
        if (nr, nc) != pos:
            path_length += 1
        collisions += col
        oob += out
        used_steps += 1
        pos = (nr, nc)
        if pos == goal:
            reached = True
            break

    return pos, used_steps, collisions, oob, reached, path_length


def cost_of(moves):
    pos, used, col, oob, reached, path_length = simulate_path(moves)
    if(reached): #hedefe ulaştıysa 
        return used + BETA*col + GAMMA*oob - EARLY_GOAL_BONUS

    return DELTA_DIST * manhattan(pos, goal) + BETA*col + GAMMA*oob + ALPHA*used


def fitness_of(moves):
    return FIT_C - cost_of(moves)


def sample_moves(q_ind): 
    """L*4 olasilik; her adim için bir hareket örneklensin"""
    seq = []
    for probs in q_ind:
        mv = random.choices(MOVES, weights=probs, k=1)[0]
        seq.append(mv)
    return seq


def init_q_individual():
    return [[0.25, 0.25, 0.25, 0.25] for _ in range(L)]  # ilk basta esit olasilikla baslattik, 



def run_and_report(moves, t1):
    pos, used, col, oob, reached, path_length = simulate_path(moves)
    c = cost_of(moves)
    # print("\n=== RESULT ===")
    # print("Reached goal?:", reached)
    # print("Steps used   :", used)
    # print("Collisions   :", col)
    # print("OOB tries    :", oob)
    # print("Cost         :", c)
    # print("Moves (prefix used):", ''.join(moves[:used]))


    result = {
        "algorithm": "QIGA",
        "gens": GEN,
        "population": population_size,
        "chromozome_lenght": L,
        "reached": reached,
        "steps_used": used,
        "collision": col,
        "oob": oob,
        "cost": c,
        "time_sec": t1,
        "path_length": path_length
    }

    print("RESULTS: ", result)

    with open("QIGA_result.json", "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False, indent=4)




if __name__ == "__main__":
    
    t0 = time.time()

    random.seed(42)

    pop = [init_q_individual() for _ in range(population_size)]

    global_best_fit = -1e9 #?
    global_best_moves = None
    no_improve = 0


    for g in range(GEN):
        
        # H, Ne = shannon_from_population(moves)
        # print("Shannon index: ", H , " Etkin Genotip: ", Ne)


        observed = []
        bucket = [[] for _ in range(population_size)]

        for qi, q in enumerate(pop):
            for _ in range(K):
                moves = sample_moves(q)
                # print("Ind: ", q1, "  moves: ", moves)
                
                f = fitness_of(moves)
                # print("Ind: ", qi, "  fitness: ", f)

                observed.append((f, moves, qi))

                bucket[qi].append((f, moves))

        # best_moves_by_q = [max(bucket[qi], key=lambda x: x[0])[1] for qi in range(population_size)]
        # H, Ne = shannon_from_population(best_moves_by_q)
        # print(f"Shannon index: {H:.6f}  Etkin Genotip: {Ne:.6f}")


        observed.sort(key=lambda x: x[0], reverse=True)
        gen_best_fit, gen_best_moves, _ = observed[0] #jenerasyonun en iyisi 

        if(gen_best_fit > global_best_fit):
            global_best_fit = gen_best_fit
            global_best_moves = gen_best_moves[:]
            no_improve = 0
        else:
            no_improve += 1


        print("global_best_fit: ", global_best_fit) 


        #olasilik guncellemesi yapilir, tum Q bireylerini global en iyiye yaklastir
        for qi, q in enumerate(pop):
            update_toward_best(q, global_best_moves, delta=DELTA)


        for q in pop:
            mutate_q(q, rate=MUT_RATE, strength=MUT_STRENGTH)


        if no_improve >= STAGNATION_PATIENCE:
            for q in pop:
                mutate_q(q, rate=0.4, strength=0.7)
            no_improve = 0

        # İzleme
        best_cost = FIT_C - global_best_fit
        print(f"Gen {g:03d}  best_cost={best_cost:.3f}")

        # if gen_best_fit > global_best_fit and simulate_path(gen_best_moves)[4]:
        #     # reached=True ve adım sayısı belirli bir eşikten küçükse kes
        #     if simulate_path(gen_best_moves)[1] <= 10:
        #         break


    t1 = time.time() - t0
    run_and_report(global_best_moves, t1)
    plot_path(global_best_moves, start, goal, obstacles, N, M)
    plt.savefig("quantum_inspired_genetic_algorithm.png")