Support High/Low Pass Fillter

.gitignore

@@ -6,3 +6,4 @@
 __pycache__
 flagged
 *.DS_Store
+*.mp4

Dockerfile

@@ -10,3 +10,8 @@ WORKDIR /app
 COPY pyproject.toml* poetry.lock* /app/
 RUN poetry install 
 RUN rm -rf /app/pyproject.toml* /app/poetry.lock*
+
+# For Huggingface
+# COPY . /app/
+
+# CMD [ "python3", "lab_tool_webui.py" ]

lab_tool_webui.py → app.py

@@ -22,6 +22,16 @@ with gr.Blocks() as main_ui:
                 column_dropdown = gr.Dropdown(["Fp1", "Fp2", "T7", "T8", "O1", "O2"], value="Fp2", label="使用する信号データ", allow_custom_value=True, info="使用する信号データを選んでください。デフォルトはFp2です。")
                 start_time = gr.Slider(minimum=0, maximum=60, value=0.0, step=0.5, label="Start Time (sec)")
                 end_time = gr.Slider(minimum=0, maximum=60, value=60.0, step=0.5, label="End Time (sec)")
+                filter_setting = gr.Radio(
+                    ["No Filter", "High PASS", "Low PASS"],
+                    label="フィルター設定",
+                    value="High PASS",
+                )
+                fp_hp = gr.Slider(minimum=0, maximum=20, value=3, step=0.1, label="通過域端周波数 [Hz]")
+                fs_hp = gr.Slider(minimum=0, maximum=20, value=1, step=0.1, label="阻止域端周波数 [Hz]")
+                gpass = gr.Slider(minimum=0, maximum=100, value=3, step=1, label="通過域端最大損失 [dB]")
+                gstop = gr.Slider(minimum=0, maximum=100, value=40, step=1, label="阻止域端最小損失 [dB]")
+
                 submit_button = gr.Button("計算開始")
 
                 file_input.change(
@@ -34,7 +44,11 @@ with gr.Blocks() as main_ui:
                 wavelet_image = gr.Image(type="filepath", label="Wavelet")
                 signal_image = gr.Image(type="filepath", label="Signal")
 
-        submit_button.click(wavelet.wavelet_ui, inputs=[file_input, fs_slider, fmax_slider, column_dropdown, start_time, end_time], outputs=[wavelet_image, signal_image])
+        submit_button.click(wavelet.wavelet_ui, inputs=[
+            file_input,
+            fs_slider, fmax_slider, column_dropdown, start_time, end_time,
+            filter_setting, fp_hp, fs_hp, gpass, gstop],
+            outputs=[wavelet_image, signal_image])
 
     with gr.Tab("1f Noise Search"):
         with gr.Row():
@@ -50,4 +64,4 @@ with gr.Blocks() as main_ui:
 
 
 if __name__ == "__main__":
-    main_ui.queue().launch(server_name="0.0.0.0")
+    main_ui.queue().launch(server_name="0.0.0.0", server_port=7860)

lab_tools/filter.py

@@ -0,0 +1,21 @@
+import scipy.signal as signal
+
+
+def apply_filter(signal_data, sample_rate, pass_freq, stop_freq, pass_gain, stop_gain, filter_type):
+    nyquist_freq = sample_rate / 2                       # ナイキスト周波数
+    normalized_pass_freq = pass_freq / nyquist_freq      # 通過域端周波数を正規化
+    normalized_stop_freq = stop_freq / nyquist_freq      # 阻止域端周波数を正規化
+    filter_order, cutoff_freq = signal.buttord(
+        normalized_pass_freq, normalized_stop_freq, pass_gain, stop_gain
+    )  # フィルタのオーダーと正規化周波数を計算
+    b, a = signal.butter(filter_order, cutoff_freq, filter_type)  # フィルタの伝達関数を計算
+    filtered_signal = signal.filtfilt(b, a, signal_data)          # 信号にフィルタを適用
+    return filtered_signal
+
+
+def lowpass(signal_data, sample_rate, pass_freq, stop_freq, pass_gain, stop_gain):
+    return apply_filter(signal_data, sample_rate, pass_freq, stop_freq, pass_gain, stop_gain, "low")
+
+
+def highpass(signal_data, sample_rate, pass_freq, stop_freq, pass_gain, stop_gain):
+    return apply_filter(signal_data, sample_rate, pass_freq, stop_freq, pass_gain, stop_gain, "high")

lab_tools/highpass.py

@@ -1,62 +0,0 @@
-import numpy as np
-from scipy import signal
-from scipy import fftpack
-from typing import Tuple
-
-
-def highpass_filter(signal_data: np.ndarray, samplerate: float, fp: float, fs: float, gpass: float, gstop: float) -> np.ndarray:
-    fn = samplerate / 2
-    wp = fp / fn
-    ws = fs / fn
-
-    N, Wn = signal.buttord(wp, ws, gpass, gstop)
-    b, a = signal.butter(N, Wn, "high")
-
-    filtered_signal = signal.filtfilt(b, a, signal_data)
-
-    return filtered_signal
-
-
-def overlap_frames(signal_data: np.ndarray, samplerate: float, frame_size: int, overlap: float) -> Tuple[np.ndarray, int]:
-    total_duration = len(signal_data) / samplerate
-    frame_duration = frame_size / samplerate
-    step_size = frame_size * (1 - overlap / 100)
-
-    num_frames = int((total_duration - (frame_duration * overlap / 100)) / (frame_duration * (1 - overlap / 100)))
-
-    frames = []
-
-    for i in range(num_frames):
-        start_idx = int(step_size * i)
-        frames.append(signal_data[start_idx:start_idx + frame_size])
-
-    return np.array(frames), num_frames
-
-
-def hanning(signal_data: np.ndarray, frame_size: int, num_frames: int) -> Tuple[np.ndarray, float]:
-    han = signal.get_window('hann', frame_size)
-    acf = 1 / (sum(han) / frame_size)
-
-    for i in range(num_frames):
-        signal_data[i] *= han
-
-    return signal_data, acf
-
-
-def fft_ave(signal_data: np.ndarray, samplerate: float, frame_size: int, num_frames: int, acf: float):
-    fft_array = []
-    for i in range(num_frames):
-        fft_result = fftpack.fft(signal_data[i]) / frame_size
-        fft_array.append(acf * np.abs(fft_result))
-
-    fft_axis = np.linspace(0, samplerate / 2, frame_size // 2)
-    fft_array = np.array(fft_array)[:, :frame_size // 2]
-    fft_mean = np.mean(fft_array, axis=0)
-
-    return fft_array, fft_mean, fft_axis
-
-
-def linear_to_db(x: float, y: float) -> float:
-    if y == 0:
-        raise ValueError("y cannot be zero in logarithmic conversion")
-    return 20 * np.log10(x / y)

lab_tools/wavelet.py

@@ -4,6 +4,7 @@ import math
 import tempfile
 
 from lab_tools import labutils
+from lab_tools import filter
 
 
 # モルレーウェーブレット関数
@@ -42,13 +43,24 @@ def plot_cwt(cwt_result, time_data, fmax):
 
 
 # グラフ描画とCWTの処理を行う関数
-def wavelet_ui(uploaded_file, Fs, fmax, column_name, start_time, end_time):
+def wavelet_ui(
+        uploaded_file,
+        Fs, fmax, column_name, start_time, end_time,
+        filter_setting, fp_hp, fs_hp, gpass, gstop):
     filepath = uploaded_file.name
     signal = labutils.load_signal(filepath, column_name)
-
     if len(signal) == 0:
         return None, None
 
+    # Filter
+    timestamps = labutils.load_signal(filepath, "Timestamp")
+    dt = (timestamps[1] - timestamps[0])
+    samplerate = 1.0 / dt
+    if filter_setting == "High PASS":
+        signal = filter.highpass(signal, samplerate, fp_hp, fs_hp, gpass, gstop)
+    elif filter_setting == "Low PASS":
+        signal = filter.lowpass(signal, samplerate, fp_hp, fs_hp, gpass, gstop)
+
     # 時間データを計算
     t_data = np.arange(0, len(signal) / Fs, 1 / Fs)
 

run.py

@@ -1,71 +0,0 @@
-import numpy as np
-import matplotlib.pyplot as plt
-from pydub import AudioSegment
-from scipy.fftpack import fft
-import yt_dlp
-import os
-
-url = "https://youtu.be/Ci_zad39Uhw?si=AhB9ArgrWUvbPiv5"
-
-ydl_opts = {
-    'postprocessors': [
-        {
-            'key': 'FFmpegExtractAudio',
-            'preferredcodec': 'mp3',
-            'preferredquality': '128',
-        }
-    ],
-    'outtmpl': '%(title)s.%(ext)s'  # ファイル名のテンプレート
-}
-
-with yt_dlp.YoutubeDL(ydl_opts) as ydl:
-    info_dict = ydl.extract_info(url, download=True)
-    file_path = ydl.prepare_filename(info_dict)
-    filename, ext = os.path.splitext(file_path)
-    filename += ".mp3"
-    print(f"Downloaded file path: {filename}")
-
-# MP3ファイルの読み込みとWAV形式への変換
-audio = AudioSegment.from_mp3(filename)
-os.remove(filename)
-data = np.array(audio.get_array_of_samples())
-sample_rate = audio.frame_rate
-
-# モノラル変換（ステレオの場合）
-if audio.channels > 1:
-    data = data.reshape((-1, audio.channels)).mean(axis=1)
-
-# フーリエ変換の実行
-N = len(data)
-T = 1.0 / sample_rate
-yf = fft(data)
-xf = np.fft.fftfreq(N, T)[:N//2]
-
-# パワースペクトルの計算
-power_spectrum = 2.0/N * np.abs(yf[:N//2])
-
-# プロット用に周波数とパワーを制限
-xf_log = xf[1:]  # 0Hzを除去 (ログスケールでは0が扱えないため)
-power_spectrum_log = power_spectrum[1:]
-
-# 縦軸の範囲を指定（例: 0から100まで）
-y_min = 0
-y_max = 1000
-
-# グラフの描画
-plt.figure(figsize=(10, 6))
-plt.plot(xf_log, power_spectrum_log)
-plt.xscale('log')
-plt.yscale('log')
-
-plt.title('Power Spectrum (Log Scale)')
-plt.xlabel('Frequency (Hz)')
-plt.ylabel('Power')
-plt.grid(True, which="both", ls="--")
-plt.xlim([1, sample_rate // 2])  # 1Hz から Nyquist周波数 (sample_rate/2) まで
-
-# 縦軸の範囲を指定
-# plt.ylim([y_min, y_max])
-
-
-plt.show()

test.py

@@ -1,93 +0,0 @@
-import pandas as pd
-import sys
-import matplotlib.pyplot as plt
-from lab_tools import highpass
-
-
-def load_signal(file_path, column_name):
-    try:
-        with open(file_path, 'r') as file:
-            # データ部分が始まる行を見つける
-            for i, line in enumerate(file):
-                if 'Timestamp' in line:
-                    header_line = i
-                    break
-
-        # 見つけたヘッダー行からデータを読み込む
-        df = pd.read_csv(file_path, skiprows=header_line)
-        signal = df[column_name].values
-        return signal
-    except FileNotFoundError as e:
-        print(f"Error: {e}", file=sys.stderr)
-        return []
-    except KeyError as e:
-        print(f"Column '{column_name}' not found in the file. ({e})", file=sys.stderr)
-        return []
-
-
-samplerate = 1000
-time_data = load_signal("./test2_143809.csv", "Timestamp")
-signal_data = load_signal("./test2_143809.csv", "Fp1")
-fp = 22                            # 通過域端周波数[Hz]※ベクトル
-fs = 10                              # 阻止域端周波数[Hz]※ベクトル
-gpass = 5                                               # 通過域端最大損失[dB]
-gstop = 40                                              # 阻止域端最小損失[dB]
-Fs = 4096                                               # フレームサイズ
-overlap = 90
-
-data_filt = highpass.highpass_filter(signal_data, samplerate, fp, fs, gpass, gstop)
-
-t_array_org, N_ave_org = highpass.overlap_frames(signal_data, samplerate, Fs, overlap)
-t_array_filt, N_ave_filt = highpass.overlap_frames(signal_data, samplerate, Fs, overlap)
-
-t_array_org, acf_org = highpass.hanning(t_array_org, Fs, N_ave_org)
-t_array_filt, acf_filt = highpass.hanning(t_array_filt, Fs, N_ave_filt)
-
-fft_array_org, fft_mean_org, fft_axis_org = highpass.fft_ave(t_array_org, samplerate, Fs, N_ave_org, acf_org)
-fft_array_filt, fft_mean_filt, fft_axis_filt = highpass.fft_ave(t_array_filt, samplerate, Fs, N_ave_filt, acf_filt)
-
-fft_mean_org = highpass.linear_to_db(fft_mean_org, 2e-5)
-fft_mean_filt = highpass.linear_to_db(fft_mean_filt, 2e-5)
-
-# フォントの種類とサイズを設定する。
-# plt.rcParams['font.size'] = 14
-# plt.rcParams['font.family'] = 'Times New Roman'
-
-# 目盛を内側にする。
-plt.rcParams['xtick.direction'] = 'in'
-plt.rcParams['ytick.direction'] = 'in'
-
-# グラフの上下左右に目盛線を付ける。
-fig = plt.figure(figsize=(20, 10))
-ax1 = fig.add_subplot(211)
-ax1.yaxis.set_ticks_position('both')
-ax1.xaxis.set_ticks_position('both')
-ax2 = fig.add_subplot(212)
-ax2.yaxis.set_ticks_position('both')
-ax2.xaxis.set_ticks_position('both')
-
-# 軸のラベルを設定する。
-ax1.set_xlabel('Time [s]')
-ax1.set_ylabel('V[μV]')
-ax2.set_xlabel('Frequency [Hz]')
-ax2.set_ylabel('Amp[dB]')
-
-# データプロットの準備とともに、ラベルと線の太さ、凡例の設置を行う。
-ax1.plot(time_data, signal_data, label='original', lw=1)
-ax1.plot(time_data, data_filt, label='filtered', lw=1)
-ax2.plot(fft_axis_org, fft_mean_org, label='original', lw=1)
-ax2.plot(fft_axis_filt, fft_mean_filt, label='filtered', lw=1)
-plt.legend()
-
-# 軸のリミットを設定する。
-# ax1.set_xlim(0,1200)
-# ax1.set_xticks(np.arange(0,1201,100))
-# ax2.set_xlim(0, max(fft_axis_org)/2)
-# ax2.set_xticks(np.arange(0,501,10))
-# ax2.set_ylim(-50, 150)
-
-# レイアウト設定
-fig.tight_layout()
-
-# グラフを表示する。
-plt.savefig("./out.png")