Buckets:

DLProjectCNNLSTM
/

eeg-seizure-project

Files

xet

DLProjectCNNLSTM/eeg-seizure-project / CNN-LSTM /verify_processed_data.py

omkorde7470

24 days ago

download

raw

13.1 kB

"""
verify_processed_data.py
Comprehensive verification script for preprocessed EEG seizure dataset.
Verifies shapes, types, metadata alignment, NaN/Inf checks, channel ordering,
and fixed subject-wise splits using memory mapping for fast zero-RAM execution.
"""

import os
import sys
import json
import time
import zipfile
import numpy as np
import pandas as pd

DATA_DIR = os.path.join("data", "processed", "10s")
NPZ_PATH = os.path.join(DATA_DIR, "eeg_windows.npz")
MMAP_PATH = os.path.join(DATA_DIR, "windows.npy")
METADATA_PATH = os.path.join(DATA_DIR, "window_metadata_full.csv")
MANIFEST_PATH = os.path.join(DATA_DIR, "processing_manifest.json")
REPORT_PATH = os.path.join(DATA_DIR, "dataset_verification_report.json")
SPLIT_MANIFEST_PATH = os.path.join(DATA_DIR, "split_manifest.json")

# Established subject assignments
TRAIN_SUBJECTS = [f"chb{i:02d}" for i in range(1, 16)] + ["chb21"]
VAL_SUBJECTS = ["chb16", "chb17", "chb18", "chb19"]
TEST_SUBJECTS = ["chb20", "chb22", "chb23", "chb24"]


def extract_members_if_needed():
    """Extract windows.npy to disk for memory-mapped access if not already present."""
    if os.path.exists(MMAP_PATH):
        print(f"Memory-mapped array already extracted at {MMAP_PATH} ({os.path.getsize(MMAP_PATH) / (1024**2):.2f} MB)", flush=True)
        return
    print(f"Extracting 'windows.npy' from {NPZ_PATH} to disk for zero-RAM memory-mapping...", flush=True)
    t0 = time.time()
    with zipfile.ZipFile(NPZ_PATH, 'r') as z:
        z.extract("windows.npy", DATA_DIR)
    t1 = time.time()
    print(f"Extraction completed in {t1 - t0:.1f}s. Size: {os.path.getsize(MMAP_PATH) / (1024**2):.2f} MB", flush=True)


