"""(c) Kevin Dunn, 2010-2026. MIT License.
Deterministic reshaping of panel data into the ``descriptive_long`` schema.
Raw panel data usually arrives wide (one column per attribute) or already long.
The front end (or a code sandbox) parses the spreadsheet into rows; this module
performs the reshape *deterministically* and self-checks it, so the bulk
transform never goes through an LLM's tokens (which would risk silent
mis-mapping on real-sized panels).
:func:`reshape_to_long` takes an explicit column mapping (the caller, possibly
an LLM, decides which column is which) and a ``layout`` flag, melts to long when
needed, and verifies a set of round-trip invariants: the grand mean, the mean
per attribute, the mean per panelist, and the count of non-missing cells must be
identical before and after the reshape. A mismatch (for example product and
attribute columns swapped) raises rather than silently corrupting every
downstream statistic.
"""
from __future__ import annotations
from typing import Any, Literal
import numpy as np
import pandas as pd
from process_improve.sensory.validation import DESCRIPTIVE_LONG_COLUMNS
Layout = Literal["long", "wide_by_attribute", "wide_by_product"]
#: Tolerances for the before/after round-trip invariant comparison.
_ABS_TOL = 1e-8
_REL_TOL = 1e-6
#: Canonical row order of the long output (sample-major). Ordering does not
#: affect any analysis, but a stable order keeps the validated content hash
#: independent of the input order.
_CANONICAL_SORT = ["product", "attribute", "panelist_id", "session", "replicate"]
def _label(value: object) -> str:
"""Return a label as the reshape writes it: a string without surrounding whitespace.
The round-trip check keys the before and after means by this form, so a header such
as ``"Salty "`` is compared with the ``"Salty"`` the long table carries.
"""
return str(value).strip()
def _series_mean_map(frame: pd.DataFrame, key: str, value: str) -> dict[str, float]:
"""Mean of ``value`` grouped by ``key``, as a plain ``{label: mean}`` dict."""
grouped = frame.groupby(key, observed=True)[value].mean()
return {_label(k): float(v) for k, v in grouped.items()}
def _compare_maps(before: dict[str, float], after: dict[str, float]) -> float:
"""Return the largest absolute difference between two ``{label: mean}`` maps.
A label on one side only, or a mean missing (NaN) on one side only, is an infinite
difference: the reshape lost or invented that label, or every score it had. A mean
missing on both sides belongs to a label with no scores, carried through unchanged.
No NaN reaches ``max``: its result over NaN depends on the order of its input, and
the order of a set of labels changes with the hash seed, so the round-trip check
used to pass or fail at random.
"""
if before.keys() != after.keys():
return float("inf")
largest = 0.0
for label, mean_before in before.items():
mean_after = after[label]
if np.isnan(mean_before) and np.isnan(mean_after):
continue
difference = abs(mean_before - mean_after)
largest = max(largest, float("inf") if np.isnan(difference) else difference)
return largest
def _refuse_missing_labels(data: pd.DataFrame, columns: dict[str, Any]) -> None:
"""Raise ``ValueError`` when a row has no panelist, product or attribute label.
``columns`` maps each role to the column that carries it. Such a row cannot be
attributed, and the round-trip check cannot see it: pandas 2 turned the missing
label into the text ``"nan"``, a label the long table appeared to invent, but pandas 3
keeps it missing, and grouping drops it from both sides of the comparison.
"""
for role, col in columns.items():
missing = data[col].isna()
if missing.any():
rows = data.index[missing].tolist()
raise ValueError(
f"{len(rows)} row(s) have no {role} in column {col!r} (index {rows[:5]}); every row needs one."
)
def _refuse_labels_alike_but_for_spaces(long_df: pd.DataFrame) -> None:
"""Raise ``ValueError`` when two labels differ only in surrounding whitespace.
The reshape strips each label (:func:`_label`), which would merge ``"P1"`` and ``"P1 "``
into one panelist: silently when their means agree, and otherwise through a round-trip
failure that blames the column mapping (#1003).
"""
for col, role in (("panelist_id", "panelist"), ("product", "product"), ("attribute", "attribute")):
first: dict[str, str] = {}
for label in long_df[col].astype(str).unique():
other = first.setdefault(label.strip(), label)
if other != label:
raise ValueError(
f"The {role} labels {other!r} and {label!r} differ only in surrounding whitespace, so they "
f"would be merged into one {role}. Write them alike if they are one {role}, or apart if not."
)
[docs]
def reshape_to_long( # noqa: C901, PLR0912, PLR0915
data: pd.DataFrame,
*,
layout: Layout,
mapping: dict[str, Any],
) -> tuple[pd.DataFrame, dict[str, Any]]:
"""Reshape panel data into the ``descriptive_long`` schema, with round-trip checks.
Parameters
----------
data : pandas.DataFrame
The parsed panel table (rows already read from the spreadsheet).
layout : {"long", "wide_by_attribute", "wide_by_product"}
``"long"`` passes through (renaming columns and canonicalising);
``"wide_by_attribute"`` melts attribute columns into rows (rows carry
the product); ``"wide_by_product"`` melts product columns into rows
(rows carry the attribute, i.e. one panelist-by-product matrix per
attribute, stacked with an attribute label column).
mapping : dict
Explicit column roles:
* ``panelist_id`` (required): column name.
* ``session`` and ``replicate`` (optional): column names; default to a
constant 1 when absent.
* For ``wide_by_attribute``: ``product`` (required) and ``attributes``
(the list of attribute columns; if omitted, all non-id columns).
* For ``wide_by_product``: ``attribute`` (required, the block-label
column) and ``products`` (the list of product columns; if omitted,
all non-id columns).
* For ``long``: ``product``, ``attribute`` and ``score`` column names.
* ``ignore`` (optional): columns to drop before reshaping (nuisance
columns such as a site or batch code). When the attribute/product
list is omitted, "all remaining columns" excludes these.
Returns
-------
long_df : pandas.DataFrame
Data in the canonical ``descriptive_long`` schema, sample-major sorted.
checks : dict
The round-trip invariants (grand mean, per-attribute and per-panelist
max differences, cell counts) and ``ok``.
Raises
------
ValueError
If required mapping columns are missing, the data is means-only (no
panelist column), or a round-trip invariant fails (which signals a
wrong column mapping).
"""
if layout not in ("long", "wide_by_attribute", "wide_by_product"):
raise ValueError(f"layout must be 'long', 'wide_by_attribute', or 'wide_by_product', got {layout!r}.")
# Drop any nuisance columns up front, so "all remaining columns" (when the
# attribute/product list is omitted) automatically excludes them.
ignore = mapping.get("ignore") or []
if ignore:
data = data.drop(columns=[c for c in ignore if c in data.columns])
panelist_col = mapping.get("panelist_id")
if not panelist_col:
raise ValueError(
"mapping must name a 'panelist_id' column. Means-only data (no panelist column) cannot be "
"used: the Mixed Assessor Model needs panelist-level scores."
)
if panelist_col not in data.columns:
raise ValueError(f"mapping['panelist_id'] = {panelist_col!r} is not a column in the data.")
product_col = mapping.get("product")
session_col = mapping.get("session")
replicate_col = mapping.get("replicate")
# --- Pre-reshape marginal aggregates -------------------------------
if layout == "long":
attribute_col = mapping.get("attribute")
score_col = mapping.get("score")
for role, col in (("product", product_col), ("attribute", attribute_col), ("score", score_col)):
if not col or col not in data.columns:
raise ValueError(f"mapping[{role!r}] = {col!r} is not a column in the data (required for long layout).")
_refuse_missing_labels(data, {"panelist id": panelist_col, "product": product_col, "attribute": attribute_col})
long_df = data.rename(
columns={
panelist_col: "panelist_id",
product_col: "product",
attribute_col: "attribute",
score_col: "score",
**({session_col: "session"} if session_col else {}),
**({replicate_col: "replicate"} if replicate_col else {}),
}
).copy()
long_df["score"] = pd.to_numeric(long_df["score"], errors="coerce").astype("float64")
grand_before = float(np.nanmean(long_df["score"].to_numpy()))
n_cells_before = int(long_df["score"].notna().sum())
attr_mean_before = _series_mean_map(long_df, "attribute", "score")
panelist_mean_before = _series_mean_map(long_df, "panelist_id", "score")
else:
# Both wide layouts melt a block of value columns into one long column.
# wide_by_attribute: rows carry the product, the columns are attributes.
# wide_by_product: rows carry the attribute (a block label), the columns
# are products - i.e. one panelist x product matrix per attribute, stacked.
if layout == "wide_by_attribute":
if not product_col or product_col not in data.columns:
raise ValueError(f"mapping['product'] = {product_col!r} is required for wide_by_attribute layout.")
row_key_col, melted_name, value_key = product_col, "attribute", "attributes"
else: # wide_by_product
row_key_col = mapping.get("attribute")
if not row_key_col or row_key_col not in data.columns:
raise ValueError(f"mapping['attribute'] = {row_key_col!r} is required for wide_by_product layout.")
melted_name, value_key = "product", "products"
row_key_target = "product" if layout == "wide_by_attribute" else "attribute"
_refuse_missing_labels(data, {"panelist id": panelist_col, row_key_target: row_key_col})
id_cols = [c for c in (panelist_col, row_key_col, session_col, replicate_col) if c]
value_cols = mapping.get(value_key)
if value_cols is None:
value_cols = [c for c in data.columns if c not in id_cols]
if not value_cols:
raise ValueError(f"No {melted_name} columns found for {layout} layout.")
missing_vals = [c for c in value_cols if c not in data.columns]
if missing_vals:
raise ValueError(f"These {melted_name} columns are not in the data: {missing_vals}.")
wide = data.copy()
for col in value_cols:
wide[col] = pd.to_numeric(wide[col], errors="coerce").astype("float64")
cell_block = wide[value_cols]
grand_before = float(np.nanmean(cell_block.to_numpy()))
n_cells_before = int(cell_block.notna().to_numpy().sum())
panelist_mean_before = {
_label(pid): float(np.nanmean(grp[value_cols].to_numpy()))
for pid, grp in wide.groupby(panelist_col, observed=True)
}
if layout == "wide_by_attribute":
attr_mean_before = {_label(a): float(np.nanmean(wide[a].to_numpy())) for a in value_cols}
else: # attribute is a row key, so group rows by it
attr_mean_before = {
_label(attr): float(np.nanmean(grp[value_cols].to_numpy()))
for attr, grp in wide.groupby(row_key_col, observed=True)
}
long_df = wide.melt(id_vars=id_cols, value_vars=value_cols, var_name=melted_name, value_name="score").rename(
columns={panelist_col: "panelist_id", row_key_col: row_key_target}
)
if session_col:
long_df = long_df.rename(columns={session_col: "session"})
if replicate_col:
long_df = long_df.rename(columns={replicate_col: "replicate"})
# --- Defaults and dtypes -------------------------------------------
if "session" not in long_df.columns:
long_df["session"] = 1
if "replicate" not in long_df.columns:
long_df["replicate"] = 1
_refuse_labels_alike_but_for_spaces(long_df)
for col in ("panelist_id", "product", "attribute"):
long_df[col] = long_df[col].astype(str).str.strip()
long_df["score"] = pd.to_numeric(long_df["score"], errors="coerce").astype("float64")
# --- Post-reshape marginal aggregates ------------------------------
grand_after = float(np.nanmean(long_df["score"].to_numpy()))
n_cells_after = int(long_df["score"].notna().sum())
attr_mean_after = _series_mean_map(long_df, "attribute", "score")
panelist_mean_after = _series_mean_map(long_df, "panelist_id", "score")
grand_diff = abs(grand_before - grand_after)
attr_diff = _compare_maps(attr_mean_before, attr_mean_after)
panelist_diff = _compare_maps(panelist_mean_before, panelist_mean_after)
tol = _ABS_TOL + _REL_TOL * abs(grand_before)
ok = grand_diff <= tol and attr_diff <= tol and panelist_diff <= tol and n_cells_before == n_cells_after
checks = {
"ok": ok,
"grand_mean_before": grand_before,
"grand_mean_after": grand_after,
"grand_mean_diff": grand_diff,
"per_attribute_max_diff": attr_diff,
"per_panelist_max_diff": panelist_diff,
"n_cells_before": n_cells_before,
"n_cells_after": n_cells_after,
}
if not ok:
raise ValueError(
"Round-trip check failed after reshaping; the column mapping is likely wrong. "
f"grand_mean_diff={grand_diff:.3g}, per_attribute_max_diff={attr_diff:.3g}, "
f"per_panelist_max_diff={panelist_diff:.3g}, "
f"cells before/after={n_cells_before}/{n_cells_after}."
)
long_df = long_df.loc[:, list(DESCRIPTIVE_LONG_COLUMNS)]
long_df = long_df.sort_values(_CANONICAL_SORT, kind="stable").reset_index(drop=True)
return long_df, checks
