Trixy7/trixy_core/audio_morpher/processor.py

# -*- coding: utf-8 -*-
"""
VoiceMorphProcessor — Pipeline-Bridge fuer Audio-Processing.

Nur laengen-neutrale Effekte: VTLP, Breathiness, Shimmer, loud_pitch_boost.
Kein pitch_shift, speed, pause_scale, loud_tempo_scale (aendern Chunk-Laenge).
"""

from __future__ import annotations

import struct
from typing import TYPE_CHECKING

import numpy as np

from trixy_core.audio.processing.processor import AudioProcessor, ProcessorPriority
from trixy_core.audio_morpher.dsp import vtlp, add_breathiness, add_shimmer
from trixy_core.audio_morpher.dynamics import loud_pitch_boost
from trixy_core.audio_morpher.morpher import AudioMorpher
from trixy_core.audio_morpher.presets import VoiceMorphPreset

if TYPE_CHECKING:
    from trixy_core.audio.processing.context import AudioProcessingContext


class VoiceMorphProcessor(AudioProcessor):
    """
    Audio-Pipeline-Prozessor fuer Voice-Morphing.

    Wendet nur laengen-neutrale Effekte an (Chunk-Laenge bleibt gleich):
    - VTLP (Formant-Verschiebung)
    - Breathiness
    - Shimmer
    - loud_pitch_boost
    """

    def __init__(
        self,
        preset: VoiceMorphPreset | str = "",
        intensity: float = 1.0,
        morpher: AudioMorpher | None = None,
    ) -> None:
        super().__init__(
            processor_id="voice_morph",
            name="Voice Morph",
            priority=ProcessorPriority.PRE_EFFECTS,
        )
        self._intensity = intensity
        self._morpher = morpher or AudioMorpher()

        # Preset aufloesen
        if isinstance(preset, str) and preset:
            self._preset = self._morpher.get_preset(preset)
        elif isinstance(preset, VoiceMorphPreset):
            self._preset = preset
        else:
            self._preset = None

    @property
    def preset(self) -> VoiceMorphPreset | None:
        return self._preset

    @preset.setter
    def preset(self, value: VoiceMorphPreset | str | None) -> None:
        if isinstance(value, str):
            self._preset = self._morpher.get_preset(value)
        else:
            self._preset = value

    @property
    def intensity(self) -> float:
        return self._intensity

    @intensity.setter
    def intensity(self, value: float) -> None:
        self._intensity = max(0.0, min(1.0, value))

    def process(
        self,
        chunk: bytes,
        context: "AudioProcessingContext",
    ) -> bytes:
        """Wendet laengen-neutrale Voice-Morph-Effekte auf PCM-Chunk an."""
        if self._preset is None or self._intensity < 0.01:
            return chunk

        # PCM 16-bit → float64
        num_samples = len(chunk) // 2
        if num_samples < 4:
            return chunk

        samples = struct.unpack(f"<{num_samples}h", chunk)
        audio = np.array(samples, dtype=np.float64) / 32768.0

        # Sample-Rate aus Kontext (Standard: 16000 fuer Pipeline)
        sample_rate = getattr(context, "sample_rate", 16000)

        # Nur laengen-neutrale Effekte anwenden
        formant = self._preset.formant_extra * self._intensity
        breath = self._preset.breathiness * self._intensity
        shim = self._preset.shimmer * self._intensity
        lpb = self._preset.loud_pitch_boost * self._intensity

        if abs(lpb) > 0.05:
            audio = loud_pitch_boost(audio, sample_rate, lpb)

        if abs(formant) > 0.01:
            alpha = 1.0 + formant
            audio = vtlp(audio, sample_rate, alpha)

        if breath > 0.005:
            audio = add_breathiness(audio, sample_rate, breath)

        if shim > 0.005:
            audio = add_shimmer(audio, sample_rate, shim)

        # float64 → PCM 16-bit
        audio_int = np.clip(audio * 32768.0, -32768, 32767).astype(np.int16)
        return struct.pack(f"<{num_samples}h", *audio_int)

    def get_config(self) -> dict:
        config = super().get_config()
        config["preset"] = self._preset.preset_id if self._preset else None
        config["intensity"] = self._intensity
        return config