klemenk commited on
Commit
d84f8d0
·
verified ·
1 Parent(s): 4c7497c

Add causal and centered WavCoch encoding modes

Browse files
README.md ADDED
@@ -0,0 +1,101 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ ---
2
+ license: apache-2.0
3
+ tags:
4
+ - audio
5
+ - speech
6
+ - tokenizer
7
+ - vocoder
8
+ - wavcoch
9
+ library_name: transformers
10
+ base_model: TuKoResearch/WavCochCausalV8192-vocoder
11
+ ---
12
+
13
+ # WavCoch V8192 vocoder with causal and centered encoding
14
+
15
+ **WavCoch** is a causal waveform-to-cochleagram tokenizer by **Greta Tuckute** and **Klemen Kotar**.
16
+
17
+
18
+
19
+ ## Model Details
20
+
21
+ | Parameter | Value |
22
+ |-----------|-------|
23
+ | Parameters | ~24.42M |
24
+ | Window Size | 1001 |
25
+ | Hop Length | 80 |
26
+ | Encoder Dim | 512 |
27
+ | Vocabulary Size | 8192 |
28
+ | Includes Vocoder | True |
29
+
30
+ ## Usage
31
+
32
+ ```python
33
+ import torch
34
+ from transformers import AutoModel
35
+
36
+ wavcoch = AutoModel.from_pretrained(
37
+ "TuKoResearch/WavCochCausalV8192-vocoder-causal-centered",
38
+ trust_remote_code=True,
39
+ )
40
+
41
+ # Default mode is identical to TuKoResearch/WavCochCausalV8192-vocoder.
42
+ causal_codes = wavcoch.quantize(waveform_tensor)
43
+ assert torch.equal(causal_codes, wavcoch.quantize(waveform_tensor, mode="causal"))
44
+
45
+ # Centered mode preserves the number of frames while centering the 1001-sample
46
+ # analysis window on the token endpoint. It uses 421 left and 500 right zeros.
47
+ centered_codes = wavcoch.quantize(waveform_tensor, mode="centered")
48
+
49
+ codes = causal_codes
50
+ coch = wavcoch.decode(codes)
51
+ embeddings = wavcoch(
52
+ input_values=waveform_tensor,
53
+ output_hidden_states=True,
54
+ sampling_rate=16000,
55
+ ).hidden_states[0]
56
+
57
+ audio = wavcoch.decode_audio(codes)
58
+ ```
59
+
60
+ ## Notes
61
+
62
+ This repo includes a bundled vocoder and supports `decode_audio(...)` for end-to-end waveform synthesis.
63
+
64
+ `mode="causal"` is the default and uses the original 921 samples of left
65
+ padding. `mode="centered"` redistributes the same total padding as 421 samples
66
+ on the left and 500 samples on the right. This keeps frame counts unchanged
67
+ while moving the analysis-window center forward by 500 samples (31.25 ms at
68
+ 16 kHz). Centered encoding uses future waveform context and is therefore not
69
+ streaming-causal.
70
+
71
+ The mode argument is accepted by `forward(...)`, `quantize(...)`, and
72
+ `wav2coch(...)`. Decoding methods operate on codes and do not take a mode.
73
+
74
+ ## Compatibility and validation
75
+
76
+ This repository uses the exact `model.safetensors` weights from
77
+ `TuKoResearch/WavCochCausalV8192-vocoder`; only the self-contained runtime,
78
+ configuration metadata, and model card add the encoding-mode API.
79
+
80
+ Validation used float32 with TF32 disabled:
81
+
82
+ - Default encoding and explicit `mode="causal"` produced bit-exact codes,
83
+ hidden states, decoded cochleagrams, and decoded audio relative to the base
84
+ model.
85
+ - Exact causal compatibility held for input lengths both divisible and not
86
+ divisible by the 80-sample hop.
87
+ - Six synthetic beep configurations produced the same 200 frames for causal
88
+ and centered modes from one second of audio.
89
+ - Centered decoded-cochleagram power advanced by 6 frames (30 ms, the nearest
90
+ 5 ms token-grid value to 31.25 ms), with shifted-curve Pearson correlations
91
+ from 0.986 to 0.999.
92
+ - AuriStream7BDeep-40Pred float32 head-1 and head-40 surprisals were bit-exact
93
+ between the base model's codes and this model's causal codes.
94
+
95
+ Centered mode changes the encoded token sequence and uses 31.25 ms of future
96
+ waveform context. Models trained on causal WavCoch codes should therefore be
97
+ evaluated explicitly before centered codes are used for likelihood estimates
98
+ or generation.
99
+
100
+ When called with `output_hidden_states=True`, WavCoch exposes a single hidden-state layer:
101
+ the post-FSQ projected embedding sequence used for direct probing.
config.json ADDED
@@ -0,0 +1,78 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ {
2
+ "model_type": "wavcoch",
3
+ "architectures": [
4
+ "WavCoch"
5
+ ],
6
+ "auto_map": {
7
+ "AutoConfig": "configuration_wavcoch.WavCochConfig",
8
+ "AutoModel": "modeling_wavcoch.WavCoch"
9
+ },
10
+ "torch_dtype": "float32",
11
+ "transformers_version": "4.40.0",
12
+ "sample_rate": 16000,
13
+ "causal_pad_mode": "repeat",
14
+ "out_channels": 211,
15
+ "has_vocoder": true,
16
+ "vocoder_upsample_rates": [
17
+ 5,
18
+ 4,
19
+ 2,
20
+ 2
21
+ ],
22
+ "vocoder_upsample_kernel_sizes": [
23
+ 10,
24
+ 8,
25
+ 4,
26
+ 4
27
+ ],
28
+ "vocoder_upsample_initial_channel": 512,
29
+ "vocoder_resblock": "1",
30
+ "vocoder_resblock_kernel_sizes": [
31
+ 11,
32
+ 7,
33
+ 3
34
+ ],
35
+ "vocoder_resblock_dilation_sizes": [
36
+ [
37
+ 1,
38
+ 3,
39
+ 5
40
+ ],
41
+ [
42
+ 1,
43
+ 3,
44
+ 5
45
+ ],
46
+ [
47
+ 1,
48
+ 3,
49
+ 5
50
+ ]
51
+ ],
52
+ "window_size": 1001,
53
+ "window_padding": 921,
54
+ "default_encoding_mode": "causal",
55
+ "encoding_modes": [
56
+ "causal",
57
+ "centered"
58
+ ],
59
+ "centered_left_padding": 421,
60
+ "centered_right_padding": 500,
61
+ "hop_length": 80,
62
+ "causal_convs": true,
63
+ "encoder_layers": 8,
64
+ "encoder_dim": 512,
65
+ "encoder_kernel_size": 3,
66
+ "decoder_layers": 8,
67
+ "decoder_dim": 512,
68
+ "decoder_kernel_size": 9,
69
+ "quantizer": "FSQ",
70
+ "channels": [
71
+ 8,
72
+ 8,
73
+ 8,
74
+ 4,
75
+ 4
76
+ ],
77
+ "vocab_size": 8192
78
+ }
configuration_wavcoch.py ADDED
@@ -0,0 +1,87 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """
2
+ WavCoch configuration for Hugging Face Transformers.
3
+ """
4
+
5
+ from transformers import PretrainedConfig
6
+
7
+
8
+ class WavCochConfig(PretrainedConfig):
9
+ """Configuration class for WavCoch checkpoints with optional vocoder."""
10
+
11
+ model_type = "wavcoch"
12
+
13
+ def __init__(
14
+ self,
15
+ window_size: int = 1001,
16
+ window_padding: int = 1000,
17
+ centered_left_padding: int = None,
18
+ centered_right_padding: int = None,
19
+ hop_length: int = 80,
20
+ out_channels: int = 211,
21
+ causal_convs: bool = True,
22
+ causal_pad_mode: str = "repeat",
23
+ encoder_layers: int = 8,
24
+ encoder_dim: int = 512,
25
+ encoder_kernel_size: int = 3,
26
+ decoder_layers: int = 8,
27
+ decoder_dim: int = 512,
28
+ decoder_kernel_size: int = 9,
29
+ quantizer: str = "FSQ",
30
+ channels=None,
31
+ vocab_size: int = None,
32
+ sample_rate: int = 16000,
33
+ has_vocoder: bool = False,
34
+ vocoder_upsample_rates=None,
35
+ vocoder_upsample_kernel_sizes=None,
36
+ vocoder_upsample_initial_channel: int = 512,
37
+ vocoder_resblock: str = "1",
38
+ vocoder_resblock_kernel_sizes=None,
39
+ vocoder_resblock_dilation_sizes=None,
40
+ **kwargs,
41
+ ):
42
+ channels = list(channels or [8, 8, 8, 4, 4])
43
+ if vocab_size is None:
44
+ vocab_size = 1
45
+ for level in channels:
46
+ vocab_size *= int(level)
47
+
48
+ self.window_size = int(window_size)
49
+ self.window_padding = int(window_padding)
50
+ half_window = (self.window_size - 1) // 2
51
+ self.centered_right_padding = int(
52
+ half_window if centered_right_padding is None else centered_right_padding
53
+ )
54
+ expected_left_padding = self.window_padding - self.centered_right_padding
55
+ self.centered_left_padding = int(
56
+ expected_left_padding if centered_left_padding is None else centered_left_padding
57
+ )
58
+ if self.centered_left_padding + self.centered_right_padding != self.window_padding:
59
+ raise ValueError(
60
+ "Centered left/right padding must sum to window_padding so encoding length stays unchanged"
61
+ )
62
+ self.hop_length = int(hop_length)
63
+ self.out_channels = int(out_channels)
64
+ self.causal_convs = bool(causal_convs)
65
+ self.causal_pad_mode = str(causal_pad_mode)
66
+ self.encoder_layers = int(encoder_layers)
67
+ self.encoder_dim = int(encoder_dim)
68
+ self.encoder_kernel_size = int(encoder_kernel_size)
69
+ self.decoder_layers = int(decoder_layers)
70
+ self.decoder_dim = int(decoder_dim)
71
+ self.decoder_kernel_size = int(decoder_kernel_size)
72
+ self.quantizer = str(quantizer)
73
+ self.channels = channels
74
+ self.vocab_size = int(vocab_size)
75
+ self.sample_rate = int(sample_rate)
76
+
77
+ self.has_vocoder = bool(has_vocoder)
78
+ self.vocoder_upsample_rates = list(vocoder_upsample_rates or [5, 4, 2, 2])
79
+ self.vocoder_upsample_kernel_sizes = list(vocoder_upsample_kernel_sizes or [10, 8, 4, 4])
80
+ self.vocoder_upsample_initial_channel = int(vocoder_upsample_initial_channel)
81
+ self.vocoder_resblock = str(vocoder_resblock)
82
+ self.vocoder_resblock_kernel_sizes = list(vocoder_resblock_kernel_sizes or [11, 7, 3])
83
+ self.vocoder_resblock_dilation_sizes = [
84
+ list(d) for d in (vocoder_resblock_dilation_sizes or [[1, 3, 5], [1, 3, 5], [1, 3, 5]])
85
+ ]
86
+
87
+ super().__init__(**kwargs)
configure_wavcoch.py ADDED
@@ -0,0 +1,8 @@
 
 
 
 
 
 
 
 
 
