"""

ProcessBehavior - Main entry point for process behavior analysis.

This module provides a user-friendly interface with IDE auto-completion for column names

and automatic chart selection driven by the analytical design state (ADS).

Usage:

import processbehavior as pb

# Step 1: formulate() - analyze structure and get recommendations

study = pb.formulate(df, response='Measurement', factors=['Operator'], time='ProductionTime')

# Step 2: execute() - run the chart

result = study.execute()

Use the ProcessBehavior class directly when you want the fluent derived-variable verbs,

which attach before formulating, or `.cols` auto-completion for column references:

pbd = ProcessBehavior(df)

study = pbd.transform('torque', 'log').formulate(response=pbd.cols.torque_log)

Note the instance is bound to `pbd`, not `pb`: `pb` is the conventional alias for the

module itself, and shadowing it makes `pb.formulate(...)` read as the module-level

function when it is an instance method.

"""

from __future__ import annotations

import logging

import warnings

from dataclasses import dataclass, field

from typing import TYPE_CHECKING

import pandas as pd

from .derivations import Derivation

from .derivations import evaluate as _evaluate_derivation

from .derivations import validate as _validate_derivation

from .exceptions import ColumnNotFoundError, ProcessBehaviorWarning, ValidationError

from .formulation_spec import FormulationSpec

from .sds_detector import SDSRegistry, SDSResult

if TYPE_CHECKING:

from .study import Study

logger = logging.getLogger(__name__)

# Threshold for auto-converting object columns to numeric.

# If >= this fraction of non-NA values convert successfully, apply the conversion.

_NUMERIC_CONVERSION_THRESHOLD = 0.8

def _is_text_like(series: pd.Series) -> bool:

"""True when a column can hold a garbage *string* token.

Enumerates the dtypes to include rather than excluding numeric ones: a categorical or

a pandas ``string`` column holds strings and must still be scanned, but neither is

``object`` and a "not numeric" test would misclassify at least one of them.

"""

from pandas.api.types import is_object_dtype, is_string_dtype

dtype = series.dtype

if isinstance(dtype, pd.CategoricalDtype):

return True

return bool(is_object_dtype(series) or is_string_dtype(series))

def _try_clean_numeric_strings(series: pd.Series) -> pd.Series | None:

"""Try to clean formatted numeric strings (currency, thousands sep, etc.).

Returns a numeric Series if cleaning succeeds for >= 80% of non-NA values,

or None if the column doesn't look numeric after cleaning.

Handles: $, EUR, GBP, JPY, unicode currency symbols, thousands commas,

accounting negatives like (1,234.56), percentage signs, and whitespace.

"""

from pandas.api.types import is_datetime64_any_dtype, is_numeric_dtype

# Don't "clean" already-typed columns. Crucially, datetime64 must be guarded:

# pd.to_numeric(datetime64) silently succeeds (int64 nanoseconds), which would

# otherwise replace a date column with huge integers. (Mirrors the datetime

# guard in DataPreparation._detect_and_convert_type.)

if is_numeric_dtype(series) or is_datetime64_any_dtype(series) or isinstance(series.dtype, pd.PeriodDtype):

return None

# Only process object columns with actual data

non_na = series.dropna()

if len(non_na) == 0:

return None

# Fast path: try direct pd.to_numeric first (handles plain numeric strings)

direct = pd.to_numeric(series, errors='coerce')

direct_success = direct.notna().sum()

original_non_na = non_na.shape[0]

if direct_success >= original_non_na * _NUMERIC_CONVERSION_THRESHOLD:

return direct

result = _coerce_formatted_numerics(series)

# Check success rate against non-NA values only

converted_count = result.notna().sum()

if converted_count >= original_non_na * _NUMERIC_CONVERSION_THRESHOLD:

return result

return None

def _coerce_formatted_numerics(series: pd.Series) -> pd.Series:

"""The value transform behind :func:`_try_clean_numeric_strings`, without the

accept/reject decision.

Split out so the decision can be scored against a different population than the one

transformed — see :func:`_clean_numeric_strings_via_uniques`, which transforms the

distinct values but must score frequency-weighted over the full column.

"""

# Convert to string for .str operations (handles mixed float/str columns)

s = series.astype(str)

# Preserve original NAs

original_na_mask = series.isna()

