"""Build the three clean processed route files covering all eight datasets.
Two sources feed the routes:
1. The five xpectrass reference datasets, loaded raw from the ``xpectrass`` package loaders
(the published, Zenodo-archived dataset) and preprocessed per dataset on the native grid
(denoise -> baseline -> atmospheric-interpolate -> normalize), then interpolated onto the
common 680-3100 cm^-1 grid with ``combine_datasets``.
2. The three external augmentation datasets (``openspecy``, ``simple``, ``awi2024``), whose
already-normalized common-grid spectra are preserved in the package catalog DB
(``xpectra/data/xpectra_catalog.sqlite``) and appended to the combined frame.
Savitzky-Golay derivatives are computed ONCE, on the full combined frame, after every dataset
sits on the identical grid in the identical (descending) column order. This is deliberate:
the derivative parameters are index-based (``delta`` per point, ``window_length`` in points),
so taking derivatives on native grids or on differently ordered columns would give
per-dataset scale/sign artifacts. Combining first makes consistency automatic.
The result is exactly three files (norm / deriv1 / deriv2), each holding all eight datasets
distinguished by the ``study`` column; downstream code selects the labeled core studies or
the external pools via ``PreprocessConfig.labeled_studies`` / ``external_studies``.
By default duplicates are removed per dataset (both ambiguous duplicate ``sample_id`` values and
exact-duplicate spectra, keep-first), so the matrices are clean at source. Identical spectra
appearing in two *different* sources (e.g. Primpke-2018 spectra present in both Open Specy and
siMPle) are legitimately kept — uniqueness is a within-study property. This module is the
single implementation shared by ``scripts/process_raw_data.py`` and the step-0 notebook.
"""
from __future__ import annotations
from pathlib import Path
import numpy as np
import pandas as pd
from xpectrass import (
combine_datasets,
load_frond_2021,
load_jung_2018,
load_kedzierski_2019,
load_kedzierski_2019_u,
load_villegas_camacho_2024_c4,
)
from .._paths import find_project_root
from .config import PreprocessConfig
from .core import derivative_route_dataframe, preprocess_raw_dataframe
# Working data belongs to the user's project directory (walked up from cwd); only the
# catalog DB ships inside the package and is addressed relative to the install location.
PROCESSED_DIR = find_project_root() / "processed_data"
CATALOG_PATH = Path(__file__).resolve().parents[1] / "data" / "xpectra_catalog.sqlite"
META_COLUMNS = ["study", "sample_id", "type", "environmental", "resolution"]
# External augmentation datasets: already-normalized common-grid spectra preserved in the
# catalog DB (not in xpectrass). They ride along in the combined route files with
# study = dataset_id; the finer per-row source_study split stays available in the DB.
EXTERNAL_DATASETS = ("openspecy", "simple", "awi2024")
# The core reference datasets come straight from the xpectrass package loaders; order matches
# the original build. No spectra are stored locally.
LOADERS = {
"jung_2018": load_jung_2018,
"kedzierski_2019": load_kedzierski_2019,
"kedzierski_2019_u": load_kedzierski_2019_u,
"frond_2021": load_frond_2021,
"villegas_camacho_2024_c4": load_villegas_camacho_2024_c4,
}
BUILD_ORDER = list(LOADERS)
# Per-dataset preprocessing overrides. The Kedzierski spectra are absorbance but look like transmittance
# and would trip xpectrass's absorbance auto-detection, so they are forced into absorbance mode
# (kedzierski_2019 additionally rescaled by 100); jung/frond/villegas use the default
# auto-detect + convert path.
PREPROCESS_OVERRIDES: dict[str, dict] = {
"kedzierski_2019": dict(force_absorbance=True, absorbance_scale_factor=100.0),
"kedzierski_2019_u": dict(force_absorbance=True, absorbance_scale_factor=None),
}
ROUTE_FILENAMES = {
"norm": "combined_norm_data.csv.xz",
"deriv1": "combined_norm_deriv1_data.csv.xz",
"deriv2": "combined_norm_deriv2_data.csv.xz",
}
[docs]
def load_dataset_frame(dataset_id: str, dedupe: bool = True) -> pd.DataFrame:
"""Load one dataset's raw spectra from the xpectrass loader.
When ``dedupe`` is set, drop both ambiguous duplicate ``sample_id`` values and exact-duplicate
spectra (keep first), so the output has unique ids and unique spectra.
"""
if dataset_id not in LOADERS:
raise KeyError(f"dataset {dataset_id!r} not in LOADERS: {list(LOADERS)}")
df = LOADERS[dataset_id]().reset_index(drop=True)
spec_cols = [c for c in df.columns if c not in META_COLUMNS]
# Replace non-finite values with 0 (matches the historical raw handling).
df.loc[:, spec_cols] = np.nan_to_num(
df[spec_cols].to_numpy(dtype=float), nan=0.0, posinf=0.0, neginf=0.0)
if dedupe:
dup_sid = df["sample_id"].duplicated(keep="first").to_numpy()
dup_spec = df[spec_cols].round(6).duplicated(keep="first").to_numpy()
df = df.loc[~(dup_sid | dup_spec)].reset_index(drop=True)
return df
[docs]
def read_frames(dedupe: bool = True, only: list[str] | None = None,
verbose: bool = True) -> dict[str, pd.DataFrame]:
"""Load every study's raw spectra from xpectrass (deduplicated by default)."""
order = only or BUILD_ORDER
frames = {}
for name in order:
df = load_dataset_frame(name, dedupe=dedupe)
frames[name] = df
if verbose:
print(f" loaded {name:28s} {len(df):5d} spectra"
f"{' [deduplicated]' if dedupe else ''}")
return frames
[docs]
def read_external_frames(catalog_path: Path = CATALOG_PATH, dedupe: bool = True,
only: list[str] | None = None,
verbose: bool = True) -> dict[str, pd.DataFrame]:
"""Load the external augmentation datasets from the catalog DB.
The catalog stores them already normalized on the common wavenumber grid, so they need
no preprocessing — only appending to the combined frame. ``study`` is set to the
dataset_id so route files can be filtered per source.
"""
import json
import sqlite3
catalog_path = Path(catalog_path)
if not catalog_path.exists():
raise FileNotFoundError(
f"Catalog DB not found at {catalog_path}. The external datasets "
f"({', '.join(EXTERNAL_DATASETS)}) live there; restore it before building.")
frames: dict[str, pd.DataFrame] = {}
con = sqlite3.connect(catalog_path)
try:
for did in (only or EXTERNAL_DATASETS):
cols = json.loads(con.execute(
"SELECT native_columns FROM datasets WHERE dataset_id=?", (did,)).fetchone()[0])
recs = con.execute(
"SELECT sample_id, polymer_type, environmental, resolution, intensities "
"FROM spectra WHERE dataset_id=? ORDER BY spectrum_id", (did,)).fetchall()
meta = pd.DataFrame([{
"study": did, "sample_id": sid, "type": t,
"environmental": "Y" if e else "N", "resolution": r}
for sid, t, e, r, _ in recs])
spec = pd.DataFrame(
np.nan_to_num(np.vstack([np.frombuffer(rec[-1], np.float64) for rec in recs]),
nan=0.0, posinf=0.0, neginf=0.0),
columns=cols)
df = pd.concat([meta, spec], axis=1)
if dedupe:
dup_sid = df["sample_id"].duplicated(keep="first").to_numpy()
dup_spec = spec.round(6).duplicated(keep="first").to_numpy()
df = df.loc[~(dup_sid | dup_spec)].reset_index(drop=True)
frames[did] = df
if verbose:
print(f" loaded {did:28s} {len(df):5d} spectra [external, pre-normalized]"
f"{' [deduplicated]' if dedupe else ''}")
finally:
con.close()
return frames
def _append_externals(combined: pd.DataFrame, externals: dict[str, pd.DataFrame],
config: PreprocessConfig) -> pd.DataFrame:
"""Row-append the pre-normalized external frames onto the combined grid.
Columns are matched by wavenumber value (not label formatting) and reordered to the
combined frame's descending order, so the derivatives computed afterwards are
consistent across all datasets — including the sign of the 1st derivative, which
would flip if an ascending-ordered frame were differentiated as stored.
"""
from .core import spectral_columns_sorted
spec_cols, _ = spectral_columns_sorted(combined)
by_wn = {round(float(str(c)), 4): c for c in spec_cols}
parts = [combined]
for name, ext in externals.items():
ext_spec, _ = spectral_columns_sorted(ext)
rename = {}
for c in ext_spec:
key = round(float(str(c)), 4)
if key not in by_wn:
raise ValueError(
f"External dataset '{name}' has wavenumber {key} not present in the "
f"combined grid; it must be normalized on the common grid.")
rename[c] = by_wn[key]
if len(rename) != len(spec_cols):
raise ValueError(
f"External dataset '{name}' covers {len(rename)} of {len(spec_cols)} "
f"combined grid points; grids must match exactly.")
ext = ext.rename(columns=rename)
parts.append(ext[list(combined.columns)])
return pd.concat(parts, ignore_index=True)
[docs]
def build_routes(frames: dict[str, pd.DataFrame], config: PreprocessConfig | None = None,
externals: dict[str, pd.DataFrame] | None = None,
verbose: bool = True) -> dict[str, pd.DataFrame]:
"""Per-dataset preprocess -> combine to common grid -> append externals -> derivatives."""
config = config or PreprocessConfig()
study_order = list(frames)
normalized = []
for name in study_order:
kw = PREPROCESS_OVERRIDES.get(name, {})
if verbose:
tag = " [forced absorbance]" if kw else ""
print(f" [preprocess] {name} ({len(frames[name])} spectra){tag} ...", flush=True)
norm = preprocess_raw_dataframe(frames[name], config=config, **kw)
if "study" in norm.columns:
norm = norm.drop(columns=["study"])
normalized.append(norm)
if verbose:
print(" [combine] interpolating to common grid ...", flush=True)
combined_norm, _ = combine_datasets(
datasets=normalized,
wn_min=config.wn_min, wn_max=config.wn_max, resolution=config.resolution,
descending=config.descending, method=config.combine_method,
label_column=config.label_column, exclude_columns=None,
add_study_column=["sample_id", "environmental", "resolution"],
study_names=study_order, show_progress=False, n_jobs=1, data_mode="normalized",
)
if externals:
if verbose:
n_ext = sum(len(v) for v in externals.values())
print(f" [append] {len(externals)} external datasets "
f"({n_ext} pre-normalized spectra) ...", flush=True)
combined_norm = _append_externals(combined_norm, externals, config)
if verbose:
print(" [derivatives] d1, d2 on the full combined frame ...", flush=True)
d1 = derivative_route_dataframe(combined_norm, order=1, config=config)
d2 = derivative_route_dataframe(combined_norm, order=2, config=config)
return {"norm": combined_norm, "deriv1": d1, "deriv2": d2}
[docs]
def build_processed_routes(dedupe: bool = True, only: list[str] | None = None,
config: PreprocessConfig | None = None,
include_external: bool = True,
verbose: bool = True) -> dict[str, pd.DataFrame]:
"""End-to-end: xpectrass raw + catalog externals -> the three route DataFrames."""
frames = read_frames(dedupe=dedupe, only=only, verbose=verbose)
externals = read_external_frames(dedupe=dedupe, verbose=verbose) if include_external else None
return build_routes(frames, config=config, externals=externals, verbose=verbose)
[docs]
def write_routes(routes: dict[str, pd.DataFrame], out_dir: Path = PROCESSED_DIR,
verbose: bool = True) -> None:
"""Write the route DataFrames as compressed CSVs."""
out_dir = Path(out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
for route, df in routes.items():
out = out_dir / ROUTE_FILENAMES[route]
df.to_csv(out, compression="xz", index=False)
if verbose:
print(f" wrote {out} {df.shape}")
[docs]
def duplicate_report(routes: dict[str, pd.DataFrame]) -> pd.DataFrame:
"""Per-route counts of rows, unique (study, spectrum) and (study, sample_id) pairs.
Uniqueness is keyed within study: the same spectrum appearing in two different sources
(e.g. a Primpke-2018 spectrum in both Open Specy and siMPle) is a legitimate property of
the sources, not a leak, and is never collapsed. All three counts should match.
"""
rows = []
for route, df in routes.items():
spec = [c for c in df.columns
if str(c).replace(".", "", 1).replace("-", "", 1).isdigit()]
key = pd.DataFrame(np.round(df[spec].to_numpy(float), 6))
key.insert(0, "__study__", df["study"].to_numpy())
u_spec = key.drop_duplicates().shape[0]
u_sid = df[["study", "sample_id"]].drop_duplicates().shape[0]
rows.append({"route": route, "rows": len(df),
"unique_spectra": u_spec, "unique_sample_ids": u_sid})
return pd.DataFrame(rows)