DeepMoD
This file contains building blocks for the deepmod framework: I) The constraint class that constrains the neural network with the obtained solution, II) The sparsity estimator class, III) Function library class on which the model discovery is performed. IV) The DeepMoD class integrates these seperate building blocks. These are all abstract classes and implement the flow logic, rather than the specifics.
Constraint
__init__(self)
special
Abstract baseclass for the constraint module.
Source code in deepymod/model/deepmod.py
def __init__(self) -> None:
"""Abstract baseclass for the constraint module."""
super().__init__()
self.sparsity_masks: TensorList = None
apply_mask(self, thetas)
Applies the sparsity mask to the feature (library) matrix.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
thetas |
<function NewType.<locals>.new_type at 0x12645b950> |
List of all library matrices of size [(n_samples, n_features) x n_outputs]. |
required |
Returns:
Type | Description |
---|---|
<function NewType.<locals>.new_type at 0x12645b950> |
TensorList: The sparse version of the library matrices of size [(n_samples, n_active_features) x n_outputs]. |
Source code in deepymod/model/deepmod.py
def apply_mask(self, thetas: TensorList) -> TensorList:
""" Applies the sparsity mask to the feature (library) matrix.
Args:
thetas (TensorList): List of all library matrices of size [(n_samples, n_features) x n_outputs].
Returns:
TensorList: The sparse version of the library matrices of size [(n_samples, n_active_features) x n_outputs].
"""
sparse_thetas = [theta[:, sparsity_mask] for theta, sparsity_mask in zip(thetas, self.sparsity_masks)]
return sparse_thetas
calculate_coeffs(self, sparse_thetas, time_derivs)
Abstract method. Specific method should return the coefficients as calculated from the sparse feature matrices and temporal derivatives.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
sparse_thetas |
<function NewType.<locals>.new_type at 0x12645b950> |
List containing the sparse feature tensors of size (n_samples, n_active_features). |
required |
time_derivs |
<function NewType.<locals>.new_type at 0x12645b950> |
List containing the time derivatives of size (n_samples, n_outputs). |
required |
Returns:
Type | Description |
---|---|
<function NewType.<locals>.new_type at 0x12645b950> |
(TensorList): Calculated coefficients of size (n_features, n_outputs). |
Source code in deepymod/model/deepmod.py
@abstractmethod
def calculate_coeffs(self, sparse_thetas: TensorList, time_derivs: TensorList) -> TensorList:
"""Abstract method. Specific method should return the coefficients as calculated from the sparse feature
matrices and temporal derivatives.
Args:
sparse_thetas (TensorList): List containing the sparse feature tensors of size (n_samples, n_active_features).
time_derivs (TensorList): List containing the time derivatives of size (n_samples, n_outputs).
Returns:
(TensorList): Calculated coefficients of size (n_features, n_outputs).
"""
pass
forward(self, input)
The forward pass of the constraint module applies the sparsity mask to the feature matrix theta, and then calculates the coefficients according to the method in the child.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
input |
Tuple[TensorList, TensorList] |
(time_derivs, library) tuple of size ([(n_samples, 1) X n_outputs], [(n_samples, n_features) x n_outputs]). |
required |
Source code in deepymod/model/deepmod.py
def forward(self, input: Tuple[TensorList, TensorList]) -> Tuple[TensorList, TensorList]:
""" The forward pass of the constraint module applies the sparsity mask to the feature matrix theta,
and then calculates the coefficients according to the method in the child.
Args:
input (Tuple[TensorList, TensorList]): (time_derivs, library) tuple of size
([(n_samples, 1) X n_outputs], [(n_samples, n_features) x n_outputs]).
"""
time_derivs, thetas = input
if self.sparsity_masks is None:
self.sparsity_masks = [torch.ones(theta.shape[1], dtype=torch.bool).to(theta.device) for theta in thetas]
sparse_thetas = self.apply_mask(thetas)
self.coeff_vectors = self.calculate_coeffs(sparse_thetas, time_derivs)
DeepMoD
sparsity_masks
property
readonly
Returns the sparsity masks which contain the active terms.
__init__(self, function_approximator, library, sparsity_estimator, constraint)
special
The DeepMoD class integrates the various buiding blocks into one module. The function approximator approximates the data, the library than builds a feature matrix from its output and the constraint constrains these. The sparsity estimator is called during training to update the sparsity mask (i.e. which terms the constraint is allowed to use.)
Parameters:
Name | Type | Description | Default |
---|---|---|---|
function_approximator |
Sequential |
[description] |
required |
library |
Library |
[description] |
required |
sparsity_estimator |
Estimator |
[description] |
required |
constraint |
Constraint |
[description] |
required |
Source code in deepymod/model/deepmod.py
def __init__(self,
function_approximator: torch.nn.Sequential,
library: Library,
sparsity_estimator: Estimator,
constraint: Constraint) -> None:
"""The DeepMoD class integrates the various buiding blocks into one module. The function approximator approximates the data,
the library than builds a feature matrix from its output and the constraint constrains these. The sparsity estimator is called
during training to update the sparsity mask (i.e. which terms the constraint is allowed to use.)
Args:
function_approximator (torch.nn.Sequential): [description]
library (Library): [description]
sparsity_estimator (Estimator): [description]
constraint (Constraint): [description]
"""
super().__init__()
self.func_approx = function_approximator
self.library = library
self.sparse_estimator = sparsity_estimator
self.constraint = constraint
constraint_coeffs(self, scaled=False, sparse=False)
Calculate the coefficients as estimated by the constraint.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
scaled |
bool |
Determine whether or not the coefficients should be normalized |
False |
sparse |
bool |
Whether to apply the sparsity mask to the coefficients. |
False |
Returns:
Type | Description |
---|---|
<function NewType.<locals>.new_type at 0x12645b950> |
(TensorList): List of coefficients of size [(n_features, 1) x n_outputs] |
Source code in deepymod/model/deepmod.py
def constraint_coeffs(self, scaled=False, sparse=False) -> TensorList:
""" Calculate the coefficients as estimated by the constraint.
Args:
scaled (bool): Determine whether or not the coefficients should be normalized
sparse (bool): Whether to apply the sparsity mask to the coefficients.
Returns:
(TensorList): List of coefficients of size [(n_features, 1) x n_outputs]
"""
coeff_vectors = self.constraint.coeff_vectors
if scaled:
coeff_vectors = [coeff / norm[:, None] for coeff, norm, mask in zip(coeff_vectors, self.library.norms, self.sparsity_masks)]
if sparse:
coeff_vectors = [sparsity_mask[:, None] * coeff for sparsity_mask, coeff in zip(self.sparsity_masks, coeff_vectors)]
return coeff_vectors
estimator_coeffs(self)
Calculate the coefficients as estimated by the sparse estimator.
Returns:
Type | Description |
---|---|
<function NewType.<locals>.new_type at 0x12645b950> |
(TensorList): List of coefficients of size [(n_features, 1) x n_outputs] |
Source code in deepymod/model/deepmod.py
def estimator_coeffs(self) -> TensorList:
""" Calculate the coefficients as estimated by the sparse estimator.
Returns:
(TensorList): List of coefficients of size [(n_features, 1) x n_outputs]
"""
coeff_vectors = self.sparse_estimator.coeff_vectors
return coeff_vectors
forward(self, input)
The forward pass approximates the data, builds the time derivative and feature matrices and applies the constraint.
It returns the prediction of the network, the time derivatives and the feature matrices.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
input |
Tensor |
Tensor of shape (n_samples, n_outputs) containing the coordinates, first column should be the time coordinate. |
required |
Returns:
Type | Description |
---|---|
Tuple[torch.Tensor, TensorList, TensorList] |
Tuple[torch.Tensor, TensorList, TensorList]: The prediction, time derivatives and and feature matrices of respective sizes ((n_samples, n_outputs), [(n_samples, 1) x n_outputs]), [(n_samples, n_features) x n_outputs]) |
Source code in deepymod/model/deepmod.py
def forward(self, input: torch.Tensor) -> Tuple[torch.Tensor, TensorList, TensorList]:
""" The forward pass approximates the data, builds the time derivative and feature matrices
and applies the constraint.
It returns the prediction of the network, the time derivatives and the feature matrices.
Args:
input (torch.Tensor): Tensor of shape (n_samples, n_outputs) containing the coordinates, first column should be the time coordinate.
Returns:
Tuple[torch.Tensor, TensorList, TensorList]: The prediction, time derivatives and and feature matrices of respective sizes
((n_samples, n_outputs), [(n_samples, 1) x n_outputs]), [(n_samples, n_features) x n_outputs])
"""
prediction, coordinates = self.func_approx(input)
time_derivs, thetas = self.library((prediction, coordinates))
self.constraint((time_derivs, thetas))
return prediction, time_derivs, thetas
Estimator
__init__(self)
special
Abstract baseclass for the sparse estimator module.
Source code in deepymod/model/deepmod.py
def __init__(self) -> None:
"""Abstract baseclass for the sparse estimator module."""
super().__init__()
self.coeff_vectors = None
fit(self, X, y)
Abstract method. Specific method should compute the coefficient based on feature matrix X and observations y. Note that we expect X and y to be numpy arrays, i.e. this module is non-differentiable.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
x |
np.ndarray |
Feature matrix of size (n_samples, n_features) |
required |
y |
ndarray |
observations of size (n_samples, n_outputs) |
required |
Returns:
Type | Description |
---|---|
ndarray |
(np.ndarray): Coefficients of size (n_samples, n_outputs) |
Source code in deepymod/model/deepmod.py
@abstractmethod
def fit(self, X: np.ndarray, y: np.ndarray) -> np.ndarray:
""" Abstract method. Specific method should compute the coefficient based on feature matrix X and observations y.
Note that we expect X and y to be numpy arrays, i.e. this module is non-differentiable.
Args:
x (np.ndarray): Feature matrix of size (n_samples, n_features)
y (np.ndarray): observations of size (n_samples, n_outputs)
Returns:
(np.ndarray): Coefficients of size (n_samples, n_outputs)
"""
pass
forward(self, thetas, time_derivs)
The forward pass of the sparse estimator module first normalizes the library matrices and time derivatives by dividing each column (i.e. feature) by their l2 norm, than calculate the coefficient vectors according to the sparse estimation algorithm supplied by the child and finally returns the sparsity mask (i.e. which terms are active) based on these coefficients.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
thetas |
<function NewType.<locals>.new_type at 0x12645b950> |
List containing the sparse feature tensors of size [(n_samples, n_active_features) x n_outputs]. |
required |
time_derivs |
<function NewType.<locals>.new_type at 0x12645b950> |
List containing the time derivatives of size [(n_samples, 1) x n_outputs]. |
required |
Returns:
Type | Description |
---|---|
<function NewType.<locals>.new_type at 0x12645b950> |
(TensorList): List containting the sparsity masks of a boolean type and size [(n_samples, n_features) x n_outputs]. |
Source code in deepymod/model/deepmod.py
def forward(self, thetas: TensorList, time_derivs: TensorList) -> TensorList:
"""The forward pass of the sparse estimator module first normalizes the library matrices
and time derivatives by dividing each column (i.e. feature) by their l2 norm, than calculate the coefficient vectors
according to the sparse estimation algorithm supplied by the child and finally returns the sparsity
mask (i.e. which terms are active) based on these coefficients.
Args:
thetas (TensorList): List containing the sparse feature tensors of size [(n_samples, n_active_features) x n_outputs].
time_derivs (TensorList): List containing the time derivatives of size [(n_samples, 1) x n_outputs].
Returns:
(TensorList): List containting the sparsity masks of a boolean type and size [(n_samples, n_features) x n_outputs].
"""
# we first normalize theta and the time deriv
with torch.no_grad():
normed_time_derivs = [(time_deriv / torch.norm(time_deriv)).detach().cpu().numpy() for time_deriv in time_derivs]
normed_thetas = [(theta / torch.norm(theta, dim=0, keepdim=True)).detach().cpu().numpy() for theta in thetas]
self.coeff_vectors = [self.fit(theta, time_deriv.squeeze())[:, None]
for theta, time_deriv in zip(normed_thetas, normed_time_derivs)]
sparsity_masks = [torch.tensor(coeff_vector != 0.0, dtype=torch.bool).squeeze().to(thetas[0].device) # move to gpu if required
for coeff_vector in self.coeff_vectors]
return sparsity_masks
Library
__init__(self)
special
Abstract baseclass for the library module.
Source code in deepymod/model/deepmod.py
def __init__(self) -> None:
"""Abstract baseclass for the library module."""
super().__init__()
self.norms = None
forward(self, input)
Compute the library (time derivatives and thetas) from a given dataset. Also calculates the norms of these, later used to calculate the normalized coefficients.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
input |
Tuple[torch.Tensor, torch.Tensor] |
(prediction, data) tuple of size ((n_samples, n_outputs), (n_samples, n_dims)) |
required |
Returns:
Type | Description |
---|---|
Tuple[TensorList, TensorList] |
Tuple[TensorList, TensorList]: Temporal derivative and libraries of size ([(n_samples, 1) x n_outputs]), [(n_samples, n_features)x n_outputs]) |
Source code in deepymod/model/deepmod.py
def forward(self, input: Tuple[torch.Tensor, torch.Tensor]) -> Tuple[TensorList, TensorList]:
"""Compute the library (time derivatives and thetas) from a given dataset. Also calculates the norms
of these, later used to calculate the normalized coefficients.
Args:
input (Tuple[TensorList, TensorList]): (prediction, data) tuple of size ((n_samples, n_outputs), (n_samples, n_dims))
Returns:
Tuple[TensorList, TensorList]: Temporal derivative and libraries of size ([(n_samples, 1) x n_outputs]), [(n_samples, n_features)x n_outputs])
"""
time_derivs, thetas = self.library(input)
self.norms = [(torch.norm(time_deriv) / torch.norm(theta, dim=0, keepdim=True)).detach().squeeze() for time_deriv, theta in zip(time_derivs, thetas)]
return time_derivs, thetas
library(self, input)
Abstract method. Specific method should calculate the temporal derivative and feature matrices. These should be a list; one temporal derivative and feature matrix per output.
input (Tuple[TensorList, TensorList]): (prediction, data) tuple of size ((n_samples, n_outputs), (n_samples, n_dims))
Returns:
Type | Description |
---|---|
Tuple[TensorList, TensorList] |
Tuple[TensorList, TensorList]: Temporal derivative and libraries of size ([(n_samples, 1) x n_outputs]), [(n_samples, n_features)x n_outputs])
Source code in deepymod/model/deepmod.py
@abstractmethod
def library(self, input: Tuple[torch.Tensor, torch.Tensor]) -> Tuple[TensorList, TensorList]:
"""Abstract method. Specific method should calculate the temporal derivative and feature matrices.
These should be a list; one temporal derivative and feature matrix per output.
Args:
input (Tuple[TensorList, TensorList]): (prediction, data) tuple of size ((n_samples, n_outputs), (n_samples, n_dims))
Returns:
Tuple[TensorList, TensorList]: Temporal derivative and libraries of size ([(n_samples, 1) x n_outputs]), [(n_samples, n_features)x n_outputs])
"""
pass