# Also treat the string 'nan' (from astype) as NA

str_nan_mask = s == 'nan'

# Strip whitespace

cleaned = s.str.strip()

# Accounting negatives: (xxx) -> -xxx

cleaned = cleaned.str.replace(r'^\((.*)\)$', r'-\1', regex=True)

# Remove currency symbols and optional adjacent whitespace

# Use literal Unicode chars (€£¥) — raw string \u escapes aren't valid in pyarrow regex

cleaned = cleaned.str.replace(r'[$€£¥]\s*', '', regex=True)

# Remove thousands separators (commas)

cleaned = cleaned.str.replace(',', '', regex=False)

# Remove percentage signs

cleaned = cleaned.str.replace('%', '', regex=False)

# Final strip (catches residual whitespace after symbol removal)

cleaned = cleaned.str.strip()

result = pd.to_numeric(cleaned, errors='coerce')

# Restore original NAs (don't count astype('str') artifacts as conversions)

result[original_na_mask | str_nan_mask] = pd.NA

return result

# Above this share of distinct values, mapping through the uniques stops paying for

# itself and the direct path is used instead. Measured at 1M rows: label-like columns

# win at every cardinality (36x at 10 uniques, still 1.4x when every value is distinct),

# but numeric-like strings — where the fast to_numeric path already short-circuits —

# break even around 0.10. 0.05 keeps both kinds on the winning side.

_UNIQUE_MAP_MAX_RATIO = 0.05

def _clean_numeric_strings_via_uniques(series: pd.Series) -> pd.Series | None:

"""``_try_clean_numeric_strings`` applied to the distinct values, then mapped back.

Cleaning is per-value, so on a low-cardinality column — factor labels, coded levels,

anything categorical — the direct path repeats identical string work millions of

times. Converting the distinct values and mapping back is 19-36x faster at 1M rows

with identical output.

The threshold is the subtle part. ``_try_clean_numeric_strings`` keeps a conversion

when >= 80% of *non-NA values* convert, which is frequency-weighted; the same test

over distinct values is not. A column of 9,900 copies of ``'1'`` plus 100 distinct

labels converts on the full column (99%) but fails over uniques (1%). So the

conversion is re-scored here against the full column, never against the uniques.

"""

from pandas.api.types import is_object_dtype

# Plain object only. Series.map preserves an extension dtype — mapping a categorical

# returns a *categorical* of ints where the direct path returns int64, and a `string`

# column maps to int64 where the direct path returns nullable Int64. Both are real

# dtype changes (a categorical is not numeric, which would drop the column out of a

# client's "numeric columns" picker), and replicating pandas' dtype rules here would

# be more fragile than simply not taking the shortcut. Object is where the win is.

if not is_object_dtype(series):

return _try_clean_numeric_strings(series)

non_na = series.dropna()

if len(non_na) == 0:

return None

uniques = pd.Series(non_na.unique())

if len(uniques) > len(series) * _UNIQUE_MAP_MAX_RATIO:

return _try_clean_numeric_strings(series)

threshold = len(non_na) * _NUMERIC_CONVERSION_THRESHOLD

def _mapped(converted_uniques: pd.Series) -> pd.Series:

# Keys come from non-NA values only, so no NaN-is-not-equal-to-itself problem.

# Values absent from the mapping (the original NAs) come back as NaN, matching

# the direct path.

return series.map(dict(zip(uniques, converted_uniques, strict=False)))

# Fast path, then slow path — the same order and the same acceptance test as

# _try_clean_numeric_strings, but scored over the full column so the frequency

# weighting is preserved.

direct = _mapped(pd.to_numeric(uniques, errors='coerce'))

if direct.notna().sum() >= threshold:

return direct

formatted = _mapped(_coerce_formatted_numerics(uniques))

if formatted.notna().sum() >= threshold:

return formatted

return None

@dataclass

class ColumnRef:

