|
| 1 | +import base64 |
| 2 | +from pickle import dumps, loads |
| 3 | +from random import randrange |
| 4 | +from typing import Dict, List |
| 5 | +from .player import Player |
| 6 | + |
| 7 | + |
| 8 | +class InsufficientParametersError(Exception): |
| 9 | + """Error indicating that insufficient parameters were specified to initialize an Evolvable Player.""" |
| 10 | + def __init__(self, *args): |
| 11 | + super().__init__(*args) |
| 12 | + |
| 13 | + |
| 14 | +class EvolvablePlayer(Player): |
| 15 | + """A class for a player that can evolve, for use in the Moran process or with reinforcement learning algorithms. |
| 16 | +
|
| 17 | + This is an abstract base class, not intended to be used directly. |
| 18 | + """ |
| 19 | + |
| 20 | + name = "EvolvablePlayer" |
| 21 | + parent_class = Player |
| 22 | + parent_kwargs = [] # type: List[str] |
| 23 | + |
| 24 | + def overwrite_init_kwargs(self, **kwargs): |
| 25 | + """Use to overwrite parameters for proper cloning and testing.""" |
| 26 | + for k, v in kwargs.items(): |
| 27 | + self.init_kwargs[k] = v |
| 28 | + |
| 29 | + def create_new(self, **kwargs): |
| 30 | + """Creates a new variant with parameters overwritten by kwargs.""" |
| 31 | + init_kwargs = self.init_kwargs.copy() |
| 32 | + init_kwargs.update(kwargs) |
| 33 | + return self.__class__(**init_kwargs) |
| 34 | + |
| 35 | + # Serialization and deserialization. You may overwrite to obtain more human readable serializations |
| 36 | + # but you must overwrite both. |
| 37 | + |
| 38 | + def serialize_parameters(self): |
| 39 | + """Serialize parameters.""" |
| 40 | + pickled = dumps(self.init_kwargs) # bytes |
| 41 | + s = base64.b64encode(pickled).decode('utf8') # string |
| 42 | + return s |
| 43 | + |
| 44 | + @classmethod |
| 45 | + def deserialize_parameters(cls, serialized): |
| 46 | + """Deserialize parameters to a Player instance.""" |
| 47 | + init_kwargs = loads(base64.b64decode(serialized)) |
| 48 | + return cls(**init_kwargs) |
| 49 | + |
| 50 | + # Optional methods for evolutionary algorithms and Moran processes. |
| 51 | + |
| 52 | + def mutate(self): |
| 53 | + """Optional method to allow Player to produce a variant (not in place).""" |
| 54 | + pass # pragma: no cover |
| 55 | + |
| 56 | + def crossover(self, other): |
| 57 | + """Optional method to allow Player to produce variants in combination with another player. Returns a new |
| 58 | + Player.""" |
| 59 | + pass # pragma: no cover |
| 60 | + |
| 61 | + # Optional methods for particle swarm algorithm. |
| 62 | + |
| 63 | + def receive_vector(self, vector): |
| 64 | + """Receive a vector of params and overwrite the Player.""" |
| 65 | + pass # pragma: no cover |
| 66 | + |
| 67 | + def create_vector_bounds(self): |
| 68 | + """Creates the bounds for the decision variables for Particle Swarm Algorithm.""" |
| 69 | + pass # pragma: no cover |
| 70 | + |
| 71 | + |
| 72 | +def copy_lists(lists: List[List]) -> List[List]: |
| 73 | + return list(map(list, lists)) |
| 74 | + |
| 75 | + |
| 76 | +def crossover_lists(list1: List, list2: List) -> List: |
| 77 | + cross_point = randrange(len(list1)) |
| 78 | + new_list = list(list1[:cross_point]) + list(list2[cross_point:]) |
| 79 | + return new_list |
| 80 | + |
| 81 | + |
| 82 | +def crossover_dictionaries(table1: Dict, table2: Dict) -> Dict: |
| 83 | + keys = list(table1.keys()) |
| 84 | + cross_point = randrange(len(keys)) |
| 85 | + new_items = [(k, table1[k]) for k in keys[:cross_point]] |
| 86 | + new_items += [(k, table2[k]) for k in keys[cross_point:]] |
| 87 | + new_table = dict(new_items) |
| 88 | + return new_table |
0 commit comments