Advanced Filtering

Kalman Filter

from vitalDSP.filtering.advanced_signal_filtering import AdvancedSignalFiltering
import plotly.io as pio
pio.renderers.default = "sphinx_gallery"
# pio.renderers.default = "plotly_mimetype"  # or "jupyterlab"
# from IPython.display import display, HTML
# display(HTML('<script src="https://cdnjs.cloudflare.com/ajax/libs/require.js/2.3.6/require.min.js"></script>'))
import numpy as np
from plotly import graph_objects as go
import os
from vitalDSP.notebooks import plot_trace, load_sample_ecg_small


# FILE_PATH = os.path.join("..","..","..","sample_data","public","ecg.csv")
# signal_col, date_col = process_in_chunks(FILE_PATH,data_type='ecg', fs=128)
signal_col, date_col = load_sample_ecg_small()
signal_col = np.array(signal_col)

af = AdvancedSignalFiltering(signal_col)
filtered_signal = af.kalman_filter(R=0.1, Q=0.01)

plot_trace(signal_col,filtered_signal)

Optimization-based filter

from vitalDSP.filtering.advanced_signal_filtering import AdvancedSignalFiltering
from vitalDSP.utils.data_processing.synthesize_data import generate_ecg_signal
import numpy as np
from vitalDSP.notebooks import load_sample_ecg_small, plot_trace

fs = 128
signal_col, date_col = load_sample_ecg_small()
signal_col = np.array(signal_col)

target = generate_ecg_signal(sfecg=128, N=30, Anoise=0.01, hrmean=70, sfint=512)
target = np.array(target[:min(len(target), len(signal_col))])
signal_col = signal_col[:len(target)]

af = AdvancedSignalFiltering(signal_col)
filtered_signal = af.optimization_based_filtering(target, loss_type="mse",
                                                   initial_guess=0, learning_rate=0.01,
                                                   iterations=20)
plot_trace(signal_col, filtered_signal)

Gradient Descent Filter

from vitalDSP.filtering.advanced_signal_filtering import AdvancedSignalFiltering
from vitalDSP.utils.data_processing.synthesize_data import generate_ecg_signal
import numpy as np
from plotly import graph_objects as go
import os
from vitalDSP.notebooks import load_sample_ecg_small, plot_trace

fs = 128
signal_col, date_col = load_sample_ecg_small()
signal_col = np.array(signal_col)

# Create a target signal
target = generate_ecg_signal(sfecg=128, N=30, Anoise=0.01, hrmean=70, sfint=512)
target = target[:min(len(target),len(signal_col))]

signal_col = signal_col[:min(len(target),len(signal_col))]
af = AdvancedSignalFiltering(signal_col)
filtered_signal = af.gradient_descent_filter(target, learning_rate=0.1, 
                                                iterations=20)

plot_trace(signal_col,filtered_signal)

Convolution-based filter

from vitalDSP.filtering.advanced_signal_filtering import AdvancedSignalFiltering
import numpy as np
from vitalDSP.notebooks import plot_trace, load_sample_ecg_small

signal_col, date_col = load_sample_ecg_small()
signal_col = np.array(signal_col)

af = AdvancedSignalFiltering(signal_col)

smoothed = af.convolution_based_filter(kernel_type="smoothing", kernel_size=5)
plot_trace(signal_col, smoothed, title="Smoothing kernel")

edge = af.convolution_based_filter(kernel_type="edge_detection")
plot_trace(signal_col, edge, title="Edge detection kernel")

sharpened = af.convolution_based_filter(kernel_type="sharpening")
plot_trace(signal_col, sharpened, title="Sharpening kernel")

Attention-based filter

from vitalDSP.filtering.advanced_signal_filtering import AdvancedSignalFiltering
import numpy as np
from vitalDSP.notebooks import load_sample_ecg_small, plot_trace

signal_col, date_col = load_sample_ecg_small()
signal_col = np.array(signal_col)

af = AdvancedSignalFiltering(signal_col)

gaussian_out = af.attention_based_filter(attention_type="gaussian", size=11, sigma=2.0)
plot_trace(signal_col, gaussian_out, title="Gaussian attention")

linear_out = af.attention_based_filter(attention_type="linear", size=11, ascending=True)
plot_trace(signal_col, linear_out, title="Linear attention")

exp_out = af.attention_based_filter(attention_type="exponential", size=7, base=2.0)
plot_trace(signal_col, exp_out, title="Exponential attention")

LMS Adaptive Filtering

from vitalDSP.filtering.advanced_signal_filtering import AdvancedSignalFiltering
from vitalDSP.utils.data_processing.synthesize_data import generate_ecg_signal
import numpy as np
from vitalDSP.notebooks import load_sample_ecg_small, plot_trace

fs = 128
signal_col, date_col = load_sample_ecg_small()
signal_col = np.array(signal_col)

target = generate_ecg_signal(sfecg=128, N=30, Anoise=0.01, hrmean=70, sfint=512)
target = np.array(target[:min(len(target), len(signal_col))])
signal_col = signal_col[:len(target)]

af = AdvancedSignalFiltering(signal_col)
# mu must satisfy mu < 1/(filter_order * max_signal_power) for LMS stability
filtered_signal = af.adaptive_filtering(target, mu=0.005, filter_order=5)

plot_trace(signal_col, filtered_signal)