"""

Column reference with level awareness for IDE discoverability.

NOT a str subclass to avoid pandas/numpy quirks, serialization issues,

and hashing/equality surprises. Implements __hash__ and __eq__ for

dict key usage. Compares equal to strings for flexibility.

Attributes

----------

name : str

The column name in the DataFrame

levels : list

Sorted unique values from the data (property)

count : int

Number of distinct levels (property)

Examples

--------

>>> pb = ProcessBehavior(df)

>>> pb.cols.Lane # Lane (4): [1, 2, 3, 4]

>>> pb.cols.Lane.levels # [1, 2, 3, 4]

>>> pb.cols.Lane.count # 4

Use in plan:

>>> study = pb.formulate(

... response=pb.cols.Weight,

... plan={pb.cols.Lane: [1, 2, 3, 4]}

... )

"""

name: str

_df: pd.DataFrame = field(repr=False, compare=False)

@property

def levels(self) -> list:

"""Sorted unique values from the data."""

values = self._df[self.name].dropna().unique()

try:

return sorted(values.tolist())

except TypeError:

# Mixed types can't be sorted

return list(values)

@property

def count(self) -> int:

"""Number of distinct levels."""

return len(self.levels)

def __str__(self) -> str:

return self.name

def __hash__(self) -> int:

return hash(self.name)

def __eq__(self, other: object) -> bool:

if isinstance(other, ColumnRef):

return self.name == other.name

if isinstance(other, str):

return self.name == other

return False

def __repr__(self) -> str:

lvls = self.levels

if len(lvls) <= 6:

return f'{self.name} ({len(lvls)}): {lvls}'

return f'{self.name} ({len(lvls)}): [{lvls[0]}..{lvls[-1]}]'

class ColumnAccessor:

"""

Provides IDE auto-completion for DataFrame column names with level awareness.

Usage:

pb = ProcessBehavior(df)

pb.cols.Height # ColumnRef with auto-completion

pb.cols.Height.levels # [1.0, 1.5, 2.0, ...]

pb.cols.Height.count # Number of unique levels

This class dynamically creates ColumnRef attributes for each column in the

DataFrame, enabling IDE auto-completion, preventing typos, and providing

level discoverability for sampling plans. Columns are sorted alphabetically

for consistent tab-completion ordering.

"""

def __init__(self, df: pd.DataFrame):

"""

Initialize accessor with DataFrame columns.

Args:

df: The DataFrame whose columns will be accessible

"""

self._df = df

# Sort by string representation to handle mixed-type column names

self._columns = sorted(df.columns, key=str)

self._attr_to_col = {} # Track sanitized_name → original_column

# Dynamically add each column as a ColumnRef attribute

for col in self._columns:

# Convert column name to valid Python identifier if needed

attr_name = self._sanitize_column_name(col)

if attr_name in self._attr_to_col:

# Collision detected - warn and skip

existing_col = self._attr_to_col[attr_name]

logger.warning(

f"Column name collision: '{col}' and '{existing_col}' "

f"both sanitize to '{attr_name}'. "

f"'{col}' will only be accessible via pb.cols['{col}']."

)

else:

# Avoid overwriting internal attributes (e.g., _df, _columns)

if hasattr(self, attr_name):

logger.warning(

f"Column '{col}' sanitizes to '{attr_name}' which conflicts "

f"with an internal attribute. Use pb.cols['{col}'] to access."

)

continue

self._attr_to_col[attr_name] = col

setattr(self, attr_name, ColumnRef(col, df))

def _sanitize_column_name(self, col_name: str) -> str:

"""

Convert column name to valid Python identifier.

Handles spaces, special characters, etc.

Args:

col_name: Original column name

Returns:

Sanitized name safe for use as Python attribute

"""

# Normalize non-string column names (e.g., integers)

col_name = str(col_name)

# Replace spaces and special chars with underscores

safe_name = col_name.replace(' ', '_').replace('-', '_')

# Remove other special characters

safe_name = ''.join(c if c.isalnum() or c == '_' else '_' for c in safe_name)

# Handle empty names

if not safe_name:

return '_empty'

# Ensure doesn't start with number

if safe_name[0].isdigit():

safe_name = f'col_{safe_name}'

return safe_name

def __repr__(self) -> str:

"""Display available columns."""

return f'ColumnAccessor({self._columns})'

def __getitem__(self, col_name: str) -> ColumnRef:

"""

Access column by original name (dict-style).

Useful for columns with names that can't be valid Python identifiers

or when collisions occur during sanitization.

Args:

col_name: Original column name

Returns:

ColumnRef for the column (for use in formulate())

Raises:

ColumnNotFoundError: If column doesn't exist

"""

