14225235949

Committed 02 Apr 2025 04:56PM UTC coverage: 90.508% (+0.5%) from 90.054%

Build # 14225235949

Build Type

push

github

Committed by

veghp

Commit Message

Bump to v3.2.14

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

101 existing lines in 35 files now uncovered.

2994 of 3308 relevant lines covered (90.51%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.07

/dnachisel/DnaOptimizationProblem/mixins/ObjectivesMaximizerMixin.py

from ...Location import Location
from ...Specification.SpecEvaluation import ProblemObjectivesEvaluations
from ..NoSolutionError import NoSolutionError


class ObjectivesMaximizerMixin:
    @property
    def objectives_before(self):
        if self._objectives_before is None:
            sequence = self.sequence
            self.sequence = self.sequence_before
            self._objectives_before = self.objectives_evaluations()
            self.sequence = sequence
        return self._objectives_before

    def objectives_evaluations(self):
        """Return a list of the evaluation of each objective of the problem"""
        return ProblemObjectivesEvaluations.from_problem(self)

    def objective_scores_sum(self):
        return self.objectives_evaluations().scores_sum()

    def objectives_text_summary(self):
        return self.objectives_evaluations().to_text()

    def optimize_by_exhaustive_search(self):
        """ """
        if not self.all_constraints_pass():
            summary = self.constraints_text_summary(failed_only=True)
            raise NoSolutionError(
                summary + "Optimization can only be done when all constraints are "
                "verified.",
                self,
            )

        if all([obj.best_possible_score is not None for obj in self.objectives]):
            best_possible_score = sum(
                [obj.best_possible_score for obj in self.objectives]
            )
        else:
            best_possible_score = None

        current_best_score = self.objective_scores_sum()
        current_best_sequence = self.sequence
        all_variants = self.mutation_space.all_variants(self.sequence)
        space_size = int(self.mutation_space.space_size)
        self.logger(mutation__total=space_size)
        for variant in self.logger.iter_bar(mutation=all_variants):
            self.sequence = variant
            if self.all_constraints_pass():
                score = self.objective_scores_sum()
                if score > current_best_score:
                    current_best_score = score
                    current_best_sequence = self.sequence
                    if (best_possible_score is not None) and (
                        current_best_score >= best_possible_score
                    ):
                        self.logger(mutation__index=space_size)
                        break
        self.sequence = current_best_sequence

    def optimize_by_random_mutations(self):
        """ """
        if not self.all_constraints_pass():
            summary = self.constraints_text_summary()
            raise ValueError(
                summary + "Optimization can only be done when all"
                " constraints are verified"
            )
        score = self.objective_scores_sum()

        if all([obj.best_possible_score is not None for obj in self.objectives]):
            best_possible_score = sum(
                [obj.best_possible_score * obj.boost for obj in self.objectives]
            )
        else:
            best_possible_score = None
        iters = self.max_random_iters
        stagnating_iterations = 0
        for iteration in self.logger.iter_bar(mutation=range(iters)):
            if (best_possible_score is not None) and (score >= best_possible_score):
                self.logger(mutation__index=iters)
                break
            if (self.optimization_stagnation_tolerance is not None) and (
                stagnating_iterations > self.optimization_stagnation_tolerance
            ):
                break

            previous_sequence = self.sequence
            self.sequence = self.mutation_space.apply_random_mutations(
                n_mutations=self.mutations_per_iteration,
                sequence=self.sequence,
            )
            if self.all_constraints_pass():
                new_score = self.objective_scores_sum()
                if new_score > score:
                    score = new_score
                    stagnating_iterations = 0
                else:
                    self.sequence = previous_sequence
            else:
                self.sequence = previous_sequence
            stagnating_iterations += 1

    def optimize_objective(self, objective):
        """Optimize the total objective score, focusing on a single objective.

        This method will attempt to increase the global objective score by
        focusing on a single objective. First the locations of under-optimal
        subsequences for this objective are identified, then these locations
        are optimized one after the other, left to right.

        For each location, a local problem is created and the optimization uses
        either a custom optimization algorithm, an exhaustive search, or a
        random search, to optimize the local problem
        """
        # EVALUATE OBJECTIVE. RETURN IF THERE IS NOTHING TO BE DONE.
        evaluation = objective.evaluate(self)
        locations = evaluation.locations
        if (objective.best_possible_score is not None) and (
            evaluation.score == objective.best_possible_score
        ):
            return

        # FOR EACH LOCATION, CREATE AND OPTIMIZE A LOCAL PROBLEM.

        for location in self.logger.iter_bar(
            location=locations, bar_message=lambda l: str(l)
        ):
            # Localize the mutation space by freezing any nucleotide outside of
            # it
            mutation_space = self.mutation_space.localized(location)
            if mutation_space.space_size == 0:
                continue

            # Update the location so it matches the span of the mutation_space
            # the resulting location will be equal or smaller to the original
            # location.
            location = Location(*mutation_space.choices_span)
            localized_constraints = [
                cst.localized(location, problem=self) for cst in self.constraints
            ]
            localized_constraints = [
                cst for cst in localized_constraints if cst is not None
            ]
            localized_objectives = [
                obj.localized(location, problem=self)
                for obj in self.objectives
                if obj.boost != 0
            ]
            localized_objectives = [
                obj for obj in localized_objectives if obj is not None
            ]
            local_problem = self.__class__(
                sequence=self.sequence,
                constraints=localized_constraints,
                mutation_space=mutation_space,
                objectives=localized_objectives,
            )
            self.logger.store(
                problem=self, local_problem=local_problem, location=location
            )
            local_problem.randomization_threshold = self.randomization_threshold
            local_problem.max_random_iters = self.max_random_iters
            local_problem.optimization_stagnation_tolerance = (
                self.optimization_stagnation_tolerance
            )
            local_problem.mutations_per_iteration = self.mutations_per_iteration

            # OPTIMIZE THE LOCAL PROBLEM

            if hasattr(objective, "optimization_heuristic"):
                # Some specifications implement their own optimization method.
                objective.optimization_heuristic(local_problem)
            else:
                # Run an exhaustive or random search depending on the size
                # of the mutation space.
                space_size = local_problem.mutation_space.space_size
                exhaustive_search = space_size < self.randomization_threshold
                if exhaustive_search:
                    local_problem.optimize_by_exhaustive_search()
                else:
                    local_problem.optimize_by_random_mutations()

            # UPDATE THE PROBLEM's SEQUENCE

            self.sequence = local_problem.sequence

    def optimize(self):
        """Maximize the total score by optimizing each objective in turn."""

        objectives = [
            obj
            for obj in self.objectives
            if not obj.optimize_passively and obj.boost != 0
        ]
        if len(objectives) == 0:
            return
        for objective in self.logger.iter_bar(
            objective=objectives, bar_message=lambda o: str(o)
        ):
            self.optimize_objective(objective=objective)

1	from ...Location import Location	1✔
2	from ...Specification.SpecEvaluation import ProblemObjectivesEvaluations	1✔
3	from ..NoSolutionError import NoSolutionError	1✔
4
5
6	class ObjectivesMaximizerMixin:	1✔
7	@property	1✔
8	def objectives_before(self):	1✔
9	if self._objectives_before is None:	1✔
10	sequence = self.sequence	1✔
11	self.sequence = self.sequence_before	1✔
12	self._objectives_before = self.objectives_evaluations()	1✔
13	self.sequence = sequence	1✔
14	return self._objectives_before	1✔
15
16	def objectives_evaluations(self):	1✔
17	"""Return a list of the evaluation of each objective of the problem"""
18	return ProblemObjectivesEvaluations.from_problem(self)	1✔
19
20	def objective_scores_sum(self):	1✔
21	return self.objectives_evaluations().scores_sum()	1✔
22
23	def objectives_text_summary(self):	1✔
24	return self.objectives_evaluations().to_text()	×
25
26	def optimize_by_exhaustive_search(self):	1✔
27	""" """
28	if not self.all_constraints_pass():	1✔
UNCOV 29	summary = self.constraints_text_summary(failed_only=True)	×
30	raise NoSolutionError(	×
31	summary + "Optimization can only be done when all constraints are "
32	"verified.",
33	self,
34	)
35
36	if all([obj.best_possible_score is not None for obj in self.objectives]):	1✔
37	best_possible_score = sum(	1✔
38	[obj.best_possible_score for obj in self.objectives]
39	)
40	else:
UNCOV 41	best_possible_score = None	×
42
43	current_best_score = self.objective_scores_sum()	1✔
44	current_best_sequence = self.sequence	1✔
45	all_variants = self.mutation_space.all_variants(self.sequence)	1✔
46	space_size = int(self.mutation_space.space_size)	1✔
47	self.logger(mutation__total=space_size)	1✔
48	for variant in self.logger.iter_bar(mutation=all_variants):	1✔
49	self.sequence = variant	1✔
50	if self.all_constraints_pass():	1✔
51	score = self.objective_scores_sum()	1✔
52	if score > current_best_score:	1✔
53	current_best_score = score	1✔
54	current_best_sequence = self.sequence	1✔
55	if (best_possible_score is not None) and (	1✔
56	current_best_score >= best_possible_score
57	):
58	self.logger(mutation__index=space_size)	1✔
59	break	1✔
60	self.sequence = current_best_sequence	1✔
61
62	def optimize_by_random_mutations(self):	1✔
63	""" """
64	if not self.all_constraints_pass():	1✔
UNCOV 65	summary = self.constraints_text_summary()	×
UNCOV 66	raise ValueError(	×
67	summary + "Optimization can only be done when all"
68	" constraints are verified"
69	)
70	score = self.objective_scores_sum()	1✔
71
72	if all([obj.best_possible_score is not None for obj in self.objectives]):	1✔
73	best_possible_score = sum(	1✔
74	[obj.best_possible_score * obj.boost for obj in self.objectives]
75	)
76	else:
77	best_possible_score = None	1✔
78	iters = self.max_random_iters	1✔
79	stagnating_iterations = 0	1✔
80	for iteration in self.logger.iter_bar(mutation=range(iters)):	1✔
81	if (best_possible_score is not None) and (score >= best_possible_score):	1✔
82	self.logger(mutation__index=iters)	1✔
83	break	1✔
84	if (self.optimization_stagnation_tolerance is not None) and (	1✔
85	stagnating_iterations > self.optimization_stagnation_tolerance
86	):
87	break	1✔
88
89	previous_sequence = self.sequence	1✔
90	self.sequence = self.mutation_space.apply_random_mutations(	1✔
91	n_mutations=self.mutations_per_iteration,
92	sequence=self.sequence,
93	)
94	if self.all_constraints_pass():	1✔
95	new_score = self.objective_scores_sum()	1✔
96	if new_score > score:	1✔
97	score = new_score	1✔
98	stagnating_iterations = 0	1✔
99	else:
100	self.sequence = previous_sequence	1✔
101	else:
102	self.sequence = previous_sequence	1✔
103	stagnating_iterations += 1	1✔
104
105	def optimize_objective(self, objective):	1✔
106	"""Optimize the total objective score, focusing on a single objective.
107
108	This method will attempt to increase the global objective score by
109	focusing on a single objective. First the locations of under-optimal
110	subsequences for this objective are identified, then these locations
111	are optimized one after the other, left to right.
112
113	For each location, a local problem is created and the optimization uses
114	either a custom optimization algorithm, an exhaustive search, or a
115	random search, to optimize the local problem
116	"""
117	# EVALUATE OBJECTIVE. RETURN IF THERE IS NOTHING TO BE DONE.
118	evaluation = objective.evaluate(self)	1✔
119	locations = evaluation.locations	1✔
120	if (objective.best_possible_score is not None) and (	1✔
121	evaluation.score == objective.best_possible_score
122	):
123	return	1✔
124
125	# FOR EACH LOCATION, CREATE AND OPTIMIZE A LOCAL PROBLEM.
126
127	for location in self.logger.iter_bar(	1✔
128	location=locations, bar_message=lambda l: str(l)
129	):
130	# Localize the mutation space by freezing any nucleotide outside of
131	# it
132	mutation_space = self.mutation_space.localized(location)	1✔
133	if mutation_space.space_size == 0:	1✔
134	continue	1✔
135
136	# Update the location so it matches the span of the mutation_space
137	# the resulting location will be equal or smaller to the original
138	# location.
139	location = Location(*mutation_space.choices_span)	1✔
140	localized_constraints = [	1✔
141	cst.localized(location, problem=self) for cst in self.constraints
142	]
143	localized_constraints = [	1✔
144	cst for cst in localized_constraints if cst is not None
145	]
146	localized_objectives = [	1✔
147	obj.localized(location, problem=self)
148	for obj in self.objectives
149	if obj.boost != 0
150	]
151	localized_objectives = [	1✔
152	obj for obj in localized_objectives if obj is not None
153	]
154	local_problem = self.__class__(	1✔
155	sequence=self.sequence,
156	constraints=localized_constraints,
157	mutation_space=mutation_space,
158	objectives=localized_objectives,
159	)
160	self.logger.store(	1✔
161	problem=self, local_problem=local_problem, location=location
162	)
163	local_problem.randomization_threshold = self.randomization_threshold	1✔
164	local_problem.max_random_iters = self.max_random_iters	1✔
165	local_problem.optimization_stagnation_tolerance = (	1✔
166	self.optimization_stagnation_tolerance
167	)
168	local_problem.mutations_per_iteration = self.mutations_per_iteration	1✔
169
170	# OPTIMIZE THE LOCAL PROBLEM
171
172	if hasattr(objective, "optimization_heuristic"):	1✔
173	# Some specifications implement their own optimization method.
UNCOV 174	objective.optimization_heuristic(local_problem)	×
175	else:
176	# Run an exhaustive or random search depending on the size
177	# of the mutation space.
178	space_size = local_problem.mutation_space.space_size	1✔
179	exhaustive_search = space_size < self.randomization_threshold	1✔
180	if exhaustive_search:	1✔
181	local_problem.optimize_by_exhaustive_search()	1✔
182	else:
183	local_problem.optimize_by_random_mutations()	1✔
184
185	# UPDATE THE PROBLEM's SEQUENCE
186
187	self.sequence = local_problem.sequence	1✔
188
189	def optimize(self):	1✔
190	"""Maximize the total score by optimizing each objective in turn."""
191
192	objectives = [	1✔
193	obj
194	for obj in self.objectives
195	if not obj.optimize_passively and obj.boost != 0
196	]
197	if len(objectives) == 0:	1✔
198	return	1✔
199	for objective in self.logger.iter_bar(	1✔
200	objective=objectives, bar_message=lambda o: str(o)
201	):
202	self.optimize_objective(objective=objective)	1✔

Edinburgh-Genome-Foundry / DnaChisel / 14225235949

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous