SATP¶

def normalize_polar_angles(angles: pd.Series) -> pd.Series:
    """
    Return polar angles in canonical (counter-clockwise) orientation.

    CircE's BFGS optimisation may converge to a mathematically equivalent
    *reflected* solution in which the PAQ attributes are arranged in clockwise
    (decreasing) order rather than the canonical counter-clockwise (increasing)
    order.  Both solutions fit the correlation data equally well, but the
    reflected form is inconsistent with the standard circumplex ordering
    (pleasant → vibrant → eventful → …) and will produce incorrect GDIFF values
    if compared against the ideal equally-spaced angles.

    Detection uses a monotonicity check on the first three angles after PAQ1:
    if the angle for PAQ2 exceeds PAQ3, or PAQ3 exceeds PAQ4, the solution is
    reflected.  This is more robust than a threshold-on-sum heuristic because
    it tests the structural property of the orientation directly.

    When reflection is detected, ``360 - angle`` is applied to PAQ2-PAQ8.
    PAQ1 is anchored at 0° and is left unchanged.

    Parameters
    ----------
    angles
        Series of polar angle estimates (degrees) with PAQ_IDS as the index.
        Typically the ``polar_angles`` attribute of a `CircE` instance.

    Returns
    -------
    :
        Polar angles in canonical (counter-clockwise) orientation, with the
        same index as the input.

    Examples
    --------
    >>> from soundscapy.surveys.survey_utils import PAQ_IDS
    >>> import pandas as pd
    >>> reflected = pd.Series(
    >>>     [0.0, 315.0, 270.0, 225.0, 180.0, 135.0, 90.0, 45.0],
    >>>     index=PAQ_IDS)
    >>> normalize_polar_angles(reflected).tolist()
    [0.0, 45.0, 90.0, 135.0, 180.0, 225.0, 270.0, 315.0]

    """
    # Monotonicity check (positional): canonical ordering has PAQ2 < PAQ3 < PAQ4.
    if angles.iloc[1] > angles.iloc[2] or angles.iloc[2] > angles.iloc[3]:
        corrected = angles.copy()
        corrected.iloc[1:] = 360 - corrected.iloc[1:]
        return corrected
    return angles