resolved_name = col_name

# Support string lookup for non-string column labels (e.g., 123 -> "123")

if resolved_name not in self._df.columns and isinstance(resolved_name, str):

matching = [c for c in self._df.columns if str(c) == resolved_name]

if len(matching) == 1:

resolved_name = matching[0]

if resolved_name not in self._df.columns:

available = list(self._df.columns)

raise ColumnNotFoundError(

f"Column '{col_name}' not found. Available: {available}", column=col_name, available=available

)

return ColumnRef(resolved_name, self._df)

def __dir__(self):

"""Support for tab-completion in IPython/Jupyter."""

return list(self._attr_to_col.keys())

# ---------------------------------------------------------------------------

# formulate() helpers (module-level so they're easy to test in isolation)

# ---------------------------------------------------------------------------

def _validate_factors_or_plan_args(

factors: list | None, plan: dict | None, time: object | None = None

) -> None:

"""Enforce the factors XOR plan contract for formulate().

Both provided is rejected (ambiguous source of factor structure).

Omitting both is fine as long as there is a ``time`` column: that is a

single measurement stream over time — the most common SPC study there is,

and an X/mR chart handles it. It used to be refused outright, while the

equivalent ``factors=[]`` was accepted and worked, so the two spellings of

"no factors" did opposite things and the refusal named neither.

With no factors, no plan *and* no time there is nothing to analyse

against: no grid, and no order to plot in.

"""

if factors is not None and plan is not None:

raise ValidationError(

"Cannot specify both 'factors' and 'plan'. Use either:\n"

' • factors=[...] to infer structure from observed data (complete designs)\n'

' • plan={col: [levels], ...} to specify expected structure (complete + incomplete designs)'

)

if factors is None and plan is None and time is None:

raise ValidationError(

'A study needs either a time order or a grouping structure — this has neither.\n\n'

'Add whichever describes your data:\n'

" • time='<column>' — a single measurement stream in order (X/mR charts)\n"

" • factors=['<column>', ...] — categorical variables defining subgroups\n"

' • plan={column: [levels], ...} — expected factor levels, which also\n'

' lets incomplete designs be detected\n\n'

'A single stream is the common case:\n'

" study = pb.formulate(df, response='weight', time='hour')"

)

def _compute_pds(

detector: SDSRegistry,

sampling_plan: dict[str, list] | None,

T_planned: int | None,

N_planned: int | None,

) -> SDSResult | None:

"""Compute the Plan Design State (PDS) when a sampling plan was supplied.

PDS is `None` when no plan exists or any of T/N are missing — in that

case the study has only an ODS (observed) and ADS (analytical) state.

"""

if sampling_plan is None or T_planned is None or N_planned is None:

return None

from math import prod

K = prod(len(v) for v in sampling_plan.values())

pds = detector.classify_from_plan(K, T_planned, N_planned)

logger.debug('PDS: SDS %s (%s)', pds.sds, pds.reason)

return pds

class ProcessBehavior:

"""

Main entry point for process behavior analysis with auto-completion.

This class makes analysis frictionless by:

1. Providing IDE auto-completion for column names

2. Auto-detecting the design-state lineage (PDS / ODS / ADS)

3. Showing valid chart types based on the analytical design state (ADS)

4. Recommending the best chart for the detected ADS

5. Two-step workflow: formulate() then execute()

Usage:

# Basic usage with auto-completion

pb = ProcessBehavior(df)

# Step 1: formulate() - analyze structure and get recommendations

study = pb.formulate(

response=pb.cols.Measurement,

time=pb.cols.Time,

factors=[pb.cols.Operator, pb.cols.Machine]

)

# Inspect the study

print(study.observed_design_state.sds) # Detected ODS

print(study.valid_charts) # What's available

print(study.recommended_chart) # Best choice

# Step 2: execute() - run the chart

result = study.execute() # Uses recommended chart

result = study.execute(chart='Xbar') # Or explicit chart

Attributes:

cols: ColumnAccessor for IDE auto-completion of column names

data: The underlying pandas DataFrame

"""

def __init__(self, df: pd.DataFrame, na_values: list[str] | None = None):

