# nRAN-T: Multi-Stream Rational-Addition Networks for Time-Domain Classification ## Abstract We introduce **nRAN-T**, a neural network architecture that generalizes the Multi-Term Rational-Addition Number (nRAN) to time-domain classification. An nRAN-T classifier represents its decision boundary as a base offset plus a finite sum of rational correction terms, each operating at an explicit temporal scale. This preserves scale-separated information throughout the forward pass, only collapsing into a scalar prediction at the final layer. We provide a complete PyTorch specification, capacity-management algorithms, and training protocols sufficient for independent implementation. --- ## 1. Preliminaries: From nRAN to Neural Streams An nRAN is defined as $$x = \sum_{i=1}^{n} \frac{a_i}{b_i} + c$$ where $(a_i, b_i)$ are rational correction terms and $c$ is a base offset. The **collapse** is the final evaluation of the sum. In the neural setting, we map this algebra directly onto a time-domain classifier: | nRAN Concept | Neural Interpretation | |-------------|----------------------| | $a_i$ | Output of stream $i$'s convolutional encoder (numerator) | | $b_i$ | Temporal scale (dilation / receptive field) of stream $i$ | | $c$ | Learnable class bias (base offset) | | Addition | Element-wise sum of stream features | | Collapse | Global pooling over time + linear classification | | Capacity $n_{\max}$ | Maximum number of active streams before compaction | **Why $n > 1$ matters.** A single convolutional stack can separate a large-scale structure from one small-scale detail. An nRAN-T can separate **many temporal scales simultaneously**: $$f(P) = c + \frac{\mathcal{F}_1(P)}{b_1} + \frac{\mathcal{F}_2(P)}{b_2} + \cdots + \frac{\mathcal{F}_n(P)}{b_n}$$ In ordinary deep networks, information at scale $b=64$ (phrase-level) and $b=1$ (note-onset level) must compete for the same channel capacity. In nRAN-T, each scale survives as a separate term until collapse. --- ## 2. Architecture Specification ### 2.1 Input Representation For a piano roll classifier, the input is a matrix $$P \in \mathbb{R}^{N \times T}$$ where $N$ is the number of pitch classes (typically 88 or 128) and $T$ is the number of time frames. The input may be binary (note on/off), velocity-valued, or multi-channel (e.g., right hand, left hand, pedal). ### 2.2 Stream Decomposition An **nRAN-T block** consists of $n$ parallel streams. Stream $i$ is defined by: 1. **Temporal scale** $b_i \in \mathbb{Z}^+$ (e.g., $[1, 4, 16, 64]$). 2. **Encoder** $\mathcal{F}_i$: a small temporal convolutional stack with dilation $b_i$. 3. **Rational term**: $\displaystyle \text{term}_i = \frac{\mathcal{F}_i(P)}{b_i}$ The division by $b_i$ is the critical operation. It ensures that coarse-scale streams (large receptive fields) cannot dominate fine-scale streams simply by having larger activation magnitudes. The scale acts as a **learnable but structured normalization**. ### 2.3 Base Offset The base offset $c \in \mathbb{R}^K$ is a learnable parameter vector where $K$ is the number of classes. It acts as a class prior, analogous to the nRAN offset. ### 2.4 Collapse The final operation is: $$\text{logits} = c + \text{MLP}\left( \text{Pool}_{t}\left( \sum_{i=1}^{n} \text{term}_i \right) \right)$$ where $\text{Pool}_t$ is adaptive average pooling over the time dimension. --- ## 3. PyTorch Implementation ### 3.1 Core Module ```python import torch import torch.nn as nn import torch.nn.functional as F from typing import List, Optional, Tuple, Union class RationalTerm(nn.Module): """ Rational correction term: a_i / b_i. Encodes input features at a specific temporal scale. """ def __init__( self, in_channels: int, out_channels: int, kernel_size: int = 3, dilation: int = 1, learnable_denominator: bool = True ): super().__init__() self.dilation = dilation self.conv = nn.Conv1d( in_channels, out_channels, kernel_size=kernel_size, padding=kernel_size // 2 * dilation, dilation=dilation, bias=True ) if learnable_denominator: self.b = nn.Parameter(torch.tensor(float(dilation))) else: self.register_buffer("b", torch.tensor(float(dilation))) def forward(self, x: torch.Tensor) -> torch.Tensor: a = self.conv(x) return a / (self.b.abs() + 1e-6) class nRANTClassifier(nn.Module): """ Multi-Stream Rational-Addition Network for Time-Domain Classification. Args: num_classes: Number of output classes in_channels: Input feature dimension (e.g., 88 for piano roll) hidden_dim: Internal feature dimension scales: Temporal scales (denominators b_i) max_n: Hard capacity limit (truncates scales list) dropout: Dropout probability """ def __init__( self, num_classes: int, in_channels: int = 88, hidden_dim: int = 128, scales: List[int] = [1, 4, 16, 64], max_n: Optional[int] = None, dropout: float = 0.1 ): super().__init__() self.num_classes = num_classes self.scales = scales[:max_n] if max_n else scales self.max_n = len(self.scales) # Input embedding self.input_proj = nn.Conv1d(in_channels, hidden_dim, kernel_size=1) # Parallel stream encoders self.streams = nn.ModuleList() for b in self.scales: stream = nn.Sequential( nn.Conv1d(hidden_dim, hidden_dim, 3, padding=b, dilation=b), nn.BatchNorm1d(hidden_dim), nn.ReLU(inplace=True), nn.Dropout1d(dropout), nn.Conv1d(hidden_dim, hidden_dim, 3, padding=b, dilation=b), nn.BatchNorm1d(hidden_dim), nn.ReLU(inplace=True), nn.Conv1d(hidden_dim, hidden_dim, 1) # numerator projection ) self.streams.append(stream) # Learnable denominators self.denominators = nn.Parameter( torch.tensor(self.scales, dtype=torch.float32) ) # Base offset (class prior) self.base_offset = nn.Parameter(torch.zeros(num_classes)) # Collapse self.pool = nn.AdaptiveAvgPool1d(1) self.head = nn.Sequential( nn.Linear(hidden_dim, hidden_dim), nn.ReLU(inplace=True), nn.Dropout(dropout), nn.Linear(hidden_dim, num_classes) ) def forward( self, x: torch.Tensor, return_terms: bool = False ) -> Union[torch.Tensor, Tuple[torch.Tensor, List[torch.Tensor]]]: """ Forward pass. Args: x: Input tensor of shape (B, C, T) return_terms: If True, return individual rational terms Returns: logits: (B, num_classes) terms: List of (B, H, T) tensors if return_terms=True """ h = self.input_proj(x) # (B, H, T) terms: List[torch.Tensor] = [] for i, stream in enumerate(self.streams): a_i = stream(h) b_i = self.denominators[i].abs() + 1e-6 term_i = a_i / b_i terms.append(term_i) # nRAN addition: sum of all terms fused = torch.stack(terms, dim=0).sum(dim=0) # (B, H, T) # Collapse over time pooled = self.pool(fused).squeeze(-1) # (B, H) # Classification head + base offset logits = self.head(pooled) + self.base_offset # (B, K) if return_terms: return logits, terms return logits def capacity_regularization(self, terms: List[torch.Tensor]) -> torch.Tensor: """ Soft capacity pressure. Encourages small terms to vanish, making them safe to merge into the base offset. """ loss = 0.0 for i, term in enumerate(terms): loss = loss + term.abs().mean() / (self.denominators[i].abs() + 1e-6) return loss / len(terms) def compact_streams(self, strategy: str = "magnitude") -> int: """ Hard capacity compaction: remove the weakest stream. Returns the index of the removed stream, or -1. """ if len(self.streams) <= 1: return -1 with torch.no_grad(): if strategy == "magnitude": mags = [ sum(p.norm().item() for p in stream.parameters()) for stream in self.streams ] weak_idx = int(torch.tensor(mags).argmin()) else: raise ValueError(f"Unknown compaction strategy: {strategy}") # Remove the weakest stream del self.streams[weak_idx] self.denominators = nn.Parameter( torch.cat([ self.denominators[:weak_idx], self.denominators[weak_idx + 1:] ]) ) self.max_n -= 1 return weak_idx ``` ### 3.2 Training Loop ```python def train_epoch( model: nn.Module, dataloader: torch.utils.data.DataLoader, optimizer: torch.optim.Optimizer, device: torch.device, lambda_cap: float = 1e-3, lambda_div: float = 1e-4 ) -> float: model.train() total_loss = 0.0 for x, y in dataloader: x, y = x.to(device), y.to(device) optimizer.zero_grad() logits, terms = model(x, return_terms=True) # Standard classification loss ce = F.cross_entropy(logits, y) # Capacity regularization: encourage small terms to vanish cap = model.capacity_regularization(terms) # Optional: diversity loss (encourage streams to be orthogonal) # This prevents all streams from learning the same scale diversity = 0.0 if len(terms) > 1: for i in range(len(terms)): for j in range(i + 1, len(terms)): cos = F.cosine_similarity( terms[i].mean(dim=(0, 2)), terms[j].mean(dim=(0, 2)), dim=0 ) diversity = diversity + cos ** 2 loss = ce + lambda_cap * cap + lambda_div * diversity loss.backward() optimizer.step() total_loss += loss.item() return total_loss / len(dataloader) ``` ### 3.3 Curriculum Capacity Scheduling Start training with a small number of streams and grow: ```python # Initialize with base + 1 stream model = nRANTClassifier(num_classes=10, scales=[1], max_n=1) # Every 20 epochs, expand capacity by adding one stream if epoch % 20 == 0 and model.max_n < 4: new_scale = [1, 4, 16, 64][model.max_n] # Append a new stream and extend denominators # ... (implementation left as module extension) ``` --- ## 4. Capacity Management Strategies ### 4.1 Soft Compaction (Training) Applied at every forward pass via regularization. The L1 term in `capacity_regularization` pushes small-magnitude corrections toward zero. When a term's mean absolute value drops below a threshold, it is safe to merge into the base offset without altering the network's decision boundary. ### 4.2 Hard Compaction (Inference / Post-Training) **Exact Rational Compaction:** Merge all integer-scale stream weights into a single convolution using weight addition. Only valid if two streams share the same dilation and kernel size. **Magnitude-Based Merge:** Remove the stream with the smallest parameter norm and absorb its expected output into the base offset bias. **Collapse into Offset:** If the input must be processed under severe latency constraints, evaluate only the base offset and the largest-scale stream. This is the nRAN equivalent of collapsing all rational terms into $c$. --- ## 5. Properties and Guarantees **Scale Separation.** Because each stream is explicitly divided by its temporal scale $b_i$, gradients with respect to fine-scale streams (small $b_i$) are larger than those with respect to coarse-scale streams. This prevents the "coarse-scale dominates" problem common in multi-resolution networks. **Additive Exactness.** Before the final pooling layer, the representation is an exact sum of independent terms. Unlike concatenation-based multi-scale architectures, no dimensionality bottleneck forces information loss between scales. **Interpretability.** Each stream can be inspected independently. A stream with $b=1$ shows onset-level decisions; a stream with $b=64$ shows phrase-level decisions. The magnitude of each term before collapse directly indicates its contribution to the final class. --- ## 6. Extensions ### 6.1 Cross-Stream Attention (Dual Beam) The basic architecture uses additive fusion. A **Dual Beam** extension introduces cross-stream attention: $$\text{fused} = \sum_{i=1}^{n} \alpha_i \cdot \text{term}_i$$ where $\alpha_i$ is computed by an attention mechanism over the temporal axis. This allows the model to "highlight" one stream at specific time steps. ### 6.2 Adaptive Denominators Instead of fixing $b_i \in \{1, 4, 16, 64\}$, treat them as learnable parameters initialized to those values. The network can then adapt its temporal scales to the dataset. To prevent collapse (all $b_i \rightarrow 1$), add a regularization penalty: $$\mathcal{L}_{\text{scale}} = \sum_{i