UmbraAudio/specs/06-normalization-watermarking.md

# Spec: Normalization & Watermarking

## Overview

Post-generation processing includes audio normalization (RMS leveling, peak limiting) and neural watermarking (author identity embedded in audio). These run as part of the generation pipeline per-utterance.

## Components

### 1. Audio Normalizer (`assembly/normalize.py`)

```python
class AudioNormalizer:
    def __init__(
        self,
        target_rms_db: float = -20.0,
        peak_limit_db: float = -3.0,
        noise_floor_db: float = -60.0,
        sample_rate: int = 24000,
    ) -> None

    def normalize_file(self, input_path: Path, output_path: Path) -> dict
    def check_acx_compliance(self, audio_path: Path) -> dict
```

#### Normalization Algorithm

```python
def normalize_file(self, input_path: Path, output_path: Path) -> dict:
    # 1. Read audio
    audio, sr = sf.read(input_path)

    # 2. Mono downmix if stereo
    if audio.ndim > 1:
        audio = audio.mean(axis=1)

    # 3. Compute current RMS in dB
    rms = np.sqrt(np.mean(audio ** 2))
    current_rms_db = 20 * np.log10(rms + 1e-10)

    # 4. Calculate gain
    gain_db = self.target_rms_db - current_rms_db
    gain = 10 ** (gain_db / 20)
    audio = audio * gain

    # 5. Peak limiting
    peak = np.max(np.abs(audio))
    peak_db = 20 * np.log10(peak + 1e-10)
    if peak_db > self.peak_limit_db:
        limiting_gain = 10 ** ((self.peak_limit_db - peak_db) / 20)
        audio = audio * limiting_gain

    # 6. Clip to [-1.0, 1.0]
    audio = np.clip(audio, -1.0, 1.0)

    # 7. Write output
    sf.write(output_path, audio, self.sample_rate)
```

**Returns:**
```python
{
    "input_path": str,
    "output_path": str,
    "original_rms_db": float,
    "normalized_rms_db": float,
    "peak_db": float,
    "gain_applied_db": float,
}
```

#### ACX Compliance Check

```python
def check_acx_compliance(self, audio_path: Path) -> dict
```

**ACX Requirements:**
| Metric | Range |
|--------|-------|
| RMS level | -23.0 dB to -18.0 dB |
| Peak level | Below -3.0 dB |

**Returns:**
```python
{
    "rms_db": float,
    "peak_db": float,
    "rms_in_range": bool,     # -23.0 <= rms <= -18.0
    "peak_below_limit": bool,  # peak < -3.0
    "compliant": bool,         # rms_in_range AND peak_below_limit
}
```

### 2. Watermarking

#### Author Watermark Bits

```python
def _author_watermark_bits(author_name: str, length: int = 128) -> np.ndarray
```

**Algorithm:**
1. SHA-256 hash of author name string
2. Convert 128 bits from the hash digest to a float array
3. Each byte → 8 bits, take first `length` bits
4. Return as `np.float32` array

**Example:**
```python
author_name = "B.D. Murphy"
digest = sha256(b"B.D. Murphy").digest()  # 32 bytes = 256 bits
bits = first 128 bits → np.float32 array of 128 values (0.0 or 1.0)
```

#### Perth Neural Watermarker

```python
class ChatterboxEngine:
    def _get_watermarker(self) -> watermarker | None
    def apply_watermark(self, wav, watermark_bits, sample_rate) -> torch.Tensor
```

**Watermarker Selection:**
1. Try `PerthImplicitWatermarker(device="cuda")` — neural watermarking
2. Fall back to `DummyWatermarker()` — no-op pass-through
3. Return `None` if neither available

**Application:**
```python
def apply_watermark(self, wav, watermark_bits, sample_rate=None):
    if self._watermarker is None:
        return wav  # No watermarking available

    try:
        audio_np = wav.squeeze(0).cpu().numpy()
        if audio_np.ndim > 1:
            audio_np = audio_np.mean(axis=0)

        watermarked = self._watermarker.embed(audio_np, watermark_bits)
        return torch.from_numpy(watermarked).unsqueeze(0)
    except Exception:
        return wav  # Fallback: return unwatermarked
```

**PerthImplicitWatermarker:**
- Located at: `~/.cache/torch/hub/snakers4_perth`
- Model: `perth_net` — implicit neural audio watermarking
- Embeds 128-bit payload into audio signal
- Robust to common audio transformations (compression, resampling)
- VRAM usage: minimal (~50MB)

**DummyWatermarker:**
- No-op implementation
- Returns audio unchanged
- Used as fallback when Perth is unavailable

### Integration with Generation

Watermarking runs inside `ChatterboxEngine.generate()`:

```python
def generate(self, text, output_path, voice_ref, instruct, watermark_bits, ...):
    wav = self._model.generate(text, ...)

    if watermark_bits is not None:
        wav = self.apply_watermark(wav, watermark_bits)

    # Write to file
    sf.write(str(output_path), audio_np, self.sample_rate)
```

### Integration with Normalization

Normalization runs inside `ChapterGenerator.generate_chapter()`:

```python
# After successful generation:
if self.normalizer:
    normalized_path = normalized_dir / out_path.name
    self.normalizer.normalize_file(out_path, normalized_path)
```

### Quality Defect Tracking

ACX compliance is checked per-file in the orchestrator:

```python
# After generation:
result = normalizer.check_acx_compliance(audio_path)
if not result["compliant"]:
    db.log_defect(project_id, "quality", "WARNING",
                  f"ACX non-compliance: RMS={result['rms_db']:.1f}dB, Peak={result['peak_db']:.1f}dB",
                  chapter_id=ch["id"])
```

## Default Parameters

| Parameter | Default | ACX Requirement |
|-----------|---------|-----------------|
| `target_rms_db` | -20.0 dB | -23.0 to -18.0 dB |
| `peak_limit_db` | -3.0 dB | Below -3.0 dB |
| `noise_floor_db` | -60.0 dB | N/A |
| `sample_rate` | 24000 Hz | N/A |
| `watermark_length` | 128 bits | N/A |