def person_center(data: pd.DataFrame, by: str = "participant") -> pd.DataFrame:
    """
    Center PAQ ratings within each participant (column-wise within-person centering).

    !!! warning "Deprecated v0.8.0"
        Use `soundscapy.surveys.ipsatize` with ``method="column_wise"``
        instead.  For the centering that matches the published SATP analysis,
        use ``method="grand_mean"`` (the default of
        `~soundscapy.surveys.ipsatize`).

    This function applies **column-wise** centering: for every PAQ column
    independently, each participant's mean across their observations is
    subtracted (8 centering scalars per participant).

    !!! note
        This is *not* the centering described in the original SATP R
        implementation (Aletta et al., 2024), which applies grand-mean
        centering (one scalar per participant across all PAQ columns and
        observations).  Use `soundscapy.surveys.ipsatize` with
        ``method="grand_mean"`` to match the R reference implementation.

    Parameters
    ----------
    data
        DataFrame containing PAQ columns and a participant grouping column.
    by
        Column to group by for centering. Default is ``"participant"``.

    Returns
    -------
    :
        DataFrame containing only the PAQ columns (not ``by``), with
        column-wise participant-centred values.

    """
    import warnings  # noqa: PLC0415

    warnings.warn(
        "person_center() is deprecated; use ipsatize(method='column_wise') instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return ipsatize(data, method="column_wise", participant_col=by)

def fit_circe(
    data: pd.DataFrame,
    language: str,
    datasource: str,
    *,
    models: list[CircModelE] | None = None,
    center_by_participant: bool = True,
    errors: Literal["raise", "warn"] = "raise",
) -> "CircEResults":
    """
    Fit circumplex SEM models to PAQ data and return a tidy DataFrame.

    Validates input data, optionally applies grand-mean within-person centering
    (matching the published SATP analysis), computes a complete-case correlation
    matrix, and fits the requested circumplex model types using Browne's BFGS
    optimisation via the R ``CircE`` package.

    Parameters
    ----------
    data
        DataFrame with PAQ1-PAQ8 and a ``participant`` column.
        Column aliases (e.g. PAQ label names, ``Participant``) are accepted
        and renamed automatically by the schema validator.
    language
        Language code for the dataset (e.g. ``"eng"``, ``"fra"``).
        Stored in the results; not used for computation.
    datasource
        Dataset identifier (e.g. ``"SATP"``, ``"ISD"``).
        Stored in the results; not used for computation.
    models
        List of model types to fit. Default: all four ``CircModelE`` variants.
        Passing ``[]`` returns an empty `CircEResults`
        (``len(result) == 0``).
    center_by_participant
        Whether to apply grand-mean within-person centering (via
        `~soundscapy.surveys.ipsatize` with ``method="grand_mean"``)
        before fitting.  Set to ``False`` if the data is already centered or
        if no centering is desired.
    errors
        How to handle rows that fail schema validation (PAQ values outside
        ``[0, 100]``, missing required columns, etc.):

        ``"raise"`` *(default)* — raise a `pandera.errors.SchemaErrors`
        immediately, listing every failing row and constraint.

        ``"warn"`` — emit a `UserWarning` describing the failing rows
        and continue with the valid rows only.

        !!! note
            If you pass *already-centered* data, set
            ``center_by_participant=False`` to skip the internal centering step;
            otherwise pass raw ``[0, 100]``-range data and use the default
            ``center_by_participant=True``.  Passing pre-centered data without
            disabling centering will cause schema validation to reject the
            negative values.

    Returns
    -------
    :
        Collection of fitted models.  Access the tidy DataFrame via
        ``.table``; access individual model results via ``.for_model()``.
        Failed models are stored in ``.error_rows`` and included in
        ``.table``.

    Examples
    --------
    >>> import soundscapy as sspy
    >>> from soundscapy.satp import fit_circe
    >>> data = sspy.isd.load()
    >>> data = data.rename(columns={'SessionID': 'participant'})
    >>> results = fit_circe(data, language='eng', datasource='ISD', errors='warn')
    >>> len(results)
    4

    """
    warnings.warn(
        "The SATP analysis module is experimental. Use with caution.",
        UserWarning,
        stacklevel=2,
    )
    if len(data) == 0:
        msg = (
            "No complete cases found: input DataFrame is empty. "
            "Check that data contains valid rows with PAQ1-PAQ8 columns."
        )
        raise ValueError(msg)

    try:
        validated = SATPSchema.validate(data, lazy=True)
    except SchemaErrors as exc:
        if errors == "raise":
            raise
        bad_idx = exc.failure_cases["index"].dropna().unique()
        clean = data.loc[~data.index.isin(bad_idx)]
        warnings.warn(
            f"Dropping {len(data) - len(clean)} rows that failed schema validation "
            f"({len(clean)} rows remain). "
            "Pass errors='raise' to raise an error instead.",
            UserWarning,
            stacklevel=2,
        )
        try:
            validated = SATPSchema.validate(clean, lazy=True)
        except SchemaErrors as exc2:
            # type: ignore [missing-argument]
            raise SchemaErrors(
                schema_errors=exc2.schema_errors,
                data=exc2.data,
            ) from exc2

    if center_by_participant and "participant" not in validated.columns:
        msg = (
            "center_by_participant=True requires a 'participant' column. "
            "Pass center_by_participant=False if your data is already centered."
        )
        raise ValueError(msg)
    processed = (
        ipsatize(validated, method="grand_mean", participant_col="participant")
        if center_by_participant
        else validated
    )

    # Use listwise deletion (complete cases only) — consistent with R's na.omit().
    complete = processed[PAQ_IDS].dropna()
    n = len(complete)
    if n == 0:
        msg = (
            "No complete cases found after validation and ipsatization. "
            "Check that PAQ1-PAQ8 are not all NaN and participant column is present."
        )
        raise ValueError(msg)
    corr = complete.corr()

    circ_models = models if models is not None else list(CircModelE)
    fitted: list[CircE] = []
    error_rows: list[dict] = []
    fit_exceptions = (ValueError, np.linalg.LinAlgError, RuntimeError, RRuntimeError)
    for model in circ_models:
        try:
            circe = CircE.compute_bfgs_fit(corr, n, datasource, language, model)
            fitted.append(circe)
        except fit_exceptions as e:
            warnings.warn(f"{model.value} raised {e}", stacklevel=2)
            # Populate all expected columns with None so that pandas does not
            # promote numeric columns (e.g. n, d) to float64 across all rows
            # when mixing sparse error dicts with full success dicts.
            error_rows.append(
                {
                    "language": language,
                    "datasource": datasource,
                    "model": model.value,
                    "n": n,
                    "m": None,
                    "chisq": None,
                    "d": None,
                    "p": None,
                    "cfi": None,
                    "gfi": None,
                    "agfi": None,
                    "srmr": None,
                    "mcsc": None,
                    "rmsea": None,
                    "rmsea_l": None,
                    "rmsea_u": None,
                    "gdiff": None,
                    **dict.fromkeys(PAQ_IDS),
                    "error": str(e),
                }
            )

    return CircEResults(
        models=fitted,
        language=language,
        datasource=datasource,
        error_rows=error_rows,
    )