/
/
/
1"""
2Smart Fades - candidate selection.
3
4A ``CandidateSelector`` scores every built candidate against the full policy
5set, folding each policy's ``Verdict`` into one ``ScoredCandidate`` scoreboard
6entry, then picks the lowest-penalty, non-rejected survivor. Every policy runs
7on every candidate - no short-circuit on the first rejection - so the debug
8log always shows the complete scoreboard, not just whichever rule fired first.
9"""
10
11from __future__ import annotations
12
13import logging
14from collections import Counter
15from collections.abc import Sequence
16from dataclasses import dataclass
17
18from music_assistant.constants import VERBOSE_LOG_LEVEL
19
20from .candidates import Candidate
21from .context import TransitionContext
22from .policies import Policy, Verdict
23
24
25@dataclass(frozen=True, slots=True)
26class ScoredCandidate:
27 """One candidate's full scoreboard entry: every policy's verdict and its resulting rank."""
28
29 candidate: Candidate
30 total_penalty: float
31 verdicts: tuple[Verdict, ...]
32 rejected: bool
33
34
35class CandidateSelector:
36 """Scores every candidate against a fixed policy set and picks the best survivor."""
37
38 def __init__(self, policies: Sequence[Policy], logger: logging.Logger) -> None:
39 """Initialize the selector with the policy set to score every candidate against."""
40 self._policies = tuple(policies)
41 self._logger = logger
42
43 def select(
44 self, candidates: Sequence[Candidate], ctx: TransitionContext
45 ) -> ScoredCandidate | None:
46 """
47 Score every candidate; return the lowest-penalty survivor, or None when all are rejected.
48
49 Ties resolve to whichever candidate appears earlier in ``candidates``.
50
51 :param candidates: Built candidates to score, in generator-declared order.
52 :param ctx: The shared per-transition facts every policy judges against.
53 """
54 scored = [self._score(candidate, ctx) for candidate in candidates]
55 survivors = [entry for entry in scored if not entry.rejected]
56 if not survivors:
57 histogram = Counter(
58 verdict.reason for entry in scored for verdict in entry.verdicts if verdict.rejected
59 )
60 reasons = ", ".join(f"{reason} x{count}" for reason, count in histogram.most_common())
61 self._logger.debug(
62 "all %d candidates rejected (%s); emergency handoff will be used",
63 len(scored),
64 reasons,
65 )
66 return None
67 winner = min(survivors, key=lambda entry: entry.total_penalty)
68 if self._logger.isEnabledFor(VERBOSE_LOG_LEVEL):
69 ranked = sorted(survivors, key=lambda entry: entry.total_penalty)
70 runner_up = ranked[1] if len(ranked) > 1 else None
71 self._logger.log(
72 VERBOSE_LOG_LEVEL,
73 "selection: scored=%d survivors=%d winner source=%s tier=%s bars=%d total=%.2f "
74 "runner_up=%s runner_up_total=%s",
75 len(scored),
76 len(survivors),
77 winner.candidate.spec.source,
78 winner.candidate.spec.tier,
79 winner.candidate.spec.bars,
80 winner.total_penalty,
81 runner_up.candidate.spec.source if runner_up is not None else None,
82 runner_up.total_penalty if runner_up is not None else None,
83 )
84 return winner
85
86 def _score(self, candidate: Candidate, ctx: TransitionContext) -> ScoredCandidate:
87 """Evaluate every policy on one candidate and log its full scoreboard entry."""
88 verdicts = tuple(policy.evaluate(candidate, ctx) for policy in self._policies)
89 total_penalty = sum(verdict.penalty for verdict in verdicts)
90 rejected = any(verdict.rejected for verdict in verdicts)
91 if self._logger.isEnabledFor(VERBOSE_LOG_LEVEL):
92 # per-policy breakdown so a tuning pass can see which penalty (or
93 # rejection) each policy contributed, not just the aggregate total
94 breakdown_entries = []
95 for policy, verdict in zip(self._policies, verdicts, strict=True):
96 name = type(policy).__name__.removesuffix("Policy")
97 value = (
98 f"REJECTED({verdict.reason})" if verdict.rejected else f"{verdict.penalty:.2f}"
99 )
100 breakdown_entries.append(f"{name}={value}")
101 plan, metrics = candidate.plan, candidate.metrics
102 self._logger.log(
103 VERBOSE_LOG_LEVEL,
104 "candidate source=%s tier=%s bars=%d duration=%.2f anchor=%.2f fadein_trim=%s "
105 "total=%.2f rejected=%s trim=%.2f collision=%.2f weighted_collision=%.2f "
106 "on_downbeat=%s %s",
107 candidate.spec.source,
108 candidate.spec.tier,
109 candidate.spec.bars,
110 plan.crossfade_duration,
111 plan.fade_out_window,
112 plan.fadein_trim_start,
113 total_penalty,
114 rejected,
115 metrics.audible_outgoing_trim,
116 metrics.collision_seconds,
117 metrics.weighted_collision_seconds,
118 metrics.anchor_on_downbeat,
119 " ".join(breakdown_entries),
120 )
121 return ScoredCandidate(
122 candidate=candidate,
123 total_penalty=total_penalty,
124 verdicts=verdicts,
125 rejected=rejected,
126 )
127