1
+ """
2
+ Backward-compatible import shim for older WavCoch repos.
3
+ """
4
+
5
+ from .configuration_wavcoch import WavCochConfig
6
+
7
+
8
+ __all__ = ["WavCochConfig"]
model.safetensors ADDED
@@ -0,0 +1,3 @@
 
 
 
 
1
+ version https://git-lfs.github.com/spec/v1
2
+ oid sha256:578c6091b20c5813eaa82571458781404aa24aedb26c2de89e1b34d8621af190
3
+ size 97726648
modeling_wavcoch.py ADDED
@@ -0,0 +1,640 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """
2
+ WavCoch model for Hugging Face Transformers.
3
+
4
+ This implementation is self-contained so HF-hosted WavCoch checkpoints do not
5
+ depend on the local auristream package or vector_quantize_pytorch.
6
+ """
7
+
8
+ import math
9
+ import os
10
+ from typing import List, Optional, Tuple
11
+
12
+ os.environ.setdefault("USE_TORCH_XLA", "0")
13
+
14
+ import torch
15
+ import torch.nn as nn
16
+ import torch.nn.functional as F
17
+ from torch.nn import Conv1d, ConvTranspose1d
18
+ from torch.nn.utils import remove_weight_norm
19
+ try:
20
+ from torch.nn.utils.parametrizations import weight_norm
21
+ except ImportError: # pragma: no cover - older PyTorch compatibility
22
+ from torch.nn.utils import weight_norm
23
+
24
+ from transformers import PreTrainedModel
25
+ from transformers.modeling_outputs import BaseModelOutput
26
+ try:
27
+ from transformers.tokenization_utils_base import BatchEncoding
28
+ except ImportError: # pragma: no cover - compatibility with older Transformers
29
+ from transformers.tokenization_utils import BatchEncoding
30
+ import transformers.modeling_utils as transformers_modeling_utils
31
+ import transformers.utils.import_utils as transformers_import_utils
32
+
33
+ transformers_import_utils.is_torch_xla_available = lambda *args, **kwargs: False
34
+ transformers_modeling_utils.is_torch_xla_available = lambda *args, **kwargs: False
35
+
36
+ try:
37
+ from .configuration_wavcoch import WavCochConfig
38
+ except ImportError: # pragma: no cover - compatibility with older repos
39
+ from .configure_wavcoch import WavCochConfig
40
+
41
+
42
+ class CausalConv1d(nn.Module):
43
+ """1D causal convolution with left-only padding."""
44
+
45
+ def __init__(
46
+ self,
47
+ in_channels: int,
48
+ out_channels: int,
49
+ kernel_size: int,
50
+ stride: int = 1,
51
+ dilation: int = 1,
52
+ bias: bool = True,
53
+ groups: int = 1,
54
+ pad_mode: str = "repeat",
55
+ constant_value: float = 0.0,
56
+ ):
57
+ super().__init__()
58
+ left_pad = dilation * (kernel_size - 1)
59
+ if pad_mode == "repeat":
60
+ self.pad = nn.ReplicationPad1d((left_pad, 0))
61
+ elif pad_mode == "constant":
62
+ self.pad = nn.ConstantPad1d((left_pad, 0), constant_value)
63
+ else:
64
+ raise ValueError(f"Unsupported pad_mode: {pad_mode}")
65
+ self.conv = nn.Conv1d(
66
+ in_channels,
67
+ out_channels,
68
+ kernel_size=kernel_size,
69
+ stride=stride,
70
+ padding=0,
71
+ dilation=dilation,
72
+ groups=groups,
73
+ bias=bias,
74
+ )
75
+
76
+ def forward(self, x: torch.Tensor) -> torch.Tensor:
77
+ return self.conv(self.pad(x))
78
+
79
+
80
+ def _build_conv1d(
81
+ in_channels: int,
82
+ out_channels: int,
83
+ kernel_size: int,
84
+ *,
85
+ causal: bool,
86
+ dilation: int = 1,
87
+ pad_mode: str = "repeat",
88
+ ):
89
+ if causal:
90
+ return CausalConv1d(
91
+ in_channels,
92
+ out_channels,
93
+ kernel_size=kernel_size,
94
+ stride=1,
95
+ dilation=dilation,
96
+ pad_mode=pad_mode,
97
+ )
98
+
99
+ padding = dilation * (kernel_size - 1) // 2
100
+ return nn.Conv1d(
101
+ in_channels,
102
+ out_channels,
103
+ kernel_size=kernel_size,
104
+ stride=1,
105
+ dilation=dilation,
106
+ padding=padding,
107
+ )
108
+
109
+
110
+ class FSQ(nn.Module):
111
+ """Finite Scalar Quantization with the subset of functionality needed for inference."""
112
+
113
+ def __init__(self, levels: List[int], dim: int):
114
+ super().__init__()
115
+ if not levels:
116
+ raise ValueError("FSQ levels must be non-empty")
117
+
118
+ self.levels = [int(level) for level in levels]
119
+ self.codebook_dim = len(self.levels)
120
+ self.dim = int(dim)
121
+
122
+ level_tensor = torch.tensor(self.levels, dtype=torch.int32)
123
+ basis = torch.cumprod(torch.tensor([1] + self.levels[:-1], dtype=torch.int32), dim=0)
124
+ self.register_buffer("_levels", level_tensor, persistent=False)
125
+ self.register_buffer("_basis", basis, persistent=False)
126
+
127
+ if self.dim != self.codebook_dim:
128
+ self.project_in = nn.Linear(self.dim, self.codebook_dim)
129
+ self.project_out = nn.Linear(self.codebook_dim, self.dim)
130
+ else:
131
+ self.project_in = nn.Identity()
132
+ self.project_out = nn.Identity()
133
+
134
+ def _refresh_level_buffers(self, device: Optional[torch.device] = None):
135
+ level_values = [int(level) for level in self.levels]
136
+ if device is None:
137
+ if isinstance(self.project_in, nn.Linear):
138
+ device = self.project_in.weight.device
139
+ elif isinstance(self.project_out, nn.Linear):
140
+ device = self.project_out.weight.device
141
+ else:
142
+ device = self._levels.device
143
+
144
+ self._levels = torch.tensor(level_values, dtype=torch.int32, device=device)
145
+ self._basis = torch.cumprod(
146
+ torch.tensor([1] + level_values[:-1], dtype=torch.int32, device=device),
147
+ dim=0,
148
+ )
149
+
150
+ def bound(self, z: torch.Tensor, eps: float = 1e-3) -> torch.Tensor:
151
+ levels = self._levels.to(dtype=z.dtype, device=z.device)
152
+ half_l = (levels - 1) * (1 + eps) / 2
153
+ offset = torch.where(
154
+ (self._levels % 2).to(device=z.device) == 0,
155
+ torch.tensor(0.5, device=z.device, dtype=z.dtype),
156
+ torch.tensor(0.0, device=z.device, dtype=z.dtype),
157
+ )
158
+ shift = (offset / half_l).atanh()
159
+ return (z + shift).tanh() * half_l - offset
160
+
161
+ def _scale_and_shift(self, zhat_normalized: torch.Tensor) -> torch.Tensor:
162
+ half_width = (self._levels // 2).to(dtype=zhat_normalized.dtype, device=zhat_normalized.device)
163
+ return (zhat_normalized * half_width) + half_width
164
+
165
+ def _scale_and_shift_inverse(self, zhat: torch.Tensor) -> torch.Tensor:
166
+ half_width = (self._levels // 2).to(dtype=zhat.dtype, device=zhat.device)
167
+ return (zhat - half_width) / half_width
168
+
169
+ def quantize_values(self, z: torch.Tensor) -> torch.Tensor:
170
+ self._refresh_level_buffers(device=z.device)
171
+ half_width = (self._levels // 2).to(dtype=z.dtype, device=z.device)
172
+ return self.bound(z).round() / half_width
173
+
174
+ def codes_to_indices(self, zhat: torch.Tensor) -> torch.Tensor:
175
+ self._refresh_level_buffers(device=zhat.device)
176
+ zhat = self._scale_and_shift(zhat)
177
+ basis = self._basis.to(device=zhat.device, dtype=zhat.dtype)
178
+ return (zhat * basis).sum(dim=-1).to(torch.int32)
179
+
180
+ def indices_to_level_indices(self, indices: torch.Tensor) -> torch.Tensor:
181
+ self._refresh_level_buffers(device=indices.device)
182
+ indices = indices.unsqueeze(-1)
183
+ levels = self._levels.to(device=indices.device)
184
+ basis = self._basis.to(device=indices.device)
185
+ return (indices // basis) % levels
186
+
187
+ def indices_to_codes(self, indices: torch.Tensor) -> torch.Tensor:
188
+ self._refresh_level_buffers(device=indices.device)
189
+ level_indices = self.indices_to_level_indices(indices)
190
+ codes = self._scale_and_shift_inverse(level_indices.to(dtype=torch.float32))
191
+ return self.project_out(codes)
192
+
193
+ def forward(self, z: torch.Tensor):
194
+ orig_dtype = z.dtype
195
+ z = self.project_in(z.to(torch.float32))
196
+ q = self.quantize_values(z)
197
+ indices = self.codes_to_indices(q)
198
+ out = self.project_out(q).to(orig_dtype)
199
+ return out, indices.long()
200
+
201
+
202
+ LRELU_SLOPE = 0.1
203
+
204
+
205
+ def get_padding(kernel_size: int, dilation: int = 1) -> int:
206
+ return int((kernel_size * dilation - dilation) / 2)
207
+
208
+
209
+ def init_weights(module, mean: float = 0.0, std: float = 0.01):
210
+ classname = module.__class__.__name__
211
+ if classname.find("Conv") != -1 and hasattr(module, "weight"):
212
+ module.weight.data.normal_(mean, std)
213
+
214
+
215
+ class ResBlock1(nn.Module):
216
+ __constants__ = ["lrelu_slope"]
217
+
218
+ def __init__(self, channels: int, kernel_size: int = 3, dilation=(1, 3, 5)):
219
+ super().__init__()
220
+ self.lrelu_slope = LRELU_SLOPE
221
+
222
+ ch = channels
223
+ ks = kernel_size
224
+ self.convs1 = nn.Sequential(
225
+ weight_norm(Conv1d(ch, ch, ks, 1, get_padding(ks, dilation[0]), dilation[0])),
226
+ weight_norm(Conv1d(ch, ch, ks, 1, get_padding(ks, dilation[1]), dilation[1])),
227
+ weight_norm(Conv1d(ch, ch, ks, 1, get_padding(ks, dilation[2]), dilation[2])),
228
+ )
229
+ self.convs2 = nn.Sequential(
230
+ weight_norm(Conv1d(ch, ch, ks, 1, get_padding(ks, 1))),
231
+ weight_norm(Conv1d(ch, ch, ks, 1, get_padding(ks, 1))),
232
+ weight_norm(Conv1d(ch, ch, ks, 1, get_padding(ks, 1))),
233
+ )
234
+ self.convs1.apply(init_weights)
235
+ self.convs2.apply(init_weights)
236
+
237
+ def forward(self, x: torch.Tensor) -> torch.Tensor:
238
+ for conv1, conv2 in zip(self.convs1, self.convs2):
239
+ xt = F.leaky_relu(x, self.lrelu_slope)
240
+ xt = conv1(xt)
241
+ xt = F.leaky_relu(xt, self.lrelu_slope)
242
+ xt = conv2(xt)
243
+ x = xt + x
244
+ return x
245
+
246
+ def remove_weight_norm(self):
247
+ for layer in self.convs1:
248
+ remove_weight_norm(layer)
249
+ for layer in self.convs2:
250
+ remove_weight_norm(layer)
251
+
252
+
253
+ class ResBlock2(nn.Module):
254
+ __constants__ = ["lrelu_slope"]
255
+
256
+ def __init__(self, channels: int, kernel_size: int = 3, dilation=(1, 3)):
257
+ super().__init__()
258
+ self.lrelu_slope = LRELU_SLOPE
259
+
260
+ ch = channels
261
+ ks = kernel_size
262
+ self.convs = nn.ModuleList(
263
+ [
264
+ weight_norm(Conv1d(ch, ch, ks, 1, get_padding(kernel_size, dilation[0]), dilation[0])),
265
+ weight_norm(Conv1d(ch, ch, ks, 1, get_padding(kernel_size, dilation[1]), dilation[1])),
266
+ ]
267
+ )
268
+ self.convs.apply(init_weights)
269
+
270
+ def forward(self, x: torch.Tensor) -> torch.Tensor:
271
+ for conv in self.convs:
272
+ xt = F.leaky_relu(x, self.lrelu_slope)
273
+ xt = conv(xt)
274
+ x = xt + x
275
+ return x
276
+
277
+ def remove_weight_norm(self):
278
+ for layer in self.convs:
279
+ remove_weight_norm(layer)
280
+
281
+
282
+ class Generator(nn.Module):
283
+ __constants__ = ["lrelu_slope", "num_kernels", "num_upsamples"]
284
+
285
+ def __init__(
286
+ self,
287
+ out_channels: int = 211,
288
+ upsample_rates=None,
289
+ upsample_kernel_sizes=None,
290
+ upsample_initial_channel: int = 512,
291
+ resblock: str = "1",
292
+ resblock_kernel_sizes=None,
293
+ resblock_dilation_sizes=None,
294
+ ):
295
+ super().__init__()
296
+ upsample_rates = list(upsample_rates or [5, 4, 2, 2])
297
+ upsample_kernel_sizes = list(upsample_kernel_sizes or [10, 8, 4, 4])
298
+ resblock_kernel_sizes = list(resblock_kernel_sizes or [11, 7, 3])
299
+ resblock_dilation_sizes = [list(d) for d in (resblock_dilation_sizes or [[1, 3, 5], [1, 3, 5], [1, 3, 5]])]
300
+
301
+ self.num_kernels = len(resblock_kernel_sizes)
302
+ self.num_upsamples = len(upsample_rates)
303
+ self.lrelu_slope = LRELU_SLOPE
304
+
305
+ self.conv_pre = weight_norm(Conv1d(out_channels, upsample_initial_channel, 7, 1, padding=3))
306
+ resblock_cls = ResBlock1 if resblock == "1" else ResBlock2
307
+
308
+ ups = []
309
+ for i, (rate, kernel) in enumerate(zip(upsample_rates, upsample_kernel_sizes)):
310
+ ups.append(
311
+ weight_norm(
312
+ ConvTranspose1d(
313
+ upsample_initial_channel // (2 ** i),
314
+ upsample_initial_channel // (2 ** (i + 1)),
315
+ kernel,
316
+ rate,
317
+ padding=(kernel - rate) // 2,
318
+ )
319
+ )
320
+ )
321
+ self.ups = nn.Sequential(*ups)
322
+
323
+ resblocks = []
324
+ for i in range(len(self.ups)):
325
+ ch = upsample_initial_channel // (2 ** (i + 1))
326
+ resblocks.append(
327
+ nn.Sequential(
328
+ *[
329
+ resblock_cls(ch, kernel, dilation)
330
+ for kernel, dilation in zip(resblock_kernel_sizes, resblock_dilation_sizes)
331
+ ]
332
+ )
333
+ )
334
+ self.resblocks = nn.Sequential(*resblocks)
335
+
336
+ self.conv_post = weight_norm(Conv1d(ch, 1, 17, 1, padding=0))
337
+ self.ups.apply(init_weights)
338
+ self.conv_post.apply(init_weights)
339
+
340
+ def load_state_dict(self, state_dict, strict: bool = True):
341
+ new_state_dict = {}
342
+ for key, value in state_dict.items():
343
+ new_key = key
344
+ if "resblocks" in key:
345
+ parts = key.split(".")
346
+ if len(parts) == 5:
347
+ layer = int(parts[1])
348
+ new_key = f"resblocks.{layer // 3}.{layer % 3}.{'.'.join(parts[2:])}"
349
+ new_state_dict[new_key] = value
350
+
351
+ current_state = self.state_dict()
352
+ for key, value in list(new_state_dict.items()):
353
+ if key not in current_state:
354
+ continue
355
+ len_diff = value.dim() - current_state[key].dim()
356
+ if len_diff == -1:
357
+ new_state_dict[key] = value.unsqueeze(-1)
358
+ elif len_diff == 1:
359
+ new_state_dict[key] = value.squeeze(-1)
360
+
361
+ super().load_state_dict(new_state_dict, strict=strict)
362
+
363
+ def forward(self, x: torch.Tensor) -> torch.Tensor:
364
+ x = self.conv_pre(x.permute(0, 2, 1))
365
+
366
+ for upsample_layer, resblock_group in zip(self.ups, self.resblocks):
367
+ x = F.leaky_relu(x, self.lrelu_slope)
368
+ x = upsample_layer(x)
369
+ xs = 0
370
+ for resblock in resblock_group:
371
+ xs = xs + resblock(x)
372
+ x = xs / self.num_kernels
373
+
374
+ x = F.leaky_relu(x)
375
+ x = self.conv_post(x)
376
+ return torch.tanh(x)
377
+
378
+ def remove_weight_norm(self):
379
+ for layer in self.ups:
380
+ remove_weight_norm(layer)
381
+ for group in self.resblocks:
382
+ for block in group:
383
+ block.remove_weight_norm()
384
+ remove_weight_norm(self.conv_pre)
385
+ remove_weight_norm(self.conv_post)
386
+
387
+
388
+ class WavCoch(PreTrainedModel):
389
+ """Causal waveform-to-cochleagram tokenizer with optional vocoder."""
390
+
391
+ config_class = WavCochConfig
392
+ main_input_name = "wav"
393
+
394
+ def __init__(self, config: WavCochConfig):
395
+ super().__init__(config)
396
+ self.config = config
397
+
398
+ self.N = int(config.window_size)
399
+ self.hop_length = int(config.hop_length)
400
+ self.window_padding = int(getattr(config, "window_padding", self.N - self.hop_length))
401
+ self.centered_right_padding = int(
402
+ getattr(config, "centered_right_padding", (self.N - 1) // 2)
403
+ )
404
+ self.centered_left_padding = int(
405
+ getattr(config, "centered_left_padding", self.window_padding - self.centered_right_padding)
406
+ )
407
+ if self.centered_left_padding + self.centered_right_padding != self.window_padding:
408
+ raise ValueError(
409
+ "Centered left/right padding must sum to window_padding so encoding length stays unchanged"
410
+ )
411
+ self.causal_convs = bool(getattr(config, "causal_convs", True))
412
+ self.causal_pad_mode = getattr(config, "causal_pad_mode", "repeat")
413
+
414
+ out_bins = self.N // 2 + 1
415
+ self.conv_real_filters = nn.Conv1d(1, out_bins, kernel_size=self.N, stride=self.hop_length)
416
+ self.conv_imag_filters = nn.Conv1d(1, out_bins, kernel_size=self.N, stride=self.hop_length)
417
+ self._initialize_conv_filters()
418
+
419
+ self.encoder = self._build_conv_stack(
420
+ in_channels=out_bins,
421
+ out_channels=config.encoder_dim,
422
+ num_layers=config.encoder_layers,
423
+ kernel_size=config.encoder_kernel_size,
424
+ causal=self.causal_convs,
425
+ )
426
+ self.quantizer = FSQ(levels=list(config.channels), dim=config.encoder_dim)
427
+ self.decoder = self._build_conv_stack(
428
+ in_channels=config.decoder_dim,
429
+ out_channels=config.out_channels,
430
+ num_layers=config.decoder_layers,
431
+ kernel_size=config.decoder_kernel_size,
432
+ causal=self.causal_convs,
433
+ )
434
+
435
+ self.has_vocoder = bool(getattr(config, "has_vocoder", False))
436
+ if self.has_vocoder:
437
+ if int(config.out_channels) != 211:
438
+ raise ValueError("Bundled vocoder currently expects 211 cochleagram channels")
439
+ self.vocoder = Generator(
440
+ out_channels=config.out_channels,
441
+ upsample_rates=config.vocoder_upsample_rates,
442
+ upsample_kernel_sizes=config.vocoder_upsample_kernel_sizes,
443
+ upsample_initial_channel=config.vocoder_upsample_initial_channel,
444
+ resblock=config.vocoder_resblock,
445
+ resblock_kernel_sizes=config.vocoder_resblock_kernel_sizes,
446
+ resblock_dilation_sizes=config.vocoder_resblock_dilation_sizes,
447
+ )
448
+ else:
449
+ self.vocoder = None
450
+
451
+ self._vocab_size = int(config.vocab_size)
452
+ self.post_init()
453
+
454
+ def _build_conv_stack(
455
+ self,
456
+ in_channels: int,
457
+ out_channels: int,
458
+ num_layers: int,
459
+ kernel_size: int,
460
+ causal: bool,
461
+ ) -> nn.Sequential:
462
+ layers = []
463
+ for layer_idx in range(int(num_layers)):
464
+ input_channels = in_channels if layer_idx == 0 else out_channels
465
+ conv = _build_conv1d(
466
+ input_channels,
467
+ out_channels,
468
+ kernel_size,
469
+ causal=causal,
470
+ pad_mode=self.causal_pad_mode,
471
+ )
472
+ layers.extend([conv, nn.ReLU()])
473
+ return nn.Sequential(*layers)
474
+
475
+ def _compute_twiddle_factors(self):
476
+ n = torch.arange(self.N, dtype=torch.float32).unsqueeze(1)
477
+ k = torch.arange(self.N, dtype=torch.float32).unsqueeze(0)
478
+ angles = -2.0 * math.pi * n * k / float(self.N)
479
+ return torch.cos(angles), torch.sin(angles)
480
+
481
+ def _initialize_conv_filters(self):
482
+ with torch.no_grad():
483
+ cos_matrix, sin_matrix = self._compute_twiddle_factors()
484
+ cos_matrix = cos_matrix[: self.N // 2 + 1, :]
485
+ sin_matrix = sin_matrix[: self.N // 2 + 1, :]
486
+ window = torch.hann_window(self.N, periodic=True).view(1, 1, -1)
487
+ real_weights = (cos_matrix.unsqueeze(1) * window).to(dtype=self.conv_real_filters.weight.dtype)
488
+ imag_weights = (sin_matrix.unsqueeze(1) * window).to(dtype=self.conv_imag_filters.weight.dtype)
489
+ self.conv_real_filters.weight.copy_(real_weights)
490
+ self.conv_imag_filters.weight.copy_(imag_weights)
491
+
492
+ for param in self.conv_real_filters.parameters():
493
+ param.requires_grad_(False)
494
+ for param in self.conv_imag_filters.parameters():
495
+ param.requires_grad_(False)
496
+
497
+ def _normalize_sample_rate(self, sample_rate: Optional[int], sampling_rate: Optional[int]) -> int:
498
+ if sample_rate is not None and sampling_rate is not None and sample_rate != sampling_rate:
499
+ raise ValueError(f"sample_rate ({sample_rate}) and sampling_rate ({sampling_rate}) conflict")
500
+ resolved = int(sample_rate or sampling_rate or self.config.sample_rate)
501
+ if resolved != int(self.config.sample_rate):
502
+ raise ValueError(
503
+ f"WavCoch expects {self.config.sample_rate} Hz audio, but received {resolved} Hz"
504
+ )
505
+ return resolved
506
+
507
+ def _prepare_wav_batch(self, wav) -> torch.Tensor:
508
+ if isinstance(wav, list):
509
+ wav = [item if isinstance(item, torch.Tensor) else torch.tensor(item) for item in wav]
510
+ normalized = []
511
+ for item in wav:
512
+ if item.ndim == 1:
513
+ normalized.append(item)
514
+ elif item.ndim == 2 and 1 in item.shape:
515
+ normalized.append(item.reshape(-1))
516
+ else:
517
+ raise ValueError(f"Unexpected list element shape {tuple(item.shape)}")
518
+ wav = torch.nn.utils.rnn.pad_sequence(normalized, batch_first=True).unsqueeze(1)
519
+ elif isinstance(wav, torch.Tensor):
520
+ if wav.ndim == 1:
521
+ wav = wav.unsqueeze(0).unsqueeze(0)
522
+ elif wav.ndim == 2:
523
+ wav = wav.unsqueeze(1)
524
+ elif wav.ndim != 3:
525
+ raise ValueError(f"Unexpected tensor shape {tuple(wav.shape)}, expected 1D, 2D or 3D")
526
+ else:
527
+ raise TypeError(f"Unsupported input type: {type(wav)}")
528
+
529
+ return wav.to(dtype=torch.float32)
530
+
531
+ @property
532
+ def vocab_size(self) -> int:
533
+ return self._vocab_size
534
+
535
+ def _resolve_wav_input(
536
+ self,
537
+ wav: Optional[torch.Tensor],
538
+ input_values: Optional[torch.Tensor],
539
+ ) -> torch.Tensor:
540
+ if wav is not None and input_values is not None:
541
+ raise ValueError("Provide either `wav` or `input_values`, not both")
542
+ resolved = wav if wav is not None else input_values
543
+ if resolved is None:
544
+ raise ValueError("WavCoch requires waveform input via `wav` or `input_values`")
545
+ return resolved
546
+
547
+ def _encoding_padding(self, mode: str) -> Tuple[int, int]:
548
+ if mode == "causal":
549
+ return self.window_padding, 0
550
+ if mode == "centered":
551
+ return self.centered_left_padding, self.centered_right_padding
552
+ raise ValueError(f"Unsupported encoding mode {mode!r}; expected 'causal' or 'centered'")
553
+
554
+ def _encode_quantized(
555
+ self,
556
+ wav: torch.Tensor,
557
+ pad: bool = True,
558
+ mode: str = "causal",
559
+ ):
560
+ wav = self._prepare_wav_batch(wav)
561
+ left_padding, right_padding = self._encoding_padding(mode)
562
+ if pad:
563
+ wav = F.pad(wav, (left_padding, right_padding), mode="constant", value=0.0)
564
+
565
+ with torch.no_grad():
566
+ real_part = self.conv_real_filters(wav)
567
+ imag_part = self.conv_imag_filters(wav)
568
+
569
+ x = real_part + imag_part
570
+ x = self.encoder(x).permute(0, 2, 1)
571
+ quantized, indices = self.quantizer(x)
572
+ return quantized, indices
573
+
574
+ def forward(
575
+ self,
576
+ wav: Optional[torch.Tensor] = None,
577
+ coch: Optional[torch.Tensor] = None,
578
+ return_tensors: str = "pt",
579
+ sample_rate: Optional[int] = None,
580
+ sampling_rate: Optional[int] = None,
581
+ pad: bool = True,
582
+ output_hidden_states: Optional[bool] = False,
583
+ return_dict: Optional[bool] = True,
584
+ input_values: Optional[torch.Tensor] = None,
585
+ mode: str = "causal",
586
+ ):
587
+ del return_tensors # unused, kept for tokenizer-like API compatibility
588
+ self._normalize_sample_rate(sample_rate, sampling_rate)
589
+ wav = self._resolve_wav_input(wav, input_values)
590
+ quantized, indices = self._encode_quantized(wav, pad=pad, mode=mode)
591
+
592
+ if output_hidden_states:
593
+ hidden_states = (quantized,)
594
+ if not return_dict:
595
+ return quantized, hidden_states
596
+ return BaseModelOutput(last_hidden_state=quantized, hidden_states=hidden_states)
597
+
598
+ if coch is None:
599
+ codes = indices.long()
600
+ return BatchEncoding({"input_values": codes, "input_ids": codes})
601
+
602
+ pred_coch = self.decoder(quantized.permute(0, 2, 1)).permute(0, 2, 1)
603
+ loss = F.l1_loss(pred_coch, coch)
604
+ return pred_coch, loss, None
605
+
606
+ @torch.no_grad()
607
+ def quantize(self, wav: torch.Tensor, pad: bool = True, mode: str = "causal") -> torch.Tensor:
608
+ _, indices = self._encode_quantized(wav, pad=pad, mode=mode)
609
+ return indices.long()
610
+
611
+ @torch.no_grad()
612
+ def decode(self, indices: torch.Tensor) -> torch.Tensor:
613
+ if indices.ndim == 1:
614
+ indices = indices.unsqueeze(0)
615
+ emb = self.quantizer.indices_to_codes(indices.long())
616
+ return self.decoder(emb.permute(0, 2, 1)).permute(0, 2, 1)
617
+
618
+ @torch.no_grad()
619
+ def wav2coch(self, wav: torch.Tensor, pad: bool = True, mode: str = "causal") -> torch.Tensor:
620
+ quantized, _ = self._encode_quantized(wav, pad=pad, mode=mode)
621
+ return self.decoder(quantized.permute(0, 2, 1)).permute(0, 2, 1)
622
+
623
+ @torch.no_grad()
624
+ def vocode(self, coch: torch.Tensor) -> torch.Tensor:
625
+ if self.vocoder is None:
626
+ raise ValueError("This WavCoch checkpoint does not include a bundled vocoder")
627
+
628
+ if coch.ndim == 2:
629
+ coch = coch.unsqueeze(0)
630
+ elif coch.ndim != 3:
631
+ raise ValueError(f"Unexpected cochleagram shape {tuple(coch.shape)}")
632
+
633
+ if coch.shape[-1] != self.config.out_channels and coch.shape[1] == self.config.out_channels:
634
+ coch = coch.transpose(1, 2)
635
+
636
+ return self.vocoder(coch)
637
+
638
+ @torch.no_grad()
639
+ def decode_audio(self, indices: torch.Tensor) -> torch.Tensor:
640
+ return self.vocode(self.decode(indices))