import time
from HyperGP.libs.primitive_set import PrimitiveSet
from HyperGP.mods.tree2graph import ExecutableGen
from HyperGP.operators.execution.tree_exec.compiler_v2 import compile_v2
from HyperGP.operators.execution.tree_exec.compiler_v1 import compile_v1
import numpy as np
from HyperGP.mods.cash_manager import CashManager
from HyperGP import Tensor
import HyperGP
class ExecMethod:
def __call__(self, *args, **kwargs):
raise NotImplementedError("No find '__call__' implement")
class Executor(ExecMethod):
def __init__(self):
self.length_clt = 0
pass
"""
input: (n_terms, dataset_size)
"""
def __call__(self, progs, input:np.array, pset: PrimitiveSet, cashset:CashManager=None):
# print("prog mean len: ", np.mean([len(prog.list()) for prog in progs]))
'''pre-conversion'''
# import time
# st = time.time()
exec_list, states = ExecutableGen()(progs, pset, cashset)
# print('t1 ', time.time() - st)
# st = time.time()
f_avec = [pset.genFunc(f_str) for f_str in pset.primitiveSet]
compile_1 = [True]
def ret_false():
compile_1[0] = False
return -2
func_list = [None] * (pset.func_count - len(f_avec)) + [f.func.idx if hasattr(f, "func") and hasattr(f.func, "idx") else ret_false() for f in f_avec] + [-1]
'''compile and run'''
if compile_1[0] and len(input.shape) == 2 and (isinstance(input, Tensor) or isinstance(input, np.ndarray) or isinstance(input, list)):
expr = compile_v1(exec_list, pset, states, func_list)
output, records = expr(input)
# print('...compile_v1...')
# expr2 = compile_v2(exec_list, pset, states)
# output2, _ = expr2(input)
# if not (output == output2).numpy().all():
# print(output.numpy()[(output != output2).numpy()])
# print(output2.numpy()[(output != output2).numpy()])
# assert 0==1
# # print(output)
# # print(output2)
# # print(input)
# print(output[0].numpy()[output[0].numpy() != output2[0].numpy()])
# print(output2[0].numpy()[output[0].numpy() != output2[0].numpy()])
# assert (output[0].numpy() == output2[0].numpy()).all()
else:
expr = compile_v2(exec_list, pset, states)
output, records = expr(input)
# print('...compile_v2...')
# print('t2 ', time.time() - st)
return output, records
[docs]
def executor(progs, input, pset: PrimitiveSet, cashset:CashManager=None):
"""
Execute the GP programs
Args:
progs(list): The GP programs need to be executed.
input(``HyperGP.Tensor`` or ``numpy.ndarray`` or ``list``): shape = (feature size, ...)
pset(``HyperGP.PrimitiveSet``): used for the progs
Returns:
outputs: The outputs generated by the progs with the given inputs.
records: The outputs of the record nodes.
Examples:
>>> from HyperGP import executor, PrimitiveSet, Population
>>> import numpy as np
>>> # Generae the primitive set
>>> pset = PrimitiveSet(
input_arity=1,
primitive_set=[
('add', HyperGP.add, 2),
('sub', HyperGP.sub, 2),
('mul', HyperGP.mul, 2),
('div', HyperGP.div, 2),
])
>>> # Generate the input data
>>> input_array = Tensor(np.random.uniform(0, 10, size=(1, 10000)))
>>> # Initialize the population
>>> pop = Population()
>>> pstates = ProgBuildStates(pset=pset, depth_rg=[2, 6], len_limit=100)
>>> pop.initPop(pop_size=pop_size, prog_paras=pstates)
>>> # Execute the individuals
>>> output, _ = executor(pop.states['progs'].indivs, input_array, pset)
"""
return Executor()(progs, input, pset, cashset)
"""TEST"""
if "__name__" == "__main__":
pass