1+ """
2+ Authors: Reza Najian Asl, https://github.com/RezaNajian
3+ Date: Dec, 2025
4+ License: FOL/LICENSE
5+ """
6+
7+ from __future__ import annotations
8+
9+ from typing import Optional , Callable , Sequence
10+
11+ import jax
12+ import jax .numpy as jnp
13+ from flax import nnx
14+
15+ # JAX/Flax NNX implementation of ChannelMLP.
16+ #
17+ # Ported from the original PyTorch implementation:
18+ # Repository: https://github.com/neuraloperator/neuraloperator
19+ # File: neuralop/layers/channel_mlp.py
20+ # Commit: 14c0f7320dc7c94e907a16fd276248df2d71407c (2025-11-14)
21+ # URL:
22+ # https://github.com/neuraloperator/neuraloperator/blob/14c0f7320dc7c94e907a16fd276248df2d71407c/neuralop/layers/channel_mlp.py
23+ #
24+ # Original code copyright (c) 2023 NeuralOperator developers
25+ # Licensed under the MIT License.
26+ #
27+ # Note:
28+ # The PyTorch implementation operates in NCHW (channels-first) format,
29+ # while JAX/Flax NNX uses NHWC (channels-last). This port includes
30+ # careful transformations between channel orders to preserve behavior.
31+
32+
33+ class ChannelMLP (nnx .Module ):
34+ """Multi-layer perceptron applied channel-wise across spatial dimensions.
35+
36+ ChannelMLP applies a sequence of linear projections and nonlinearities to the
37+ channel dimension of input tensors, making it invariant to spatial resolution.
38+ This is particularly useful in neural operators where spatial dimensions may vary
39+ but channel-wise processing should remain consistent across all positions.
40+
41+ Unlike the original PyTorch implementation—which uses 1D convolutions with
42+ kernel size 1 applied to (B, C, L) channel-first tensors—this JAX/Flax NNX
43+ version operates on channels-last tensors of shape (B, d1, ..., dn, C). In this
44+ layout, a standard Linear layer naturally performs the same per-channel mixing
45+ as a Conv1d with kernel size 1, since both are equivalent to applying an MLP
46+ independently at each spatial location.
47+
48+ Using Linear layers in NNX offers several advantages:
49+
50+ * It directly matches the channels-last convention of JAX/Flax.
51+ * It avoids unnecessary convolution machinery and is computationally simpler.
52+ * It more clearly expresses the intended abstraction: an MLP over channels,
53+ broadcast across all spatial positions.
54+ * It preserves numerical equivalence to the Conv1d(kernel=1) operation.
55+
56+ Internally, the spatial dimensions are flattened, the MLP is applied over the
57+ channel axis, and the output is reshaped back to the original spatial form.
58+
59+ Parameters
60+ ----------
61+ in_channels : int
62+ Number of input channels.
63+ out_channels : int, optional
64+ Number of output channels. If None, defaults to in_channels.
65+ hidden_channels : int, optional
66+ Number of hidden channels in intermediate layers. If None, defaults to in_channels.
67+ n_layers : int, optional
68+ Number of linear layers in the MLP, by default 2.
69+ n_dim : int, optional
70+ Spatial dimensionality (unused but kept for API compatibility), by default 2.
71+ non_linearity : callable, optional
72+ Activation function applied between layers, by default jax.nn.gelu.
73+ dropout : float, optional
74+ Dropout probability applied after each layer (except the last). If 0, no dropout is applied,
75+ by default 0.0.
76+ """
77+
78+ def __init__ (
79+ self ,
80+ in_channels : int ,
81+ out_channels : Optional [int ] = None ,
82+ hidden_channels : Optional [int ] = None ,
83+ n_layers : int = 2 ,
84+ n_dim : int = 2 , # unused, kept for API compatibility
85+ non_linearity : Optional [Callable ] = jax .nn .gelu ,
86+ dropout : float = 0.0 ,
87+ * ,
88+ rngs : nnx .Rngs ,
89+ ):
90+ super ().__init__ ()
91+
92+ self .n_layers = n_layers
93+ self .in_channels = in_channels
94+ self .out_channels = in_channels if out_channels is None else out_channels
95+ self .hidden_channels = in_channels if hidden_channels is None else hidden_channels
96+
97+ self .non_linearity = non_linearity
98+
99+ # Dropout layers (one per layer, like PyTorch)
100+ if dropout > 0.0 :
101+ self .dropouts : Optional [nnx .List [nnx .Module ]] = nnx .List ([
102+ nnx .Dropout (rate = dropout , rngs = rngs ) for _ in range (n_layers )
103+ ])
104+ else :
105+ self .dropouts = None
106+
107+ # Build linear layers
108+ self .fcs = nnx .List ([])
109+ for i in range (n_layers ):
110+ if i == 0 and i == (n_layers - 1 ):
111+ # Single layer: input -> output
112+ lin = nnx .Linear (
113+ rngs = rngs ,
114+ in_features = self .in_channels ,
115+ out_features = self .out_channels ,
116+ )
117+ elif i == 0 :
118+ # First layer: input -> hidden
119+ lin = nnx .Linear (
120+ rngs = rngs ,
121+ in_features = self .in_channels ,
122+ out_features = self .hidden_channels ,
123+ )
124+ elif i == (n_layers - 1 ):
125+ # Last layer: hidden -> output
126+ lin = nnx .Linear (
127+ rngs = rngs ,
128+ in_features = self .hidden_channels ,
129+ out_features = self .out_channels ,
130+ )
131+ else :
132+ # Internal: hidden -> hidden
133+ lin = nnx .Linear (
134+ rngs = rngs ,
135+ in_features = self .hidden_channels ,
136+ out_features = self .hidden_channels ,
137+ )
138+ self .fcs .append (lin )
139+
140+ def __call__ (self , x : jnp .ndarray ) -> jnp .ndarray :
141+ """
142+ Parameters
143+ ----------
144+ x : jnp.ndarray
145+ Input tensor of shape (batch, d1, ..., dn, in_channels)
146+ train : bool
147+ Whether we are in training mode (affects dropout).
148+ """
149+ size = x .shape # (B, d1, ..., dn, C_in)
150+ B = size [0 ]
151+ spatial_dims = size [1 :- 1 ]
152+ C_in = size [- 1 ]
153+
154+ assert (
155+ C_in == self .in_channels
156+ ), f"Expected last dim (channels) = { self .in_channels } , got { C_in } "
157+
158+ # Flatten spatial dims: (B, d1, ..., dn, C) -> (B, L, C)
159+ x = x .reshape ((B , - 1 , C_in ))
160+
161+ # Apply MLP across the last dimension (C), broadcasting over (B, L)
162+ for i , fc in enumerate (self .fcs ):
163+ x = fc (x ) # Linear along last axis
164+ if i < self .n_layers - 1 :
165+ x = self .non_linearity (x )
166+ if self .dropouts is not None :
167+ x = self .dropouts [i ](x )
168+
169+ # Restore original spatial dims with new channels
170+ x = x .reshape ((B , * spatial_dims , self .out_channels ))
171+ return x
172+
173+ class LinearChannelMLP (nnx .Module ):
174+ """
175+ Multi-layer perceptron (MLP) for channel processing using fully connected layers.
176+
177+ This is a Flax NNX port of the corresponding PyTorch implementation. It is an
178+ alternative to a convolution-based ChannelMLP, using standard Linear layers.
179+
180+ The network is defined by `layers = [in_channels, hidden1, ..., out_channels]`
181+ and applies:
182+ - a Linear transformation at every layer,
183+ - `non_linearity` after every layer except the last,
184+ - Dropout after every Linear layer *including the last* (to match the PyTorch code)
185+ when `dropout > 0`.
186+
187+ Parameters
188+ ----------
189+ layers : Sequence[int]
190+ Architecture definition: [in_channels, hidden1, ..., out_channels].
191+ Must have at least 2 elements (input and output sizes).
192+ non_linearity : Callable[[jnp.ndarray], jnp.ndarray], optional
193+ Activation function applied after each Linear layer except the last.
194+ Defaults to `jax.nn.gelu`.
195+ dropout : float, optional
196+ Dropout probability. If > 0, dropout is applied after each Linear layer
197+ (including the last) to match the PyTorch implementation. If 0, no dropout
198+ is applied. Defaults to 0.0.
199+ rngs : nnx.Rngs
200+ Random number generators used for parameter initialization and dropout.
201+ """
202+
203+ def __init__ (
204+ self ,
205+ layers : Sequence [int ],
206+ non_linearity : Callable [[jnp .ndarray ], jnp .ndarray ] = jax .nn .gelu ,
207+ dropout : float = 0.0 ,
208+ * ,
209+ rngs : nnx .Rngs ,
210+ ):
211+ super ().__init__ ()
212+
213+ self .n_layers = len (layers ) - 1
214+ assert self .n_layers >= 1 , (
215+ "Error: trying to instantiate a LinearChannelMLP "
216+ "with only one linear layer."
217+ )
218+
219+ self .non_linearity = non_linearity
220+
221+ # Fully connected layers
222+ self .fcs = nnx .List ([])
223+ for j in range (self .n_layers ):
224+ self .fcs .append (
225+ nnx .Linear (
226+ in_features = layers [j ],
227+ out_features = layers [j + 1 ],
228+ rngs = rngs ,
229+ )
230+ )
231+
232+ # Dropout layers (one per linear layer) or None
233+ if dropout > 0.0 :
234+ self .dropout : Optional [nnx .List [nnx .Dropout ]] = nnx .List ([
235+ nnx .Dropout (rate = dropout , rngs = rngs ) for _ in range (self .n_layers )
236+ ])
237+ else :
238+ self .dropout = None
239+
240+ def __call__ (self , x : jnp .ndarray ) -> jnp .ndarray :
241+ """
242+ Apply the linear channel MLP.
243+
244+ The input is assumed to be a 2D array where the last dimension corresponds
245+ to channels/features. This method preserves the leading dimension(s) by
246+ applying per-row Linear transformations.
247+
248+ Dropout (if enabled) is applied after each Linear layer, including the last,
249+ matching the behavior of the original PyTorch implementation.
250+
251+ Parameters
252+ ----------
253+ x : jnp.ndarray
254+ Input array of shape (batch, in_channels) or (batch * spatial, in_channels).
255+
256+ Returns
257+ -------
258+ jnp.ndarray
259+ Output array of shape (batch, out_channels) or (batch * spatial, out_channels).
260+ """
261+ for i , fc in enumerate (self .fcs ):
262+ x = fc (x ) # Linear transformation
263+
264+ # Nonlinearity on all but last layer
265+ if i < self .n_layers - 1 :
266+ x = self .non_linearity (x )
267+
268+ # Dropout after each layer (including last), matching the PyTorch code
269+ if self .dropout is not None :
270+ x = self .dropout [i ](x )
271+
272+ return x
0 commit comments