class Individual(Saveable, Copyable):
    """An individual in an evolutionary population.

    Attributes
    ----------
    genome : Genome
        The Genome of the Individual.
    error_vector : np.array
        An array of error values produced by evaluating the Individual's program.
    total_error : float
        The sum of all error values in the Individual's error_vector.
    error_vector_bytes:
        Hashable Byte representation of the Individual's error vector.

    """

    __slots__ = [
        "id", "genome", "signature",
        "_program", "_error_vector", "_total_error", "_error_vector_bytes"
    ]

    def __init__(self, genome: Genome, signature: ProgramSignature):
        self.id = uuid4()
        self.genome = genome
        self.signature = signature
        self._program = None
        self._error_vector = None
        self._total_error = None
        self._error_vector_bytes = None

    @property
    def program(self) -> Program:
        """Push program associated with this Individual. Taken from Plush genome."""
        if self._program is None:
            cb = genome_to_code(self.genome)
            self._program = Program(code=cb, signature=self.signature)
        return self._program

    @property
    def error_vector(self) -> np.ndarray:
        """Numpy array of numeric error values."""
        return self._error_vector

    @error_vector.setter
    def error_vector(self, error_vector: np.ndarray):
        self._error_vector = error_vector

    @property
    def total_error(self) -> Union[np.int64, np.float64]:
        """Numeric sum of the error vector."""
        if self._total_error is None:
            try:
                self._total_error = np.sum(self.error_vector)
            except OverflowError:
                self._total_error = np.inf
        return self._total_error

    @total_error.setter
    def total_error(self, value: Union[np.int64, np.float64]):
        raise AttributeError("Cannot set total_error directly. Must set error_vector.")

    @property
    def error_vector_bytes(self):
        """Hashable Byte representation of the Individual's error vector."""
        if self._error_vector_bytes is None:
            self._error_vector_bytes = self._error_vector.data.tobytes()
        return self._error_vector_bytes

    @error_vector_bytes.setter
    def error_vector_bytes(self, value):
        raise AttributeError("Cannot set error_vector_bytes directly. Must set error_vector.")

    def __lt__(self, other):
        return self.total_error < other.total_error

    def __eq__(self, other):
        return isinstance(other, Individual) and self.genome == other.genome class DatasetEvaluator(Evaluator):
    """Evaluator driven by a labeled dataset."""

    def __init__(self,
                 X, y,
                 interpreter: PushInterpreter = "default",
                 penalty: float = 1e6):
        """Create Evaluator based on a labeled dataset. Inspired by sklearn.

        Parameters
        ----------
        X : list, array-like, or pandas dataframe of shape = [n_samples, n_features]
            The inputs to evaluate each program on.

        y : list, array-like, or pandas dataframe.
            The target values. Shape = [n_samples] or [n_samples, n_outputs]

        interpreter : PushInterpreter or {"default"}
            The interpreter used to run the push programs.

        penalty : float
            If no response is given by the program on a given input, assign this
            error as the error.

        """
        super().__init__(interpreter, penalty)
        self.X = pd.DataFrame(X)
        self.y = pd.DataFrame(y)

    @tap
    def evaluate(self, program: Program) -> np.array:
        """Evaluate the program and return the error vector.

        Parameters
        ----------
        program
            Program (CodeBlock of Push code) to evaluate.

        Returns
        -------
        np.ndarray
            The error vector of the program.

        """
        super().evaluate(program)
        errors = []
        for ndx in range(self.X.shape[0]):
            inputs = self.X.iloc[ndx].to_list()
            expected = self.y.iloc[ndx].to_list()
            actual = self.interpreter.run(program, inputs)
            errors.append(self.default_error_function(actual, expected))
        return np.array(errors).flatten() class GeneSpawner:
    """A factory of random Genes (Atoms) and Genomes.

    When  spawning a random gene, the result can be one of three types of Atoms.
    An Instruction, a Closer, or a Literal. If the Atom is a Literal, it may
    be one of the supplied Literals, or it may be the result of running one of
    the Ephemeral Random Constant generators.

    Reference for ERCs:
    "A field guide to genetic programming", Section 3.1
    Riccardo Poli and William B. Langdon and Nicholas Freitag McPhee,
    http://www.gp-field-guide.org.uk/

    Attributes
    ----------
    n_input : int
        Number of input instructions that could appear the genomes.
    instruction_set : pyshgp.push.instruction_set.InstructionSet
        InstructionSet containing instructions to use when spawning genes and
        genomes.
    literals : Sequence[pyshgp.push.instruction_set.atoms.Literal]
        A list of Literal objects to pull from when spawning genes and genomes.
    erc_generator : Sequence[Callable]
        A list of functions (aka Ephemeral Random Constant generators). When one
        of these functions is called, the output is placed in a Literal and
        returned as the spawned gene.
    distribution : pyshgp.utils.DiscreteProbDistrib
        A probability distribution describing how frequently to produce
        Instructions, Closers, Literals, and ERCs.

    """

    def __init__(self,
                 n_inputs: int,
                 instruction_set: Union[InstructionSet, str],
                 literals: Sequence[Any],
                 erc_generators: Sequence[Callable],
                 distribution: DiscreteProbDistrib = "proportional"):
        self.n_inputs = n_inputs
        self.erc_generators = erc_generators

        self.instruction_set = instruction_set
        if self.instruction_set == "core":
            self.instruction_set = InstructionSet(register_core=True)
        self.type_library = self.instruction_set.type_library
        self.literals = [lit if isinstance(lit, Literal) else infer_literal(lit, self.type_library) for lit in literals]

        if distribution == "proportional":
            self.distribution = (
                DiscreteProbDistrib()
                .add(GeneTypes.INPUT, self.n_inputs)
                .add(GeneTypes.INSTRUCTION, len(self.instruction_set))
                .add(GeneTypes.CLOSE, sum([i.code_blocks for i in self.instruction_set.values()]))
                .add(GeneTypes.LITERAL, len(literals))
                .add(GeneTypes.ERC, len(erc_generators))
            )
        else:
            self.distribution = distribution

    def random_input(self) -> Input:
        """Return a random ``Input``.

        Returns
        -------
        pyshgp.push.atoms.Input

        """
        return Input(input_index=np.random.randint(self.n_inputs))

    def random_instruction(self) -> InstructionMeta:
        """Return a random Instruction from the InstructionSet.

        Returns
        -------
        pyshgp.push.atoms.InstructionMeta
            A randomly selected Literal.

        """
        i = np.random.choice(list(self.instruction_set.values()))
        return InstructionMeta(name=i.name, code_blocks=i.code_blocks)

    def random_literal(self) -> Literal:
        """Return a random Literal from the set of Literals.

        Returns
        -------
        pyshgp.push.atoms.Literal
            A randomly selected Literal.

        """
        lit = np.random.choice(self.literals)
        if not isinstance(lit, Literal):
            lit = infer_literal(lit, self.type_library)
        return lit

    def random_erc(self) -> Literal:
        """Materialize a random ERC generator into a Literal and return it.

        Returns
        -------
        pyshgp.push.atoms.Literal
            A Literal whose value comes from running a ERC generator function.

        """
        erc_value = np.random.choice(self.erc_generators)()
        if not isinstance(erc_value, Literal):
            erc_value = infer_literal(erc_value, self.type_library)
        return erc_value

    def random_gene(self) -> Atom:
        """Return a random Atom based on the GenomeSpawner's distribution.

        Returns
        -------
        pyshgp.push.atoms.Atom
            An random Atom. Either an Instruction, Closer, or Literal.

        """
        atom_type = self.distribution.sample()
        if atom_type is GeneTypes.INPUT:
            return self.random_input()
        elif atom_type is GeneTypes.INSTRUCTION:
            return self.random_instruction()
        elif atom_type is GeneTypes.CLOSE:
            return Closer()
        elif atom_type is GeneTypes.LITERAL:
            return self.random_literal()
        elif atom_type is GeneTypes.ERC:
            return self.random_erc()
        else:
            raise ValueError("GenomeSpawner distribution bad atom type {t}".format(t=str(atom_type)))

    def spawn_genome(self, size: Union[int, Sequence[int]]) -> Genome:
        """Return a random Genome based on the GenomeSpawner's distribution.

        The genome will contain the specified number of Atoms if size is an
        integer. If size is a pair of integers, the genome will be of a random
        size in the range of the two integers.

        Parameters
        ----------
        size
            The resulting genome will contain this many Atoms if size is an
            integer. If size is a pair of integers, the genome will be of a random
            size in the range of the two integers.

        Returns
        -------
        pyshgp.gp.genome.Genome
            A Genome with random contents of a given size.

        """
        if isinstance(size, Sequence):
            size = np.random.randint(size[0], size[1]) + 1
        return Genome([self.random_gene() for _ in range(size)]) class PushInterpreter:
    """An interpreter capable of running Push programs.

    Parameters
    ----------
    instruction_set : Union[InstructionSet, str], optional
        The ``InstructionSet`` to use for executing programs. Default is "core"
        which instantiates an ``InstructionSet`` using all the core instructions.

    Attributes
    ----------
    instruction_set : InstructionSet
        The ``InstructionSet`` to use for executing programs.
    state : PushState
        The current ``PushState``. Contains one stack for each ``PushType``
        mentioned by the instructions in the instruction set.
    status : PushInterpreterStatus
        A string denoting if the interpreter has encountered a situation
        where non-standard termination was required.

    """

    def __init__(self,
                 instruction_set: Union[InstructionSet, str] = "core",
                 reset_on_run: bool = True):
        self.reset_on_run = reset_on_run
        # If no instruction set given, create one and register all instructions.
        if instruction_set == "core":
            self.instruction_set = InstructionSet(register_core=True)
        else:
            self.instruction_set = instruction_set

        self.type_library = self.instruction_set.type_library

        # Initialize the PushState and status
        self.state: PushState = None
        self.status: PushInterpreterStatus = None
        self._validate()

    def _validate(self):
        library_type_names = set(self.type_library.keys())
        required_stacks = self.instruction_set.required_stacks() - {"stdout", "exec", "untyped"}
        if not required_stacks <= library_type_names:
            raise ValueError(
                "PushInterpreter instruction_set and type_library are incompatible. {iset} vs {tlib}. Diff: {d}".format(
                    iset=required_stacks,
                    tlib=library_type_names,
                    d=required_stacks - library_type_names,
                ))

    def _evaluate_instruction(self, instruction: Instruction, config: PushConfig):
        self.state = instruction.evaluate(self.state, config)

    def untyped_to_typed(self):
        """Infer ``PushType`` of items on state's untyped queue and push to corresponding stacks."""
        while len(self.state.untyped) > 0:
            el = self.state.untyped.popleft()
            push_type = self.type_library.push_type_of(el, error_on_not_found=True)
            self.state[push_type.name].push(el)

    @tap
    def evaluate_atom(self, atom: Atom, config: PushConfig):
        """Evaluate an ``Atom``.

        Parameters
        ----------
        atom : Atom
            The Atom (``Literal``, ``InstructionMeta``, ``Input``, or ``CodeBlock``) to
            evaluate against the current ``PushState``.
        config : PushConfig
            The configuration of the Push program being run.

        """
        try:
            if isinstance(atom, InstructionMeta):
                self._evaluate_instruction(self.instruction_set[atom.name], config)
            elif isinstance(atom, Input):
                input_value = self.state.inputs[atom.input_index]
                self.state.untyped.append(input_value)
            elif isinstance(atom, CodeBlock):
                for a in atom[::-1]:
                    self.state["exec"].push(a)
            elif isinstance(atom, Literal):
                self.state[atom.push_type.name].push(atom.value)
            elif isinstance(atom, Closer):
                raise PushError("Closers should not be in push programs. Only genomes.")
            else:
                raise PushError("Cannot evaluate {t}, require a subclass of Atom".format(t=type(atom)))
            self.untyped_to_typed()
        except Exception as e:
            err_type = type(e)
            err_msg = str(e)
            traceback.print_exc()
            raise PushError(
                "{t} raised while evaluating {atom}. Original message: \"{m}\"".format(
                    t=err_type.__name__,
                    atom=atom,
                    m=err_msg
                ))

    @tap
    def run(self,
            program: Program,
            inputs: list,
            print_trace: bool = False) -> list:
        """Run a Push ``Program`` given some inputs and desired output ``PushTypes``.

        The general flow of this method is:
            1. Create a new push state
            2. Load the program and inputs.
            3. If the exec stack is empty, return the outputs.
            4. Else, pop the exec stack and process the atom.
            5. Return to step 3.

        Parameters
        ----------
        program : Program
            Program to run.
        inputs : list
            A sequence of values to use as inputs to the push program.
        print_trace : bool
            If True, each step of program execution will be summarized in stdout.

        Returns
        -------
        Sequence
            A sequence of values pulled from the final push state. May contain
            pyshgp.utils.Token.no_stack_item if output stacks are empty.

        """
        push_config = program.signature.push_config

        if self.reset_on_run or self.state is None:
            self.state = PushState(self.type_library, push_config)
            self.status = PushInterpreterStatus.normal

        # Setup
        self.state.load_code(program.code)
        self.state.load_inputs(inputs)
        stop_time = time.time() + push_config.runtime_limit
        steps = 0

        if print_trace:
            print("Initial State:")
            self.state.pretty_print()

        # Iterate atom evaluation until entire program is evaluated.
        while len(self.state["exec"]) > 0:
            # Stopping conditions
            if steps > push_config.step_limit:
                self.status = PushInterpreterStatus.step_limit_exceeded
                break
            if time.time() > stop_time:
                self.status = PushInterpreterStatus.runtime_limit_exceeded
                break

            # Next atom in the program to evaluate.
            next_atom = self.state["exec"].pop()

            if print_trace:
                start = time.time()
                print("\nCurrent Atom: " + str(next_atom))

            # Evaluate atom.
            old_size = self.state.size()
            self.evaluate_atom(next_atom, push_config)
            if self.state.size() > old_size + push_config.growth_cap:
                self.status = PushInterpreterStatus.growth_cap_exceeded
                break

            if print_trace:
                duration = time.time() - start
                print("Current State (step {step}):".format(step=steps))
                self.state.pretty_print()
                print("Step duration:", duration)
            steps += 1

        if print_trace:
            print("Finished program evaluation.")

        return self.state.observe_stacks(program.signature.output_stacks) class PushConfig(PRecord):
    """A configuration for a Push program.

    Attributes
    ----------
    step_limit : int
        Max number of atoms to process before terminating the execution of a
        given program. Default is 500
    runtime_limit : int
        Max number of seconds to run a push program before forcing
        termination. Default is 10.
    growth_cap : int
        Max number of elements that can be added to a PushState at any given
        step of program execution. If exceeded, program terminates. Default is 500.
    collection_size_cap : int, optional
        Max size of any collection (code blocks, vectors, strings, etc). Default is 1000.
    numeric_magnitude_limit : float, optional
        Max magnitude of numbers. Default is 1 million.

    """

    step_limit = field(type=int, initial=500, mandatory=True)
    runtime_limit = field(type=int, initial=10, mandatory=True)
    growth_cap = field(type=int, initial=500, mandatory=True)
    collection_size_cap = field(type=int, initial=1000, mandatory=True)
    numeric_magnitude_limit = field(type=float, initial=1e12, mandatory=True) class ProgramSignature(PRecord):
    """A specification of a Push program.

    Attributes
    ----------
    arity : int
        The number of inputs that the program will take.
    output_stacks : List[str]
        The names of the stack(s) which the output values will be pulled from after program execution.
    push_config : PushConfig
        The configuration of the PushInterpreter to use when executing the program.

    """

    arity = field(type=int, mandatory=True)
    output_stacks = field(type=list, mandatory=True)
    push_config = field(type=PushConfig, mandatory=True, initial=PushConfig()) def tap(fn):
    """Decorate a function/method to call any associated taps that have been registered in the ``TapManager``.

    Functional behavior is not changed.

    """
    fn_id = inspect.getmodule(fn).__name__ + "." + fn.__qualname__

    @wraps(fn)
    def tapped(*args, **kwargs):
        tap = TapManager.get(fn_id)
        if tap is not None:
            tap.pre(fn_id, args, kwargs)
        result = fn(*args, **kwargs)
        if tap is not None:
            tap.post(fn_id, args, kwargs, result)
        return result

    return tapped def set_verbosity(level: int):
    """Register some ``Tap`` objects in the ``TapManger`` that print to stdout during an evolutionary run.

    Verbosity level 0 prints nothing to stdout.

    Verbosity level 1 prints minimal run configuration as well as population statistics every 5 generations.

    Verbosity level 2+ prints full run configuration, population statics every generation, and traces the genome
    simplification process.

    """
    if level > 0:
        TapManager.register("pyshgp.gp.genome.GenomeSimplifier.simplify", StdOutSimplification())

    if level == 1:
        TapManager.register("pyshgp.gp.search.SearchAlgorithm.step", StdOutSearchStepTap(every_n_steps=5))
        TapManager.register("pyshgp.gp.search.SearchAlgorithm.run", StdOutRun(pre_print_config=True,
                                                                              pre_print_atoms=False,
                                                                              post_print_best=False))
    if level > 1:
        TapManager.register("pyshgp.gp.search.SearchAlgorithm.step", StdOutSearchStepTap(every_n_steps=1))
        TapManager.register("pyshgp.gp.search.SearchAlgorithm.run", StdOutRun(pre_print_config=True,
                                                                              pre_print_atoms=True,
                                                                              post_print_best=True))
        TapManager.register("pyshgp.gp.genome.GenomeSimplifier._step", StdOutSimplificationStep()) def list_rindex(lst, el):
    """Index of the last occurrence of an item in a list. Return None is item not found."""
    for i in reversed(range(len(lst))):
        if lst[i] == el:
            return i
    return None def check_is_fitted(estimator, attribute):
    """Check the given estimator has already been fit.

    Inspired by the scikit-learn function with the same name.
    https://scikit-learn.org/stable/modules/generated/sklearn.utils.validation.check_is_fitted.html#sklearn.utils.validation.check_is_fitted
    """
    if not hasattr(estimator, 'fit'):
        raise TypeError("{e} is not an estimator.".format(e=estimator))

    if not hasattr(estimator, attribute):
        raise ValueError(
            "Estimator has not been fit yet. Call 'fit' with appropriate arguments before using this method."
        ) def check_X_y(X, y) -> Tuple[Sequence, Sequence, int, Sequence]:
    """Check the given X and y datasets are prepared to be passed to a PushEstimator.

    Inspired by the scikit-learn function with the same name.
    https://scikit-learn.org/stable/modules/generated/sklearn.utils.check_X_y.html#sklearn.utils.check_X_y

    """
    X = check_2d(X)
    y = check_2d(y)
    X_column_count = check_num_columns(X)
    y_dim_types = check_column_types(y)
    return X, y, X_column_count, y_dim_types
