import time
import numpy as np
[docs]
class ELM:
def __init__(self, input_dim, hidden_dim, output_dim, r_m=1):
self.input_dim = input_dim
self.hidden_dim = hidden_dim
self.output_dim = output_dim
self._activation = np.tanh
self._beta = np.random.uniform(-r_m, r_m, size=(self.hidden_dim, self.output_dim))
self._w = np.random.uniform(-r_m, r_m, size=(self.input_dim, self.hidden_dim))
self._bias = np.zeros(self.hidden_dim) #np.random.uniform(-r_m, r_m, size=(self.hidden_dim,))
print('Bias shape:', self._bias.shape)
print('W shape:', self._w.shape)
print('Beta shape:', self._beta.shape)
[docs]
def fit(self, X, Y, display_time=False):
H = self._activation(X.dot(self._w) + self._bias)
if display_time:
start = time.time()
H_pinv = np.linalg.pinv(H)
if display_time:
stop = time.time()
print(f'Train time: {stop-start}')
self._beta = H_pinv.dot(Y)
# print('Fit Beta shape:', self._beta.shape)
def __call__(self, X):
H = self._activation(X.dot(self._w) + self._bias)
return H.dot(self._beta)
[docs]
class LocELM:
def __init__(self, X, num_subdomains, num_hidden_nodes, r_m, splitting_dimension_range):
"""
Implements the locELM approach for function representation.
Parameters:
X (np.ndarray): Input data
y (np.ndarray): Output data
num_subdomains (int): Number of subdomains to partition the domain into
num_hidden_nodes (int): Number of nodes in the last hidden layer of each local ELM
r_m (float): Maximum magnitude of the random weights and biases
splitting_dimension_range (tuple): Range of the dimension along which to split the data into subdomains
Returns:
np.ndarray: Trained weights for the output layers of the local ELMs
"""
self.subdomain_indices = self._get_subdomain_indices(X, num_subdomains, splitting_dimension_range)
self.num_subdomains = num_subdomains
self.splitting_dimension_range = splitting_dimension_range
self.hidden_weights = [np.random.uniform(-r_m, r_m, size=(X.shape[1], num_hidden_nodes))
for _ in range(num_subdomains)]
self.hidden_biases = [np.random.uniform(-r_m, r_m, size=(num_hidden_nodes)) # np.zeros(num_hidden_nodes)
for _ in range(num_subdomains)]
# activation_func = lambda x: 1 / (1 + np.exp(-x))
self.activation_func = np.tanh
self.hidden_outputs = [self.activation_func(X[idx] @ self.hidden_weights[i] + self.hidden_biases[i])
for i, idx in enumerate(self.subdomain_indices)]
def _get_subdomain_indices(self, X, num_subdomains, splitting_dimension_range):
"""
Splits the domain into subdomains along the first dimension of X.
Parameters:
X (np.ndarray): Input data
num_subdomains (int): Number of subdomains to partition the domain into
splitting_dimension_range (tuple): Range of the dimension along which to split the data into subdomains
Returns:
np.ndarray: Indices of the subdomains
"""
splitting_intervals = np.linspace(splitting_dimension_range[0], splitting_dimension_range[1], num_subdomains + 1)
subdomain_indices = []
for i in range(num_subdomains):
# If the splitting needs to be done on other dimensions, ej. time, add more conditions here
if i == num_subdomains - 1:
subdomain_indices.append(np.where((X[:, 0] >= splitting_intervals[i]) & (X[:, 0] <= splitting_intervals[i + 1])))
else:
subdomain_indices.append(np.where((X[:, 0] >= splitting_intervals[i]) & (X[:, 0] < splitting_intervals[i + 1])))
return subdomain_indices
[docs]
def fit(self, y):
self.betas = []
for i, idx in enumerate(self.subdomain_indices):
A = self.hidden_outputs[i]
b = y[idx]
self.betas.append(np.linalg.lstsq(A, b, rcond=None)[0])
# return np.concatenate(self.output_weights)
def __call__(self, X):
subdomain_indices = self._get_subdomain_indices(X, self.num_subdomains, self.splitting_dimension_range)
for i, idx in enumerate(subdomain_indices):
H = self.activation_func(X[idx] @ self.hidden_weights[i] + self.hidden_biases[i])
pred = H @ self.betas[i]
# stack idx and pred into a single array
if i == 0:
output = np.hstack((idx[0].reshape(-1, 1), pred))
else:
output = np.vstack((output, np.hstack((idx[0].reshape(-1, 1), pred))))
# sort the output array by the first column and keep only the predictions
output = output[output[:, 0].argsort()][:, 1:]
return output