"""

Initialize ProcessBehavior with data and optional NA value handling.

Args:

df: pandas DataFrame containing process data

na_values: Additional values to treat as NA/missing (beyond pandas defaults).

Common garbage characters are handled automatically.

Examples: ['*', '?', '--', 'ND', 'BDL', '<LOD']

Examples:

# Basic usage - automatic garbage character handling

>>> pb = ProcessBehavior(df)

# Custom NA indicators (combined with defaults)

>>> pb = ProcessBehavior(df, na_values=['-999', '9999', 'MISSING'])

"""

if not isinstance(df, pd.DataFrame):

raise ValidationError(f'Expected pandas DataFrame, got {type(df).__name__}')

# Default garbage characters commonly found in real-world data

# These are NOT recognized by pandas by default

default_na = [

'*', # Common in lab data for missing/invalid

'?', # Question mark for unknown

'--', # Double dash for missing

'ND', # Not Detected

'BDL', # Below Detection Limit

'BQL', # Below Quantification Limit

'<LOD', # Below Limit of Detection

'>ULQ', # Above Upper Limit of Quantification

'N/D', # Not Detected (variant)

'n/d', # Not detected (lowercase)

'MISSING',

'missing',

]

# Combine default with user-specified NA values

all_na_values = default_na + (na_values or [])

# Clean the data - replace garbage characters with pd.NA

cleaned_df = df.copy()

# Track which columns had NA values for informative warning

columns_with_na = []

na_counts = {}

for col in cleaned_df.columns:

# A garbage token is a string, so only text-like columns can hold one. Scanning

# numeric/datetime/bool columns is a guaranteed no-op and was ~1.6s of a 3.6s

# init at 1M x 50. Note the test is "is text-like", not "is not numeric":

# a CATEGORICAL OF STRINGS can match, and excluding on numeric-ness alone

# would silently stop cleaning it.

if not _is_text_like(cleaned_df[col]):

continue

# Count how many garbage values we find

na_mask = cleaned_df[col].isin(all_na_values)

na_count = na_mask.sum()

if na_count > 0:

columns_with_na.append(col)

na_counts[col] = na_count

# Replace with pd.NA

cleaned_df.loc[na_mask, col] = pd.NA

# Try to convert to numeric if it was originally numeric

# This handles cases like ['235.5', '*', '237.2'] -> [235.5, NaN, 237.2]

try:

# Try conversion - if it fails, keep original dtype

numeric_col = pd.to_numeric(cleaned_df[col])

cleaned_df[col] = numeric_col

except (ValueError, TypeError):

# Keep as-is if conversion fails (likely string data)

pass

# Tell the user what we changed on their behalf. This is a warning, not

# a log record: the library configures no logging handler, so a

# logger.warning here was invisible in most sessions and could not be

# filtered or escalated by callers.

if columns_with_na:

total_na = sum(na_counts.values())

warnings.warn(

f'Found {total_na} garbage/NA values across {len(columns_with_na)} column(s):\n'

+ '\n'.join([f' • {col}: {count} values' for col, count in na_counts.items()])

+ '\n\nThese values were converted to NA and will be excluded from analysis.',

ProcessBehaviorWarning,

stacklevel=3,

)

# Phase 2: Clean numeric formatting (currency symbols, thousands

# separators, accounting negatives, percentages) in object columns

formatting_cleaned = {}

for col in cleaned_df.columns:

result = _clean_numeric_strings_via_uniques(cleaned_df[col])

if result is not None:

formatting_cleaned[col] = int(result.notna().sum())

cleaned_df[col] = result

if formatting_cleaned:

warnings.warn(

f'Cleaned numeric formatting in {len(formatting_cleaned)} column(s):\n'

+ '\n'.join([f' • {col}: {count} values converted' for col, count in formatting_cleaned.items()])

+ '\n\nCurrency symbols, thousands separators, and '

'accounting negatives were removed.',

ProcessBehaviorWarning,

stacklevel=3,

)

self.data = cleaned_df

self.cols = ColumnAccessor(self.data)

# Pending derived-variable specs, accumulated by the fluent verbs.

# Immutable: each verb returns a new ProcessBehavior with an extended

# tuple. Materialized into the analytic dataset at formulate().

self._derivations: tuple[Derivation, ...] = ()

logger.info(f'ProcessBehavior: {len(df)} rows, {len(df.columns)} columns')

