music-assistant-server

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