Source code for Population.PopulationInitialization

  1"""Population initialization factories and strategies.
  2
  3Provides protocols and concrete implementations for creating initial
  4populations of :class:`~Individual.BitstreamIndividual.BitstreamIndividual`
  5objects with configurable post-construction and randomization strategies.
  6"""
  7
  8from pathlib import Path
  9from random import Random
 10from typing import Protocol
 11from BitstreamEvolutionProtocols import Circuit, Individual, Population
 12from Circuit.FileBasedCircuit import FileBasedCircuit
 13from Directories import Directories
 14from Individual.BitstreamIndividual import BitstreamIndividual
 15from Logger import Logger
 16from utilities import wipe_folder
 17import os
 18
 19SEED_HARDWARE_FILEPATH = Path("data/seed-hardware.asc")
 20

[docs]
 21class GenerateInitialPopulations(Protocol):
 22    "Somehow gets you an initial implementation."
 23    def __call__(self) -> list[Population]: ...

 24

[docs]
 25class GenerateInitialPopulation(Protocol):
 26    "The special case of GenerateInitialPopulations that constructs only a single population"
 27    def __call__(self) -> Population: ...

 28

[docs]
 29class GenerateSinglePopulationWrapper:
 30    """
 31    A wrapper around GenerateInitialPopulation (singular) that follows
 32    GenerateInitialPopulations (plural) which is the protocol used in Evolution
 33    Returns a one-element list of the population created
 34    Example Usage: Evolution(..., GenerateSinglePopulationWrapper(GenerateRandomPopulation(...)), ...)
 35    """
 36    def __init__(self, gen_func: GenerateInitialPopulation):
 37        self.__gen_func = gen_func
 38    def generate(self) -> list[Population]:
 39        return [self.__gen_func()]

 40

[docs]
 41class PostConstructionStrategy(Protocol):
 42    '''Consumes the incoming parameter'''
 43    def run(self, individual: BitstreamIndividual) -> BitstreamIndividual: ...

 44

[docs]
 45class RandomizationStrategy(Protocol):
 46    '''Consumes the incoming parameter'''
 47    def randomize(self, individuals: list[BitstreamIndividual]) -> list[BitstreamIndividual]: ...

 48
 49class GenerateBitstreamPopulation:
 50    # force everything to be passed by keyword
 51    def __init__(self, *, sz: int, bitstream_sz: int,
 52                 post_construction_strategy: PostConstructionStrategy,
 53                 randomization_strategy: RandomizationStrategy,
 54                 mutation_prob: float, rand: Random):
 55        self.__sz = sz
 56        self.__bitstream_sz = bitstream_sz
 57        self.__post_construction_strategy = post_construction_strategy
 58        self.__randomization_strategy = randomization_strategy
 59        self.__rand = rand
 60
 61        self.__mutation_prob = mutation_prob
 62        
 63    def generate(self) -> Population:
 64        """
 65        Creates initial population based on the config.
 66        1. Clears the files used to keep track of circuit
 67        2. Uses appropriate initialization method specified by config.
 68        3. Handles randomization until condition in config is met.
 69        """
 70
 71        individuals = []
 72
 73        for index in range(1, self.__sz + 1):
 74            individual = BitstreamIndividual(self.__bitstream_sz, self.__rand, self.__mutation_prob)
 75            individual = self.__post_construction_strategy.run(individual)
 76
 77            individuals.append(individual)
 78
 79        circuits = self.__randomization_strategy.randomize(individuals)
 80
 81        return Population(circuits, None)
 82
 83# used for CLONE_SEED
 84class NoPostConstructionStrategy:
 85    def run(self, individual: BitstreamIndividual) -> BitstreamIndividual:
 86        return individual
 87
 88# used for CLONE_SEED_MUTATE
 89class MutateOncePostConstructionStrategy:
 90    def run(self, individual: BitstreamIndividual) -> BitstreamIndividual:
 91        individual.mutate()
 92        return individual
 93    
 94# used for RANDOM
 95class RandomizeBitstreamPostConstructionStrategy:
 96    def run(self, individual: BitstreamIndividual) -> BitstreamIndividual:
 97        individual.randomize()
 98        return individual
 99
100class NoRandomizationStrategy:
101    def randomize(self, individuals: list[BitstreamIndividual]) -> list[BitstreamIndividual]:
102        return individuals
103
104class FileBasedCircuitFactory:
105    def __init__(self, *, sz: int, logger: Logger, directories: Directories, routing_type: str, accessed_columns: list[int]):
106        self.__logger = logger
107        self.__directories = directories
108        self.__routing_type = routing_type
109        self.__accessed_columns = accessed_columns
110
111        wipe_folder(self.__directories.asc_dir)
112        wipe_folder(self.__directories.bin_dir)
113        wipe_folder(self.__directories.data_dir)
114
115        self.__circuits = []
116        for i in range(sz):
117            circuit = FileBasedCircuit(
118                index=i,
119                filename="hardware" + str(i),
120                template=SEED_HARDWARE_FILEPATH,
121                logger=self.__logger,
122                directories=self.__directories,
123                routing_type=self.__routing_type,
124                accessed_columns=self.__accessed_columns
125            )
126            self.__circuits.append(circuit)
127
128    def create(self, populations: list[Population]) -> dict[Circuit, list[tuple[Population, Individual]]]:
129        res: dict[Circuit, list[tuple[Population, Individual]]] = {}
130        index = 0
131        for p in populations:
132            pop = p.population_list
133            for (individual, _) in pop:
134                circuit = self.__circuits[index]
135                bitstream: list[bool] = individual.get_bitstream() # type: ignore
136                circuit.set_bitstream(bitstream)
137                res[circuit] = [(p, individual)]
138                index += 1
139
140        return res
141
142# TODO: to add in randomize_until modes, we need a way to get circuit fitness/measurements AT initialization time