calcium_wavelet_pipeline.py

from pathlib import Path
import pandas as pd
import numpy as np

# ====== ENTER REPLICATE FOLDERS HERE ======
DATA_FOLDERS = [
    Path(r"..."),
]

# ====== OUTPUT FILE ====== 
# put the correct sample name 
OUTPUT_FILE = Path(
    r"..."
) / "wavelet_summary_SAMPLE.csv"

# ====== FILE PATTERN TO SEARCH FOR ======
PATTERN = "Y_avWavelet*.csv"

results = []

for data_folder in DATA_FOLDERS:
    if not data_folder.exists():
        print(f"[SKIP] Folder does not exist: {data_folder}")
        continue

    files = sorted(data_folder.glob(PATTERN))

    if not files:
        print(f"[SKIP] No {PATTERN} files found in: {data_folder}")
        continue

    # use folder name as replicate identifier
    rep_name = data_folder.name

    for f in files:
        try:
            df = pd.read_csv(f, sep=None, engine="python")

            # remove unwanted columns such as 'Unnamed'
            df = df.loc[:, ~df.columns.astype(str).str.contains(r"^Unnamed")]

            # keep only numeric columns
            numeric_df = df.select_dtypes(include=[np.number])

            if numeric_df.shape[1] < 2:
                print(f"[SKIP] {f.name}: not enough numeric columns")
                continue

            # try to identify columns by name
            cols_lower = {str(c).lower(): c for c in numeric_df.columns}

            period_col = None
            power_col = None

            for key in cols_lower:
                if period_col is None and ("period" in key or "periods" in key):
                    period_col = cols_lower[key]
                if power_col is None and ("power" in key or "count" in key or "ampl" in key or "value" in key):
                    power_col = cols_lower[key]

            # if column names cannot be identified, use the first two numeric columns
            if period_col is None or power_col is None:
                period_col = numeric_df.columns[0]
                power_col = numeric_df.columns[1]

            # find row with maximum power
            idx_max = numeric_df[power_col].idxmax()

            best_period = float(numeric_df.loc[idx_max, period_col])
            best_power = float(numeric_df.loc[idx_max, power_col])

            best_frequency = 1 / best_period if best_period != 0 else np.nan
            best_bpm = 60 / best_period if best_period != 0 else np.nan

            results.append({
                "replicate": rep_name,
                "source_folder": str(data_folder),
                "file": f.name,
                "period_at_max": best_period,
                "max_power": best_power,
                "frequency_hz": best_frequency,
                "bpm": best_bpm
            })

            print(
                f"[OK] {rep_name} | {f.name} -> "
                f"period={best_period:.6g}, "
                f"frequency={best_frequency:.6g} Hz, bpm={best_bpm:.6g}"
            )

        except Exception as e:
            print(f"[ERROR] {rep_name} | {f.name}: {e}")

out = pd.DataFrame(results)

# sort by replicate and then by power
if not out.empty:
    out = out.sort_values(["replicate", "max_power"], ascending=[True, False])

out.to_csv(OUTPUT_FILE, index=False)

print("\nResults saved to:")
print(OUTPUT_FILE)
print("\nPreview:")
print(out)