Coverage for britney2/installability/tester.py: 98%
341 statements
« prev ^ index » next coverage.py v7.6.0, created at 2025-10-17 17:32 +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 builtins import frozenset as _frozenset
15from collections import defaultdict
16from collections.abc import Callable, Iterable, Iterator, MutableSet, Sized
17from functools import partial
18from itertools import chain, filterfalse
19from typing import (
20 TYPE_CHECKING,
21 Any,
22 Literal,
23 Optional,
24 Union,
25 cast,
26)
28from britney2.utils import add_transitive_dependencies_flatten, iter_except
30if TYPE_CHECKING: 30 ↛ 31line 30 didn't jump to line 31 because the condition on line 30 was never true
31 from .. import BinaryPackageId
32 from .universe import BinaryPackageUniverse
35class InstallabilityTester:
36 def __init__(
37 self, universe: "BinaryPackageUniverse", suite_contents: set["BinaryPackageId"]
38 ) -> None:
39 """Create a new installability tester
41 suite_contents is a (mutable) set of package ids that determines
42 which of the packages in universe are currently in the suite.
44 Package id: (pkg_name, pkg_version, pkg_arch)
45 - NB: arch:all packages are "re-mapped" to given architecture.
46 (simplifies caches and dependency checking)
47 """
49 self._universe = universe
50 # FIXME: Move this field to TargetSuite
51 self._suite_contents = suite_contents
52 self._stats = InstallabilityStats()
53 logger_name = ".".join((self.__class__.__module__, self.__class__.__name__))
54 self.logger = logging.getLogger(logger_name)
56 # Cache of packages known to be broken - we deliberately do not
57 # include "broken" in it. See _optimize for more info.
58 self._cache_broken: set["BinaryPackageId"] = set()
59 # Cache of packages known to be installable
60 self._cache_inst: set["BinaryPackageId"] = set()
61 # Per "arch" cache of the "minimal" (possibly incomplete)
62 # pseudo-essential set. This includes all the packages that
63 # are essential and packages that will always follow.
64 #
65 # It may not be a complete essential set, since alternatives
66 # are not always resolved. Noticeably cases like "awk" may be
67 # left out (since it could be either gawk, mawk or
68 # original-awk) unless something in this sets depends strictly
69 # on one of them
70 self._cache_ess: dict[
71 str,
72 tuple[
73 frozenset["BinaryPackageId"],
74 frozenset["BinaryPackageId"],
75 frozenset[frozenset["BinaryPackageId"]],
76 ],
77 ] = {}
79 essential_w_transitive_deps: MutableSet["BinaryPackageId"] = set(
80 universe.essential_packages
81 )
82 add_transitive_dependencies_flatten(universe, essential_w_transitive_deps)
83 self._cache_essential_transitive_dependencies = essential_w_transitive_deps
85 def compute_installability(self) -> None:
86 """Computes the installability of all the packages in the suite
88 This method computes the installability of all packages in
89 the suite and caches the result. This has the advantage of
90 making "is_installable" queries very fast for all packages
91 in the suite.
92 """
94 universe = self._universe
95 check_inst = self._check_inst
96 cbroken = self._cache_broken
97 cache_inst = self._cache_inst
98 suite_contents = self._suite_contents
99 tcopy = [x for x in suite_contents]
100 for t in filterfalse(cache_inst.__contains__, tcopy):
101 if t in cbroken: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true
102 continue
103 res = check_inst(t)
104 if t in universe.equivalent_packages:
105 eqv = (
106 x for x in universe.packages_equivalent_to(t) if x in suite_contents
107 )
108 if res:
109 cache_inst.update(eqv)
110 else:
111 eqv_set = frozenset(eqv)
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(
193 [pkg_id in choices for choices in ess_choices]
194 ):
195 # Removes a package from the "pseudo-essential set"
196 del self._cache_ess[pkg_id.architecture]
198 if not self._universe.reverse_dependencies_of(pkg_id):
199 # no reverse relations - safe
200 return True
201 if (
202 pkg_id not in self._universe.broken_packages
203 and pkg_id in self._cache_inst
204 ):
205 # It is in our cache (and not guaranteed to be broken) - throw out the cache
206 self._cache_inst = set()
207 self._stats.cache_drops += 1
209 return True
211 def is_installable(self, pkg_id: "BinaryPackageId") -> bool:
212 """Test if a package is installable in this package set
214 The package is assumed to be in the suite and only packages in
215 the suite can be used to satisfy relations.
217 :param pkg_id: The id of the package
218 Returns True iff the package is installable.
219 Returns False otherwise.
220 """
222 self._stats.is_installable_calls += 1
224 if pkg_id not in self._universe: # pragma: no cover
225 raise KeyError(str(pkg_id))
227 if (
228 pkg_id not in self._suite_contents
229 or pkg_id in self._universe.broken_packages
230 ):
231 self._stats.cache_hits += 1
232 return False
234 if pkg_id in self._cache_inst:
235 self._stats.cache_hits += 1
236 return True
238 self._stats.cache_misses += 1
239 return self._check_inst(pkg_id)
241 def _check_inst(
242 self,
243 t: "BinaryPackageId",
244 musts: set["BinaryPackageId"] | None = None,
245 never: set["BinaryPackageId"] | None = None,
246 choices: set[frozenset["BinaryPackageId"]] | None = None,
247 ) -> bool:
248 # See the explanation of musts, never and choices below.
249 stats = self._stats
250 universe = self._universe
251 suite_contents = self._suite_contents
252 cbroken = self._cache_broken
254 # Our installability verdict - start with "yes" and change if
255 # prove otherwise.
256 verdict = True
258 # set of packages that must be installed with this package
259 if musts is None:
260 musts = set()
261 musts.add(t)
262 # set of packages we can *never* choose (e.g. due to conflicts)
263 if never is None:
264 never = set()
265 # set of relations were we have a choice, but where we have not
266 # committed ourselves yet. Hopefully some choices may be taken
267 # for us (if one of the alternatives appear in "musts")
268 if choices is None:
269 choices = set()
271 # The subset of musts we haven't checked yet.
272 check = [t]
274 if len(musts) == 1:
275 # Include the essential packages in the suite as a starting point.
276 if t.architecture not in self._cache_ess:
277 # The minimal essential set cache is not present -
278 # compute it now.
279 (start, ess_never, ess_choices) = self._get_min_pseudo_ess_set(
280 t.architecture
281 )
282 else:
283 (start, ess_never, ess_choices) = self._cache_ess[t.architecture]
285 if t in ess_never:
286 # t conflicts with something in the essential set or the essential
287 # set conflicts with t - either way, t is f***ed
288 cbroken.add(t)
289 suite_contents.remove(t)
290 stats.conflicts_essential += 1
291 return False
292 musts.update(start)
293 never.update(ess_never)
294 choices.update(ess_choices)
296 # curry check_loop
297 check_loop = partial(
298 self._check_loop, universe, suite_contents, stats, musts, never, cbroken
299 )
301 # Useful things to remember:
302 #
303 # * musts and never are disjointed at all times
304 # - if not, t cannot be installable. Either t, or one of
305 # its dependencies conflict with t or one of its (other)
306 # dependencies.
307 #
308 # * choices should generally be avoided as much as possible.
309 # - picking a bad choice requires backtracking
310 # - sometimes musts/never will eventually "solve" the choice.
311 #
312 # * check never includes choices (these are always in choices)
313 #
314 # * A package is installable if never and musts are disjointed
315 # and both check and choices are empty.
316 # - exception: resolve_choices may determine the installability
317 # of t via recursion (calls _check_inst). In this case
318 # check and choices are not (always) empty.
320 def _prune_choices(
321 rebuild: set[frozenset["BinaryPackageId"]],
322 len: Callable[[Sized], int] = len,
323 ) -> bool:
324 """Picks a choice from choices and updates rebuild.
326 Prunes the choices and updates "rebuild" to reflect the
327 pruned choices.
329 Returns True if t is installable (determined via recursion).
330 Returns False if a choice was picked and added to check.
331 Returns None if t is uninstallable (no choice can be picked).
333 NB: If this returns False, choices should be replaced by
334 rebuild.
335 """
337 assert musts is not None
338 assert choices is not None
339 assert never is not None
340 # We already satisfied/chosen at least one of the literals
341 # in the choice, so the choice is gone
342 for choice in filter(musts.isdisjoint, choices):
343 # cbroken is needed here because (in theory) it could
344 # have changed since the choice was discovered and it
345 # is smaller than suite_contents (so presumably faster)
346 remain = choice - never - cbroken
348 if len(remain) == 1:
349 # the choice was reduced to one package we haven't checked - check that
350 check.extend(remain)
351 musts.update(remain)
352 stats.choice_presolved += 1
353 continue
355 if not remain:
356 # all alternatives would violate the conflicts or are uninstallable
357 # => package is not installable
358 stats.choice_presolved += 1
359 return False
361 # The choice is still deferred
362 rebuild.add(frozenset(remain))
364 return True
366 # END _prune_choices
368 while check:
369 if not check_loop(choices, check):
370 verdict = False
371 break
373 if choices:
374 rebuild: set[frozenset["BinaryPackageId"]] = set()
376 if not _prune_choices(rebuild):
377 verdict = False
378 break
380 if not check and rebuild:
381 # We have to "guess" now, which is always fun, but not cheap. We
382 # stop guessing:
383 # - once we run out of choices to make (obviously), OR
384 # - if one of the choices exhaust all but one option
385 if self.resolve_choices(check, musts, never, rebuild):
386 # The recursive call have already updated the
387 # cache so there is not point in doing it again.
388 return True
389 choices = rebuild
391 if verdict:
392 # if t is installable, then so are all packages in musts
393 self._cache_inst.update(musts)
394 stats.solved_installable += 1
395 else:
396 stats.solved_uninstallable += 1
398 return verdict
400 def resolve_choices(
401 self,
402 check: list["BinaryPackageId"],
403 musts: set["BinaryPackageId"],
404 never: set["BinaryPackageId"],
405 choices: set[frozenset["BinaryPackageId"]],
406 ) -> bool:
407 universe = self._universe
408 suite_contents = self._suite_contents
409 stats = self._stats
410 cbroken = self._cache_broken
412 while choices:
413 choice_options = choices.pop()
415 choice = iter(choice_options)
416 last = next(choice) # pick one to go last
417 solved = False
418 for p in choice:
419 musts_copy = musts.copy()
420 never_tmp: set["BinaryPackageId"] = set()
421 choices_tmp: set[frozenset["BinaryPackageId"]] = set()
422 check_tmp = [p]
423 # _check_loop assumes that "musts" is up to date
424 musts_copy.add(p)
425 if not self._check_loop(
426 universe,
427 suite_contents,
428 stats,
429 musts_copy,
430 never_tmp,
431 cbroken,
432 choices_tmp,
433 check_tmp,
434 ):
435 # p cannot be chosen/is broken (unlikely, but ...)
436 continue
438 # Test if we can pick p without any consequences.
439 # - when we can, we avoid a backtrack point.
440 if never_tmp <= never and choices_tmp <= choices:
441 # we can pick p without picking up new conflicts
442 # or unresolved choices. Therefore we commit to
443 # using p.
444 musts.update(musts_copy)
445 stats.choice_resolved_without_restore_point += 1
446 solved = True
447 break
449 if not musts.isdisjoint(never_tmp):
450 # If we pick p, we will definitely end up making
451 # t uninstallable, so p is a no-go.
452 continue
454 stats.backtrace_restore_point_created += 1
455 # We are not sure that p is safe, setup a backtrack
456 # point and recurse.
457 never_tmp |= never
458 choices_tmp |= choices
459 if self._check_inst(p, musts_copy, never_tmp, choices_tmp):
460 # Success, p was a valid choice and made it all
461 # installable
462 return True
464 # If we get here, we failed to find something that
465 # would satisfy choice (without breaking the
466 # installability of t). This means p cannot be used
467 # to satisfy the dependencies, so pretend to conflict
468 # with it - hopefully it will reduce future choices.
469 never.add(p)
470 stats.backtrace_restore_point_used += 1
472 if not solved:
473 # Optimization for the last case; avoid the recursive call
474 # and just assume the last will lead to a solution. If it
475 # doesn't there is no solution and if it does, we don't
476 # have to back-track anyway.
477 check.append(last)
478 musts.add(last)
479 stats.backtrace_last_option += 1
480 return False
481 return False
483 def _check_loop(
484 self,
485 universe: "BinaryPackageUniverse",
486 suite_contents: set["BinaryPackageId"],
487 stats: "InstallabilityStats",
488 musts: set["BinaryPackageId"],
489 never: set["BinaryPackageId"],
490 cbroken: set["BinaryPackageId"],
491 choices: set[frozenset["BinaryPackageId"]],
492 check: list["BinaryPackageId"],
493 len: Callable[[Sized], int] = len,
494 frozenset: Callable[..., Any] = frozenset,
495 ) -> bool:
496 """Finds all guaranteed dependencies via "check".
498 If it returns False, t is not installable. If it returns True
499 then "check" is exhausted. If "choices" are empty and this
500 returns True, then t is installable.
501 """
502 # Local variables for faster access...
503 not_satisfied: partial[filter["BinaryPackageId"]] = partial(
504 filter, musts.isdisjoint
505 )
507 # While we have guaranteed dependencies (in check), examine all
508 # of them.
509 for cur in iter_except(check.pop, IndexError):
510 relations = universe.relations_of(cur)
512 if relations.negative_dependencies:
513 # Conflicts?
514 if cur in never:
515 # cur adds a (reverse) conflict, so check if cur
516 # is in never.
517 #
518 # - there is a window where two conflicting
519 # packages can be in check. Example "A" depends
520 # on "B" and "C". If "B" conflicts with "C",
521 # then both "B" and "C" could end in "check".
522 return False
523 # We must install cur for the package to be installable,
524 # so "obviously" we can never choose any of its conflicts
525 never.update(relations.negative_dependencies & suite_contents)
527 # depgroup can be satisfied by picking something that is
528 # already in musts - lets pick that (again). :)
529 for depgroup in cast(
530 set[_frozenset["BinaryPackageId"]],
531 not_satisfied(relations.dependencies),
532 ):
533 # Of all the packages listed in the relation remove those that
534 # are either:
535 # - not in the suite
536 # - known to be broken (by cache)
537 # - in never
538 candidates = (depgroup & suite_contents) - never
540 if not candidates:
541 # We got no candidates to satisfy it - this
542 # package cannot be installed with the current
543 # (version of the) suite
544 if cur not in cbroken and depgroup.isdisjoint(never):
545 # cur's dependency cannot be satisfied even if never was empty.
546 # This means that cur itself is broken (as well).
547 cbroken.add(cur)
548 suite_contents.remove(cur)
549 return False
550 if len(candidates) == 1:
551 # only one possible solution to this choice and we
552 # haven't seen it before
553 check.extend(candidates)
554 musts.update(candidates)
555 else:
556 possible_eqv = {
557 x for x in candidates if x in universe.equivalent_packages
558 }
559 if len(possible_eqv) > 1:
560 # Exploit equivalency to reduce the number of
561 # candidates if possible. Basically, this
562 # code maps "similar" candidates into a single
563 # candidate that will give a identical result
564 # to any other candidate it eliminates.
565 #
566 # See InstallabilityTesterBuilder's
567 # _build_eqv_packages_table method for more
568 # information on how this works.
569 new_cand = {x for x in candidates if x not in possible_eqv}
570 stats.eqv_table_times_used += 1
572 for chosen in iter_except(possible_eqv.pop, KeyError):
573 new_cand.add(chosen)
574 possible_eqv -= universe.packages_equivalent_to(chosen)
575 stats.eqv_table_total_number_of_alternatives_eliminated += len(
576 candidates
577 ) - len(new_cand)
578 if len(new_cand) == 1:
579 check.extend(new_cand)
580 musts.update(new_cand)
581 stats.eqv_table_reduced_to_one += 1
582 continue
583 elif len(candidates) == len(new_cand):
584 stats.eqv_table_reduced_by_zero += 1
586 candidates = frozenset(new_cand)
587 else:
588 # Candidates have to be a frozenset to be added to choices
589 candidates = frozenset(candidates)
590 # defer this choice till later
591 choices.add(candidates)
592 return True
594 def _get_min_pseudo_ess_set(self, arch: str) -> tuple[
595 frozenset["BinaryPackageId"],
596 frozenset["BinaryPackageId"],
597 frozenset[frozenset["BinaryPackageId"]],
598 ]:
599 if arch not in self._cache_ess: 599 ↛ 660line 599 didn't jump to line 660 because the condition on line 599 was always true
600 # The minimal essential set cache is not present -
601 # compute it now.
602 suite_contents = self._suite_contents
603 cbroken = self._cache_broken
604 universe = self._universe
605 stats = self._stats
607 ess_base = [
608 x
609 for x in self._universe.essential_packages
610 if x.architecture == arch and x in suite_contents
611 ]
612 start = set(ess_base)
613 ess_never: set["BinaryPackageId"] = set()
614 ess_choices: set[frozenset["BinaryPackageId"]] = set()
615 not_satisfied: partial[filter["BinaryPackageId"]] = partial(
616 filter, start.isdisjoint
617 )
619 while ess_base:
620 self._check_loop(
621 universe,
622 suite_contents,
623 stats,
624 start,
625 ess_never,
626 cbroken,
627 ess_choices,
628 ess_base,
629 )
630 if ess_choices:
631 # Try to break choices where possible
632 nchoice = set()
633 for choice in cast(
634 set[frozenset["BinaryPackageId"]], not_satisfied(ess_choices)
635 ):
636 b = False
637 for c in choice:
638 relations = universe.relations_of(c)
639 if (
640 relations.negative_dependencies <= ess_never
641 and not any(not_satisfied(relations.dependencies))
642 ):
643 ess_base.append(c)
644 b = True
645 break
646 if not b:
647 nchoice.add(choice)
648 ess_choices = nchoice
649 else:
650 break
652 for x in start:
653 ess_never.update(universe.negative_dependencies_of(x))
654 self._cache_ess[arch] = (
655 frozenset(start),
656 frozenset(ess_never),
657 frozenset(ess_choices),
658 )
660 return self._cache_ess[arch]
662 def compute_stats(self) -> dict[str, "ArchStats"]:
663 universe = self._universe
664 graph_stats: dict[str, ArchStats] = defaultdict(ArchStats)
665 seen_eqv: dict[str, set["BinaryPackageId"]] = defaultdict(set)
667 for pkg in universe:
668 (pkg_name, pkg_version, pkg_arch) = pkg
669 relations = universe.relations_of(pkg)
670 arch_stats = graph_stats[pkg_arch]
672 arch_stats.nodes += 1
674 if pkg in universe.equivalent_packages and pkg not in seen_eqv[pkg_arch]:
675 eqv = [
676 e
677 for e in universe.packages_equivalent_to(pkg)
678 if e.architecture == pkg_arch
679 ]
680 arch_stats.eqv_nodes += len(eqv)
682 arch_stats.add_dep_edges(relations.dependencies)
683 arch_stats.add_con_edges(relations.negative_dependencies)
685 for stat in graph_stats.values():
686 stat.compute_all()
688 return graph_stats
691class InstallabilityStats:
692 def __init__(self) -> None:
693 self.cache_hits = 0
694 self.cache_misses = 0
695 self.cache_drops = 0
696 self.backtrace_restore_point_created = 0
697 self.backtrace_restore_point_used = 0
698 self.backtrace_last_option = 0
699 self.choice_presolved = 0
700 self.choice_resolved_without_restore_point = 0
701 self.is_installable_calls = 0
702 self.solved_installable = 0
703 self.solved_uninstallable = 0
704 self.conflicts_essential = 0
705 self.eqv_table_times_used = 0
706 self.eqv_table_reduced_to_one = 0
707 self.eqv_table_reduced_by_zero = 0
708 self.eqv_table_total_number_of_alternatives_eliminated = 0
710 def stats(self) -> list[str]:
711 formats = [
712 "Requests - is_installable: {is_installable_calls}",
713 "Cache - hits: {cache_hits}, misses: {cache_misses}, drops: {cache_drops}",
714 "Choices - pre-solved: {choice_presolved}, No RP: {choice_resolved_without_restore_point}",
715 "Backtrace - RP created: {backtrace_restore_point_created}, RP used: {backtrace_restore_point_used}, reached last option: {backtrace_last_option}", # nopep8
716 "Solved - installable: {solved_installable}, uninstallable: {solved_uninstallable}, conflicts essential: {conflicts_essential}", # nopep8
717 "Eqv - times used: {eqv_table_times_used}, perfect reductions: {eqv_table_reduced_to_one}, failed reductions: {eqv_table_reduced_by_zero}, total no. of alternatives pruned: {eqv_table_total_number_of_alternatives_eliminated}", # nopep8
718 ]
719 return [x.format(**self.__dict__) for x in formats]
722class ArchStats:
723 def __init__(self) -> None:
724 self.nodes = 0
725 self.eqv_nodes = 0
726 self.dep_edges: list[frozenset[frozenset["BinaryPackageId"]]] = []
727 self.con_edges: list[frozenset["BinaryPackageId"]] = []
728 self.stats: dict[str, dict[str, int | float]] = defaultdict(
729 lambda: defaultdict(int)
730 )
732 def stat(self, statname: str) -> dict[str, int | float]:
733 return self.stats[statname]
735 def stat_summary(self) -> list[str]:
736 text = []
737 values: list[Any] | tuple[Any, ...]
738 for statname in [
739 "nodes",
740 "dependency-clauses",
741 "dependency-clause-alternatives",
742 "negative-dependency-clauses",
743 ]:
744 stat = self.stats[statname]
745 if statname != "nodes":
746 format_str = "%s, max: %d, min: %d, median: %d, average: %f (%d/%d)"
747 values = [
748 statname,
749 stat["max"],
750 stat["min"],
751 stat["median"],
752 stat["average"],
753 stat["sum"],
754 stat["size"],
755 ]
756 if "average-per-node" in stat:
757 format_str += ", average-per-node: %f"
758 values.append(stat["average-per-node"])
759 else:
760 format_str = "nodes: %d, eqv-nodes: %d"
761 values = (self.nodes, self.eqv_nodes)
762 text.append(format_str % tuple(values))
763 return text
765 def add_dep_edges(self, edges: frozenset[frozenset["BinaryPackageId"]]) -> None:
766 self.dep_edges.append(edges)
768 def add_con_edges(self, edges: frozenset["BinaryPackageId"]) -> None:
769 self.con_edges.append(edges)
771 def _list_stats(
772 self, stat_name: str, sorted_list: list[int], average_per_node: bool = False
773 ) -> None:
774 if sorted_list:
775 stats = self.stats[stat_name]
776 stats["max"] = sorted_list[-1]
777 stats["min"] = sorted_list[0]
778 stats["sum"] = sum(sorted_list)
779 stats["size"] = len(sorted_list)
780 stats["average"] = float(stats["sum"]) / len(sorted_list)
781 stats["median"] = sorted_list[len(sorted_list) // 2]
782 if average_per_node:
783 stats["average-per-node"] = float(stats["sum"]) / self.nodes
785 def compute_all(self) -> None:
786 dep_edges = self.dep_edges
787 con_edges = self.con_edges
788 sorted_no_dep_edges = sorted(len(x) for x in dep_edges)
789 sorted_size_dep_edges = sorted(len(x) for x in chain.from_iterable(dep_edges))
790 sorted_no_con_edges = sorted(len(x) for x in con_edges)
791 self._list_stats("dependency-clauses", sorted_no_dep_edges)
792 self._list_stats(
793 "dependency-clause-alternatives",
794 sorted_size_dep_edges,
795 average_per_node=True,
796 )
797 self._list_stats("negative-dependency-clauses", sorted_no_con_edges)