"""Ingestion: load a spreadsheet as-is and infer a campaign schema.
DENIA's first design commitment is that the user's file is never reformatted.
The table is loaded in its native shape and a :class:`~denia.schema.CampaignSchema`
is inferred on top of it - by an LLM agent when available
(:class:`denia.agents.SchemaAgent`), or by the deterministic heuristics below,
which also serve as the agent's verifiable baseline.
"""
from __future__ import annotations
import re
from pathlib import Path
from typing import List, Optional, Union
import numpy as np
import pandas as pd
from .schema import CampaignSchema, ColumnRole, ColumnSpec
# Column-name patterns that typically denote targets/objectives in
# experimental tables (yields, figures of merit), vs auxiliary measurements.
_TARGET_PATTERNS = [r"^y\(", r"yield", r"figure of merit", r"fom", r"efficiency"]
_AUX_PATTERNS = [r"^x\(", r"^s\(", r"conversion", r"selectivity"]
_META_PATTERNS = [r"^nr", r"^id$", r"reference", r"publication", r"author", r"doi", r"year", r"comment"]
_UNIT_IN_NAME = re.compile(r",\s*([^,]+)$") # e.g. "Temperature, K" -> unit "K"
[docs]
def load_table(path: Union[str, Path], sheet: Optional[str] = None) -> pd.DataFrame:
"""Load an .xlsx/.csv experimental table without altering its structure.
For Excel files with several sheets, the largest sheet is assumed to be
the data sheet unless ``sheet`` is given (reference lists and column
descriptions typically live in the smaller sheets).
"""
path = Path(path)
if path.suffix.lower() in {".csv", ".tsv"}:
sep = "\t" if path.suffix.lower() == ".tsv" else ","
return pd.read_csv(path, sep=sep)
xl = pd.ExcelFile(path)
if sheet is None:
sizes = {s: pd.read_excel(xl, sheet_name=s, nrows=0).shape[1] * _sheet_len(xl, s) for s in xl.sheet_names}
sheet = max(sizes, key=sizes.get)
return pd.read_excel(xl, sheet_name=sheet)
def _sheet_len(xl: pd.ExcelFile, sheet: str) -> int:
try:
return len(pd.read_excel(xl, sheet_name=sheet, usecols=[0]))
except Exception:
return 0
def _matches(name: str, patterns: List[str]) -> bool:
low = name.lower().strip()
return any(re.search(p, low) for p in patterns)
def _extract_unit(name: str) -> Optional[str]:
m = _UNIT_IN_NAME.search(name)
return m.group(1).strip() if m else None
[docs]
def infer_schema(
df: pd.DataFrame,
target: Optional[Union[str, List[str]]] = None,
) -> CampaignSchema:
"""Deterministic, dependency-free schema inference.
Heuristics encoded here:
* columns whose name matches metadata patterns (ids, references) -> METADATA
* user-specified ``target`` (or yield-like names) -> TARGET
* conversion/selectivity-like names -> AUXILIARY
* object-dtype columns -> CATEGORY_INPUT
* numeric columns named like ``"<Category> mol%"`` and sharing a prefix with
a category column -> FRACTION_INPUT paired to it
* remaining numeric columns -> NUMERIC_INPUT
"""
targets = [target] if isinstance(target, str) else list(target or [])
cols: List[ColumnSpec] = []
names = list(df.columns)
for name in names:
unit = _extract_unit(str(name))
s = df[name]
if name in targets or (not targets and _matches(str(name), _TARGET_PATTERNS)):
cols.append(ColumnSpec(name, ColumnRole.TARGET, unit=unit))
elif _matches(str(name), _META_PATTERNS):
cols.append(ColumnSpec(name, ColumnRole.METADATA))
elif _matches(str(name), _AUX_PATTERNS):
cols.append(ColumnSpec(name, ColumnRole.AUXILIARY, unit=unit))
elif pd.api.types.is_object_dtype(s) or pd.api.types.is_string_dtype(s):
cols.append(ColumnSpec(name, ColumnRole.CATEGORY_INPUT))
elif pd.api.types.is_numeric_dtype(s):
partner = _fraction_partner(str(name), names)
if partner:
cols.append(ColumnSpec(name, ColumnRole.FRACTION_INPUT, unit="mol%", partner=partner))
else:
cols.append(ColumnSpec(name, ColumnRole.NUMERIC_INPUT, unit=unit))
else:
cols.append(ColumnSpec(name, ColumnRole.IGNORE))
schema = CampaignSchema(columns=cols, provenance="heuristic")
if not schema.targets:
raise ValueError(
"Could not identify a target column. Pass target='...' explicitly, "
"or use SchemaAgent for agentic inference."
)
return schema
def _fraction_partner(name: str, all_names: List[str]) -> Optional[str]:
"""'Cation 1 mol%' -> 'Cation 1' if that column exists."""
low = name.lower()
for suffix in (" mol%", " wt%", " %", " fraction"):
if low.endswith(suffix):
stem = name[: -len(suffix)].strip()
for cand in all_names:
if str(cand).strip().lower() == stem.lower():
return str(cand)
return None