# =========================================================================

# Factory Methods: Read from files

# =========================================================================

@classmethod

def read_csv(cls, path: str, na_values: list[str] | None = None, **kwargs) -> ProcessBehavior:

"""

Read data from a CSV file.

Parameters

----------

path : str

Path to the CSV file.

na_values : list of str, optional

Additional values to treat as NA/missing.

**kwargs

Additional arguments passed to pandas.read_csv().

Returns

-------

ProcessBehavior

ProcessBehavior instance with loaded data.

Examples

--------

>>> pb = ProcessBehavior.read_csv('fillweight_data.csv')

>>> pb = ProcessBehavior.read_csv('data.csv', encoding='latin-1')

"""

df = pd.read_csv(path, **kwargs)

return cls(df, na_values=na_values)

@classmethod

def read_excel(

cls, path: str, sheet_name: str | int = 0, na_values: list[str] | None = None, **kwargs

) -> ProcessBehavior:

"""

Read data from an Excel file.

Parameters

----------

path : str

Path to the Excel file (.xlsx, .xls).

sheet_name : str or int, default 0

Sheet name or index to read.

na_values : list of str, optional

Additional values to treat as NA/missing.

**kwargs

Additional arguments passed to pandas.read_excel().

Returns

-------

ProcessBehavior

ProcessBehavior instance with loaded data.

Examples

--------

>>> pb = ProcessBehavior.read_excel('data.xlsx')

>>> pb = ProcessBehavior.read_excel('data.xlsx', sheet_name='Sheet2')

"""

df = pd.read_excel(path, sheet_name=sheet_name, **kwargs)

return cls(df, na_values=na_values)

@classmethod

def read_parquet(cls, path: str, na_values: list[str] | None = None, **kwargs) -> ProcessBehavior:

"""

Read data from a Parquet file.

Parameters

----------

path : str

Path to the Parquet file.

na_values : list of str, optional

Additional values to treat as NA/missing.

**kwargs

Additional arguments passed to pandas.read_parquet().

Returns

-------

ProcessBehavior

ProcessBehavior instance with loaded data.

Examples

--------

>>> pb = ProcessBehavior.read_parquet('data.parquet')

"""

try:

df = pd.read_parquet(path, **kwargs)

except ImportError:

raise ImportError(

'Reading Parquet files requires pyarrow or fastparquet. Install with: pip install pyarrow'

) from None

return cls(df, na_values=na_values)

@classmethod

def read_clipboard(cls, na_values: list[str] | None = None, **kwargs) -> ProcessBehavior:

"""

Read data from the system clipboard.

Useful for quickly pasting data from Excel or Google Sheets.

Parameters

----------

na_values : list of str, optional

Additional values to treat as NA/missing.

**kwargs

Additional arguments passed to pandas.read_clipboard().

Returns

-------

ProcessBehavior

ProcessBehavior instance with clipboard data.

Examples

--------

Copy data from Excel, then:

>>> pb = ProcessBehavior.read_clipboard()

"""

df = pd.read_clipboard(**kwargs)

return cls(df, na_values=na_values)

@staticmethod

def _to_column_name(col: str | ColumnRef) -> str:

"""Extract column name from str or ColumnRef."""

return col.name if isinstance(col, ColumnRef) else col

def _validate_plan(self, plan: dict) -> tuple[dict[str, list], list[str], int, int]:

"""

Validate and normalize sampling plan.

Parameters

----------

plan : dict

Sampling plan with required 'factors' key and optional 'T', 'N'.

Example: {'factors': {'Lane': [1,2,3,4], 'Phase': [1,2,3]}, 'T': 10, 'N': 2}

Returns

-------

tuple[dict[str, list], list[str], int | None, int | None]

(normalized_factors, factor_order, T_planned, N_planned)

Raises

------

ValidationError

If 'factors' key is missing

ColumnNotFoundError

If a plan column doesn't exist in the data

"""

# Require 'factors' key

if 'factors' not in plan:

raise ValidationError(

"Sampling plan must have 'factors' key.\n"

"Example: plan={'factors': {'Lane': [1,2,3,4], 'Phase': [1,2,3]}, 'T': 10, 'N': 2}"

)

plan_factors = plan['factors']

# Require at least one factor

if not plan_factors:

