API Reference

def execute_test(self, test) -> Tuple[float, str]:
    """
    Executes a given test, evaluates its fitness, and logs execution details.

    Arguments:
        test: The test input to be executed, typically a genotype representation.

    Returns:
        Tuple[float, str]: A tuple containing the fitness value (as a float) and additional information (as a string).

    Notes:
        - Converts the genotype to phenotype using the generator.
        - Validates the test before execution; if invalid, returns a fitness of 0.
        - If an exception occurs during execution, logs the error and updates the test info accordingly.
    """

    self.exec_counter += 1  # counts how many executions have been

    fitness = 0

    self.test_dict[self.exec_counter] = {"test": list(test), "fitness": None, "info": None, "timestamp": time.time() }

    test = self.generator.genotype2phenotype(test)

    valid, info = self.generator.is_valid(test)
    if not valid:
        log.debug("The generated road is invalid")
        self.test_dict[self.exec_counter]["fitness"] = fitness
        self.test_dict[self.exec_counter]["info"] = info
        return float(fitness)

    try:
        start = time.time()
        fitness = self._execute(test)
        end = time.time()
        self.test_dict[self.exec_counter]["execution_time"] = end - start
        log.debug(f"Execution time: {end - start} seconds")
        self.test_dict[self.exec_counter]["fitness"] = fitness
        self.test_dict[self.exec_counter]["info"] = info

    except Exception as e:
        log.info(f"Error {e} found")
        log.info(f"Error {traceback.format_exc()} found")
        log.error("Error during execution of test.", exc_info=True)
        self.test_dict[self.exec_counter]["info"] = f"ERROR: {e}"


    return float(fitness)

def __str__(self) -> str:
    """String representation of the generator."""
    return f"{self.__class__.__name__}(name='{self.name}', genotype_size={self.genotype_size})"

@abc.abstractmethod
def cmp_func(self, test1: typing.List[float], test2: typing.List[float]) -> int:
    """Compare two tests.

    Args:
        test1 (np.array): First test to compare.
        test2 (np.array): Second test to compare.

    Returns:
        float: the degree of similarity between the two tests based on the distance function (cosine similarity by default)
    """
    pass

@abc.abstractmethod
def generate_random_test(self) -> (typing.List[float]):
    """Generate samples from the generator

    Returns:
        np.array: Generated samples.
    """
    pass

@abc.abstractmethod
def genotype2phenotype(self, genotype: typing.List[float]) -> typing.List[float]:
    """Convert a genotype to a phenotype.

    Args:
        genotype (np.array): Genotype to convert.

    Returns:
        np.array: Phenotype representation of the genotype.
    """
    pass

@abc.abstractmethod
def is_valid(self, test: typing.List[float]) -> bool:
    """Check if a test is valid.

    Args:
        test (np.array): Test to check.

    Returns:
        bool: True if the test is valid, False otherwise.
    """
    pass

def phenotype2genotype(self, phenotype: typing.List[float]) -> typing.List[float]:
    """Convert a phenotype to a genotype.

    Args:
        phenotype (np.array): Phenotype to convert.

    Returns:
        np.array: Genotype representation of the phenotype.
    """
    pass

@abc.abstractmethod
def visualize_test(self, test: typing.List[float], save_path: str = None):
    """Visualize a test.

    Args:
        test (np.array): Test to visualize.
    """
    pass

def denormalize_flattened_test(self, norm_test: typing.List[float]) -> typing.List[float]:
    '''
    Use min and max values to denormalize the test case.
    Args:
        norm_test (list): Flattened normalized test case to denormalize.
    '''
    result = norm_test*(self._u_b - self._l_b) + self._l_b
    result =  np.array(result, dtype=object)
    result[0] = int(round(result[0]))  # Ensure the first element is an integer (number of boxes)

    return result

def is_valid(self, test) -> tuple[bool, str]:
    """
    Check if the given test is valid.

    Args:
        test: The test to be validated.

    Returns:
        A tuple containing a boolean indicating the validity of the test and a string with additional information.
    """
    #log.info(f"test {test}")
    boxes  = test.test.simulation.obstacles

    valid, info  = self.obstacles_fit(boxes)

    return valid, info

def normalize_flattened_test(self, test: typing.List[float]) -> typing.List[float]:
    '''
    Use min and max values to normalize the test case.
    Args:
        test (list): Flattened test case to normalize.'''
    result = (test - self._l_b)/(self._u_b - self._l_b)
    return result

