Coverage for britney2/installability/tester.py: 98%
340 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-18 12:43 +0000
1# Copyright (C) 2012 Niels Thykier <niels@thykier.net>
3# This program is free software; you can redistribute it and/or modify
4# it under the terms of the GNU General Public License as published by
5# the Free Software Foundation; either version 2 of the License, or
6# (at your option) any later version.
8# This program is distributed in the hope that it will be useful,
9# but WITHOUT ANY WARRANTY; without even the implied warranty of
10# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11# GNU General Public License for more details.
13import logging
14from collections import defaultdict
15from collections.abc import Iterable, Iterator, MutableSet
16from dataclasses import dataclass, field
17from functools import partial
18from itertools import chain, filterfalse
19from typing import (
20 TYPE_CHECKING,
21 Literal,
22 cast,
23)
25from more_itertools import iter_except
27from britney2.utils import add_transitive_dependencies_flatten
29if TYPE_CHECKING: 29 ↛ 30line 29 didn't jump to line 30 because the condition on line 29 was never true
30 from .. import BinaryPackageId
31 from .universe import BinaryPackageUniverse
34class InstallabilityTester:
35 def __init__(
36 self, universe: "BinaryPackageUniverse", suite_contents: set["BinaryPackageId"]
37 ) -> None:
38 """Create a new installability tester
40 suite_contents is a (mutable) set of package ids that determines
41 which of the packages in universe are currently in the suite.
43 Package id: (pkg_name, pkg_version, pkg_arch)
44 - NB: arch:all packages are "re-mapped" to given architecture.
45 (simplifies caches and dependency checking)
46 """
48 self._universe = universe
49 # FIXME: Move this field to TargetSuite
50 self._suite_contents = suite_contents
51 self._stats = InstallabilityStats()
52 logger_name = ".".join((self.__class__.__module__, self.__class__.__name__))
53 self.logger = logging.getLogger(logger_name)
55 # Cache of packages known to be broken - we deliberately do not
56 # include "broken" in it.
57 self._cache_broken: set["BinaryPackageId"] = set()
58 # Cache of packages known to be installable
59 self._cache_inst: set["BinaryPackageId"] = set()
60 # Per "arch" cache of the "minimal" (possibly incomplete)
61 # pseudo-essential set. This includes all the packages that
62 # are essential and packages that will always follow.
63 #
64 # It may not be a complete essential set, since alternatives
65 # are not always resolved. Noticeably cases like "awk" may be
66 # left out (since it could be either gawk, mawk or
67 # original-awk) unless something in this sets depends strictly
68 # on one of them
69 self._cache_ess: dict[
70 str,
71 tuple[
72 frozenset["BinaryPackageId"],
73 frozenset["BinaryPackageId"],
74 frozenset[frozenset["BinaryPackageId"]],
75 ],
76 ] = {}
78 essential_w_transitive_deps: MutableSet["BinaryPackageId"] = set(
79 universe.essential_packages
80 )
81 add_transitive_dependencies_flatten(universe, essential_w_transitive_deps)
82 self._cache_essential_transitive_dependencies = essential_w_transitive_deps
84 def compute_installability(self) -> None:
85 """Computes the installability of all the packages in the suite
87 This method computes the installability of all packages in
88 the suite and caches the result. This has the advantage of
89 making "is_installable" queries very fast for all packages
90 in the suite.
91 """
93 universe = self._universe
94 check_inst = self._check_inst
95 cbroken = self._cache_broken
96 cache_inst = self._cache_inst
97 suite_contents = self._suite_contents
98 tcopy = [x for x in suite_contents]
99 for t in filterfalse(cache_inst.__contains__, tcopy):
100 if t in cbroken: 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true
101 continue
102 res = check_inst(t)
103 if t in universe.equivalent_packages:
104 if res:
105 cache_inst.update(
106 x
107 for x in universe.packages_equivalent_to(t)
108 if x in suite_contents
109 )
110 else:
111 eqv_set = universe.packages_equivalent_to(t) & suite_contents
112 suite_contents -= eqv_set
113 cbroken |= eqv_set
115 @property
116 def stats(self) -> "InstallabilityStats":
117 return self._stats
119 def any_of_these_are_in_the_suite(self, pkgs: Iterable["BinaryPackageId"]) -> bool:
120 """Test if at least one package of a given set is in the suite
122 :param pkgs: A set of package ids (as defined in the constructor)
123 :return: True if any of the packages in pkgs are currently in the suite
124 """
125 return not self._suite_contents.isdisjoint(pkgs)
127 def is_pkg_in_the_suite(self, pkg_id: "BinaryPackageId") -> bool:
128 """Test if the package of is in the suite
130 :param pkg_id: A package id (as defined in the constructor)
131 :return: True if the pkg is currently in the suite
132 """
133 return pkg_id in self._suite_contents
135 def which_of_these_are_in_the_suite(
136 self, pkgs: Iterable["BinaryPackageId"]
137 ) -> Iterator["BinaryPackageId"]:
138 """Iterate over all packages that are in the suite
140 :param pkgs: An iterable of package ids
141 :return: An iterable of package ids that are in the suite
142 """
143 yield from (x for x in pkgs if x in self._suite_contents)
145 def add_binary(self, pkg_id: "BinaryPackageId") -> Literal[True]:
146 """Add a binary package to the suite
148 If the package is not known, this method will throw an
149 KeyError.
150 """
152 if pkg_id not in self._universe: # pragma: no cover
153 raise KeyError(str(pkg_id))
155 if pkg_id in self._universe.broken_packages:
156 self._suite_contents.add(pkg_id)
157 elif pkg_id not in self._suite_contents: 157 ↛ 174line 157 didn't jump to line 174 because the condition on line 157 was always true
158 self._suite_contents.add(pkg_id)
159 if self._cache_inst:
160 self._stats.cache_drops += 1
161 self._cache_inst = set()
162 if self._cache_broken:
163 # Re-add broken packages as some of them may now be installable
164 self._suite_contents |= self._cache_broken
165 self._cache_broken = set()
166 if (
167 pkg_id in self._cache_essential_transitive_dependencies
168 and pkg_id.architecture in self._cache_ess
169 ):
170 # Adds new possibly pseudo-essential => "pseudo-essential" set needs to be
171 # recomputed
172 del self._cache_ess[pkg_id.architecture]
174 return True
176 def remove_binary(self, pkg_id: "BinaryPackageId") -> Literal[True]:
177 """Remove a binary from the suite
179 :param pkg_id: The id of the package
180 :raises KeyError: if the package is not known
181 """
183 if pkg_id not in self._universe: # pragma: no cover
184 raise KeyError(str(pkg_id))
186 self._cache_broken.discard(pkg_id)
188 if pkg_id in self._suite_contents:
189 self._suite_contents.remove(pkg_id)
190 if pkg_id.architecture in self._cache_ess:
191 (start, ess_never, ess_choices) = self._cache_ess[pkg_id.architecture]
192 if pkg_id in start or any(pkg_id in choices for choices in ess_choices):
193 # Removes a package from the "pseudo-essential set"
194 del self._cache_ess[pkg_id.architecture]
196 if not self._universe.reverse_dependencies_of(pkg_id):
197 # no reverse relations - safe
198 return True
199 if (
200 pkg_id not in self._universe.broken_packages
201 and pkg_id in self._cache_inst
202 ):
203 # It is in our cache (and not guaranteed to be broken) - throw out the cache
204 self._cache_inst = set()
205 self._stats.cache_drops += 1
207 return True
209 def is_installable(self, pkg_id: "BinaryPackageId") -> bool:
210 """Test if a package is installable in this package set
212 The package is assumed to be in the suite and only packages in
213 the suite can be used to satisfy relations.
215 :param pkg_id: The id of the package
216 Returns True iff the package is installable.
217 Returns False otherwise.
218 """
220 self._stats.is_installable_calls += 1
222 if pkg_id not in self._universe: # pragma: no cover
223 raise KeyError(str(pkg_id))
225 if (
226 pkg_id not in self._suite_contents
227 or pkg_id in self._universe.broken_packages
228 ):
229 self._stats.cache_hits += 1
230 return False
232 if pkg_id in self._cache_inst:
233 self._stats.cache_hits += 1
234 return True
236 self._stats.cache_misses += 1
237 return self._check_inst(pkg_id)
239 def _check_inst(
240 self,
241 t: "BinaryPackageId",
242 musts: set["BinaryPackageId"] | None = None,
243 never: set["BinaryPackageId"] | None = None,
244 choices: set[frozenset["BinaryPackageId"]] | None = None,
245 ) -> bool:
246 # See the explanation of musts, never and choices below.
247 stats = self._stats
248 universe = self._universe
249 suite_contents = self._suite_contents
250 cbroken = self._cache_broken
252 # Our installability verdict - start with "yes" and change if
253 # prove otherwise.
254 verdict = True
256 # set of packages that must be installed with this package
257 if musts is None:
258 musts = set()
259 musts.add(t)
260 # set of packages we can *never* choose (e.g. due to conflicts)
261 if never is None:
262 never = set()
263 # set of relations were we have a choice, but where we have not
264 # committed ourselves yet. Hopefully some choices may be taken
265 # for us (if one of the alternatives appear in "musts")
266 if choices is None:
267 choices = set()
269 # The subset of musts we haven't checked yet.
270 check = [t]
272 if len(musts) == 1:
273 # Include the essential packages in the suite as a starting point.
274 if t.architecture not in self._cache_ess:
275 # The minimal essential set cache is not present -
276 # compute it now.
277 (start, ess_never, ess_choices) = self._get_min_pseudo_ess_set(
278 t.architecture
279 )
280 else:
281 (start, ess_never, ess_choices) = self._cache_ess[t.architecture]
283 if t in ess_never:
284 # t conflicts with something in the essential set or the essential
285 # set conflicts with t - either way, t is f***ed
286 cbroken.add(t)
287 suite_contents.remove(t)
288 stats.conflicts_essential += 1
289 return False
290 musts.update(start)
291 never.update(ess_never)
292 choices.update(ess_choices)
294 # curry check_loop
295 check_loop = partial(
296 self._check_loop, universe, suite_contents, stats, musts, never, cbroken
297 )
299 # Useful things to remember:
300 #
301 # * musts and never are disjointed at all times
302 # - if not, t cannot be installable. Either t, or one of
303 # its dependencies conflict with t or one of its (other)
304 # dependencies.
305 #
306 # * choices should generally be avoided as much as possible.
307 # - picking a bad choice requires backtracking
308 # - sometimes musts/never will eventually "solve" the choice.
309 #
310 # * check never includes choices (these are always in choices)
311 #
312 # * A package is installable if never and musts are disjointed
313 # and both check and choices are empty.
314 # - exception: resolve_choices may determine the installability
315 # of t via recursion (calls _check_inst). In this case
316 # check and choices are not (always) empty.
318 def _prune_choices(rebuild: set[frozenset["BinaryPackageId"]]) -> bool:
319 """Picks a choice from choices and updates rebuild.
321 Prunes the choices and updates "rebuild" to reflect the
322 pruned choices.
324 Returns True if t is installable (determined via recursion).
325 Returns False if a choice was picked and added to check.
326 Returns None if t is uninstallable (no choice can be picked).
328 NB: If this returns False, choices should be replaced by
329 rebuild.
330 """
332 assert musts is not None
333 assert choices is not None
334 assert never is not None
335 # We already satisfied/chosen at least one of the literals
336 # in the choice, so the choice is gone
337 for choice in filter(musts.isdisjoint, choices):
338 # cbroken is needed here because (in theory) it could
339 # have changed since the choice was discovered and it
340 # is smaller than suite_contents (so presumably faster)
341 remain = choice - never - cbroken
343 if len(remain) == 1:
344 # the choice was reduced to one package we haven't checked - check that
345 check.extend(remain)
346 musts.update(remain)
347 stats.choice_presolved += 1
348 continue
350 if not remain:
351 # all alternatives would violate the conflicts or are uninstallable
352 # => package is not installable
353 stats.choice_presolved += 1
354 return False
356 # The choice is still deferred
357 rebuild.add(frozenset(remain))
359 return True
361 # END _prune_choices
363 while check:
364 if not check_loop(choices, check):
365 verdict = False
366 break
368 if choices:
369 rebuild: set[frozenset["BinaryPackageId"]] = set()
371 if not _prune_choices(rebuild):
372 verdict = False
373 break
375 if not check and rebuild:
376 # We have to "guess" now, which is always fun, but not cheap. We
377 # stop guessing:
378 # - once we run out of choices to make (obviously), OR
379 # - if one of the choices exhaust all but one option
380 if self.resolve_choices(check, musts, never, rebuild):
381 # The recursive call have already updated the
382 # cache so there is not point in doing it again.
383 return True
384 choices = rebuild
386 if verdict:
387 # if t is installable, then so are all packages in musts
388 self._cache_inst.update(musts)
389 stats.solved_installable += 1
390 else:
391 stats.solved_uninstallable += 1
393 return verdict
395 def resolve_choices(
396 self,
397 check: list["BinaryPackageId"],
398 musts: set["BinaryPackageId"],
399 never: set["BinaryPackageId"],
400 choices: set[frozenset["BinaryPackageId"]],
401 ) -> bool:
402 universe = self._universe
403 suite_contents = self._suite_contents
404 stats = self._stats
405 cbroken = self._cache_broken
407 while choices:
408 choice_options = choices.pop()
410 choice = iter(choice_options)
411 last = next(choice) # pick one to go last
412 solved = False
413 for p in choice:
414 musts_copy = musts.copy()
415 never_tmp: set["BinaryPackageId"] = set()
416 choices_tmp: set[frozenset["BinaryPackageId"]] = set()
417 check_tmp = [p]
418 # _check_loop assumes that "musts" is up to date
419 musts_copy.add(p)
420 if not self._check_loop(
421 universe,
422 suite_contents,
423 stats,
424 musts_copy,
425 never_tmp,
426 cbroken,
427 choices_tmp,
428 check_tmp,
429 ):
430 # p cannot be chosen/is broken (unlikely, but ...)
431 continue
433 # Test if we can pick p without any consequences.
434 # - when we can, we avoid a backtrack point.
435 if never_tmp <= never and choices_tmp <= choices:
436 # we can pick p without picking up new conflicts
437 # or unresolved choices. Therefore we commit to
438 # using p.
439 musts.update(musts_copy)
440 stats.choice_resolved_without_restore_point += 1
441 solved = True
442 break
444 if not musts.isdisjoint(never_tmp):
445 # If we pick p, we will definitely end up making
446 # t uninstallable, so p is a no-go.
447 continue
449 stats.backtrace_restore_point_created += 1
450 # We are not sure that p is safe, setup a backtrack
451 # point and recurse.
452 never_tmp |= never
453 choices_tmp |= choices
454 if self._check_inst(p, musts_copy, never_tmp, choices_tmp):
455 # Success, p was a valid choice and made it all
456 # installable
457 return True
459 # If we get here, we failed to find something that
460 # would satisfy choice (without breaking the
461 # installability of t). This means p cannot be used
462 # to satisfy the dependencies, so pretend to conflict
463 # with it - hopefully it will reduce future choices.
464 never.add(p)
465 stats.backtrace_restore_point_used += 1
467 if not solved:
468 # Optimization for the last case; avoid the recursive call
469 # and just assume the last will lead to a solution. If it
470 # doesn't there is no solution and if it does, we don't
471 # have to back-track anyway.
472 check.append(last)
473 musts.add(last)
474 stats.backtrace_last_option += 1
475 return False
476 return False
478 def _check_loop(
479 self,
480 universe: "BinaryPackageUniverse",
481 suite_contents: set["BinaryPackageId"],
482 stats: "InstallabilityStats",
483 musts: set["BinaryPackageId"],
484 never: set["BinaryPackageId"],
485 cbroken: set["BinaryPackageId"],
486 choices: set[frozenset["BinaryPackageId"]],
487 check: list["BinaryPackageId"],
488 ) -> bool:
489 """Finds all guaranteed dependencies via "check".
491 If it returns False, t is not installable. If it returns True
492 then "check" is exhausted. If "choices" are empty and this
493 returns True, then t is installable.
494 """
495 # Local variables for faster access...
496 not_satisfied: partial[filter["BinaryPackageId"]] = partial(
497 filter, musts.isdisjoint
498 )
500 # While we have guaranteed dependencies (in check), examine all
501 # of them.
502 for cur in iter_except(check.pop, IndexError):
503 relations = universe.relations_of(cur)
505 if relations.negative_dependencies:
506 # Conflicts?
507 if cur in never:
508 # cur adds a (reverse) conflict, so check if cur
509 # is in never.
510 #
511 # - there is a window where two conflicting
512 # packages can be in check. Example "A" depends
513 # on "B" and "C". If "B" conflicts with "C",
514 # then both "B" and "C" could end in "check".
515 return False
516 # We must install cur for the package to be installable,
517 # so "obviously" we can never choose any of its conflicts
518 never |= relations.negative_dependencies & suite_contents
520 # depgroup can be satisfied by picking something that is
521 # already in musts - lets pick that (again). :)
522 for depgroup in cast(
523 set[frozenset["BinaryPackageId"]],
524 not_satisfied(relations.dependencies),
525 ):
526 # Of all the packages listed in the relation remove those that
527 # are either:
528 # - not in the suite
529 # - known to be broken (by cache)
530 # - in never
531 candidates = suite_contents & depgroup - never
533 if not candidates:
534 # We got no candidates to satisfy it - this
535 # package cannot be installed with the current
536 # (version of the) suite
537 if cur not in cbroken and depgroup.isdisjoint(never):
538 # cur's dependency cannot be satisfied even if never was empty.
539 # This means that cur itself is broken (as well).
540 cbroken.add(cur)
541 suite_contents.remove(cur)
542 return False
543 if len(candidates) == 1:
544 # only one possible solution to this choice and we
545 # haven't seen it before
546 check.extend(candidates)
547 musts |= candidates
548 else:
549 possible_eqv = candidates & universe.equivalent_packages
550 if len(possible_eqv) > 1:
551 # Exploit equivalency to reduce the number of
552 # candidates if possible. Basically, this
553 # code maps "similar" candidates into a single
554 # candidate that will give a identical result
555 # to any other candidate it eliminates.
556 #
557 # See InstallabilityTesterBuilder's
558 # _build_eqv_packages_table method for more
559 # information on how this works.
560 new_cand = candidates - possible_eqv
561 stats.eqv_table_times_used += 1
563 for chosen in iter_except(possible_eqv.pop, KeyError):
564 new_cand.add(chosen)
565 possible_eqv -= universe.packages_equivalent_to(chosen)
566 stats.eqv_table_total_number_of_alternatives_eliminated += len(
567 candidates
568 ) - len(new_cand)
569 if len(new_cand) == 1:
570 check.extend(new_cand)
571 musts |= new_cand
572 stats.eqv_table_reduced_to_one += 1
573 continue
574 elif len(candidates) == len(new_cand):
575 stats.eqv_table_reduced_by_zero += 1
577 candidates = new_cand
578 # defer this choice till later
579 choices.add(frozenset(candidates))
580 return True
582 def _get_min_pseudo_ess_set(self, arch: str) -> tuple[
583 frozenset["BinaryPackageId"],
584 frozenset["BinaryPackageId"],
585 frozenset[frozenset["BinaryPackageId"]],
586 ]:
587 if arch not in self._cache_ess: 587 ↛ 648line 587 didn't jump to line 648 because the condition on line 587 was always true
588 # The minimal essential set cache is not present -
589 # compute it now.
590 suite_contents = self._suite_contents
591 cbroken = self._cache_broken
592 universe = self._universe
593 stats = self._stats
595 ess_base = [
596 x
597 for x in self._universe.essential_packages
598 if x.architecture == arch and x in suite_contents
599 ]
600 start = set(ess_base)
601 ess_never: set["BinaryPackageId"] = set()
602 ess_choices: set[frozenset["BinaryPackageId"]] = set()
603 not_satisfied: partial[filter["BinaryPackageId"]] = partial(
604 filter, start.isdisjoint
605 )
607 while ess_base:
608 self._check_loop(
609 universe,
610 suite_contents,
611 stats,
612 start,
613 ess_never,
614 cbroken,
615 ess_choices,
616 ess_base,
617 )
618 if ess_choices:
619 # Try to break choices where possible
620 nchoice = set()
621 for choice in cast(
622 set[frozenset["BinaryPackageId"]], not_satisfied(ess_choices)
623 ):
624 b = False
625 for c in choice:
626 relations = universe.relations_of(c)
627 if (
628 relations.negative_dependencies is None
629 or relations.negative_dependencies <= ess_never
630 ) and not any(not_satisfied(relations.dependencies)):
631 ess_base.append(c)
632 b = True
633 break
634 if not b:
635 nchoice.add(choice)
636 ess_choices = nchoice
637 else:
638 break
640 for x in start:
641 ess_never.update(universe.negative_dependencies_of(x))
642 self._cache_ess[arch] = (
643 frozenset(start),
644 frozenset(ess_never),
645 frozenset(ess_choices),
646 )
648 return self._cache_ess[arch]
650 def compute_stats(self) -> dict[str, "ArchStats"]:
651 universe = self._universe
652 graph_stats: dict[str, ArchStats] = defaultdict(ArchStats)
653 seen_eqv: dict[str, set["BinaryPackageId"]] = defaultdict(set)
655 for pkg in universe:
656 pkg_arch = pkg.architecture
657 relations = universe.relations_of(pkg)
658 arch_stats = graph_stats[pkg_arch]
660 arch_stats.nodes += 1
662 if pkg in universe.equivalent_packages and pkg not in seen_eqv[pkg_arch]:
663 arch_stats.eqv_nodes += sum(
664 1
665 for e in universe.packages_equivalent_to(pkg)
666 if e.architecture == pkg_arch
667 )
669 arch_stats.add_dep_edges(relations.dependencies)
670 if relations.negative_dependencies is not None:
671 arch_stats.add_con_edges(relations.negative_dependencies)
673 for stat in graph_stats.values():
674 stat.compute_all()
676 return graph_stats
679@dataclass(slots=True)
680class InstallabilityStats:
681 cache_hits: int = 0
682 cache_misses: int = 0
683 cache_drops: int = 0
684 backtrace_restore_point_created: int = 0
685 backtrace_restore_point_used: int = 0
686 backtrace_last_option: int = 0
687 choice_presolved: int = 0
688 choice_resolved_without_restore_point: int = 0
689 is_installable_calls: int = 0
690 solved_installable: int = 0
691 solved_uninstallable: int = 0
692 conflicts_essential: int = 0
693 eqv_table_times_used: int = 0
694 eqv_table_reduced_to_one: int = 0
695 eqv_table_reduced_by_zero: int = 0
696 eqv_table_total_number_of_alternatives_eliminated: int = 0
698 def stats(self) -> list[str]:
699 return [
700 f"Requests - is_installable: {self.is_installable_calls}",
701 f"Cache - hits: {self.cache_hits}, misses: {self.cache_misses}, drops: {self.cache_drops}",
702 f"Choices - pre-solved: {self.choice_presolved}, No RP: {self.choice_resolved_without_restore_point}",
703 f"Backtrace - RP created: {self.backtrace_restore_point_created}, RP used: {self.backtrace_restore_point_used}, reached last option: {self.backtrace_last_option}", # nopep8
704 f"Solved - installable: {self.solved_installable}, uninstallable: {self.solved_uninstallable}, conflicts essential: {self.conflicts_essential}", # nopep8
705 f"Eqv - times used: {self.eqv_table_times_used}, perfect reductions: {self.eqv_table_reduced_to_one}, failed reductions: {self.eqv_table_reduced_by_zero}, total no. of alternatives pruned: {self.eqv_table_total_number_of_alternatives_eliminated}", # nopep8
706 ]
709@dataclass(slots=True)
710class _Stats:
711 max: int = 0
712 min: int = 0
713 median: int = 0
714 average: float = 0
715 sum: int = 0
716 size: int = 0
717 average_per_node: float | None = None
719 def __str__(self) -> str:
720 s = f"max: {self.max}, min: {self.min}, median: {self.median}, average: {self.average:.6f} ({self.sum}/{self.size})"
721 if self.average_per_node is not None:
722 s = f"{s}, average-per-node: {self.average_per_node:.6f}"
723 return s
726@dataclass(slots=True)
727class ArchStats:
728 nodes: int = 0
729 eqv_nodes: int = 0
730 dep_edges: list[frozenset[frozenset["BinaryPackageId"]]] = field(
731 default_factory=list
732 )
733 con_edges: list[frozenset["BinaryPackageId"]] = field(default_factory=list)
734 dependency_clauses: _Stats = field(default_factory=_Stats)
735 dependency_clause_alternatives: _Stats = field(default_factory=_Stats)
736 negative_dependency_clauses: _Stats = field(default_factory=_Stats)
738 def stat_summary(self) -> list[str]:
739 return [
740 f"nodes: {self.nodes}, eqv-nodes: {self.eqv_nodes}",
741 f"dependency-clauses, {self.dependency_clauses}",
742 f"dependency-clause-alternatives, {self.dependency_clause_alternatives}",
743 f"negative-dependency-clauses, {self.negative_dependency_clauses}",
744 ]
746 def add_dep_edges(self, edges: frozenset[frozenset["BinaryPackageId"]]) -> None:
747 self.dep_edges.append(edges)
749 def add_con_edges(self, edges: frozenset["BinaryPackageId"]) -> None:
750 self.con_edges.append(edges)
752 def _list_stats(
753 self, stats: _Stats, sorted_list: list[int], average_per_node: bool = False
754 ) -> None:
755 if sorted_list:
756 stats.max = sorted_list[-1]
757 stats.min = sorted_list[0]
758 stats.sum = sum(sorted_list)
759 stats.size = len(sorted_list)
760 stats.average = float(stats.sum) / len(sorted_list)
761 stats.median = sorted_list[len(sorted_list) // 2]
762 if average_per_node:
763 stats.average_per_node = float(stats.sum) / self.nodes
765 def compute_all(self) -> None:
766 dep_edges = self.dep_edges
767 con_edges = self.con_edges
768 sorted_no_dep_edges = sorted(len(x) for x in dep_edges)
769 sorted_size_dep_edges = sorted(len(x) for x in chain.from_iterable(dep_edges))
770 sorted_no_con_edges = sorted(len(x) for x in con_edges)
771 self._list_stats(self.dependency_clauses, sorted_no_dep_edges)
772 self._list_stats(
773 self.dependency_clause_alternatives,
774 sorted_size_dep_edges,
775 average_per_node=True,
776 )
777 self._list_stats(self.negative_dependency_clauses, sorted_no_con_edges)