raise ValidationError(

"Sampling plan 'factors' must contain at least one factor.\n"

"Example: plan={'factors': {'Lane': [1,2,3,4]}, 'T': 10}"

)

T_planned = plan.get('T')

N_planned = plan.get('N')

if T_planned is None:

raise ValidationError(

"Sampling plan must include 'T' (planned number of time periods).\n"

"Example: plan={'factors': {'Lane': [1,2,3,4], 'Phase': [1,2,3]}, 'T': 10, 'N': 2}"

)

if N_planned is None:

raise ValidationError(

"Sampling plan must include 'N' (planned observations per cell).\n"

"Example: plan={'factors': {'Lane': [1,2,3,4], 'Phase': [1,2,3]}, 'T': 10, 'N': 2}"

)

normalized: dict[str, list] = {}

factor_order: list[str] = []

for col, levels in plan_factors.items():

col_name = self._to_column_name(col)

# Validate column exists

if col_name not in self.data.columns:

available = list(self.data.columns)

raise ColumnNotFoundError(

f"Plan column '{col_name}' not found in data. Available: {available}",

column=col_name,

available=available,

)

# Validate levels is non-empty

if not levels:

raise ValidationError(

f"Factor '{col_name}' has empty level list in plan.\n"

f'Each factor must have at least one planned level.'

)

normalized[col_name] = list(levels)

factor_order.append(col_name)

# Check for extra observed levels (warn, don't error)

observed = set(self.data[col_name].dropna().unique())

planned = set(levels)

extra = observed - planned

if extra:

extra_list = sorted(extra, key=lambda x: (type(x).__name__, x))

observed_sorted = sorted(observed, key=lambda x: (type(x).__name__, x))

logger.warning(

f"Factor '{col_name}' has observed levels not in plan: {extra_list}\n"

f' Your plan: {levels}\n'

f' Observed: {observed_sorted}\n'

f'\n'

f' To update your plan:\n'

f" plan['factors']['{col_name}'] = pb.cols['{col_name}'].levels # Use observed\n"

f' # or\n'

f" plan['factors']['{col_name}'] = {observed_sorted} # Add manually"

)

return normalized, factor_order, T_planned, N_planned

# =========================================================================

# Derived variables (transforms + binning) — fluent, immutable verbs

# =========================================================================

@property

def derivations(self) -> tuple[Derivation, ...]:

"""Pending derived-variable specs attached to this ProcessBehavior."""

return self._derivations

def _with_derivations(self, specs: tuple[Derivation, ...]) -> ProcessBehavior:

"""Return a new ProcessBehavior sharing ``self.data`` with ``specs`` attached."""

new = ProcessBehavior.__new__(ProcessBehavior)

new.data = self.data

new.cols = self.cols

new._derivations = specs

return new

def _attach(self, spec: Derivation) -> ProcessBehavior:

"""Validate a spec against the current dataset context and attach it.

Raises ``ValidationError`` at attach time for column-not-found / non-numeric

source / output-name collision (the fluent path's commit point).

"""

existing = {d.output_name for d in self._derivations}

result = _validate_derivation(spec, self.data, existing_names=existing)

if not result.ok:

raise ValidationError(

f'Cannot add derived variable {spec.output_name!r}: {result.summary()}'

)

return self._with_derivations((*self._derivations, spec))

def transform(

self,

column: str | ColumnRef,

function: str,

*,

label: str | None = None,

shift: float | None = None,

exponent: float | None = None,

on_invalid: str = 'error',

) -> ProcessBehavior:

"""Attach a continuous→continuous transform; returns a new ProcessBehavior.

``function`` is one of ``log`` (``ln`` alias), ``log10``, ``sqrt``,

``arcsin`` (= ``arcsin(√x)``, x a proportion in ``[0, 1]``), ``inverse``,

``square``, ``power`` (needs ``exponent``), ``zscore``. The new column is

named ``label`` or ``{column}_{function}``. Domain violations are resolved

at formulate() per ``on_invalid`` (``'error'`` raises with a structured

count; ``'na'`` sets offenders missing); ``shift`` applies ``f(x + shift)``.

"""

spec = Derivation.transform(

self._to_column_name(column), function,

label=label, shift=shift, exponent=exponent, on_invalid=on_invalid,

)

return self._attach(spec)