def obstacles_fit(self, box_list: typing.List[Obstacle]) -> tuple[bool, str]:
    """
    Check if the boxes in the given list fit within the defined bounds.

    Args:
        box_list: A list of boxes to be checked.

    Returns:
        A tuple containing a boolean indicating if the boxes fit within the bounds and a string with additional information.
    """
    existing_boxes_geometry_list = [obstacle.geometry for obstacle in box_list]#[obstacle.position.x, obstacle.position.y, obstacle.position.r]

    min_pos = [self.min_position.x, self.min_position.y]
    max_pos = [self.max_position.x, self.max_position.y]

    outer_polygon = geometry.Polygon([min_pos, [min_pos[0], max_pos[1]], max_pos, [max_pos[0], min_pos[1]]])

    n = len(existing_boxes_geometry_list)

    for i in range(n):
        for j in range(i + 1, n):
            poly1 = existing_boxes_geometry_list[i]
            poly2 = existing_boxes_geometry_list[j]

            # Check if polygons intersect
            if poly1.intersects(poly2):

                return False, f"Obstacles {i} and {j} intersect."

    for inner_polygon in existing_boxes_geometry_list:
        if not(inner_polygon.within(outer_polygon)):
            return False, "Obstacle out of the defined bounds"


    return True, "Valid test"

@abc.abstractmethod
def do_mutation(self, x) -> np.ndarray:
    """
    Apply a mutation operation to the input array.

    Parameters:
        x: array-like
            The input data to be mutated.

    Returns:
        np.ndarray
            The mutated version of the input array.
    """
    pass

def do_mutation(self, x) -> np.ndarray:
    """
    Applies a random mutation to the given solution `x` and returns the mutated solution if it is valid.
    The mutation is selected randomly from the following options:
        - Random modification of the solution.
        - Addition of an obstacle.
        - Removal of an obstacle.
    If the mutated solution is not valid according to the validator, the original solution is returned.
    Args:
        x: The solution to be mutated.
    Returns:
        np.ndarray: The mutated solution if valid, otherwise the original solution.
    """
    if not self.generator:
        raise ValueError("Generator not set.")

    possible_mutations = [
        self._random_modification, 
        self._add_obstacle,
        self._remove_obstacle
    ]

    x = self.generator.denormalize_flattened_test(x)

    mutator = np.random.choice(possible_mutations)

    mutated_x = mutator(x)
    mutated_x = self.generator.normalize_flattened_test(mutated_x)

    return mutated_x

def configure_algorithm(self):
    """
    Configures and initializes the evolutionary algorithm with the specified crossover, mutation,
    and sampling strategies based on the current config file.
    This method also sets the DuplicateElimination strategy to remove duplicates
    during the optimization process. Experimentally obtained threshold should be used, 0.025 by default.
    """
    cross_prob = self.config["search_based"]["crossover_prob"]
    mut_prob = self.config["search_based"]["mutation_prob"]
    if self.crossover == "sbx":
        crossover = CROSSOVERS[self.crossover](prob_var=cross_prob, eta=3.0, vtype=float) # crossover defined by pymoo
    else:
        crossover = CROSSOVERS[self.crossover](cross_prob=cross_prob)
    if self.mutation == "pm":
        mutation = MUTATIONS[self.mutation](prob=mut_prob, eta=3.0, vtype=float) # mutation defined by pymoo
    else:
        mutation = MUTATIONS[self.mutation](mut_prob=mut_prob)

    self.method = ALGORITHMS[self.alg](
        pop_size=self.pop_size,
        n_offsprings=int(round(self.pop_size / 2)),
        sampling=SAMPLERS[self.sampling](self.generator),
        n_points_per_iteration=int(round(self.pop_size / 2)),
        crossover=crossover,
        mutation=mutation,
        eliminate_duplicates=AbstractDuplicateElimination(
            generator=self.generator, threshold=0.025
        ),
    )

def initialize_parameters(self):
    """
    Initializes the parameters required for the test generation process.
    This method sets up the random seed, population size, algorithm, sampling method,
    crossover, and mutation strategy based on the provided configuration. If a seed is
    not specified in the configuration, a random seed is generated.
    """
    log.info("Starting test generation, initializing parameters")

    if self.config["common"]["seed"] != "None":
        self.seed = self.config["common"]["seed"]
        log.info(f"Using provided seed: {self.seed}")
    else:
        log.info("No seed provided, generating a random seed")
        # Generate a random seed if not provided in the config
        self.seed = get_random_seed()
    self.pop_size = self.config["search_based"]["pop_size"]
    log.info(f"Population size: {self.pop_size}")
    self.alg = self.config["search_based"]["algorithm"]
    self.sampling = "abstract"
    self.crossover = self.config["search_based"]["crossover"]
    self.mutation = self.config["search_based"]["mutation"]

def initialize_problem(self):
    """
    Initializes the pymoo optimization problem by creating an instance of AbstractProblem.
    """
    self.problem = AbstractProblem(
        self.executors,
        self.generator, 
        n_var=self.generator.size, # number of variables in the genotype
        xl=self.generator.lower_bound,
        xu=self.generator.upper_bound
    )