def verify_dataset():
    report = {
        "verification_timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "files_checked": {},
        "npz_verification": {},
        "metadata_verification": {},
        "data_integrity": {},
        "split_verification": {},
        "passed": False,
        "errors": []
    }
    
    # 1. Check file existence and sizes
    for fname, fpath in [
        ("npz", NPZ_PATH),
        ("metadata", METADATA_PATH),
        ("manifest", MANIFEST_PATH)
    ]:
        if not os.path.exists(fpath):
            err = f"Missing required file: {fpath}"
            report["errors"].append(err)
            print(f"ERROR: {err}", flush=True)
            return report
        report["files_checked"][fname] = {
            "path": fpath,
            "size_bytes": os.path.getsize(fpath),
            "size_mb": round(os.path.getsize(fpath) / (1024**2), 2)
        }

    # 2. Inspect processing manifest
    with open(MANIFEST_PATH, "r") as f:
        manifest_data = json.load(f)
    report["manifest_specs"] = manifest_data
    expected_channels = manifest_data.get("channel_count", 18)
    expected_samples = manifest_data.get("samples_per_window", 2560)
    expected_ordering = manifest_data.get("channel_ordering", [])

    # 3. Check NPZ archive members
    print("Checking NPZ structure...", flush=True)
    with zipfile.ZipFile(NPZ_PATH, 'r') as z:
        archived_files = z.namelist()
        report["npz_verification"]["archived_members"] = archived_files
        print(f"NPZ archived members: {archived_files}", flush=True)

    # Extract windows.npy for zero-RAM mmap
    extract_members_if_needed()

    # Memory-map windows array
    mmap_windows = np.load(MMAP_PATH, mmap_mode="r")
    windows_shape = list(mmap_windows.shape)
    windows_dtype = str(mmap_windows.dtype)
    num_windows = windows_shape[0]

    # Open labels and window_ids directly from zipfile without loading windows
    with zipfile.ZipFile(NPZ_PATH, 'r') as z:
        with z.open("labels.npy") as f_l:
            labels_arr = np.load(f_l)
        with z.open("window_ids.npy") as f_ids:
            window_ids_arr = np.load(f_ids)

    labels_shape = list(labels_arr.shape)
    labels_dtype = str(labels_arr.dtype)
    window_ids_shape = list(window_ids_arr.shape)

    report["npz_verification"]["windows_shape"] = windows_shape
    report["npz_verification"]["windows_dtype"] = windows_dtype
    report["npz_verification"]["labels_shape"] = labels_shape
    report["npz_verification"]["labels_dtype"] = labels_dtype
    report["npz_verification"]["window_ids_shape"] = window_ids_shape

    print(f"Windows shape: {windows_shape}, dtype: {windows_dtype}", flush=True)
    print(f"Labels shape: {labels_shape}, dtype: {labels_dtype}", flush=True)
    print(f"Window IDs count: {window_ids_shape[0]}", flush=True)

    if windows_shape[1] != expected_channels or windows_shape[2] != expected_samples:
        err = f"Windows dimensions ({windows_shape}) do not match expected channels ({expected_channels}) or samples ({expected_samples})"
        report["errors"].append(err)

    # 4. Check Metadata CSV
    print("Loading and verifying metadata CSV...", flush=True)
    df = pd.read_csv(METADATA_PATH)
    meta_rows = len(df)
    report["metadata_verification"]["row_count"] = meta_rows
    report["metadata_verification"]["columns"] = list(df.columns)
    
    if meta_rows != num_windows:
        err = f"Metadata row count ({meta_rows}) does not match NPZ window count ({num_windows})"
        report["errors"].append(err)
        print(f"ERROR: {err}", flush=True)
    else:
        print(f"Metadata row count ({meta_rows}) matches NPZ windows ({num_windows}).", flush=True)

    # Verify ID and label alignment
    npz_ids = [str(x) for x in window_ids_arr]
    csv_ids = [str(x) for x in df["window_id"]]
    ids_match = (npz_ids == csv_ids)
    report["metadata_verification"]["window_ids_match"] = ids_match
    if not ids_match:
        report["errors"].append("Window IDs in NPZ do not exactly match CSV window_ids order.")

    csv_labels = df["label"].to_numpy()
    labels_match = np.array_equal(labels_arr, csv_labels)
    report["metadata_verification"]["labels_match"] = bool(labels_match)
    if not labels_match:
        report["errors"].append("Labels in NPZ do not exactly match CSV labels order.")

    # Calculate actual label counts
    unique_labels, counts = np.unique(labels_arr, return_counts=True)
    actual_label_dist = {int(k): int(v) for k, v in zip(unique_labels, counts)}
    report["metadata_verification"]["actual_label_distribution"] = actual_label_dist
    print(f"Actual label distribution: {actual_label_dist}", flush=True)

    # Check channels in metadata
    first_channels = df["channel_names"].iloc[0].split(";")
    channel_match = (first_channels == expected_ordering)
    report["metadata_verification"]["channel_ordering_match"] = channel_match
    print(f"Channel ordering matches manifest: {channel_match}", flush=True)

    # 5. Data integrity: NaN / Inf checks via mmap in chunks
    print("Checking for NaN and Inf in EEG windows via memory-mapping...", flush=True)
    has_nan = False
    has_inf = False
    chunk_size = 5000
    total_chunks = (num_windows + chunk_size - 1) // chunk_size

    min_val = float("inf")
    max_val = float("-inf")
    
    t_start = time.time()
    for c in range(total_chunks):
        start_idx = c * chunk_size
        end_idx = min(start_idx + chunk_size, num_windows)
        chunk = mmap_windows[start_idx:end_idx]
        
        c_nan = np.isnan(chunk).any()
        c_inf = np.isinf(chunk).any()
        if c_nan:
            has_nan = True
        if c_inf:
            has_inf = True
            
        c_min = float(np.min(chunk))
        c_max = float(np.max(chunk))
        if c_min < min_val: min_val = c_min
        if c_max > max_val: max_val = c_max
        print(f"  Scanned windows {start_idx} to {end_idx} ({round((end_idx/num_windows)*100, 1)}%)", flush=True)

    print(f"Scan completed in {time.time() - t_start:.2f}s", flush=True)
    report["data_integrity"]["has_nan"] = bool(has_nan)
    report["data_integrity"]["has_inf"] = bool(has_inf)
    report["data_integrity"]["global_min"] = min_val
    report["data_integrity"]["global_max"] = max_val
    print(f"Integrity check: NaN={has_nan}, Inf={has_inf}, Min={min_val:.4f}, Max={max_val:.4f}", flush=True)

    if has_nan or has_inf:
        report["errors"].append("Windows array contains NaN or Inf values.")

    # 6. Verify Fixed Subject-Wise Splits
    print("Verifying subject-wise split assignments...", flush=True)
    unique_subjects = sorted(df["subject_id"].unique())
    report["split_verification"]["total_unique_subjects"] = len(unique_subjects)
    report["split_verification"]["subjects_list"] = unique_subjects

    # Check partition overlaps
    set_train = set(TRAIN_SUBJECTS)
    set_val = set(VAL_SUBJECTS)
    set_test = set(TEST_SUBJECTS)

    train_val_overlap = set_train.intersection(set_val)
    train_test_overlap = set_train.intersection(set_test)
    val_test_overlap = set_val.intersection(set_test)

    if train_val_overlap or train_test_overlap or val_test_overlap:
        err = f"Subject partition overlap detected! TV: {train_val_overlap}, TT: {train_test_overlap}, VT: {val_test_overlap}"
        report["errors"].append(err)
        print(f"ERROR: {err}", flush=True)

    # Check chb01 and chb21 pairing
    if "chb01" in TRAIN_SUBJECTS and "chb21" in TRAIN_SUBJECTS:
        report["split_verification"]["chb01_chb21_co_partitioned"] = True
    else:
        report["split_verification"]["chb01_chb21_co_partitioned"] = False
        report["errors"].append("chb01 and chb21 are not co-partitioned in Training!")

    # Assign partitions to DataFrame
    df["partition"] = None
    df.loc[df["subject_id"].isin(TRAIN_SUBJECTS), "partition"] = "train"
    df.loc[df["subject_id"].isin(VAL_SUBJECTS), "partition"] = "val"
    df.loc[df["subject_id"].isin(TEST_SUBJECTS), "partition"] = "test"

    unassigned = df[df["partition"].isnull()]
    if len(unassigned) > 0:
        report["errors"].append(f"{len(unassigned)} windows have unassigned subjects!")

    # Check partition window IDs disjointness
    train_ids = set(df[df["partition"] == "train"]["window_id"])
    val_ids = set(df[df["partition"] == "val"]["window_id"])
    test_ids = set(df[df["partition"] == "test"]["window_id"])

    id_overlap = (train_ids.intersection(val_ids) or
                  train_ids.intersection(test_ids) or
                  val_ids.intersection(test_ids))
    report["split_verification"]["window_id_disjoint"] = not bool(id_overlap)

    # Calculate per-partition statistics
    split_summary = {}
    for part in ["train", "val", "test"]:
        part_df = df[df["partition"] == part]
        total_w = len(part_df)
        c0 = int((part_df["label"] == 0).sum())
        c1 = int((part_df["label"] == 1).sum())
        c_amb = int((part_df["label"] == -1).sum())
        sup_w = c0 + c1
        
        split_summary[part] = {
            "subjects": sorted(part_df["subject_id"].unique()),
            "total_windows": total_w,
            "supervised_windows": sup_w,
            "label_0_non_seizure": c0,
            "label_1_seizure": c1,
            "label_minus_1_ambiguous": c_amb,
            "seizure_percentage": round(c1 / sup_w * 100, 4) if sup_w > 0 else 0.0
        }

    report["split_verification"]["partitions"] = split_summary
    print("\nPartition Breakdown:", flush=True)
    for part, stats in split_summary.items():
        print(f"  {part.upper()}: {stats['total_windows']} windows (Sup: {stats['supervised_windows']}, Non-seizure: {stats['label_0_non_seizure']}, Seizure: {stats['label_1_seizure']} [{stats['seizure_percentage']:.2f}%], Ambiguous: {stats['label_minus_1_ambiguous']})", flush=True)

    # Generate split files and split manifest
    split_manifest = {
        "dataset_name": manifest_data.get("dataset_name"),
        "total_windows": num_windows,
        "partitions": split_summary,
        "established_assignments": {
            "train": TRAIN_SUBJECTS,
            "val": VAL_SUBJECTS,
            "test": TEST_SUBJECTS
        }
    }
    with open(SPLIT_MANIFEST_PATH, "w") as f:
        json.dump(split_manifest, f, indent=2)
    print(f"Split manifest saved to {SPLIT_MANIFEST_PATH}", flush=True)

    # Save individual partition CSVs for fast lookup
    for part in ["train", "val", "test"]:
        part_csv = os.path.join(DATA_DIR, f"{part}_split.csv")
        part_df = df[df["partition"] == part].copy()
        part_df.to_csv(part_csv, index=False)
        print(f"Saved {part} split metadata to {part_csv}", flush=True)

    # Final verdict
    report["passed"] = (len(report["errors"]) == 0)
    with open(REPORT_PATH, "w") as f:
        json.dump(report, f, indent=2)
    print(f"\nVerification report written to {REPORT_PATH}", flush=True)
    print(f"VERIFICATION STATUS: {'PASSED' if report['passed'] else 'FAILED'}", flush=True)

    return report


