Level 2: Core functionality and level_2 script wrappers added

This commit is contained in:
2026-02-08 19:43:26 +02:00
parent 399abebd2c
commit 9931e3830a
39 changed files with 4720 additions and 913 deletions
+88 -2
View File
@@ -30,6 +30,7 @@ import soundfile as sf
from core.aac_configuration import WIN_TYPE
from core.aac_filterbank import aac_filter_bank
from core.aac_ssc import aac_SSC
from core.aac_tns import aac_tns
from core.aac_types import *
@@ -144,8 +145,8 @@ def aac_coder_1(filename_in: Union[str, Path]) -> AACSeq1:
AACSeq1
List of encoded frames (Level 1 schema).
"""
x, fs = aac_read_wav_stereo_48k(filename_in)
_ = fs # kept for clarity; The assignment assumes 48 kHz
x, _ = aac_read_wav_stereo_48k(filename_in)
# The assignment assumes 48 kHz
hop = 1024
win = 2048
@@ -196,3 +197,88 @@ def aac_coder_1(filename_in: Union[str, Path]) -> AACSeq1:
prev_frame_type = frame_type
return aac_seq
def aac_coder_2(filename_in: Union[str, Path]) -> AACSeq2:
"""
Level-2 AAC encoder (Level 1 + TNS).
Parameters
----------
filename_in : Union[str, Path]
Input WAV filename (stereo, 48 kHz).
Returns
-------
AACSeq2
Encoded AAC sequence (Level 2 payload schema).
For each frame i:
- "frame_type": FrameType
- "win_type": WinType
- "chl"/"chr":
- "frame_F": FrameChannelF (after TNS)
- "tns_coeffs": TnsCoeffs
"""
filename_in = Path(filename_in)
x, _ = aac_read_wav_stereo_48k(filename_in)
# The assignment assumes 48 kHz
hop = 1024
win = 2048
pad_pre = np.zeros((hop, 2), dtype=np.float64)
pad_post = np.zeros((hop, 2), dtype=np.float64)
x_pad = np.vstack([pad_pre, x, pad_post])
K = int((x_pad.shape[0] - win) // hop + 1)
if K <= 0:
raise ValueError("Input too short for framing.")
aac_seq: AACSeq2 = []
prev_frame_type: FrameType = "OLS"
for i in range(K):
start = i * hop
frame_t: FrameT = x_pad[start : start + win, :]
if frame_t.shape != (win, 2):
raise ValueError("Internal framing error: frame_t has wrong shape.")
next_t = x_pad[start + hop : start + hop + win, :]
if next_t.shape[0] < win:
tail = np.zeros((win - next_t.shape[0], 2), dtype=np.float64)
next_t = np.vstack([next_t, tail])
frame_type = aac_SSC(frame_t, next_t, prev_frame_type)
# Level 1 analysis (packed stereo container)
frame_f_stereo = aac_filter_bank(frame_t, frame_type, WIN_TYPE)
# Unpack to per-channel (as you already do in Level 1)
if frame_type == "ESH":
chl_f = np.empty((128, 8), dtype=np.float64)
chr_f = np.empty((128, 8), dtype=np.float64)
for j in range(8):
chl_f[:, j] = frame_f_stereo[:, 2 * j + 0]
chr_f[:, j] = frame_f_stereo[:, 2 * j + 1]
else:
chl_f = frame_f_stereo[:, 0:1].astype(np.float64, copy=False)
chr_f = frame_f_stereo[:, 1:2].astype(np.float64, copy=False)
# Level 2: apply TNS per channel
chl_f_tns, chl_tns_coeffs = aac_tns(chl_f, frame_type)
chr_f_tns, chr_tns_coeffs = aac_tns(chr_f, frame_type)
aac_seq.append(
{
"frame_type": frame_type,
"win_type": WIN_TYPE,
"chl": {"frame_F": chl_f_tns, "tns_coeffs": chl_tns_coeffs},
"chr": {"frame_F": chr_f_tns, "tns_coeffs": chr_tns_coeffs},
}
)
prev_frame_type = frame_type
return aac_seq
+10 -1
View File
@@ -17,6 +17,15 @@ from __future__ import annotations
# Imports
from core.aac_types import WinType
# Filterbank
# ------------------------------------------------------------
# Window type
# Options: "SIN", "KBD"
WIN_TYPE: WinType = "SIN"
WIN_TYPE: WinType = "SIN"
# TNS
# ------------------------------------------------------------
PRED_ORDER = 4
QUANT_STEP = 0.1
QUANT_MAX = 0.7 # 4-bit symmetric with step 0.1 -> clamp to [-0.7, +0.7]
+91
View File
@@ -28,6 +28,7 @@ from typing import Union
import soundfile as sf
from core.aac_filterbank import aac_i_filter_bank
from core.aac_tns import aac_i_tns
from core.aac_types import *
@@ -164,3 +165,93 @@ def aac_decoder_1(aac_seq_1: AACSeq1, filename_out: Union[str, Path]) -> StereoS
sf.write(str(filename_out), y, 48000)
return y
def aac_decoder_2(aac_seq_2: AACSeq2, filename_out: Union[str, Path]) -> StereoSignal:
"""
Level-2 AAC decoder (inverse of aac_coder_2).
Behavior matches Level 1 decoder pipeline, with additional iTNS stage:
- Per frame/channel: inverse TNS using stored coefficients
- Re-pack to stereo frame_F
- IMDCT + windowing
- Overlap-add over frames
- Remove Level-1 padding (hop samples start/end)
- Write output WAV (48 kHz)
Parameters
----------
aac_seq_2 : AACSeq2
Encoded sequence as produced by aac_coder_2().
filename_out : Union[str, Path]
Output WAV filename.
Returns
-------
StereoSignal
Decoded audio samples (time-domain), stereo, shape (N, 2), dtype float64.
"""
filename_out = Path(filename_out)
hop = 1024
win = 2048
K = len(aac_seq_2)
if K <= 0:
raise ValueError("aac_seq_2 must contain at least one frame.")
n_pad = (K - 1) * hop + win
y_pad = np.zeros((n_pad, 2), dtype=np.float64)
for i, fr in enumerate(aac_seq_2):
frame_type: FrameType = fr["frame_type"]
win_type: WinType = fr["win_type"]
chl_f_tns = np.asarray(fr["chl"]["frame_F"], dtype=np.float64)
chr_f_tns = np.asarray(fr["chr"]["frame_F"], dtype=np.float64)
chl_coeffs = np.asarray(fr["chl"]["tns_coeffs"], dtype=np.float64)
chr_coeffs = np.asarray(fr["chr"]["tns_coeffs"], dtype=np.float64)
# Inverse TNS per channel
chl_f = aac_i_tns(chl_f_tns, frame_type, chl_coeffs)
chr_f = aac_i_tns(chr_f_tns, frame_type, chr_coeffs)
# Re-pack to the stereo container expected by aac_i_filter_bank
if frame_type == "ESH":
if chl_f.shape != (128, 8) or chr_f.shape != (128, 8):
raise ValueError("ESH channel frame_F must have shape (128, 8).")
frame_f: FrameF = np.empty((128, 16), dtype=np.float64)
for j in range(8):
frame_f[:, 2 * j + 0] = chl_f[:, j]
frame_f[:, 2 * j + 1] = chr_f[:, j]
else:
# Accept either (1024,1) or (1024,) from your internal convention.
if chl_f.shape == (1024,):
chl_col = chl_f.reshape(1024, 1)
elif chl_f.shape == (1024, 1):
chl_col = chl_f
else:
raise ValueError("Non-ESH left channel frame_F must be shape (1024,) or (1024, 1).")
if chr_f.shape == (1024,):
chr_col = chr_f.reshape(1024, 1)
elif chr_f.shape == (1024, 1):
chr_col = chr_f
else:
raise ValueError("Non-ESH right channel frame_F must be shape (1024,) or (1024, 1).")
frame_f = np.empty((1024, 2), dtype=np.float64)
frame_f[:, 0] = chl_col[:, 0]
frame_f[:, 1] = chr_col[:, 0]
frame_t_hat: FrameT = aac_i_filter_bank(frame_f, frame_type, win_type)
start = i * hop
y_pad[start : start + win, :] += frame_t_hat
y = aac_remove_padding(y_pad, hop=hop)
sf.write(str(filename_out), y, 48000)
return y
+60
View File
@@ -0,0 +1,60 @@
# ------------------------------------------------------------
# AAC Coder/Decoder - SNR dB calculator
#
# Multimedia course at Aristotle University of
# Thessaloniki (AUTh)
#
# Author:
# Christos Choutouridis (ΑΕΜ 8997)
# cchoutou@ece.auth.gr
#
# Description:
# This module implements SNR calculation in dB
# ------------------------------------------------------------
from __future__ import annotations
from core.aac_types import StereoSignal
import numpy as np
def snr_db(x_ref: StereoSignal, x_hat: StereoSignal) -> float:
"""
Compute overall SNR (dB) over all samples and channels after aligning lengths.
Parameters
----------
x_ref : StereoSignal
Reference stereo stream.
x_hat : StereoSignal
Reconstructed stereo stream.
Returns
-------
float
SNR in dB.
- Returns +inf if noise power is zero.
- Returns -inf if signal power is zero.
"""
x_ref = np.asarray(x_ref, dtype=np.float64)
x_hat = np.asarray(x_hat, dtype=np.float64)
if x_ref.ndim == 1:
x_ref = x_ref.reshape(-1, 1)
if x_hat.ndim == 1:
x_hat = x_hat.reshape(-1, 1)
n = min(x_ref.shape[0], x_hat.shape[0])
c = min(x_ref.shape[1], x_hat.shape[1])
x_ref = x_ref[:n, :c]
x_hat = x_hat[:n, :c]
err = x_ref - x_hat
ps = float(np.sum(x_ref * x_ref))
pn = float(np.sum(err * err))
if pn <= 0.0:
return float("inf")
if ps <= 0.0:
return float("-inf")
return float(10.0 * np.log10(ps / pn))
+549
View File
@@ -0,0 +1,549 @@
# ------------------------------------------------------------
# AAC Coder/Decoder - Temporal Noise Shaping (TNS)
#
# Multimedia course at Aristotle University of
# Thessaloniki (AUTh)
#
# Author:
# Christos Choutouridis (ΑΕΜ 8997)
# cchoutou@ece.auth.gr
#
# Description:
# Temporal Noise Shaping (TNS) module (Level 2).
#
# Public API:
# frame_F_out, tns_coeffs = aac_tns(frame_F_in, frame_type)
# frame_F_out = aac_i_tns(frame_F_in, frame_type, tns_coeffs)
#
# Notes (per assignment):
# - TNS is applied per channel (not stereo).
# - For ESH, TNS is applied independently to each of the 8 short subframes.
# - Bark band tables are taken from TableB.2.1.9a (long) and TableB.2.1.9b (short)
# provided in TableB219.mat.
# - Predictor order is fixed to p = 4.
# - Coefficients are quantized with a 4-bit uniform symmetric quantizer, step = 0.1.
# - Forward TNS applies FIR: H_TNS(z) = 1 - a1 z^-1 - ... - ap z^-p
# - Inverse TNS applies the inverse IIR filter using the same quantized coefficients.
# ------------------------------------------------------------
from __future__ import annotations
from pathlib import Path
from typing import Tuple
import numpy as np
from scipy.io import loadmat
from core.aac_configuration import PRED_ORDER, QUANT_STEP, QUANT_MAX
from core.aac_types import *
# -----------------------------------------------------------------------------
# Private helpers
# -----------------------------------------------------------------------------
_B219_CACHE: dict[str, FloatArray] | None = None
def _load_b219_tables() -> dict[str, FloatArray]:
"""
Load TableB219.mat and cache the contents.
The project layout guarantees that a 'material' directory is discoverable
from the current working directory (tests and level_123 entrypoints).
Returns
-------
dict[str, FloatArray]
Keys:
- "B219a": long bands table (for K=1024 MDCT lines)
- "B219b": short bands table (for K=128 MDCT lines)
"""
global _B219_CACHE
if _B219_CACHE is not None:
return _B219_CACHE
mat_path = Path("material") / "TableB219.mat"
if not mat_path.exists():
raise FileNotFoundError("Could not locate material/TableB219.mat in the current working directory.")
d = loadmat(str(mat_path))
if "B219a" not in d or "B219b" not in d:
raise ValueError("TableB219.mat missing required variables B219a and/or B219b.")
_B219_CACHE = {
"B219a": np.asarray(d["B219a"], dtype=np.float64),
"B219b": np.asarray(d["B219b"], dtype=np.float64),
}
return _B219_CACHE
def _band_ranges_for_kcount(k_count: int) -> BandRanges:
"""
Return Bark band index ranges [start, end] (inclusive) for the given MDCT line count.
Parameters
----------
k_count : int
Number of MDCT lines:
- 1024 for long frames
- 128 for short subframes (ESH)
Returns
-------
BandRanges (list[tuple[int, int]])
Each tuple is (start_k, end_k) inclusive.
"""
tables = _load_b219_tables()
if k_count == 1024:
tbl = tables["B219a"]
elif k_count == 128:
tbl = tables["B219b"]
else:
raise ValueError("TNS supports only k_count=1024 (long) or k_count=128 (short).")
start = tbl[:, 1].astype(int)
end = tbl[:, 2].astype(int)
ranges: list[tuple[int, int]] = [(int(s), int(e)) for s, e in zip(start, end)]
for s, e in ranges:
if s < 0 or e < s or e >= k_count:
raise ValueError("Invalid band table ranges for given k_count.")
return ranges
# -----------------------------------------------------------------------------
# Core DSP helpers
# -----------------------------------------------------------------------------
def _smooth_sw_inplace(sw: MdctCoeffs) -> None:
"""
Smooth Sw(k) to reduce discontinuities between adjacent Bark bands.
The assignment applies two passes:
- Backward: Sw(k) = (Sw(k) + Sw(k+1))/2
- Forward: Sw(k) = (Sw(k) + Sw(k-1))/2
Parameters
----------
sw : MdctCoeffs
1-D array of length K (float64). Modified in-place.
"""
k_count = int(sw.shape[0])
for k in range(k_count - 2, -1, -1):
sw[k] = 0.5 * (sw[k] + sw[k + 1])
for k in range(1, k_count):
sw[k] = 0.5 * (sw[k] + sw[k - 1])
def _compute_sw(x: MdctCoeffs) -> MdctCoeffs:
"""
Compute Sw(k) from band energies P(j) and apply boundary smoothing.
Parameters
----------
x : MdctCoeffs
1-D MDCT line array, length K.
Returns
-------
MdctCoeffs
Sw(k), 1-D array of length K, float64.
"""
x = np.asarray(x, dtype=np.float64).reshape(-1)
k_count = int(x.shape[0])
bands = _band_ranges_for_kcount(k_count)
sw = np.zeros(k_count, dtype=np.float64)
for s, e in bands:
seg = x[s : e + 1]
p_j = float(np.sum(seg * seg))
sw_val = float(np.sqrt(p_j))
sw[s : e + 1] = sw_val
_smooth_sw_inplace(sw)
return sw
def _autocorr(x: MdctCoeffs, p: int) -> MdctCoeffs:
"""
Autocorrelation r(m) for m=0..p.
Parameters
----------
x : MdctCoeffs
1-D signal.
p : int
Maximum lag.
Returns
-------
MdctCoeffs
r, shape (p+1,), float64.
"""
x = np.asarray(x, dtype=np.float64).reshape(-1)
n = int(x.shape[0])
r = np.zeros(p + 1, dtype=np.float64)
for m in range(p + 1):
r[m] = float(np.dot(x[m:], x[: n - m]))
return r
def _lpc_coeffs(xw: MdctCoeffs, p: int) -> MdctCoeffs:
"""
Solve Yule-Walker normal equations for LPC coefficients of order p.
Parameters
----------
xw : MdctCoeffs
1-D normalized sequence Xw(k).
p : int
Predictor order.
Returns
-------
MdctCoeffs
LPC coefficients a[0..p-1], shape (p,), float64.
"""
r = _autocorr(xw, p)
R = np.empty((p, p), dtype=np.float64)
for i in range(p):
for j in range(p):
R[i, j] = r[abs(i - j)]
rhs = r[1 : p + 1].reshape(p)
reg = 1e-12
R_reg = R + reg * np.eye(p, dtype=np.float64)
a = np.linalg.solve(R_reg, rhs)
return a
def _quantize_coeffs(a: MdctCoeffs) -> MdctCoeffs:
"""
Quantize LPC coefficients with uniform symmetric quantizer and clamp.
Parameters
----------
a : MdctCoeffs
LPC coefficient array, shape (p,).
Returns
-------
MdctCoeffs
Quantized coefficients, shape (p,), float64.
"""
a = np.asarray(a, dtype=np.float64).reshape(-1)
q = np.round(a / QUANT_STEP) * QUANT_STEP
q = np.clip(q, -QUANT_MAX, QUANT_MAX)
return q.astype(np.float64, copy=False)
def _is_inverse_stable(a_q: MdctCoeffs) -> bool:
"""
Check stability of the inverse TNS filter H_TNS^{-1}.
Forward filter:
H_TNS(z) = 1 - a1 z^-1 - ... - ap z^-p
Inverse filter poles are roots of:
A(z) = 1 - a1 z^-1 - ... - ap z^-p
Multiply by z^p:
z^p - a1 z^{p-1} - ... - ap = 0
Stability condition:
all roots satisfy |z| < 1.
Parameters
----------
a_q : MdctCoeffs
Quantized predictor coefficients, shape (p,).
Returns
-------
bool
True if stable, else False.
"""
a_q = np.asarray(a_q, dtype=np.float64).reshape(-1)
p = int(a_q.shape[0])
# Polynomial in z: z^p - a1 z^{p-1} - ... - ap
poly = np.empty(p + 1, dtype=np.float64)
poly[0] = 1.0
poly[1:] = -a_q
roots = np.roots(poly)
# Strictly inside unit circle for stability. Add tiny margin for numeric safety.
margin = 1e-12
return bool(np.all(np.abs(roots) < (1.0 - margin)))
def _stabilize_quantized_coeffs(a_q: MdctCoeffs) -> MdctCoeffs:
"""
Make quantized predictor coefficients stable for inverse filtering.
Policy:
- If already stable: return as-is.
- Else: iteratively shrink coefficients by gamma and re-quantize to the 0.1 grid.
- If still unstable after attempts: fall back to all-zero coefficients (disable TNS).
Parameters
----------
a_q : MdctCoeffs
Quantized predictor coefficients, shape (p,).
Returns
-------
MdctCoeffs
Stable quantized coefficients, shape (p,).
"""
a_q = np.asarray(a_q, dtype=np.float64).reshape(-1)
if _is_inverse_stable(a_q):
return a_q
# Try a few shrinking factors. Re-quantize after shrinking to keep coefficients on-grid.
gammas = (0.9, 0.8, 0.7, 0.6, 0.5, 0.4, 0.3, 0.2, 0.1)
for g in gammas:
cand = _quantize_coeffs(g * a_q)
if _is_inverse_stable(cand):
return cand
# Last resort: disable TNS for this vector
return np.zeros_like(a_q, dtype=np.float64)
def _apply_tns_fir(x: MdctCoeffs, a_q: MdctCoeffs) -> MdctCoeffs:
"""
Apply forward TNS FIR filter:
y[k] = x[k] - sum_{l=1..p} a_l * x[k-l]
Parameters
----------
x : MdctCoeffs
1-D MDCT lines, length K.
a_q : MdctCoeffs
Quantized LPC coefficients, shape (p,).
Returns
-------
MdctCoeffs
Filtered MDCT lines y, length K.
"""
x = np.asarray(x, dtype=np.float64).reshape(-1)
a_q = np.asarray(a_q, dtype=np.float64).reshape(-1)
p = int(a_q.shape[0])
k_count = int(x.shape[0])
y = np.zeros(k_count, dtype=np.float64)
for k in range(k_count):
acc = x[k]
for l in range(1, p + 1):
if k - l >= 0:
acc -= a_q[l - 1] * x[k - l]
y[k] = acc
return y
def _apply_itns_iir(y: MdctCoeffs, a_q: MdctCoeffs) -> MdctCoeffs:
"""
Apply inverse TNS IIR filter:
x_hat[k] = y[k] + sum_{l=1..p} a_l * x_hat[k-l]
Parameters
----------
y : MdctCoeffs
1-D MDCT lines after TNS, length K.
a_q : MdctCoeffs
Quantized LPC coefficients, shape (p,).
Returns
-------
MdctCoeffs
Reconstructed MDCT lines x_hat, length K.
"""
y = np.asarray(y, dtype=np.float64).reshape(-1)
a_q = np.asarray(a_q, dtype=np.float64).reshape(-1)
p = int(a_q.shape[0])
k_count = int(y.shape[0])
x_hat = np.zeros(k_count, dtype=np.float64)
for k in range(k_count):
acc = y[k]
for l in range(1, p + 1):
if k - l >= 0:
acc += a_q[l - 1] * x_hat[k - l]
x_hat[k] = acc
return x_hat
def _tns_one_vector(x: MdctCoeffs) -> tuple[MdctCoeffs, MdctCoeffs]:
"""
TNS for a single MDCT vector (one long frame or one short subframe).
Steps:
1) Compute Sw(k) from Bark band energies and smooth it.
2) Normalize: Xw(k) = X(k) / Sw(k) (safe when Sw=0).
3) Compute LPC coefficients (order p=PRED_ORDER) on Xw.
4) Quantize coefficients (4-bit symmetric, step QUANT_STEP).
5) Apply FIR filter on original X(k) using quantized coefficients.
Parameters
----------
x : MdctCoeffs
1-D MDCT vector.
Returns
-------
y : MdctCoeffs
TNS-processed MDCT vector (same length).
a_q : MdctCoeffs
Quantized LPC coefficients, shape (PRED_ORDER,).
"""
x = np.asarray(x, dtype=np.float64).reshape(-1)
sw = _compute_sw(x)
eps = 1e-12
xw = np.where(sw > eps, x / sw, 0.0)
a = _lpc_coeffs(xw, PRED_ORDER)
a_q = _quantize_coeffs(a)
# Ensure inverse stability (assignment requirement)
a_q = _stabilize_quantized_coeffs(a_q)
y = _apply_tns_fir(x, a_q)
return y, a_q
# -----------------------------------------------------------------------------
# Public Functions (Level 2)
# -----------------------------------------------------------------------------
def aac_tns(frame_F_in: FrameChannelF, frame_type: FrameType) -> Tuple[FrameChannelF, TnsCoeffs]:
"""
Temporal Noise Shaping (TNS) for ONE channel.
Parameters
----------
frame_F_in : FrameChannelF
Per-channel MDCT coefficients.
Expected (typical) shapes:
- If frame_type == "ESH": (128, 8)
- Else: (1024, 1) or (1024,)
frame_type : FrameType
Frame type code ("OLS", "LSS", "ESH", "LPS").
Returns
-------
frame_F_out : FrameChannelF
Per-channel MDCT coefficients after applying TNS.
Same shape convention as input.
tns_coeffs : TnsCoeffs
Quantized TNS predictor coefficients.
Expected shapes:
- If frame_type == "ESH": (PRED_ORDER, 8)
- Else: (PRED_ORDER, 1)
"""
x = np.asarray(frame_F_in, dtype=np.float64)
if frame_type == "ESH":
if x.shape != (128, 8):
raise ValueError("For ESH, frame_F_in must have shape (128, 8).")
y = np.empty_like(x, dtype=np.float64)
a_out = np.empty((PRED_ORDER, 8), dtype=np.float64)
for j in range(8):
y[:, j], a_out[:, j] = _tns_one_vector(x[:, j])
return y, a_out
if x.shape == (1024,):
x_vec = x
out_shape = (1024,)
elif x.shape == (1024, 1):
x_vec = x[:, 0]
out_shape = (1024, 1)
else:
raise ValueError('For non-ESH, frame_F_in must have shape (1024,) or (1024, 1).')
y_vec, a_q = _tns_one_vector(x_vec)
if out_shape == (1024,):
y_out = y_vec
else:
y_out = y_vec.reshape(1024, 1)
a_out = a_q.reshape(PRED_ORDER, 1)
return y_out, a_out
def aac_i_tns(frame_F_in: FrameChannelF, frame_type: FrameType, tns_coeffs: TnsCoeffs) -> FrameChannelF:
"""
Inverse Temporal Noise Shaping (iTNS) for ONE channel.
Parameters
----------
frame_F_in : FrameChannelF
Per-channel MDCT coefficients after TNS.
Expected (typical) shapes:
- If frame_type == "ESH": (128, 8)
- Else: (1024, 1) or (1024,)
frame_type : FrameType
Frame type code ("OLS", "LSS", "ESH", "LPS").
tns_coeffs : TnsCoeffs
Quantized TNS predictor coefficients.
Expected shapes:
- If frame_type == "ESH": (PRED_ORDER, 8)
- Else: (PRED_ORDER, 1)
Returns
-------
FrameChannelF
Per-channel MDCT coefficients after inverse TNS.
Same shape convention as input frame_F_in.
"""
x = np.asarray(frame_F_in, dtype=np.float64)
a = np.asarray(tns_coeffs, dtype=np.float64)
if frame_type == "ESH":
if x.shape != (128, 8):
raise ValueError("For ESH, frame_F_in must have shape (128, 8).")
if a.shape != (PRED_ORDER, 8):
raise ValueError("For ESH, tns_coeffs must have shape (PRED_ORDER, 8).")
y = np.empty_like(x, dtype=np.float64)
for j in range(8):
y[:, j] = _apply_itns_iir(x[:, j], a[:, j])
return y
if a.shape != (PRED_ORDER, 1):
raise ValueError("For non-ESH, tns_coeffs must have shape (PRED_ORDER, 1).")
if x.shape == (1024,):
x_vec = x
out_shape = (1024,)
elif x.shape == (1024, 1):
x_vec = x[:, 0]
out_shape = (1024, 1)
else:
raise ValueError('For non-ESH, frame_F_in must have shape (1024,) or (1024, 1).')
y_vec = _apply_itns_iir(x_vec, a[:, 0])
if out_shape == (1024,):
return y_vec
return y_vec.reshape(1024, 1)
+97 -8
View File
@@ -10,7 +10,6 @@
#
# Description:
# This module implements Public Type aliases
#
# ------------------------------------------------------------
from __future__ import annotations
@@ -39,7 +38,7 @@ Window type codes (AAC):
"""
ChannelKey: TypeAlias = Literal["chl", "chr"]
"""Channel dictionary keys used in Level 1 payloads."""
"""Channel dictionary keys used in Level payloads."""
# -----------------------------------------------------------------------------
@@ -105,6 +104,40 @@ Examples:
dtype: float64
"""
MdctFrameChannel: TypeAlias = FloatArray
"""
Per-channel MDCT container used in Level-1/2 sequences.
Typical shapes:
- If frame_type in {"OLS","LSS","LPS"}: (1024, 1) or (1024,)
- If frame_type == "ESH": (128, 8) (8 short subframes for one channel)
dtype: float64
Notes
-----
Some parts of the assignment store long-frame coefficients as a column vector
(1024, 1) to match MATLAB conventions. Internally you may also use (1024,)
when convenient, but the semantic meaning is identical.
"""
TnsCoeffs: TypeAlias = FloatArray
"""
Quantized TNS predictor coefficients (one channel).
Typical shapes (Level 2):
- If frame_type == "ESH": (4, 8) (order p=4 for each of the 8 short subframes)
- Else: (4, 1) (order p=4 for the long frame)
dtype: float64
Notes
-----
The assignment uses a 4-bit uniform symmetric quantizer with step size 0.1.
We store the quantized coefficient values as float64 (typically multiples of 0.1)
to keep the pipeline simple and readable.
"""
FrameT: TypeAlias = FloatArray
"""
@@ -142,17 +175,23 @@ Rationale for ESH (128, 16):
dtype: float64
"""
FrameChannelF: TypeAlias = FloatArray
FrameChannelF: TypeAlias = MdctFrameChannel
"""
Frequency-domain single-channel frame (MDCT coefficients).
Frequency-domain single-channel MDCT coefficients.
Typical shapes (Level 1):
- If frame_type in {"OLS","LSS","LPS"}: (1024,)
- If frame_type == "ESH": (128, 8) (8 short subframes for one channel)
Typical shapes (Level 1/2):
- If frame_type in {"OLS","LSS","LPS"}: (1024, 1) or (1024,)
- If frame_type == "ESH": (128, 8)
dtype: float64
"""
BandRanges: TypeAlias = list[tuple[int, int]]
"""
Bark-band index ranges [start, end] (inclusive) for MDCT lines.
Used by TNS to map MDCT indices k to Bark bands.
"""
# -----------------------------------------------------------------------------
# Level 1 AAC sequence payload types
@@ -168,7 +207,7 @@ class AACChannelFrameF(TypedDict):
The MDCT coefficients for ONE channel.
Typical shapes:
- ESH: (128, 8) (8 short subframes)
- else: (1024, )
- else: (1024, 1) or (1024,)
"""
frame_F: FrameChannelF
@@ -191,3 +230,53 @@ List of length K (K = number of frames).
Each element is a dict with keys:
- "frame_type", "win_type", "chl", "chr"
"""
# -----------------------------------------------------------------------------
# Level 2 AAC sequence payload types (TNS)
# -----------------------------------------------------------------------------
class AACChannelFrameF2(TypedDict):
"""
Per-channel payload for aac_seq_2[i]["chl"] or ["chr"] (Level 2).
Keys
----
frame_F:
The TNS-processed MDCT coefficients for ONE channel.
Typical shapes:
- ESH: (128, 8)
- else: (1024, 1) or (1024,)
tns_coeffs:
Quantized TNS predictor coefficients for ONE channel.
Typical shapes:
- ESH: (PRED_ORDER, 8)
- else: (PRED_ORDER, 1)
"""
frame_F: FrameChannelF
tns_coeffs: TnsCoeffs
class AACSeq2Frame(TypedDict):
"""
One frame dictionary element of aac_seq_2 (Level 2).
"""
frame_type: FrameType
win_type: WinType
chl: AACChannelFrameF2
chr: AACChannelFrameF2
AACSeq2: TypeAlias = List[AACSeq2Frame]
"""
AAC sequence for Level 2:
List of length K (K = number of frames).
Each element is a dict with keys:
- "frame_type", "win_type", "chl", "chr"
Level 2 adds:
- per-channel "tns_coeffs"
and stores:
- per-channel "frame_F" after applying TNS.
"""
+117 -48
View File
@@ -19,59 +19,15 @@ import numpy as np
import pytest
import soundfile as sf
from core.aac_coder import aac_coder_1
from core.aac_decoder import aac_decoder_1
from core.aac_coder import aac_coder_1, aac_coder_2, aac_read_wav_stereo_48k
from core.aac_decoder import aac_decoder_1, aac_decoder_2, aac_remove_padding
from core.aac_types import *
from core.aac_snr_db import snr_db
# Helper "fixtures" for aac_coder_1 / i_aac_coder_1
# -----------------------------------------------------------------------------
def _snr_db(x_ref: StereoSignal, x_hat: StereoSignal) -> float:
"""
Compute overall SNR (dB) over all samples and channels after aligning lengths.
Parameters
----------
x_ref : StereoSignal
Reference signal, shape (N, 2) typical.
x_hat : StereoSignal
Reconstructed signal, shape (M, 2) typical.
Returns
-------
float
SNR in dB.
- Returns +inf if noise power is zero.
- Returns -inf if signal power is zero.
"""
x_ref = np.asarray(x_ref, dtype=np.float64)
x_hat = np.asarray(x_hat, dtype=np.float64)
# Be conservative: align lengths and common channels.
if x_ref.ndim == 1:
x_ref = x_ref.reshape(-1, 1)
if x_hat.ndim == 1:
x_hat = x_hat.reshape(-1, 1)
n = min(x_ref.shape[0], x_hat.shape[0])
c = min(x_ref.shape[1], x_hat.shape[1])
x_ref = x_ref[:n, :c]
x_hat = x_hat[:n, :c]
err = x_ref - x_hat
ps = float(np.sum(x_ref * x_ref))
pn = float(np.sum(err * err))
if pn <= 0.0:
return float("inf")
if ps <= 0.0:
return float("-inf")
return float(10.0 * np.log10(ps / pn))
@pytest.fixture()
def tmp_stereo_wav(tmp_path: Path) -> Path:
"""
@@ -89,6 +45,56 @@ def tmp_stereo_wav(tmp_path: Path) -> Path:
return wav_path
# -----------------------------------------------------------------------------
# Helper-function tests
# -----------------------------------------------------------------------------
def test_aac_read_wav_stereo_48k_roundtrip(tmp_stereo_wav: Path) -> None:
"""
Contract test for aac_read_wav_stereo_48k():
- Reads stereo WAV
- Returns float64 array with shape (N,2)
- Returns fs = 48000
"""
x, fs = aac_read_wav_stereo_48k(tmp_stereo_wav)
assert int(fs) == 48000
assert isinstance(x, np.ndarray)
assert x.dtype == np.float64
assert x.ndim == 2
assert x.shape[1] == 2
assert x.shape[0] > 0
def test_aac_remove_padding_removes_hop_from_both_ends() -> None:
"""
Contract test for aac_remove_padding():
- Removes 'hop' samples from start and end.
"""
hop = 1024
n = 10000
y_pad: StereoSignal = np.zeros((n, 2), dtype=np.float64)
y: StereoSignal = aac_remove_padding(y_pad, hop=hop)
assert y.shape == (n - 2 * hop, 2)
assert y.dtype == np.float64
def test_aac_remove_padding_errors_on_too_short_input() -> None:
"""
aac_remove_padding must raise if y_pad is shorter than 2*hop.
"""
hop = 1024
y_pad: StereoSignal = np.zeros((2 * hop - 1, 2), dtype=np.float64)
with pytest.raises(ValueError):
_ = aac_remove_padding(y_pad, hop=hop)
# -----------------------------------------------------------------------------
# Level 1 tests
# -----------------------------------------------------------------------------
def test_aac_coder_seq_schema_and_shapes(tmp_stereo_wav: Path) -> None:
"""
Module-level contract test:
@@ -152,5 +158,68 @@ def test_end_to_end_aac_coder_decoder_high_snr(tmp_stereo_wav: Path, tmp_path: P
assert int(fs_hat) == 48000
# SNR against returned array (file should match closely, but we do not require it here).
snr = _snr_db(x_ref, x_hat)
snr = snr_db(x_ref, x_hat)
assert snr > 80.0
# -----------------------------------------------------------------------------
# Level 2 tests (new)
# -----------------------------------------------------------------------------
def test_aac_coder_2_seq_schema_and_shapes(tmp_stereo_wav: Path) -> None:
"""
Module-level contract test (Level 2):
Ensure aac_seq_2 follows the expected schema and per-frame shapes, including tns_coeffs.
"""
aac_seq: AACSeq2 = aac_coder_2(tmp_stereo_wav)
assert isinstance(aac_seq, list)
assert len(aac_seq) > 0
for fr in aac_seq:
assert "frame_type" in fr
assert "win_type" in fr
assert "chl" in fr
assert "chr" in fr
frame_type: FrameType = fr["frame_type"]
assert frame_type in ("OLS", "LSS", "ESH", "LPS")
for ch_key in ("chl", "chr"):
ch = fr[ch_key]
assert "frame_F" in ch
assert "tns_coeffs" in ch
frame_f = np.asarray(ch["frame_F"], dtype=np.float64)
coeffs = np.asarray(ch["tns_coeffs"], dtype=np.float64)
if frame_type == "ESH":
assert frame_f.shape == (128, 8)
assert coeffs.shape[0] == 4
assert coeffs.shape[1] == 8
else:
assert frame_f.shape == (1024, 1)
assert coeffs.shape == (4, 1)
def test_end_to_end_level_2_high_snr(tmp_stereo_wav: Path, tmp_path: Path) -> None:
"""
End-to-end test (Level 2):
Encode + decode and check SNR remains very high.
Level 2 is still floating-point (TNS is reversible), so reconstruction
should remain numerical-noise only.
"""
x_ref, fs = sf.read(str(tmp_stereo_wav), always_2d=True)
x_ref = np.asarray(x_ref, dtype=np.float64)
assert int(fs) == 48000
out_wav = tmp_path / "out_l2.wav"
aac_seq = aac_coder_2(tmp_stereo_wav)
x_hat: StereoSignal = aac_decoder_2(aac_seq, out_wav)
assert out_wav.exists()
_, fs_hat = sf.read(str(out_wav), always_2d=True)
assert int(fs_hat) == 48000
snr = snr_db(x_ref, x_hat)
assert snr > 75.0
+4 -17
View File
@@ -17,6 +17,7 @@ from typing import Sequence
import pytest
from core.aac_filterbank import aac_filter_bank, aac_i_filter_bank
from core.aac_snr_db import snr_db
from core.aac_types import *
# Helper fixtures for filterbank
@@ -56,20 +57,6 @@ def _ola_reconstruct(x: StereoSignal, frame_types: Sequence[FrameType], win_type
return y
def _snr_db(x: StereoSignal, y: StereoSignal) -> float:
"""
Compute SNR in dB over all samples/channels.
"""
err = x - y
ps = float(np.sum(x * x))
pn = float(np.sum(err * err))
if pn <= 0.0:
return float("inf")
if ps <= 0.0:
return float("-inf")
return 10.0 * float(np.log10(ps / pn))
# -----------------------------------------------------------------------------
# Forward filterbank tests
# -----------------------------------------------------------------------------
@@ -223,7 +210,7 @@ def test_ola_reconstruction_ols_high_snr(win_type: WinType) -> None:
a = 1024
b = N - 1024
snr = _snr_db(x[a:b, :], y[a:b, :])
snr = snr_db(x[a:b, :], y[a:b, :])
assert snr > 50.0
@@ -244,7 +231,7 @@ def test_ola_reconstruction_esh_high_snr(win_type: WinType) -> None:
a = 1024
b = N - 1024
snr = _snr_db(x[a:b, :], y[a:b, :])
snr = snr_db(x[a:b, :], y[a:b, :])
assert snr > 45.0
@@ -265,5 +252,5 @@ def test_ola_reconstruction_transition_sequence(win_type: WinType) -> None:
a = 1024
b = N - 1024
snr = _snr_db(x[a:b, :], y[a:b, :])
snr = snr_db(x[a:b, :], y[a:b, :])
assert snr > 40.0
+98
View File
@@ -0,0 +1,98 @@
# ------------------------------------------------------------
# AAC Coder/Decoder - SNR dB Tests
#
# Multimedia course at Aristotle University of
# Thessaloniki (AUTh)
#
# Author:
# Christos Choutouridis (ΑΕΜ 8997)
# cchoutou@ece.auth.gr
#
# Description:
# Basic tests for SNR calculation utility.
# ------------------------------------------------------------
from __future__ import annotations
import numpy as np
import pytest
from core.aac_snr_db import snr_db
from core.aac_types import StereoSignal
def test_snr_perfect_reconstruction_returns_inf() -> None:
"""
If x_hat == x_ref exactly, noise power is zero and SNR must be +inf.
"""
rng = np.random.default_rng(0)
x: StereoSignal = rng.normal(size=(1024, 2)).astype(np.float64)
snr = snr_db(x, x)
assert snr == float("inf")
def test_snr_zero_reference_returns_minus_inf() -> None:
"""
If reference signal is identically zero, signal power is zero
and SNR must be -inf (unless noise is also zero, which is degenerate).
"""
x_ref: StereoSignal = np.zeros((1024, 2), dtype=np.float64)
x_hat: StereoSignal = np.ones((1024, 2), dtype=np.float64)
snr = snr_db(x_ref, x_hat)
assert snr == float("-inf")
def test_snr_known_noise_level_matches_expected_value() -> None:
"""
Deterministic test with known signal and noise power.
Let:
x_ref = ones
x_hat = ones + noise
With noise variance sigma^2, expected SNR:
10 * log10(Ps / Pn)
"""
n = 1000
sigma = 0.1
x_ref: StereoSignal = np.ones((n, 2), dtype=np.float64)
noise = sigma * np.ones((n, 2), dtype=np.float64)
x_hat: StereoSignal = x_ref + noise
ps = float(np.sum(x_ref * x_ref))
pn = float(np.sum(noise * noise))
expected = 10.0 * np.log10(ps / pn)
snr = snr_db(x_ref, x_hat)
assert np.isclose(snr, expected, rtol=1e-12, atol=1e-12)
def test_snr_aligns_different_lengths_and_channels() -> None:
"""
The function must:
- align to minimum length
- align to minimum channel count
without crashing.
"""
rng = np.random.default_rng(1)
x_ref: StereoSignal = rng.normal(size=(1000, 2)).astype(np.float64)
x_hat: StereoSignal = rng.normal(size=(800, 1)).astype(np.float64)
snr = snr_db(x_ref, x_hat)
assert np.isfinite(snr)
def test_snr_accepts_1d_inputs() -> None:
"""
1-D inputs must be accepted and treated as single-channel signals.
"""
rng = np.random.default_rng(2)
x_ref = rng.normal(size=1024).astype(np.float64)
x_hat = x_ref + 0.01 * rng.normal(size=1024).astype(np.float64)
snr = snr_db(x_ref, x_hat)
assert np.isfinite(snr)
+196
View File
@@ -0,0 +1,196 @@
# ------------------------------------------------------------
# AAC Coder/Decoder - TNS Tests
#
# Multimedia course at Aristotle University of
# Thessaloniki (AUTh)
#
# Author:
# Christos Choutouridis (ΑΕΜ 8997)
# cchoutou@ece.auth.gr
#
# Description:
# Tests for Temporal Noise Shaping (TNS) module (Level 2).
#
# Validates:
# - I/O shapes for long and ESH modes
# - Quantization grid and clamping of predictor coefficients
# - Inverse-filter stability (all poles inside unit circle)
# - Functional correctness: iTNS(TNS(X)) ≈ X
# ------------------------------------------------------------
from __future__ import annotations
import pytest
from core.aac_configuration import PRED_ORDER, QUANT_MAX, QUANT_STEP
from core.aac_tns import aac_tns, aac_i_tns
from core.aac_types import *
# -----------------------------------------------------------------------------
# Helper utilities
# -----------------------------------------------------------------------------
def _is_inverse_stable_from_coeffs(a_q: MdctCoeffs) -> bool:
"""
Check stability of the inverse TNS filter H_TNS^{-1}.
Poles are roots of:
z^p - a1 z^{p-1} - ... - ap = 0
Stability condition:
|root| < 1 for all roots.
"""
a_q = np.asarray(a_q, dtype=np.float64).reshape(-1)
p = int(a_q.shape[0])
poly = np.empty(p + 1, dtype=np.float64)
poly[0] = 1.0
poly[1:] = -a_q
roots = np.roots(poly)
margin = 1e-12
return bool(np.all(np.abs(roots) < (1.0 - margin)))
def _assert_quantized_and_clamped(a_q: MdctCoeffs) -> None:
"""
Assert that coefficients:
- lie on the QUANT_STEP grid
- are clamped to [-QUANT_MAX, +QUANT_MAX]
"""
a_q = np.asarray(a_q, dtype=np.float64)
assert np.max(np.abs(a_q)) <= (QUANT_MAX + 1e-12)
grid = a_q / float(QUANT_STEP)
assert np.max(np.abs(grid - np.round(grid))) < 1e-12
# -----------------------------------------------------------------------------
# Shape / contract tests
# -----------------------------------------------------------------------------
@pytest.mark.parametrize("frame_type", ["OLS", "LSS", "LPS"])
def test_tns_shapes_long_sequences(frame_type: FrameType) -> None:
"""
Contract test (long frames):
- Input shape: (1024, 1)
- Output shape preserved
- Predictor coeffs shape: (PRED_ORDER, 1)
"""
rng = np.random.default_rng(0)
frame_F_in: FrameChannelF = rng.normal(size=(1024, 1)).astype(np.float64)
frame_F_out, tns_coeffs = aac_tns(frame_F_in, frame_type)
assert frame_F_out.shape == frame_F_in.shape
assert tns_coeffs.shape == (PRED_ORDER, 1)
def test_tns_shapes_esh() -> None:
"""
Contract test (ESH):
- Input shape: (128, 8)
- Output shape preserved
- Predictor coeffs shape: (PRED_ORDER, 8)
"""
rng = np.random.default_rng(1)
frame_F_in: FrameChannelF = rng.normal(size=(128, 8)).astype(np.float64)
frame_F_out, tns_coeffs = aac_tns(frame_F_in, "ESH")
assert frame_F_out.shape == (128, 8)
assert tns_coeffs.shape == (PRED_ORDER, 8)
# -----------------------------------------------------------------------------
# Coefficient properties
# -----------------------------------------------------------------------------
@pytest.mark.parametrize("frame_type", ["OLS", "LSS", "LPS"])
def test_tns_coeffs_quantized_clamped_and_stable_long(frame_type: FrameType) -> None:
"""
Long-frame predictor coefficients must be:
- quantized on QUANT_STEP grid
- clamped to [-QUANT_MAX, +QUANT_MAX]
- stable for inverse filtering
"""
rng = np.random.default_rng(2)
frame_F_in: FrameChannelF = rng.normal(size=(1024, 1)).astype(np.float64)
_, tns_coeffs = aac_tns(frame_F_in, frame_type)
a_q: MdctCoeffs = tns_coeffs[:, 0]
_assert_quantized_and_clamped(a_q)
assert _is_inverse_stable_from_coeffs(a_q)
def test_tns_coeffs_quantized_clamped_and_stable_esh() -> None:
"""
ESH predictor coefficients must satisfy quantization and stability
independently for each of the 8 short subframes.
"""
rng = np.random.default_rng(3)
frame_F_in: FrameChannelF = rng.normal(size=(128, 8)).astype(np.float64)
_, tns_coeffs = aac_tns(frame_F_in, "ESH")
for j in range(8):
a_q: MdctCoeffs = tns_coeffs[:, j]
_assert_quantized_and_clamped(a_q)
assert _is_inverse_stable_from_coeffs(a_q)
# -----------------------------------------------------------------------------
# Functional correctness (round-trip)
# -----------------------------------------------------------------------------
@pytest.mark.parametrize("frame_type", ["OLS", "LSS", "LPS"])
def test_tns_roundtrip_long_is_close(frame_type: FrameType) -> None:
"""
Functional test:
iTNS(TNS(X)) ≈ X for long frames.
"""
rng = np.random.default_rng(4)
frame_F_in: FrameChannelF = rng.normal(size=(1024, 1)).astype(np.float64)
frame_F_tns, tns_coeffs = aac_tns(frame_F_in, frame_type)
frame_F_hat = aac_i_tns(frame_F_tns, frame_type, tns_coeffs)
np.testing.assert_allclose(frame_F_hat, frame_F_in, rtol=1e-9, atol=1e-9)
def test_tns_roundtrip_esh_is_close() -> None:
"""
Functional test:
iTNS(TNS(X)) ≈ X for ESH frames (8 independent subframes).
"""
rng = np.random.default_rng(5)
frame_F_in: FrameChannelF = rng.normal(size=(128, 8)).astype(np.float64)
frame_F_tns, tns_coeffs = aac_tns(frame_F_in, "ESH")
frame_F_hat = aac_i_tns(frame_F_tns, "ESH", tns_coeffs)
np.testing.assert_allclose(frame_F_hat, frame_F_in, rtol=1e-9, atol=1e-9)
# -----------------------------------------------------------------------------
# Sanity
# -----------------------------------------------------------------------------
def test_tns_outputs_are_finite() -> None:
"""
Sanity test: no NaN or inf in outputs.
"""
rng = np.random.default_rng(6)
frame_F_long: FrameChannelF = rng.normal(size=(1024, 1)).astype(np.float64)
out_long, coeffs_long = aac_tns(frame_F_long, "OLS")
assert np.isfinite(out_long).all()
assert np.isfinite(coeffs_long).all()
frame_F_esh: FrameChannelF = rng.normal(size=(128, 8)).astype(np.float64)
out_esh, coeffs_esh = aac_tns(frame_F_esh, "ESH")
assert np.isfinite(out_esh).all()
assert np.isfinite(coeffs_esh).all()