@abc.abstractmethod
def initialize_test_executors(self):
    """
    To be implemented specifically for your problem."""
    pass

@abc.abstractmethod
def initialize_test_generator(self):
    '''
    To be implemented specifically for your problem.'''
    pass

def run_optimization(self):
    """
    This method initializes and executes the optimization algorithm with the provided settings,
    including termination criteria, random seed, verbosity, duplicate elimination, and history saving.
    The optimization result is stored in the `self.res` attribute.
    """
    self.res = minimize(
        self.problem,
        self.method,
        termination=get_termination(
            self.config["common"]["termination"], self.config["common"]["budget"]
        ),
        seed=self.seed,
        verbose=True,
        eliminate_duplicates=True,
        save_history=True,
    )
    return self.res

def set_up_search_algorithm(self):
    """
    Sets up the search algorithm by initializing parameters, configuring the algorithm,
    and initializing the problem. 
    """
    self.initialize_parameters()
    self.configure_algorithm()
    self.initialize_problem()

def ask(self) -> List:
    """
    Generates and returns a new population of candidate solutions using the underlying evolutionary method.

    Returns:
        list: A list representing the new population of candidate solutions.
    """
    pop = self.method.ask()

    return pop

def initialize_test_executors(self):
    """
    To be implemented specifically for your problem."""
    pass

@abc.abstractmethod
def initialize_test_generator(self):
    '''
    To be implemented specifically for your problem.'''
    pass

def set_up_search_algorithm(self):
    """
    Sets up the search algorithm by initializing parameters, configuring the algorithm,
    and initializing the problem. 
    """
    self.initialize_parameters()
    self.configure_algorithm()
    self.initialize_problem()
    termination  = get_termination(
            self.config["common"]["termination"], self.config["common"]["budget"]
        )
    self.method.setup(self.problem, termination=termination, verbose=True, eliminate_duplicates=True, save_history=True)
    np.random.seed(self.seed)
    self.current_individual = self.pop_size
    self.ind_num = self.pop_size
    self.pop = None

def tell(self, population: List):
    """Provides the test results back to the tester."""
    self.method.tell(population)

def get_results(self):
    """
    Retrieves the results of the test generation and optimization process. To be used with ask and tell interface.
    """
    return self.test_dict, self.method.result()

def initialize(self):
    """
    Initializes the tester by setting up the necessary components. To be used with ask and tell interface.
    """
    self.initialize_test_generator()
    self.initialize_test_executors()
    self.set_up_search_algorithm()

@abc.abstractmethod
def initialize_test_executors(self) -> AbstractExecutor:
    """
    Initializes the test executor. This method should be implemented
    specifically for your problem. It should launch and evaluate the target
    system or function. 
    """
    pass

@abc.abstractmethod
def initialize_test_generator(self) -> AbstractGenerator:
    """
    Initializes the test generator. 
    This method should be implemented specifically for your your problem.
    It should generate random tests, check their validity, do conversion 
    between genotype and phenotype representations. For more details,
    refer to the ```AbstractGenerator``` class.

    Raises:
        NotImplementedError: If the method is not overridden by a subclass.
    """
    pass

@abc.abstractmethod
def run_optimization(self):
    """
    Runs the optimization process to generate test cases.

    This method should be implemented by subclasses to execute the optimization algorithm.
    """
    pass

@abc.abstractmethod
def set_up_search_algorithm(self):
    """
    Initializes or configures the search algorithm to be used in the testing process.

    This method should be implemented by subclasses.
    """
    pass

def start(self):
    """
    Function to start the test generation and optimization process. To be used when evaluation happens within the optimization process.

    Returns:
        tuple: A tuple containing the test results (`self.res`) and the test executor (`self.executor`).
    """

    self.initialize_test_generator()
    self.initialize_test_executors()
    self.set_up_search_algorithm()
    self.run_optimization()
    return self.executors[0].test_dict, self.res

def initialize_test_executors(self):
    """Initializes the executor for the UAV test generator.
    The number of executors in the list corresponds to the number of search objectives.
    Minimum fitness is set to 25.0, which is the minimum path length for the RRT algorithm to identify to count the test as challenging"""

    #self.executor = ObstacleSceneExecutor(self.generator)
    self.executors = [RRTExecutor(self.generator, min_fitness=25.0)]

def initialize_test_generator(self):
    """
    Initializes the test generator for UAV testing.
    This method sets up an ```ObstacleGenerator``` instance. 
    """

    case_study = "case_studies/mission1.yaml"
    self.generator = ObstacleGenerator(
        case_study_file=case_study,
        max_box_num=3,
    )