Homec4science
Diffusion Optical_Trapping_ML (master) Browse

R13271/GRAM_measurements/feature_extraction.py
995d2644e47cmaster

Last Change
Raw File

feature_extraction.py
View Options

import numpy as np
from scipy.stats import skew, kurtosis
from scipy.fftpack import fft, ifft
from joblib import Parallel, delayed
# Find the cluster centers for each family
def find_cluster_centers(train_docs):
cluster_centers = {}
bacteria_families = set([doc['bacteria'] for doc in train_docs])
for family in bacteria_families:
family_docs = [doc for doc in train_docs if doc['bacteria'] == family]
mean_values = [np.mean(doc['transmission_normalized']) for doc in family_docs]
std_values = [np.std(doc['transmission_normalized']) for doc in family_docs]
center_mean = np.mean(mean_values)
center_std = np.mean(std_values)
cluster_centers[family] = (center_mean, center_std)
return cluster_centers
# Let's calculate the mean and this confidence interval here the doc is supposed to be normalized
def calculate_means_and_interval_model1(docs):
gram_pos_means = []
gram_neg_means = []
for doc in docs:
mean_value = np.mean(doc['transmission_normalized'])
if doc['label'] == 1:
gram_pos_means.append(mean_value)
elif doc['label'] == 0:
gram_neg_means.append(mean_value)
interval_min = max(gram_pos_means)
interval_max = min(gram_neg_means)
return interval_min, interval_max
# Compute cross-distance (auto-distance), feature suggested in the thesis epfl n10236 p 120
# This measures the self-similarity of the signal over time.
# Compute cross-distance using FFT (fast Fourier transform)
# Purpose: Provides a compact representation of the signal self similarity. It captures how the signal behaves when shifted over time
# The autocorrelation helps in understanding periodicity and repeating patterns within the transmission signal. How the signal correlates with itself over different lags
# We used FFT and IFFT and not other time domain correlation because of computational efficiency
def compute_cross_distance_fft(transmission):
T = len(transmission)
f = fft(transmission) # Computes the FFT of the transmission signal transforming it from the time domain to the frequency domain
power_spectrum = np.abs(f) ** 2 # Shows the distribution of power into the frequency components making up the signal
autocorrelation = ifft(power_spectrum).real # Inverse FFT : IFFT, obtains the autocorrelation of the signal, which measures how the signal correlates with a delayed version of itself over varying delays or lags
cross_distances = np.sqrt(np.sum((transmission - autocorrelation[:T]) ** 2)) # Euclidean distance between the original transmission signal and its autocorrelation, measures the difference between the transmission signal and the autocorrelated signal up to length T
return np.min(cross_distances)
# Now let's extract features from normalized transmission data
# We use parallelized feature extraction to improve computation time
def extract_features_model2(docs):
def process_doc(doc):
transmission = doc['transmission_normalized']
mean_transmission = np.mean(transmission)
std_transmission = np.std(transmission)
max_transmission = np.max(transmission)
min_transmission = np.min(transmission)
skewness = skew(transmission)
kurt = kurtosis(transmission)
cross_distance = compute_cross_distance_fft(transmission)
feature_vector = [mean_transmission, std_transmission, max_transmission, min_transmission, skewness, kurt, cross_distance]
return feature_vector, doc['label'], doc['bacteria']
results = Parallel(n_jobs=-1)(delayed(process_doc)(doc) for doc in docs)
features, gram_labels, bacteria_labels = zip(*results)
return np.array(features), np.array(gram_labels), np.array(bacteria_labels)
#function to separate each time series in 6 chunks
def chunk_time_series(docs, num_chunks=6):
chunked_docs = []
for doc in docs:
transmission = np.array(doc['transmission_normalized'])
chunk_size = len(transmission) // num_chunks
for i in range(num_chunks):
chunk = transmission[i * chunk_size:(i + 1) * chunk_size]
if len(chunk) == chunk_size: # Ensure the chunk is of correct size
chunked_docs.append({
'transmission_normalized': chunk,
'label': doc['label'],
'bacteria': doc['bacteria'],
'name': f"{doc['name']}_chunk_{i}"
})
return chunked_docs
c4science · Help