from typing import Union, Tuple
from pyshgp.gp.individual import Individual
from pyshgp.gp.evaluation import DatasetEvaluator
from pyshgp.gp.genome import GeneSpawner
from pyshgp.push.interpreter import PushInterpreter
from pyshgp.push.config import PushConfig
from pyshgp.push.program import ProgramSignature
from pyshgp.tap import tap, set_verbosity
from pyshgp.utils import list_rindex
from pyshgp.validation import check_is_fitted, check_X_y
import pyshgp.gp.search as sr
import pyshgp.gp.selection as sl
import pyshgp.gp.variation as vr
                 simplification_steps: int = 2000,
                 last_str_from_stdout: bool = False,
                 interpreter: PushInterpreter = "default",
                 parallelism: Union[int, bool] = False,
                 push_config: PushConfig = "default",
                 verbose: int = 0,
                 **kwargs):
        self._search_name = search
        self.spawner = spawner
        self.selector = selector
        self.variation_strategy = variation_strategy
        self.population_size = population_size
        self.max_generations = max_generations
        self.initial_genome_size = initial_genome_size
        self.simplification_steps = simplification_steps
        self.last_str_from_stdout = last_str_from_stdout
        self.parallelism = parallelism
        self.verbose = verbose
        self.ext = kwargs
        set_verbosity(self.verbose)

        # Initialize attributes that will be set later.
        self.evaluator = None
        self.signature = None
        self.search = None
        self.solution = None

        if interpreter == "default":
            self.interpreter = PushInterpreter()
        else:
            self.interpreter = interpreter

        if push_config == "default":
            self.push_config = PushConfig()
        else:
            self.push_config = push_config

    def _build_search_algo(self):
        if isinstance(self.variation_strategy, dict):
            var_strat = vr.VariationStrategy()
            for op_name, prob in self.variation_strategy.items():
                var_op = vr.get_variation_operator(op_name)
                var_strat.add(var_op, prob)
            self.variation_strategy = var_strat

        search_config = sr.SearchConfiguration(
            signature=self.signature,
            spawner=self.spawner,
            evaluator=self.evaluator,
            selection=self.selector,
            variation=self.variation_strategy,
            population_size=self.population_size,
            max_generations=self.max_generations,
            initial_genome_size=self.initial_genome_size,
            simplification_steps=self.simplification_steps,
            parallelism=self.parallelism,
            push_config=self.push_config
        )
        self.search = sr.get_search_algo(self._search_name, config=search_config, **self.ext)

    @tap
    def fit(self, X, y):
        """Run the search algorithm to synthesize a push program.

        Parameters
        ----------
        X : pandas dataframe of shape = [n_samples, n_features]
            The training input samples.
        y : list, array-like, or pandas dataframe.
            The target values (class labels in classification, real numbers in
            regression). Shape = [n_samples] or [n_samples, n_outputs]

        """
        # arity is the number of inputs that the program takes.
        X, y, arity, y_types = check_X_y(X, y)
        output_types = [self.interpreter.type_library.push_type_for_type(t).name for t in y_types]
        if self.last_str_from_stdout:
            ndx = list_rindex(output_types, "str")
            if ndx is not None:
                output_types[ndx] = "stdout"
        self.signature = ProgramSignature(arity=arity, output_stacks=output_types, push_config=self.push_config)
        self.evaluator = DatasetEvaluator(X, y, interpreter=self.interpreter)
        self._build_search_algo()
        self.solution = self.search.run()
        self.search.config.tear_down()

    def predict(self, X):
        """Execute the synthesized push program on a dataset.

        Parameters
        ----------
        X : pandas dataframe of shape = [n_samples, n_features]
            The set of cases to predict.

        Returns
        -------
        y_hat : pandas dataframe of shape = [n_samples, n_outputs]

        """
        check_is_fitted(self, "solution")
        return [
            self.interpreter.run(
                self.solution.program,
                inputs
            ) for inputs in X
        ]

    def score(self, X, y):
        """Run the search algorithm to synthesize a push program.

        Parameters
        ----------
        X : pandas dataframe of shape = [n_samples, n_features]
            The training input samples.

        y : list, array-like, or pandas dataframe.
            The target values (class labels in classification, real numbers in
            regression). Shape = [n_samples] or [n_samples, n_outputs]

        """

        check_is_fitted(self, "solution")