if __name__ == "__main__":
    report = verify_dataset()
    if not report["passed"]:
        print("VERIFICATION FAILED with errors:", flush=True)
        for e in report["errors"]:
            print(f"  - {e}", flush=True)
        sys.exit(1)
    else:
        print("DATASET VERIFICATION COMPLETE AND VALIDATED!", flush=True)

Xet Storage Details

Size:: 13.1 kB
Xet hash:: bae0905db8d1c75ebba0f2d64e8bc97f60c1558e4025dc074c28a358a64d6ce8

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	"""
	verify_processed_data.py
	Comprehensive verification script for preprocessed EEG seizure dataset.
	Verifies shapes, types, metadata alignment, NaN/Inf checks, channel ordering,
	and fixed subject-wise splits using memory mapping for fast zero-RAM execution.
	"""

	import os
	import sys
	import json
	import time
	import zipfile
	import numpy as np
	import pandas as pd

	DATA_DIR = os.path.join("data", "processed", "10s")
	NPZ_PATH = os.path.join(DATA_DIR, "eeg_windows.npz")
	MMAP_PATH = os.path.join(DATA_DIR, "windows.npy")
	METADATA_PATH = os.path.join(DATA_DIR, "window_metadata_full.csv")
	MANIFEST_PATH = os.path.join(DATA_DIR, "processing_manifest.json")
	REPORT_PATH = os.path.join(DATA_DIR, "dataset_verification_report.json")
	SPLIT_MANIFEST_PATH = os.path.join(DATA_DIR, "split_manifest.json")

	# Established subject assignments
	TRAIN_SUBJECTS = [f"chb{i:02d}" for i in range(1, 16)] + ["chb21"]
	VAL_SUBJECTS = ["chb16", "chb17", "chb18", "chb19"]
	TEST_SUBJECTS = ["chb20", "chb22", "chb23", "chb24"]


	def extract_members_if_needed():
	"""Extract windows.npy to disk for memory-mapped access if not already present."""
	if os.path.exists(MMAP_PATH):
	print(f"Memory-mapped array already extracted at {MMAP_PATH} ({os.path.getsize(MMAP_PATH) / (1024**2):.2f} MB)", flush=True)
	return
	print(f"Extracting 'windows.npy' from {NPZ_PATH} to disk for zero-RAM memory-mapping...", flush=True)
	t0 = time.time()
	with zipfile.ZipFile(NPZ_PATH, 'r') as z:
	z.extract("windows.npy", DATA_DIR)
	t1 = time.time()
	print(f"Extraction completed in {t1 - t0:.1f}s. Size: {os.path.getsize(MMAP_PATH) / (1024**2):.2f} MB", flush=True)


	def verify_dataset():
	report = {
	"verification_timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
	"files_checked": {},
	"npz_verification": {},
	"metadata_verification": {},
	"data_integrity": {},
	"split_verification": {},
	"passed": False,
	"errors": []
	}

	# 1. Check file existence and sizes
	for fname, fpath in [
	("npz", NPZ_PATH),
	("metadata", METADATA_PATH),
	("manifest", MANIFEST_PATH)
	]:
	if not os.path.exists(fpath):
	err = f"Missing required file: {fpath}"
	report["errors"].append(err)
	print(f"ERROR: {err}", flush=True)
	return report
	report["files_checked"][fname] = {
	"path": fpath,
	"size_bytes": os.path.getsize(fpath),
	"size_mb": round(os.path.getsize(fpath) / (1024**2), 2)
	}

	# 2. Inspect processing manifest
	with open(MANIFEST_PATH, "r") as f:
	manifest_data = json.load(f)
	report["manifest_specs"] = manifest_data
	expected_channels = manifest_data.get("channel_count", 18)
	expected_samples = manifest_data.get("samples_per_window", 2560)
	expected_ordering = manifest_data.get("channel_ordering", [])

	# 3. Check NPZ archive members
	print("Checking NPZ structure...", flush=True)
	with zipfile.ZipFile(NPZ_PATH, 'r') as z:
	archived_files = z.namelist()
	report["npz_verification"]["archived_members"] = archived_files
	print(f"NPZ archived members: {archived_files}", flush=True)

	# Extract windows.npy for zero-RAM mmap
	extract_members_if_needed()

	# Memory-map windows array
	mmap_windows = np.load(MMAP_PATH, mmap_mode="r")
	windows_shape = list(mmap_windows.shape)
	windows_dtype = str(mmap_windows.dtype)
	num_windows = windows_shape[0]

	# Open labels and window_ids directly from zipfile without loading windows
	with zipfile.ZipFile(NPZ_PATH, 'r') as z:
	with z.open("labels.npy") as f_l:
	labels_arr = np.load(f_l)
	with z.open("window_ids.npy") as f_ids:
	window_ids_arr = np.load(f_ids)

	labels_shape = list(labels_arr.shape)
	labels_dtype = str(labels_arr.dtype)
	window_ids_shape = list(window_ids_arr.shape)

	report["npz_verification"]["windows_shape"] = windows_shape
	report["npz_verification"]["windows_dtype"] = windows_dtype
	report["npz_verification"]["labels_shape"] = labels_shape
	report["npz_verification"]["labels_dtype"] = labels_dtype
	report["npz_verification"]["window_ids_shape"] = window_ids_shape

	print(f"Windows shape: {windows_shape}, dtype: {windows_dtype}", flush=True)
	print(f"Labels shape: {labels_shape}, dtype: {labels_dtype}", flush=True)
	print(f"Window IDs count: {window_ids_shape[0]}", flush=True)

	if windows_shape[1] != expected_channels or windows_shape[2] != expected_samples:
	err = f"Windows dimensions ({windows_shape}) do not match expected channels ({expected_channels}) or samples ({expected_samples})"
	report["errors"].append(err)

	# 4. Check Metadata CSV
	print("Loading and verifying metadata CSV...", flush=True)
	df = pd.read_csv(METADATA_PATH)
	meta_rows = len(df)
	report["metadata_verification"]["row_count"] = meta_rows
	report["metadata_verification"]["columns"] = list(df.columns)

	if meta_rows != num_windows:
	err = f"Metadata row count ({meta_rows}) does not match NPZ window count ({num_windows})"
	report["errors"].append(err)
	print(f"ERROR: {err}", flush=True)
	else:
	print(f"Metadata row count ({meta_rows}) matches NPZ windows ({num_windows}).", flush=True)

	# Verify ID and label alignment
	npz_ids = [str(x) for x in window_ids_arr]
	csv_ids = [str(x) for x in df["window_id"]]
	ids_match = (npz_ids == csv_ids)
	report["metadata_verification"]["window_ids_match"] = ids_match
	if not ids_match:
	report["errors"].append("Window IDs in NPZ do not exactly match CSV window_ids order.")

	csv_labels = df["label"].to_numpy()
	labels_match = np.array_equal(labels_arr, csv_labels)
	report["metadata_verification"]["labels_match"] = bool(labels_match)
	if not labels_match:
	report["errors"].append("Labels in NPZ do not exactly match CSV labels order.")

	# Calculate actual label counts
	unique_labels, counts = np.unique(labels_arr, return_counts=True)
	actual_label_dist = {int(k): int(v) for k, v in zip(unique_labels, counts)}
	report["metadata_verification"]["actual_label_distribution"] = actual_label_dist
	print(f"Actual label distribution: {actual_label_dist}", flush=True)

	# Check channels in metadata
	first_channels = df["channel_names"].iloc[0].split(";")
	channel_match = (first_channels == expected_ordering)
	report["metadata_verification"]["channel_ordering_match"] = channel_match
	print(f"Channel ordering matches manifest: {channel_match}", flush=True)

	# 5. Data integrity: NaN / Inf checks via mmap in chunks
	print("Checking for NaN and Inf in EEG windows via memory-mapping...", flush=True)
	has_nan = False
	has_inf = False
	chunk_size = 5000
	total_chunks = (num_windows + chunk_size - 1) // chunk_size

	min_val = float("inf")
	max_val = float("-inf")

	t_start = time.time()
	for c in range(total_chunks):
	start_idx = c * chunk_size
	end_idx = min(start_idx + chunk_size, num_windows)
	chunk = mmap_windows[start_idx:end_idx]

	c_nan = np.isnan(chunk).any()
	c_inf = np.isinf(chunk).any()
	if c_nan:
	has_nan = True
	if c_inf:
	has_inf = True

	c_min = float(np.min(chunk))
	c_max = float(np.max(chunk))
	if c_min < min_val: min_val = c_min
	if c_max > max_val: max_val = c_max
	print(f" Scanned windows {start_idx} to {end_idx} ({round((end_idx/num_windows)*100, 1)}%)", flush=True)

	print(f"Scan completed in {time.time() - t_start:.2f}s", flush=True)
	report["data_integrity"]["has_nan"] = bool(has_nan)
	report["data_integrity"]["has_inf"] = bool(has_inf)
	report["data_integrity"]["global_min"] = min_val
	report["data_integrity"]["global_max"] = max_val
	print(f"Integrity check: NaN={has_nan}, Inf={has_inf}, Min={min_val:.4f}, Max={max_val:.4f}", flush=True)

	if has_nan or has_inf:
	report["errors"].append("Windows array contains NaN or Inf values.")

	# 6. Verify Fixed Subject-Wise Splits
	print("Verifying subject-wise split assignments...", flush=True)
	unique_subjects = sorted(df["subject_id"].unique())
	report["split_verification"]["total_unique_subjects"] = len(unique_subjects)
	report["split_verification"]["subjects_list"] = unique_subjects

	# Check partition overlaps
	set_train = set(TRAIN_SUBJECTS)
	set_val = set(VAL_SUBJECTS)
	set_test = set(TEST_SUBJECTS)

	train_val_overlap = set_train.intersection(set_val)
	train_test_overlap = set_train.intersection(set_test)
	val_test_overlap = set_val.intersection(set_test)

	if train_val_overlap or train_test_overlap or val_test_overlap:
	err = f"Subject partition overlap detected! TV: {train_val_overlap}, TT: {train_test_overlap}, VT: {val_test_overlap}"
	report["errors"].append(err)
	print(f"ERROR: {err}", flush=True)

	# Check chb01 and chb21 pairing
	if "chb01" in TRAIN_SUBJECTS and "chb21" in TRAIN_SUBJECTS:
	report["split_verification"]["chb01_chb21_co_partitioned"] = True
	else:
	report["split_verification"]["chb01_chb21_co_partitioned"] = False
	report["errors"].append("chb01 and chb21 are not co-partitioned in Training!")

	# Assign partitions to DataFrame
	df["partition"] = None
	df.loc[df["subject_id"].isin(TRAIN_SUBJECTS), "partition"] = "train"
	df.loc[df["subject_id"].isin(VAL_SUBJECTS), "partition"] = "val"
	df.loc[df["subject_id"].isin(TEST_SUBJECTS), "partition"] = "test"

	unassigned = df[df["partition"].isnull()]
	if len(unassigned) > 0:
	report["errors"].append(f"{len(unassigned)} windows have unassigned subjects!")

	# Check partition window IDs disjointness
	train_ids = set(df[df["partition"] == "train"]["window_id"])
	val_ids = set(df[df["partition"] == "val"]["window_id"])
	test_ids = set(df[df["partition"] == "test"]["window_id"])

	id_overlap = (train_ids.intersection(val_ids) or
	train_ids.intersection(test_ids) or
	val_ids.intersection(test_ids))
	report["split_verification"]["window_id_disjoint"] = not bool(id_overlap)

	# Calculate per-partition statistics
	split_summary = {}
	for part in ["train", "val", "test"]:
	part_df = df[df["partition"] == part]
	total_w = len(part_df)
	c0 = int((part_df["label"] == 0).sum())
	c1 = int((part_df["label"] == 1).sum())
	c_amb = int((part_df["label"] == -1).sum())
	sup_w = c0 + c1

	split_summary[part] = {
	"subjects": sorted(part_df["subject_id"].unique()),
	"total_windows": total_w,
	"supervised_windows": sup_w,
	"label_0_non_seizure": c0,
	"label_1_seizure": c1,
	"label_minus_1_ambiguous": c_amb,
	"seizure_percentage": round(c1 / sup_w * 100, 4) if sup_w > 0 else 0.0
	}

	report["split_verification"]["partitions"] = split_summary
	print("\nPartition Breakdown:", flush=True)
	for part, stats in split_summary.items():
	print(f" {part.upper()}: {stats['total_windows']} windows (Sup: {stats['supervised_windows']}, Non-seizure: {stats['label_0_non_seizure']}, Seizure: {stats['label_1_seizure']} [{stats['seizure_percentage']:.2f}%], Ambiguous: {stats['label_minus_1_ambiguous']})", flush=True)

	# Generate split files and split manifest
	split_manifest = {
	"dataset_name": manifest_data.get("dataset_name"),
	"total_windows": num_windows,
	"partitions": split_summary,
	"established_assignments": {
	"train": TRAIN_SUBJECTS,
	"val": VAL_SUBJECTS,
	"test": TEST_SUBJECTS
	}
	}
	with open(SPLIT_MANIFEST_PATH, "w") as f:
	json.dump(split_manifest, f, indent=2)
	print(f"Split manifest saved to {SPLIT_MANIFEST_PATH}", flush=True)

	# Save individual partition CSVs for fast lookup
	for part in ["train", "val", "test"]:
	part_csv = os.path.join(DATA_DIR, f"{part}_split.csv")
	part_df = df[df["partition"] == part].copy()
	part_df.to_csv(part_csv, index=False)
	print(f"Saved {part} split metadata to {part_csv}", flush=True)

	# Final verdict
	report["passed"] = (len(report["errors"]) == 0)
	with open(REPORT_PATH, "w") as f:
	json.dump(report, f, indent=2)
	print(f"\nVerification report written to {REPORT_PATH}", flush=True)
	print(f"VERIFICATION STATUS: {'PASSED' if report['passed'] else 'FAILED'}", flush=True)

	return report


	if __name__ == "__main__":
	report = verify_dataset()
	if not report["passed"]:
	print("VERIFICATION FAILED with errors:", flush=True)
	for e in report["errors"]:
	print(f" - {e}", flush=True)
	sys.exit(1)
	else:
	print("DATASET VERIFICATION COMPLETE AND VALIDATED!", flush=True)