/
/
/
1"""
2Smart Fades - the candidate/policy transition planner.
3
4``SmartCrossFadePlanner.plan()`` is a thin orchestration of the pipeline:
5build the immutable ``TransitionContext``, let the generators propose
6candidate specs, build each into a timed candidate, score them all with the
7rejection/penalty policies, finalize the winner's EQ - or, when every
8candidate is rejected, retry with late-anchored rescue candidates (the
9ungated audible-end ladder plus a modest rescue rung) before falling back to
10the click-free emergency handoff as a last resort. Alternative strategies
11slot in as sibling ``TransitionPlanner`` subclasses.
12"""
13
14from __future__ import annotations
15
16from abc import ABC, abstractmethod
17from collections import Counter
18from dataclasses import replace
19from typing import TYPE_CHECKING
20
21from music_assistant.constants import VERBOSE_LOG_LEVEL
22from music_assistant.controllers.streams.smart_fades.models import SmartFadeNotApplicable
23
24from .assembly import EmergencyHandoffFactory, PlanAssembler
25from .candidates import (
26 CandidateFactory,
27 RescueAnchorGenerator,
28 TrimClosingAnchorGenerator,
29 default_generators,
30)
31from .context import build_transition_context
32from .policies import default_policies
33from .selection import CandidateSelector
34
35if TYPE_CHECKING:
36 import logging
37
38 from music_assistant.controllers.streams.smart_fades.models import TransitionPlan
39 from music_assistant.models.audio_analysis import AudioAnalysisData
40
41
42class TransitionPlanner(ABC):
43 """Abstract base class for transition planners."""
44
45 def __init__(self, logger: logging.Logger) -> None:
46 """Initialize the planner."""
47 self.logger = logger
48
49 @abstractmethod
50 def plan(
51 self,
52 fade_out_analysis: AudioAnalysisData,
53 fade_in_analysis: AudioAnalysisData,
54 buffer_duration: float,
55 ) -> TransitionPlan:
56 """
57 Build a ``TransitionPlan`` from the two tracks' analysis data.
58
59 Pure over the analysis rows and the available holdback window â touches
60 no audio bytes. Raises ``SmartFadeNotApplicable`` when the tracks cannot
61 yield this transition and the caller should fall back.
62
63 :param fade_out_analysis: Analysis data for the outgoing track.
64 :param fade_in_analysis: Analysis data for the incoming track.
65 :param buffer_duration: Length in seconds of the available fade-out holdback.
66 """
67
68
69class SmartCrossFadePlanner(TransitionPlanner):
70 """Plans a defensive, musically-aligned crossfade that never edits the music."""
71
72 def plan(
73 self,
74 fade_out_analysis: AudioAnalysisData,
75 fade_in_analysis: AudioAnalysisData,
76 buffer_duration: float,
77 ) -> TransitionPlan:
78 """
79 Build a smart-crossfade ``TransitionPlan`` from the two tracks' analysis.
80
81 Vocal-aware protections engage per deck: each track with a validated
82 FireRed vocal-activity timeline gets its vocals protected, while a
83 track without one is planned on energy facts alone.
84
85 :param fade_out_analysis: Analysis data for the outgoing track.
86 :param fade_in_analysis: Analysis data for the incoming track.
87 :param buffer_duration: Length in seconds of the available fade-out holdback.
88 """
89 ctx = build_transition_context(
90 fade_out_analysis, fade_in_analysis, buffer_duration, self.logger
91 )
92 factory = CandidateFactory(ctx, self.logger)
93 specs = [spec for generator in default_generators() for spec in generator.generate(ctx)]
94 candidates = [candidate for spec in specs if (candidate := factory.build(spec)) is not None]
95 if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
96 self.logger.log(
97 VERBOSE_LOG_LEVEL,
98 "generated %d specs (%s), %d built",
99 len(specs),
100 dict(Counter(spec.source for spec in specs)),
101 len(candidates),
102 )
103 if not candidates:
104 raise SmartFadeNotApplicable("no feasible transition candidate")
105 selector = CandidateSelector(default_policies(), self.logger)
106 winner = selector.select(candidates, ctx)
107 if winner is None:
108 # every phrased candidate breached a hard rejection: retry with the
109 # ungated audible-end ladder plus a modest late-anchored rescue rung
110 # before falling back to the handoff
111 rescue_specs = [
112 *TrimClosingAnchorGenerator(min_gap=0.0).generate(ctx),
113 *RescueAnchorGenerator().generate(ctx),
114 ]
115 rescue_candidates = [
116 candidate for spec in rescue_specs if (candidate := factory.build(spec)) is not None
117 ]
118 winner = selector.select(rescue_candidates, ctx) if rescue_candidates else None
119 if winner is not None:
120 self.logger.debug(
121 "shipping a rescue-pass candidate (source=%s) instead of the emergency handoff",
122 winner.candidate.spec.source,
123 )
124 if winner is None:
125 self.logger.debug("shipping click-free emergency handoff")
126 plan = EmergencyHandoffFactory(ctx, factory, self.logger).build()
127 else:
128 plan = PlanAssembler(ctx, self.logger).finalize(winner.candidate)
129 # the caller reads the outgoing grid off the planner after a successful
130 # plan and expects it masked to the plan's own anchor
131 self.outgoing = replace(
132 ctx.outgoing,
133 beats=ctx.outgoing.beats[ctx.outgoing.beats <= plan.fade_out_window],
134 downbeats=ctx.outgoing.downbeats[ctx.outgoing.downbeats <= plan.fade_out_window],
135 )
136 return plan
137