main.py

from typing import Any
from typing_extensions import override

import volpe_py as volpe
import grpc
import concurrent.futures
import threading
import array
import random

from opfunu.cec_based.cec2022 import *
from deap import base, creator, tools, algorithms

NDIM=20

func = F122022(ndim=NDIM)

LOW = func.lb[0]
HIGH = func.ub[0]
MIN_STRATEGY = 0.5
MAX_STRATEGY = 3.0
MUTATION_RATE = 0.3
CXPROB = 0.6

BASE_POPULATION_SIZE = 100
LAMBDA_SIZE = 7*BASE_POPULATION_SIZE

# Setup DEAP with ES
creator.create("FitnessMin", base.Fitness, weights=(-1.0,))
creator.create("Individual", array.array, typecode="d", fitness=creator.FitnessMin, strategy=None)
creator.create("Strategy", array.array, typecode="d")

def generateES(icls, scls, size, imin, imax, smin, smax):
    ind = icls(random.uniform(imin, imax) for _ in range(size))
    ind.strategy = scls(random.uniform(smin, smax) for _ in range(size))
    return ind

def checkStrategy(minstrategy):
    def decorator(func):
        def wrapper(*args, **kargs):
            children = func(*args, **kargs)
            for child in children:
                for i, s in enumerate(child.strategy):
                    if s < minstrategy:
                        child.strategy[i] = minstrategy
            return children
        return wrapper
    return decorator

toolbox = base.Toolbox()
toolbox.register("individual", generateES, creator.Individual, creator.Strategy, NDIM, LOW, HIGH, MIN_STRATEGY, MAX_STRATEGY)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)

def fitness(x):
    if any(v > HIGH for v in x) or any(v < LOW for v in x):
        return (float('inf'),)
    return (float(func.evaluate(list(x))),)

toolbox.register("evaluate", fitness)
toolbox.register("mate", tools.cxESBlend, alpha=0.1)
toolbox.register("mutate", tools.mutESLogNormal, c=1.0, indpb=0.03)
toolbox.register("select", tools.selTournament, tournsize=3)

toolbox.decorate("mate", checkStrategy(MIN_STRATEGY))
toolbox.decorate("mutate", checkStrategy(MIN_STRATEGY))

def gen_ind():
    ind = toolbox.individual()
    ind.fitness.values = fitness(ind)
    return ind

def popListTostring(popln: list):
    indList : list[volpe.ResultIndividual] = []
    for mem in popln:
        indList.append(
                volpe.ResultIndividual(representation=str(list(mem)), 
                                    fitness=mem.fitness.values[0])
                )
    return volpe.ResultPopulation(members=indList)

def bstringToPopln(popln: volpe.Population):
    popList = []
    for memb in popln.members:
        # Split the genotype bytes: first NDIM doubles for individual, next NDIM for strategy
        bytes_per_double = 8
        total_bytes = NDIM * bytes_per_double * 2
        
        ind_bytes = memb.genotype[:NDIM * bytes_per_double]
        strategy_bytes = memb.genotype[NDIM * bytes_per_double:]
        
        ind = creator.Individual(array.array('d', ind_bytes))
        ind.strategy = creator.Strategy(array.array('d', strategy_bytes) if len(strategy_bytes) > 0 else [random.uniform(MIN_STRATEGY, MAX_STRATEGY) for _ in range(NDIM)])
        ind.fitness.values = (memb.fitness,)
        popList.append(ind)
    return popList

def popListToBytes(popln: list):
    indList : list[volpe.Individual] = []
    for mem in popln:
        # Combine genotype and strategy into single byte representation
        combined = bytes(mem) + bytes(mem.strategy)
        indList.append(volpe.Individual(genotype=combined, fitness=mem.fitness.values[0]))
    return volpe.Population(members=indList, problemID="p1")

class VolpeGreeterServicer(volpe.VolpeContainerServicer):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.popln = [ gen_ind() for _ in range(BASE_POPULATION_SIZE)  ]
        self.poplock = threading.Lock()

    @override
    def SayHello(self, request: volpe.HelloRequest, context: grpc.ServicerContext):
        return volpe.HelloReply(message="hello " + request.name)
    @override
    def InitFromSeed(self, request: volpe.Seed, context: grpc.ServicerContext):
        """Missing associated documentation comment in .proto file."""
        with self.poplock:
            self.popln = [ gen_ind() for _ in range(BASE_POPULATION_SIZE)  ]
            return volpe.Reply(success=True)
    @override
    def InitFromSeedPopulation(self, request: volpe.Population, context: grpc.ServicerContext):
        """Missing associated documentation comment in .proto file."""
        with self.poplock:
            ogLen = len(self.popln)
            seedPop = bstringToPopln(request)
            if self.popln == None:
                self.popln = []
            self.popln.extend(seedPop)

            self.popln = toolbox.select(self.popln, ogLen)

            return volpe.Reply(success=True)
    @override
    def GetBestPopulation(self, request: volpe.PopulationSize, context):
        """Missing associated documentation comment in .proto file."""
        with self.poplock:
            if self.popln is None:
                return volpe.Population(members=[], problemID="p1")
            popSorted = sorted(self.popln, key=lambda x: x.fitness.values[0])
            return popListToBytes(popSorted[:request.size])
    @override
    def GetResults(self, request: volpe.PopulationSize, context):
        with self.poplock:
            if self.popln is None:
                return volpe.Population(members=[], problemID="p1")
            popSorted = sorted(self.popln, key=lambda x: x.fitness.values[0])
            return popListTostring(popSorted[:request.size])
    @override
    def GetRandom(self, request: volpe.PopulationSize, context):
        with self.poplock:
            if self.popln is None:
                return volpe.Population(members=[], problemID="p1")
            popList = [self.popln[random.randint(0, len(self.popln) - 1)] for _ in range(request.size)]
            return popListToBytes(popList)
    @override
    def AdjustPopulationSize(self, request: volpe.PopulationSize, context: grpc.ServicerContext):
        """Missing associated documentation comment in .proto file."""
        pass
    @override
    def RunForGenerations(self, request: volpe.PopulationSize, context):
        """Missing associated documentation comment in .proto file."""
        with self.poplock:
            # Use DEAP's eaMuPlusLambda algorithm
            self.popln, _ = algorithms.eaMuPlusLambda(
                self.popln, 
                toolbox, 
                mu=len(self.popln),
                lambda_=LAMBDA_SIZE,
                cxpb=CXPROB,
                mutpb=MUTATION_RATE,
                ngen=request.size,
                stats=None,
                halloffame=None,
                verbose=False
            )
                
        return volpe.Reply(success=True)

if __name__=='__main__':
    server = grpc.server(concurrent.futures.ThreadPoolExecutor(max_workers=10))
    volpe.add_VolpeContainerServicer_to_server(VolpeGreeterServicer(), server)
    server.add_insecure_port("0.0.0.0:8081")
    server.start()
    server.wait_for_termination()