| """Core data loading and analysis for the leveling-effect study (NS372 and delta22). No plotting. |
| |
| Reads the two datasets, decodes the stored shieldings, and runs the per-nucleus correlation / PCA |
| analysis. The figure notebook (built by build_nb_leveling.py) and test_leveling.py import from here. |
| |
| The claim tested, on each dataset separately (never pooled): once a linear scaling is allowed for, |
| any two shielding methods are near-perfectly correlated, PC1 carries almost all between-method |
| variance, and the methods fall on a 1D curve in (PC1, PC2). |
| """ |
| import numpy as np |
| import pandas as pd |
| import h5py |
| import openpyxl |
| from sklearn.decomposition import PCA |
| from sklearn.preprocessing import StandardScaler |
|
|
| from fixed_point import decode_fixed_point as _decode_fixed_point |
| from stats import rmse as _rmse |
|
|
| |
| FAMILY_ORDER = ['LDA', 'GGA', 'mGGA', 'GH', 'RSH', 'LH', 'DH', 'WFT', 'ref'] |
|
|
| def sort_by_family(M, methods, families): |
| """Stable-sort method columns of M into FAMILY_ORDER (reference family last).""" |
| rank = {f: i for i, f in enumerate(FAMILY_ORDER)} |
| idx = sorted(range(len(methods)), key=lambda i: (rank.get(families[i], 99), i)) |
| return M[:, idx], [methods[i] for i in idx], [families[i] for i in idx] |
|
|
| def log_r(R): |
| """-log10(1-|r|): 1 -> r=0.9, 2 -> 0.99, 3 -> 0.999, ...""" |
| return -np.log10(np.clip(1 - np.abs(R), 1e-6, None)) |
|
|
| def scaled_rmse(method, ref): |
| """RMSE of residuals after the per-method linear fit method ~ a*ref + b.""" |
| a, b = np.polyfit(ref, method, 1) |
| return _rmse(method, a * ref + b) |
|
|
| def _parabola_fit(xn, yn, n_angles=720): |
| |
| best = (0.0, -np.inf, None) |
| for th in np.linspace(0, np.pi, n_angles): |
| c, s = np.cos(th), np.sin(th) |
| xr, yr = c * xn + s * yn, -s * xn + c * yn |
| A = np.column_stack([np.ones_like(xr), xr, xr ** 2]) |
| coef, *_ = np.linalg.lstsq(A, yr, rcond=None) |
| ss_res = ((yr - A @ coef) ** 2).sum() |
| ss_tot = ((yr - yr.mean()) ** 2).sum() |
| r2 = 1 - ss_res / ss_tot if ss_tot > 0 else 0.0 |
| if r2 > best[1]: |
| best = (th, r2, coef) |
| return best |
|
|
| def analyze_nucleus(M, methods, ref=None): |
| """Correlation, PCA, and parabola-fit diagnostics for one nucleus. |
| |
| M is an (n_observations, n_methods) shielding matrix; ref (optional) is the reference method's |
| values, used to compute each method's scaled RMSE against it. |
| """ |
| R = np.corrcoef(M, rowvar=False) |
| iu = np.triu_indices_from(R, k=1) |
| rv = R[iu] |
| Z = StandardScaler().fit_transform(M) |
| pca = PCA().fit(Z) |
| ev = pca.explained_variance_ratio_ |
| pc1, pc2 = pca.components_[0].copy(), pca.components_[1].copy() |
| if pc1.mean() < 0: |
| pc1 = -pc1 |
| mx, my = pc1.mean(), pc2.mean() |
| sx, sy = max(pc1.std(), 1e-12), max(pc2.std(), 1e-12) |
| xn, yn = (pc1 - mx) / sx, (pc2 - my) / sy |
| th, pr2, coef = _parabola_fit(xn, yn) |
| c, s = np.cos(th), np.sin(th) |
| par = dict(theta=th, coef=coef, r2=pr2, mx=mx, sx=sx, my=my, sy=sy, |
| xr=c * xn + s * yn) |
| out = dict(R=R, methods=list(methods), pc1=pc1, pc2=pc2, ev=ev, par=par, |
| r_min=float(rv.min()), r_med=float(np.median(rv)), |
| r_max=float(rv.max())) |
| if ref is not None: |
| out['scaled_rmse'] = {m: scaled_rmse(M[:, i], ref) |
| for i, m in enumerate(methods)} |
| return out |
|
|
| def parabola_curve(par, npts=300): |
| """Sample points along the fitted parabola (from analyze_nucleus's `par`) in PC1/PC2 space.""" |
| th, coef = par['theta'], par['coef'] |
| c, s = np.cos(th), np.sin(th) |
| xr = par['xr'] |
| vertex = -coef[1] / (2 * coef[2]) if coef[2] != 0 else float(xr.mean()) |
| lo, hi = min(xr.min(), vertex), max(xr.max(), vertex) |
| pad = 0.15 * (hi - lo + 1e-9) |
| t = np.linspace(lo - pad, hi + pad, npts) |
| yr = coef[0] + coef[1] * t + coef[2] * t ** 2 |
| xn, yn = c * t - s * yr, s * t + c * yr |
| return xn * par['sx'] + par['mx'], yn * par['sy'] + par['my'] |
|
|
| def summarize(data, res): |
| """Per-nucleus leveling diagnostics (analyze_nucleus results) as one summary DataFrame.""" |
| import pandas as pd |
| rows = [] |
| for nuc, r in res.items(): |
| rows.append(dict(nucleus=nuc, |
| n_obs=data[nuc]['M'].shape[0], |
| n_methods=data[nuc]['M'].shape[1], |
| r_min=r['r_min'], r_median=r['r_med'], |
| PC1_pct=r['ev'][0] * 100, PC2_pct=r['ev'][1] * 100, |
| PC3plus_pct=(1 - r['ev'][0] - r['ev'][1]) * 100, |
| parabola_R2=r['par']['r2'])) |
| return pd.DataFrame(rows) |
|
|
| def dataset_logr_max(res): |
| """Largest -log10(1-|r|) over every nucleus of a dataset (lower triangle of each R).""" |
| vm = 0.0 |
| for r in res.values(): |
| nm = len(r['methods']); m = nm - 1 |
| sub = r['R'][1:, :-1] |
| hide = np.triu(np.ones((m, m), dtype=bool), k=1) |
| vm = max(vm, float(np.nanmax(np.where(hide, np.nan, log_r(sub))))) |
| return float(np.ceil(vm * 10) / 10) |
|
|
|
|
| |
| |
| |
| |
| |
| |
| METHODS = [ |
| ('SVWN', None, 0, 'LDA', 'SVWN'), |
| ('BLYP', None, 0, 'GGA', 'BLYP'), ('BP86', None, 0, 'GGA', 'BP86'), |
| ('PBE', None, 0, 'GGA', 'PBE'), ('B97D', None, 0, 'GGA', 'B97D'), |
| ('HCTH', None, 0, 'GGA', 'HCTH'), |
| ('KT1', None, 0, 'GGA', 'KT1'), ('KT2', None, 0, 'GGA', 'KT2'), ('KT3', None, 0, 'GGA', 'KT3'), |
| ('TPSS', 'τC', 0, 'mGGA', 'TPSS/τC'), ('τ-HCTH', 'τC', 0, 'mGGA', 'τ-HCTH/τC'), |
| ('M06-L', 'τC', 0, 'mGGA', 'M06-L/τC'), ('VSXC', 'τC', 0, 'mGGA', 'VSXC/τC'), |
| ('MN15-L', 'τC', 0, 'mGGA', 'MN15-L/τC'), ('B97M-V', 'τC', 0, 'mGGA', 'B97M-V/τC'), |
| ('SCAN', 'τC', 0, 'mGGA', 'SCAN/τC'), ('rSCAN', 'τC', 0, 'mGGA', 'rSCAN/τC'), |
| ('r2SCAN', 'τC', 0, 'mGGA', 'r2SCAN/τC'), |
| ('TPSSh', 'τC', 10, 'GH', 'TPSSh/τC'), ('B3LYP', None, 20, 'GH', 'B3LYP'), |
| ('B97-2', None, 21, 'GH', 'B97-2'), ('PBE0', None, 25, 'GH', 'PBE0'), |
| ('M06', 'τC', 27, 'GH', 'M06/τC'), ('PW6B95', 'τC', 28, 'GH', 'PW6B95/τC'), |
| ('MN15', 'τC', 44, 'GH', 'MN15/τC'), ('BHLYP', None, 50, 'GH', 'BHLYP'), |
| ('M06-2X', 'τC', 54, 'GH', 'M06-2X/τC'), |
| ('CAM-B3LYP', None, 65, 'RSH', 'CAM-B3LYP'), |
| ('ωB97X-D', None, 100, 'RSH', 'ωB97X-D'), ('ωB97X-V', None, 100, 'RSH', 'ωB97X-V'), |
| ('ωB97M-V', 'τC', 100, 'RSH', 'ωB97M-V/τC'), |
| ('LH07s-SVWN', None, 22, 'LH', 'LH07s-SVWN'), |
| ('MPSTS', 'τD', 30, 'LH', 'mPSTS/τD'), |
| ('LHJ14', 'τD', 10, 'LH', 'LHJ14/τD'), |
| ('LH07t-SVWN', 'τD', 48, 'LH', 'LH07t-SVWN/τD'), |
| ('LH12ct-SsirPW92', 'τD', 65, 'LH', 'LH12ct-SsirPW92/τD'), |
| ('LH12ct-SsifPW92', 'τD', 71, 'LH', 'LH12ct-SsifPW92/τD'), |
| ('LH14t-calPBE', 'τD', 50, 'LH', 'LH14t-calPBE/τD'), |
| ('LH20t', 'τD', 71, 'LH', 'LH20t/τD'), |
| ('B2PLYP', None, 53, 'DH', 'B2PLYP'), ('B2GP-PLYP', None, 65, 'DH', 'B2GP-PLYP'), |
| ('DSD-PBEP86', None, 70, 'DH', 'DSD-PBEP86'), |
| ('Hartree-Fock', None, 100, 'WFT', 'HF'), ('MP2', None, 100, 'WFT', 'MP2'), |
| ] |
| |
| NUCLEI = [ |
| ('1H', 'S6 - 1H Shieldings', None, ()), |
| ('11B', 'S7 - 11B Shieldings', None, ('BH',)), |
| ('13C', 'S8 - 13C Shieldings', None, ()), |
| ('15N', 'S9 - 15N Shieldings', None, ()), |
| ('17O', 'S10 - 17O Shieldings', None, ('O3',)), |
| ('19F', 'S11 - 19F Shieldings', None, ('F3-',)), |
| ('31P', 'S12 - 31P - 33S Shieldings', '31P', ()), |
| ('33S', 'S12 - 31P - 33S Shieldings', '33S', ()), |
| ] |
|
|
| def read_nucleus(wb, sheet, nuc_filter=None, exclude=()): |
| """Read the CCSD(T) reference plus each method's shielding vector from one Kaupp SI sheet.""" |
| ws = wb[sheet] |
| rows = []; last_nuc = None |
| for r in range(4, ws.max_row + 1): |
| v = ws.cell(r, 1).value |
| if v: |
| last_nuc = v |
| if isinstance(ws.cell(r, 4).value, (int, float)): |
| mol = str(ws.cell(r, 2).value or '').strip() |
| if mol in exclude: |
| continue |
| if nuc_filter is None or last_nuc == nuc_filter: |
| rows.append(r) |
| ref = np.array([ws.cell(r, 4).value for r in rows], dtype=float) |
| last_f = None; cols = [] |
| for c in range(7, ws.max_column + 1): |
| h = ws.cell(1, c).value |
| if h is not None: |
| last_f = h |
| cols.append((c, last_f, ws.cell(2, c).value)) |
| data = {} |
| for c, f, sub in cols: |
| if f is None: |
| continue |
| vals = []; ok = True |
| for r in rows: |
| v = ws.cell(r, c).value |
| if not isinstance(v, (int, float)): |
| ok = False; break |
| vals.append(v) |
| if not ok or (f, sub) in data: |
| continue |
| data[(f, sub)] = np.array(vals, dtype=float) |
| return ref, data |
|
|
| def load_ns372(path): |
| """Load the NS372 (Kaupp) shielding matrix, sorted by family, for each of the 8 nuclei.""" |
| wb = openpyxl.load_workbook(path, data_only=True) |
| out = {} |
| for nuc, sheet, filt, excl in NUCLEI: |
| ref, data = read_nucleus(wb, sheet, filt, excl) |
| used = [m for m in METHODS if (m[0], m[1]) in data] |
| cols = [data[(m[0], m[1])] for m in used] |
| labels = [m[4] for m in used] + ['CCSD(T)'] |
| fams = [m[3] for m in used] + ['ref'] |
| M = np.column_stack(cols + [ref]) |
| M, labels, fams = sort_by_family(M, labels, fams) |
| out[nuc] = dict(M=M, methods=labels, ref=ref, families=fams) |
| return out |
|
|
|
|
| |
| |
| |
| |
| |
| |
| def load_ns372_1h_convergence(path): |
| """1H CCSD(T) reference plus each benchmarked functional's shieldings, as columns. |
| |
| Tau policy differs from the SI S3 leveling analysis: Figure 2D uses the tauD kinetic-energy-density |
| treatment for every tau-dependent functional, whereas load_ns372/METHODS uses tauC for meta-GGAs. |
| So this selects tauD when present, otherwise the functional's single column, keyed by raw |
| functional name ('Hartree-Fock' renamed 'HF' for the labels). |
| """ |
| wb = openpyxl.load_workbook(path, data_only=True) |
| ref, data = read_nucleus(wb, 'S6 - 1H Shieldings') |
| names = {m[0] for m in METHODS} |
| variants = {} |
| for (func, tau), vals in data.items(): |
| if func in names: |
| variants.setdefault(func, {})[tau] = vals |
| cols = {'CCSD(T)': ref} |
| for func, by_tau in variants.items(): |
| vals = by_tau['τD'] if 'τD' in by_tau else next(iter(by_tau.values())) |
| cols['HF' if func == 'Hartree-Fock' else func] = vals |
| return pd.DataFrame(cols) |
|
|
| def convergence_statistics(cc_df, reference='CCSD(T)'): |
| """Per method: the slope of method ~ a*CCSD(T) + b, and the scaled RMSE that Figure 2D plots. |
| |
| The prediction is rescaled back into reference units, y_scaled = (method - b) / a, and compared |
| to CCSD(T); this equals the residual RMSE divided by the slope, so it differs from scaled_rmse, |
| which the S3 bar plot leaves in the method's own units. These are the x/y coordinates of the |
| Figure 2D points. |
| """ |
| ref = cc_df[reference].to_numpy(float) |
| rows = [] |
| for method in cc_df.columns: |
| if method == reference: |
| continue |
| y = cc_df[method].to_numpy(float) |
| slope, intercept = np.polyfit(ref, y, 1) |
| y_scaled = (y - intercept) / slope |
| rows.append({'Method': method, 'Slope': slope, |
| 'RMSE (scaled)': _rmse(ref, y_scaled)}) |
| return pd.DataFrame(rows) |
|
|
|
|
| |
| |
| |
| DELTA22_REF = 'DSD-PBEP86' |
|
|
| def load_delta22(path): |
| """Load the delta22 shielding matrix, sorted by family, for 1H and 13C sites.""" |
| with h5py.File(path, 'r') as f: |
| mn = [m.decode() for m in f['conventional_nmr_method_names'][:]] |
| order = ['hf', 'blyp_d3bj', 'bp86_d3bj', 'b97d3_d3bj', 'tpsstpss_d3bj', |
| 'b3lyp_d3bj', 'pbe0_d3bj', 'm062x_d3', 'wb97xd', 'wp04', 'wc04', |
| 'B2PLYP', 'mPW2PLYP', 'B2GP_PLYP', 'dsd_pbep86', 'revdsd_pbep86', |
| 'mp2', 'dlpno_mp2'] |
| disp = {'hf': 'HF', 'blyp_d3bj': 'BLYP-D3', 'bp86_d3bj': 'BP86-D3', |
| 'b97d3_d3bj': 'B97D3', 'tpsstpss_d3bj': 'TPSS-D3', 'b3lyp_d3bj': 'B3LYP-D3', |
| 'pbe0_d3bj': 'PBE0-D3', 'm062x_d3': 'M06-2X-D3', 'wb97xd': 'ωB97X-D', |
| 'wp04': 'WP04', 'wc04': 'WC04', 'B2PLYP': 'B2PLYP', 'mPW2PLYP': 'mPW2PLYP', |
| 'B2GP_PLYP': 'B2GP-PLYP', 'dsd_pbep86': 'DSD-PBEP86', |
| 'revdsd_pbep86': 'revDSD-PBEP86', 'mp2': 'MP2', 'dlpno_mp2': 'DLPNO-MP2'} |
| fam = {'hf': 'WFT', 'blyp_d3bj': 'GGA', 'bp86_d3bj': 'GGA', 'b97d3_d3bj': 'GGA', |
| 'tpsstpss_d3bj': 'mGGA', 'b3lyp_d3bj': 'GH', 'pbe0_d3bj': 'GH', |
| 'm062x_d3': 'GH', 'wb97xd': 'RSH', 'wp04': 'GH', 'wc04': 'GH', |
| 'B2PLYP': 'DH', 'mPW2PLYP': 'DH', 'B2GP_PLYP': 'DH', 'dsd_pbep86': 'DH', |
| 'revdsd_pbep86': 'DH', 'mp2': 'WFT', 'dlpno_mp2': 'WFT'} |
| gas = {} |
| for i, m in enumerate(mn): |
| func, basis, smodel, _ = m.split(',') |
| if smodel == 'gas': |
| gas.setdefault(func, {})[basis] = i |
| sel, labels, families = [], [], [] |
| for func in order: |
| b = gas[func] |
| big = 'pcSseg3' if 'pcSseg3' in b else ('pcSseg2' if 'pcSseg2' in b else 'pcSseg1') |
| sel.append(b[big]); labels.append(disp[func]); families.append(fam[func]) |
| GEOM = 1 |
| rows = {1: [], 6: []} |
| for s in f['solutes']: |
| an = f['solutes'][s]['atomic_numbers'][:] |
| cs = _decode_fixed_point( |
| f['solutes'][s]['stationary_and_pcm']['conventional_shieldings'][:, GEOM, :]) |
| for ai, z in enumerate(an): |
| if z in (1, 6): |
| rows[int(z)].append(cs[sel, ai]) |
| out = {} |
| for nuc, z in (('1H', 1), ('13C', 6)): |
| M = np.array(rows[z]) |
| M, labs, fams = sort_by_family(M, list(labels), list(families)) |
| |
| |
| ref_idx = labs.index(DELTA22_REF) |
| out[nuc] = dict(M=M, methods=labs, ref=M[:, ref_idx], families=fams) |
| return out |
|
|