def bin(

self,

column: str | ColumnRef,

*,

method: str = 'equal_freq',

n: int = 4,

breaks: list[float] | None = None,

bin_labels='range',

label: str | None = None,

right: bool = False,

) -> ProcessBehavior:

"""Attach a continuous→categorical binning; returns a new ProcessBehavior.

``method`` is ``equal_freq`` (default, n quantile bins), ``equal_width``,

``breaks`` (explicit cut points, out-of-range falls into below/above

categories), or ``sd`` (±1σ, ±2σ about the mean). The new column is an

ordered categorical named ``label`` or ``{column}_bin``. ``bin_labels`` is

``'range'`` (default, from fitted edges), ``'ordinal'``, ``'number'``, or

an explicit list. Intervals are left-closed ``[a, b)`` unless ``right=True``.

"""

spec = Derivation.bin(

self._to_column_name(column),

method=method, n=n, breaks=breaks, bin_labels=bin_labels, label=label, right=right,

)

return self._attach(spec)

def add_derived(self, *specs: Derivation) -> ProcessBehavior:

"""Attach one or more pre-built :class:`Derivation` specs (programmatic / app path)."""

pb = self

for spec in specs:

pb = pb._attach(spec)

return pb

def remove_derived(self, id: str) -> ProcessBehavior:

"""Return a new ProcessBehavior with the derivation ``id`` dropped."""

kept = tuple(d for d in self._derivations if d.id != id)

if len(kept) == len(self._derivations):

available = [d.id for d in self._derivations]

raise ValidationError(f'No derivation with id {id!r}. Attached ids: {available}.')

return self._with_derivations(kept)

def replace_derived(self, id: str, spec: Derivation) -> ProcessBehavior:

"""Return a new ProcessBehavior with derivation ``id`` swapped for ``spec`` (keeps position)."""

if id not in {d.id for d in self._derivations}:

available = [d.id for d in self._derivations]

raise ValidationError(f'No derivation with id {id!r}. Attached ids: {available}.')

# Validate the replacement against the dataset + the *other* pending names.

others = {d.output_name for d in self._derivations if d.id != id}

result = _validate_derivation(spec, self.data, existing_names=others)

if not result.ok:

raise ValidationError(

f'Cannot replace derivation {id!r} with {spec.output_name!r}: {result.summary()}'

)

swapped = tuple(spec if d.id == id else d for d in self._derivations)

return self._with_derivations(swapped)

def _materialize_derivations(self) -> tuple[pd.DataFrame, tuple[Derivation, ...]]:

"""Resolve pending specs into columns on a copy of the data (the augmented frame).

Evaluates each spec in attach order against the **original** columns

(chaining is deferred to v2), adds each ``output_name`` column, and returns

the augmented frame plus the fit-frozen specs. With no pending specs this

returns ``self.data`` untouched (the no-derivation path is identity).

``on_invalid='error'`` with domain violations raises a structured

``ValidationError`` here — the only commit-time raise.

"""

if not self._derivations:

return self.data, ()

frame = self.data.copy()

resolved: list[Derivation] = []

for spec in self._derivations:

res = _evaluate_derivation(spec, self.data[spec.column])

if (

spec.family == 'transform'

and spec.params.get('on_invalid', 'error') == 'error'

and res.n_invalid > 0

):

sample = list(res.invalid_index[:10])

raise ValidationError(

f"Derived variable '{spec.output_name}' ({spec.function} of "

f"'{spec.column}') has {res.n_invalid} domain violation(s) "

f'at rows {sample}{"..." if res.n_invalid > 10 else ""}. '

f"Resolve with shift=, on_invalid='na', or a different transform."

)

frame[spec.output_name] = res.values

resolved.append(spec.with_fitted(res.fitted))

return frame, tuple(resolved)

def formulate(

self,

response: str | ColumnRef,

factors: list[str | ColumnRef] | None = None,

time: str | ColumnRef | None = None,

plan: dict | None = None,

precision: int = 3,

unit_of_analysis: str | None = None,

) -> Study:

"""

Formulate a study for process behavior analysis.

This method prepares and enriches the dataset (including residuals and

effects as applicable) and returns a Study describing what analyses are

valid. Call study.execute() to perform chart-specific calculations and

produce an AnalysisResult.