Enforce the population size limit by removing worst programs if needed Args: exclude_program_id: Program ID to never remove (e.g., newly added program)
(self, exclude_program_id: Optional[str] = None)
| 1680 | return inspirations[:n] |
| 1681 | |
| 1682 | def _enforce_population_limit(self, exclude_program_id: Optional[str] = None) -> None: |
| 1683 | """ |
| 1684 | Enforce the population size limit by removing worst programs if needed |
| 1685 | |
| 1686 | Args: |
| 1687 | exclude_program_id: Program ID to never remove (e.g., newly added program) |
| 1688 | """ |
| 1689 | if len(self.programs) <= self.config.population_size: |
| 1690 | return |
| 1691 | |
| 1692 | # Calculate how many programs to remove |
| 1693 | num_to_remove = len(self.programs) - self.config.population_size |
| 1694 | |
| 1695 | logger.info( |
| 1696 | f"Population size ({len(self.programs)}) exceeds limit ({self.config.population_size}), removing {num_to_remove} programs" |
| 1697 | ) |
| 1698 | |
| 1699 | # Get programs sorted by fitness (worst first) |
| 1700 | all_programs = list(self.programs.values()) |
| 1701 | |
| 1702 | # Sort by combined_score if available, otherwise by average metric (worst first) |
| 1703 | sorted_programs = sorted( |
| 1704 | all_programs, |
| 1705 | key=lambda p: get_fitness_score(p.metrics, self.config.feature_dimensions), |
| 1706 | ) |
| 1707 | |
| 1708 | # Remove worst programs, but never remove the best program or excluded program |
| 1709 | programs_to_remove = [] |
| 1710 | protected_ids = {self.best_program_id, exclude_program_id} - {None} |
| 1711 | |
| 1712 | for program in sorted_programs: |
| 1713 | if len(programs_to_remove) >= num_to_remove: |
| 1714 | break |
| 1715 | # Don't remove the best program or excluded program |
| 1716 | if program.id not in protected_ids: |
| 1717 | programs_to_remove.append(program) |
| 1718 | |
| 1719 | # If we still need to remove more and only have protected programs, |
| 1720 | # remove from the remaining programs anyway (but keep the protected ones) |
| 1721 | if len(programs_to_remove) < num_to_remove: |
| 1722 | remaining_programs = [ |
| 1723 | p |
| 1724 | for p in sorted_programs |
| 1725 | if p not in programs_to_remove and p.id not in protected_ids |
| 1726 | ] |
| 1727 | additional_removals = remaining_programs[: num_to_remove - len(programs_to_remove)] |
| 1728 | programs_to_remove.extend(additional_removals) |
| 1729 | |
| 1730 | # Remove the selected programs |
| 1731 | for program in programs_to_remove: |
| 1732 | program_id = program.id |
| 1733 | |
| 1734 | # Remove from main programs dict |
| 1735 | if program_id in self.programs: |
| 1736 | del self.programs[program_id] |
| 1737 | |
| 1738 | # Remove from island feature maps |
| 1739 | for island_idx, island_map in enumerate(self.island_feature_maps): |
no test coverage detected