From fda27f2a374309995a9ddb23dfdeee7fc1648c99 Mon Sep 17 00:00:00 2001 From: Maximilian Schmidt Date: Fri, 27 Mar 2020 14:29:32 +0900 Subject: [PATCH] Fix more type hint issues and imports --- gp/cartesian_graph.py | 13 ++++--- gp/genome.py | 26 ++++++++------ gp/hl_api.py | 28 ++++++++------- gp/node.py | 81 ++++++++++++++++++++++++------------------- gp/population.py | 14 ++++++-- gp/primitives.py | 4 ++- 6 files changed, 97 insertions(+), 69 deletions(-) diff --git a/gp/cartesian_graph.py b/gp/cartesian_graph.py index 293f793d..d8241d09 100644 --- a/gp/cartesian_graph.py +++ b/gp/cartesian_graph.py @@ -20,12 +20,11 @@ def __init__(self, genome: gp.Genome) -> None: genome: Genome Genome defining graph connectivity and node operations. """ - self._n_outputs = None - self._n_inputs = None - self._n_columns = None - self._n_rows = None - self._nodes = None - self._gnome = None + self._n_outputs: int = 0 + self._n_inputs: int = 0 + self._n_columns: int = 0 + self._n_rows: int = 0 + self._nodes: List = [] self.parse_genome(genome) self._genome = genome @@ -163,7 +162,7 @@ def determine_active_regions(self) -> List[int]: return active_regions - def __call__(self, x: float) -> List[float]: + def __call__(self, x: List[float]) -> List[float]: # store values of x in input nodes for i, xi in enumerate(x): assert isinstance(self._nodes[i], InputNode) diff --git a/gp/genome.py b/gp/genome.py index fd0d40ba..e0ed1570 100644 --- a/gp/genome.py +++ b/gp/genome.py @@ -1,7 +1,7 @@ import gp import numpy as np -from typing import Generator, List, Optional +from typing import Generator, List, Optional, Tuple from .primitives import Primitives @@ -65,14 +65,14 @@ def __init__( 1 + self._primitives.max_arity ) # one function gene + multiple input genes - self._dna = None # stores dna as list of alleles for all regions + self._dna: List[int] = [] # stores dna as list of alleles for all regions # constants used as identifiers for input and output nodes self._id_input_node = -1 self._id_output_node = -2 - self._non_coding_allele = None + self._non_coding_allele: int = 0 - def __getitem__(self, key: int) -> None: + def __getitem__(self, key: int) -> int: if self._dna is None: raise RuntimeError("dna not initialized") return self._dna[key] @@ -126,7 +126,7 @@ def _create_input_region(self) -> list: return region def _create_random_hidden_region( - self, rng: np.random.RandomState, permissable_inputs: int + self, rng: np.random.RandomState, permissable_inputs: List[int] ) -> list: # construct dna region consisting of function allele and @@ -146,7 +146,7 @@ def _create_random_hidden_region( return region def _create_random_output_region( - self, rng: np.random.RandomState, permissable_inputs: int + self, rng: np.random.RandomState, permissable_inputs: List[int] ) -> list: # fill region with identifier for output node and single @@ -193,7 +193,7 @@ def randomize(self, rng: np.random.RandomState) -> None: self._validate_dna(dna) self._dna = dna - def _permissable_inputs(self, region_idx: int) -> list: + def _permissable_inputs(self, region_idx: int) -> List[int]: assert not self._is_input_region(region_idx) @@ -281,7 +281,9 @@ def _hidden_column_idx(self, region_idx: int) -> int: assert hidden_column_idx < self._n_columns return hidden_column_idx - def iter_input_regions(self, dna: Optional[List[int]] = None) -> Generator[int, list]: + def iter_input_regions( + self, dna: Optional[List[int]] = None + ) -> Generator[Tuple[int, list], None, None]: if dna is None: dna = self.dna for i in range(self._n_inputs): @@ -291,7 +293,9 @@ def iter_input_regions(self, dna: Optional[List[int]] = None) -> Generator[int, ] yield region_idx, region - def iter_hidden_regions(self, dna: Optional[List[int]] = None) -> Generator[int]: + def iter_hidden_regions( + self, dna: Optional[List[int]] = None + ) -> Generator[Tuple[int, List[int]], None, None]: if dna is None: dna = self.dna for i in range(self._n_hidden): @@ -301,7 +305,9 @@ def iter_hidden_regions(self, dna: Optional[List[int]] = None) -> Generator[int] ] yield region_idx, region - def iter_output_regions(self, dna: Optional[List[int]] = None) -> Generator[int]: + def iter_output_regions( + self, dna: Optional[List[int]] = None + ) -> Generator[Tuple[int, List[int]], None, None]: if dna is None: dna = self.dna for i in range(self._n_outputs): diff --git a/gp/hl_api.py b/gp/hl_api.py index aa3ea934..a9687d58 100644 --- a/gp/hl_api.py +++ b/gp/hl_api.py @@ -1,18 +1,21 @@ import numpy as np +from typing import Optional, Callable +from .abstract_individual import AbstractIndividual +from .abstract_population import AbstractPopulation +from .ea import MuPlusLambda def evolve( - pop, - objective, - ea, - max_generations, - min_fitness, - print_progress=False, - *, - callback=None, - label=None, - n_processes=1, -): + pop: AbstractPopulation, + objective: Callable[[AbstractIndividual], AbstractIndividual], + ea: MuPlusLambda, + max_generations: int, + min_fitness: float, + print_progress: Optional[bool] = False, + callback: Optional[Callable[[AbstractPopulation], None]] = None, + label: Optional[str] = None, + n_processes: int = 1, +) -> None: """ Evolves a population and returns the history of fitness of parents. @@ -45,8 +48,7 @@ def evolve( Returns ------- - dict - History of the evolution. + None """ ea.initialize_fitness_parents(pop, objective, label=label) diff --git a/gp/node.py b/gp/node.py index d7c570ef..15bb8ee0 100644 --- a/gp/node.py +++ b/gp/node.py @@ -1,10 +1,13 @@ -import gp -from typing import Any, List +from __future__ import annotations + +from typing import List, Union + +from cgp_graph import CGPGraph primitives_dict = {} # maps string of class names to classes -def register(cls: gp.CGPNode) -> None: +def register(cls: CGPNode) -> None: """Register a primitive in the global dictionary of primitives Parameters @@ -25,12 +28,15 @@ class Node: """Base class for primitive functions used in Cartesian computational graphs. """ - _arity = None - _active = False - _inputs = None - _output = None - _idx = None - _is_parameter = False + __name__ = "CGPNode" + _arity: int = 0 + _active: bool = False + _inputs: List[int] = [] + _output: float = 0.0 + _idx: int = 0 + _is_parameter: bool = False + _output_str: str = "" + _parameter_str: str = "" def __init__(self, idx: int, inputs: List[int]) -> None: """Init function. @@ -47,12 +53,15 @@ def __init__(self, idx: int, inputs: List[int]) -> None: assert idx not in inputs - def __init_subclass__(cls: gp.CGPNode, **kwargs: Any) -> None: - super().__init_subclass__(**kwargs) - register(cls) + # def __init_subclass__(cls: CGPNode, **kwargs: Any) -> None: + # super().__init_subclass__(**kwargs) + # register(cls) + + def __call__(self, x: float, graph: CGPGraph) -> None: + raise NotImplementedError @property - def arity(self) -> int: + def arity(self) -> Union[None, int]: return self._arity @property @@ -64,13 +73,13 @@ def inputs(self) -> List[int]: return self._inputs[: self._arity] @property - def idx(self) -> int: + def idx(self) -> Union[None, int]: return self._idx def __repr__(self) -> str: return ( f"{self.__class__.__name__}(idx: {self.idx}, active: {self._active}, " - f"arity: {self._arity}, inputs {self._inputs}, output {self._outputs})" + f"arity: {self._arity}, inputs {self._inputs}, output {self._output})" ) def pretty_str(self, n: int) -> str: @@ -114,7 +123,7 @@ def pretty_str(self, n: int) -> str: return s.ljust(n) @property - def output(self) -> int: + def output(self) -> float: return self._output def activate(self) -> None: @@ -122,7 +131,7 @@ def activate(self) -> None: """ self._active = True - def format_output_str(self, graph: gp.CGPGraph) -> None: + def format_output_str(self, graph: CGPGraph) -> None: """Format output string of the node. """ raise NotImplementedError() @@ -144,7 +153,7 @@ def output_str(self) -> str: @property def output_str_torch(self) -> str: - return self.output_str + return self._output_str @property def is_parameter(self) -> bool: @@ -164,10 +173,10 @@ class Add(Node): def __init__(self, idx: int, inputs: List[int]) -> None: super().__init__(idx, inputs) - def __call__(self, x: float, graph: gp.CGPGraph) -> None: + def __call__(self, x: float, graph: CGPGraph) -> None: self._output = graph[self._inputs[0]].output + graph[self._inputs[1]].output - def format_output_str(self, graph: gp.CGPGraph) -> None: + def format_output_str(self, graph: CGPGraph) -> None: self._output_str = ( f"({graph[self._inputs[0]].output_str} + {graph[self._inputs[1]].output_str})" ) @@ -182,10 +191,10 @@ class Sub(Node): def __init__(self, idx: int, inputs: List[int]) -> None: super().__init__(idx, inputs) - def __call__(self, x: float, graph: gp.CGPGraph) -> None: + def __call__(self, x: float, graph: CGPGraph) -> None: self._output = graph[self._inputs[0]].output - graph[self._inputs[1]].output - def format_output_str(self, graph: gp.CGPGraph) -> None: + def format_output_str(self, graph: CGPGraph) -> None: self._output_str = ( f"({graph[self._inputs[0]].output_str} - {graph[self._inputs[1]].output_str})" ) @@ -200,10 +209,10 @@ class Mul(Node): def __init__(self, idx: int, inputs: List[int]) -> None: super().__init__(idx, inputs) - def __call__(self, x: float, graph: gp.CGPGraph) -> None: + def __call__(self, x: float, graph: CGPGraph) -> None: self._output = graph[self._inputs[0]].output * graph[self._inputs[1]].output - def format_output_str(self, graph: gp.CGPGraph) -> None: + def format_output_str(self, graph: CGPGraph) -> None: self._output_str = ( f"({graph[self._inputs[0]].output_str} * {graph[self._inputs[1]].output_str})" ) @@ -218,11 +227,11 @@ class Div(Node): def __init__(self, idx: int, inputs: List[int]) -> None: super().__init__(idx, inputs) - def __call__(self, x: float, graph: gp.CGPGraph) -> None: + def __call__(self, x: float, graph: CGPGraph) -> None: self._output = graph[self._inputs[0]].output / graph[self._inputs[1]].output - def format_output_str(self, graph: gp.CGPGraph) -> None: + def format_output_str(self, graph: CGPGraph) -> None: self._output_str = ( f"({graph[self._inputs[0]].output_str} / {graph[self._inputs[1]].output_str})" ) @@ -240,13 +249,13 @@ def __init__(self, idx: int, inputs: List[int]) -> None: self._output = 1.0 - def __call__(self, x: float, graph: gp.CGPGraph) -> None: + def __call__(self, x: float, graph: CGPGraph) -> None: pass - def format_output_str(self, graph: gp.CGPGraph) -> None: + def format_output_str(self, graph: CGPGraph) -> None: self._output_str = f"{self._output}" - def format_output_str_torch(self, graph: gp.CGPGraph) -> None: + def format_output_str_torch(self, graph: CGPGraph) -> None: self._output_str = f"self._p{self._idx}.expand(x.shape[0])" def format_parameter_str(self) -> None: @@ -264,13 +273,13 @@ class InputNode(Node): def __init__(self, idx: int, inputs: List[int]) -> None: super().__init__(idx, inputs) - def __call__(self, x: float, graph: gp.CGPGraph) -> None: + def __call__(self, x: float, graph: CGPGraph) -> None: assert False - def format_output_str(self, graph: gp.CGPGraph) -> None: + def format_output_str(self, graph: CGPGraph) -> None: self._output_str = f"x[{self._idx}]" - def format_output_str_torch(self, graph: gp.CGPGraph) -> None: + def format_output_str_torch(self, graph: CGPGraph) -> None: self._output_str = f"x[:, {self._idx}]" @@ -283,10 +292,10 @@ class OutputNode(Node): def __init__(self, idx: int, inputs: List[int]) -> None: super().__init__(idx, inputs) - def __call__(self, x: float, graph: gp.CGPGraph) -> None: + def __call__(self, x: float, graph: CGPGraph) -> None: self._output = graph[self._inputs[0]].output - def format_output_str(self, graph: gp.CGPGraph) -> None: + def format_output_str(self, graph: CGPGraph) -> None: self._output_str = f"{graph[self._inputs[0]].output_str}" @@ -299,10 +308,10 @@ class Pow(Node): def __init__(self, idx: int, inputs: List[int]) -> None: super().__init__(idx, inputs) - def __call__(self, x: float, graph: gp.CGPGraph) -> None: + def __call__(self, x: float, graph: CGPGraph) -> None: self._output = graph[self._inputs[0]].output ** graph[self._inputs[1]].output - def format_output_str(self, graph: gp.CGPGraph) -> None: + def format_output_str(self, graph: CGPGraph) -> None: self._output_str = ( f"({graph[self._inputs[0]].output_str} ** {graph[self._inputs[1]].output_str})" ) diff --git a/gp/population.py b/gp/population.py index e456b710..81e043e9 100755 --- a/gp/population.py +++ b/gp/population.py @@ -3,6 +3,7 @@ from .individual import Individual, IndividualMultiGenome from typing import List +from .abstract_individual import AbstractIndividual @@ -37,7 +38,7 @@ def __init__(self, n_parents: int, mutation_rate: float, seed: int) -> None: self._genome_params = genome_params - self._parents = None # list of parent individuals + self._parents: List[Individual] = [] # list of parent individuals # keeps track of the number of generations, increases with # every new offspring generation @@ -73,7 +74,7 @@ def _label_new_individuals(self, individuals: List[gp.AbstractIndividual]) -> No ind.idx = self._max_idx + 1 self._max_idx += 1 - def _generate_random_individuals(self, n: int) -> None: + def _generate_random_individuals(self, n: int) -> List[Individual]: individuals = [] for i in range(n): if isinstance(self._genome_params, dict): @@ -106,6 +107,7 @@ def crossover( List[gp.Individual] List of offspring individuals. """ +<<<<<<< HEAD:gp/population.py # in principle crossover would rely on a procedure like the # following: # offsprings = [] @@ -125,6 +127,14 @@ def crossover( # pages 1135–1142. Morgan Kaufmann Publishers Inc. assert len(breeding_pool) >= n_offsprings return sorted(breeding_pool, key=lambda x: -x.fitness)[:n_offsprings] +======= + offsprings: List[AbstractIndividual] = [] + while len(offsprings) < n_offsprings: + first_parent, second_parent = self.rng.permutation(breeding_pool)[:2] + offsprings.append(first_parent.crossover(second_parent, self.rng)) + + return offsprings +>>>>>>> Fix more type hint issues and imports:gp/abstract_population.py def mutate(self, offsprings: List[gp.AbstractIndividual]) -> List[gp.AbstractIndividual]: """Mutate a list of offspring invididuals. diff --git a/gp/primitives.py b/gp/primitives.py index 3f8f7cc8..6b5c3ca2 100644 --- a/gp/primitives.py +++ b/gp/primitives.py @@ -1,7 +1,9 @@ from .node import Node -import numpy as np from typing import Hashable, List, Optional, Tuple +import numpy as np # type: ignore + +from typing import List, Tuple class Primitives: