AOI perturbation API¶

The AOI perturbation surface is vendor-neutral and exported from the eyeprocesspy package front door.

aoi_perturbation ¶

AOI perturbation and uncertainty analysis public API.

AMBIGUOUS `module-attribute` ¶

AMBIGUOUS = '__ambiguous__'

OUTSIDE `module-attribute` ¶

OUTSIDE = '__outside__'

all `module-attribute` ¶

__all__ = ['OUTSIDE', 'AMBIGUOUS', 'aoi_perturbation_spec', 'validate_aoi_geometry', 'convert_aoi_margin_to_degrees', 'convert_aoi_margin_to_pixels', 'dilate_aoi', 'erode_aoi', 'translate_aoi', 'jitter_aoi', 'perturb_aoi_geometry', 'create_aoi_perturbation_grid', 'apply_aoi_perturbation_grid', 'compare_aoi_assignments', 'estimate_aoi_assignment_stability', 'estimate_fixation_assignment_probability', 'recompute_aoi_features', 'run_aoi_sensitivity_analysis', 'summarise_aoi_sensitivity', 'assess_aoi_inference_stability', 'report_aoi_sensitivity', 'plot_aoi_perturbations', 'plot_aoi_assignment_stability', 'plot_aoi_coefficient_stability', 'plot_aoi_robustness_surface']

EyeProcessValidationError ¶

Bases: EyeProcessError

Raised when a canonical dataset violates validation contracts.

Source code in src/eyeprocesspy/exceptions.py

class EyeProcessValidationError(EyeProcessError):
    """Raised when a canonical dataset violates validation contracts."""

EyeResult ¶

Bases: dict

Small R-list-like result object with stable class metadata.

Source code in src/eyeprocesspy/irt.py

class EyeResult(dict):
    """Small R-list-like result object with stable class metadata."""

    def __init__(self, *args: Any, eyeprocess_class: str | None = None, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self.eyeprocess_class = eyeprocess_class

    def __getattr__(self, name: str) -> Any:
        try:
            return self[name]
        except KeyError as exc:
            raise AttributeError(name) from exc

_AssignmentVector ¶

Bases: list

List-compatible AOI assignment vector.

Source code in src/eyeprocesspy/aoi_perturbation.py

class _AssignmentVector(list):
    """List-compatible AOI assignment vector."""

    def tolist(self) -> list[Any]:
        return list(self)

_assign_points ¶

_assign_points(data: DataFrame, geometry: DataFrame, *, x_col: str, y_col: str, overlap_policy: str = 'ambiguous') -> pd.Series

Source code in src/eyeprocesspy/_aoi_assignment_core.py

def _assign_points(
    data: pd.DataFrame,
    geometry: pd.DataFrame,
    *,
    x_col: str,
    y_col: str,
    overlap_policy: str = "ambiguous",
) -> pd.Series:
    frame = _frame(data, "data")
    if x_col not in frame.columns or y_col not in frame.columns:
        raise EyeProcessValidationError(f"`data` must contain `{x_col}` and `{y_col}`.")
    overlap_policy = str(overlap_policy).lower()
    if overlap_policy not in {"ambiguous", "all", "error"}:
        raise EyeProcessValidationError("`overlap_policy` must be ambiguous, all, or error.")
    geom = validate_aoi_geometry(geometry)["geometry"]
    x = pd.to_numeric(frame[x_col], errors="coerce").to_numpy(dtype=float)
    y = pd.to_numeric(frame[y_col], errors="coerce").to_numpy(dtype=float)
    missing = ~np.isfinite(x) | ~np.isfinite(y)
    memberships = np.zeros((len(frame), len(geom)), dtype=bool)
    for j, (_, row) in enumerate(geom.iterrows()):
        memberships[:, j] = _aoi_contains(row, x, y) & ~missing
    counts = memberships.sum(axis=1)
    labels = np.full(len(frame), OUTSIDE, dtype=object)
    labels[missing] = None
    for i in np.where(counts == 1)[0]:
        labels[i] = str(geom.iloc[np.flatnonzero(memberships[i])[0]]["aoi_id"])
    ambiguous = np.where(counts > 1)[0]
    if len(ambiguous):
        if overlap_policy == "error":
            raise EyeProcessValidationError(
                f"{len(ambiguous)} observations have ambiguous overlapping AOI membership."
            )
        if overlap_policy == "ambiguous":
            labels[ambiguous] = AMBIGUOUS
        else:
            for i in ambiguous:
                labels[i] = "|".join(
                    geom.iloc[np.flatnonzero(memberships[i])]["aoi_id"].astype(str).tolist()
                )
    return pd.Series(labels, index=frame.index, dtype="object", name="aoi_assignment")

_validate_model_table ¶

_validate_model_table(table: Any, perturbation_id: str) -> pd.DataFrame

Source code in src/eyeprocesspy/_aoi_assignment_core.py

def _validate_model_table(table: Any, perturbation_id: str) -> pd.DataFrame:
    frame = _frame(table, "model callback result")
    required = ["term", "estimate", "SE", "CI_low", "CI_high", "p_value", "model_converged", "N"]
    missing = [c for c in required if c not in frame.columns]
    if missing:
        raise EyeProcessValidationError(
            "Model callback result is missing required column(s): " + ", ".join(missing)
        )
    out = frame.copy()
    terms = out["term"].astype("string")
    if terms.isna().any() or (terms.str.len() == 0).any():
        raise EyeProcessValidationError(
            "Model callback term values must be non-missing and non-empty."
        )
    out["term"] = terms.astype(str)

    raw_convergence = out["model_converged"]
    valid_boolean = raw_convergence.map(
        lambda v: pd.isna(v) or isinstance(v, (bool, np.bool_)) or v in (0, 1)
    )
    if not bool(valid_boolean.all()):
        raise EyeProcessValidationError(
            "`model_converged` must contain booleans, 0/1, or missing values; strings are not accepted."
        )
    out["model_converged"] = raw_convergence.astype("boolean")

    for column in ("estimate", "SE", "CI_low", "CI_high", "p_value", "N"):
        original = out[column]
        converted = pd.to_numeric(original, errors="coerce")
        bad_conversion = original.notna() & converted.isna()
        if bad_conversion.any():
            raise EyeProcessValidationError(
                f"Model callback `{column}` must be numeric or missing."
            )
        out[column] = converted
    if (out["N"].dropna() < 0).any():
        raise EyeProcessValidationError("Model callback `N` must be non-negative when supplied.")
    n_values = out["N"].dropna().to_numpy(dtype=float)
    if len(n_values) and not np.allclose(n_values, np.round(n_values), atol=1e-9, rtol=0.0):
        raise EyeProcessValidationError("Model callback `N` must be integer-valued when supplied.")

    converged = out["model_converged"].fillna(False).astype(bool)
    if converged.any():
        required_finite = ["estimate", "SE", "CI_low", "CI_high", "N"]
        bad = ~np.isfinite(out.loc[converged, required_finite].to_numpy(dtype=float))
        if bad.any():
            raise EyeProcessValidationError(
                "Converged model rows must contain finite estimate, SE, CI_low, CI_high, and N values."
            )
        if (out.loc[converged, "N"] <= 0).any():
            raise EyeProcessValidationError("Converged model rows must report N > 0.")

    out["perturbation_id"] = perturbation_id
    est = out["estimate"].to_numpy(dtype=float)
    out["direction"] = np.where(
        ~np.isfinite(est),
        "missing",
        np.where(est > 0, "positive", np.where(est < 0, "negative", "zero")),
    )
    ordered = ["perturbation_id"] + required + ["direction"]
    extras = [c for c in out.columns if c not in set(ordered)]
    return out[ordered + extras]

compare_aoi_assignments ¶

compare_aoi_assignments(baseline: Sequence[Any], perturbed: Sequence[Any], *, ids: Sequence[Any] | None = None) -> EyeResult

Source code in src/eyeprocesspy/_aoi_assignment_core.py

def compare_aoi_assignments(
    baseline: Sequence[Any],
    perturbed: Sequence[Any],
    *,
    ids: Sequence[Any] | None = None,
) -> EyeResult:
    base = pd.Series(list(baseline), dtype="object")
    alt = pd.Series(list(perturbed), dtype="object")
    if len(base) != len(alt):
        raise EyeProcessValidationError(
            "Baseline and perturbed assignments must have equal length."
        )
    resolved_ids = list(range(1, len(base) + 1)) if ids is None else list(ids)
    if len(resolved_ids) != len(base):
        raise EyeProcessValidationError("`ids` must have the same length as assignments.")
    missing = base.isna() | alt.isna()
    unchanged = (~missing) & base.eq(alt)
    newly = (~missing) & base.eq(OUTSIDE) & ~alt.eq(OUTSIDE)
    lost = (~missing) & ~base.eq(OUTSIDE) & alt.eq(OUTSIDE)
    reassigned = (~missing) & ~unchanged & ~newly & ~lost
    detail = pd.DataFrame(
        {
            "observation_id": resolved_ids,
            "baseline_aoi": base,
            "perturbed_aoi": alt,
            "unchanged": unchanged,
            "newly_assigned": newly,
            "lost_assignment": lost,
            "reassigned": reassigned,
            "missing_comparison": missing,
        }
    )
    comparable = ~missing
    n = int(comparable.sum())
    summary = pd.DataFrame(
        [
            {
                "n_total": len(base),
                "n_comparable": n,
                "proportion_unchanged": float(unchanged[comparable].mean()) if n else np.nan,
                "proportion_newly_assigned": float(newly[comparable].mean()) if n else np.nan,
                "proportion_lost": float(lost[comparable].mean()) if n else np.nan,
                "proportion_reassigned": float(reassigned[comparable].mean()) if n else np.nan,
                "proportion_missing_comparison": float(missing.mean()) if len(base) else np.nan,
            }
        ]
    )
    valid_pairs = detail.loc[comparable, ["baseline_aoi", "perturbed_aoi"]]
    matrix = pd.crosstab(valid_pairs["baseline_aoi"], valid_pairs["perturbed_aoi"], dropna=False)
    return _result(
        "eye_aoi_assignment_comparison", detail=detail, summary=summary, reassignment_matrix=matrix
    )

estimate_aoi_assignment_stability ¶

estimate_aoi_assignment_stability(comparisons: Mapping[str, Any] | DataFrame, *, metadata: DataFrame | None = None, group_cols: Sequence[str] | None = None) -> EyeResult

Source code in src/eyeprocesspy/_aoi_assignment_core.py

def estimate_aoi_assignment_stability(
    comparisons: Mapping[str, Any] | pd.DataFrame,
    *,
    metadata: pd.DataFrame | None = None,
    group_cols: Sequence[str] | None = None,
) -> EyeResult:
    rows: list[pd.DataFrame] = []
    if isinstance(comparisons, pd.DataFrame):
        required = {"perturbation_id", "baseline_aoi", "perturbed_aoi"}
        if not required.issubset(comparisons.columns):
            raise EyeProcessValidationError(
                "Comparison data must contain perturbation_id, baseline_aoi, and perturbed_aoi."
            )
        source = comparisons.copy()
    else:
        for pid, comparison in comparisons.items():
            detail = comparison["detail"].copy()
            detail["perturbation_id"] = str(pid)
            rows.append(detail)
        source = pd.concat(rows, ignore_index=True) if rows else pd.DataFrame()
    if source.empty:
        raise EyeProcessValidationError("No assignment comparisons are available.")
    if "observation_id" not in source.columns:
        source["observation_id"] = source.groupby("perturbation_id").cumcount() + 1
    source_ids = source["observation_id"].drop_duplicates().tolist()
    if pd.Series(source_ids).isna().any():
        raise EyeProcessValidationError("Comparison observation IDs must be non-missing.")
    if metadata is not None:
        meta = _frame(metadata, "metadata")
        if len(meta) != len(source_ids):
            raise EyeProcessValidationError("`metadata` must contain one row per observation.")
        if "observation_id" not in meta.columns:
            meta["observation_id"] = source_ids
        else:
            if meta["observation_id"].isna().any() or meta["observation_id"].duplicated().any():
                raise EyeProcessValidationError(
                    "Metadata observation IDs must be unique and non-missing."
                )
            if set(meta["observation_id"].tolist()) != set(source_ids):
                raise EyeProcessValidationError(
                    "Metadata observation IDs must match the comparison observation IDs exactly."
                )
        source = source.merge(meta, on="observation_id", how="left", validate="many_to_one")
    source["comparable"] = ~(source["baseline_aoi"].isna() | source["perturbed_aoi"].isna())
    source["unchanged"] = source["comparable"] & source["baseline_aoi"].eq(source["perturbed_aoi"])
    source["newly_assigned"] = (
        source["comparable"]
        & source["baseline_aoi"].eq(OUTSIDE)
        & ~source["perturbed_aoi"].eq(OUTSIDE)
    )
    source["lost_assignment"] = (
        source["comparable"]
        & ~source["baseline_aoi"].eq(OUTSIDE)
        & source["perturbed_aoi"].eq(OUTSIDE)
    )
    source["reassigned"] = source["comparable"] & ~(
        source["unchanged"] | source["newly_assigned"] | source["lost_assignment"]
    )

    def summarise(df: pd.DataFrame, keys: list[str]) -> pd.DataFrame:
        out_rows = []
        grouped = [((), df)] if not keys else df.groupby(keys, dropna=False, sort=True)
        for key, z in grouped:
            key = (key,) if keys and not isinstance(key, tuple) else key
            comparable = z["comparable"]
            n = int(comparable.sum())
            row = {k: v for k, v in zip(keys, key)}
            row.update(
                {
                    "n_total": len(z),
                    "n_comparable": n,
                    "proportion_unchanged": float(z.loc[comparable, "unchanged"].mean())
                    if n
                    else np.nan,
                    "proportion_newly_assigned": float(z.loc[comparable, "newly_assigned"].mean())
                    if n
                    else np.nan,
                    "proportion_lost": float(z.loc[comparable, "lost_assignment"].mean())
                    if n
                    else np.nan,
                    "proportion_reassigned": float(z.loc[comparable, "reassigned"].mean())
                    if n
                    else np.nan,
                }
            )
            out_rows.append(row)
        return pd.DataFrame(out_rows)

    overall = summarise(source, ["perturbation_id"])
    by_group: dict[str, pd.DataFrame] = {}
    for col in group_cols or []:
        if col not in source.columns:
            raise EyeProcessValidationError(
                f"Grouping column `{col}` is not available in comparison metadata."
            )
        by_group[col] = summarise(source, ["perturbation_id", col])
    aoi_level = summarise(source.assign(aoi=source["baseline_aoi"]), ["perturbation_id", "aoi"])
    return _result(
        "eye_aoi_assignment_stability",
        detail=source,
        overall=overall,
        group_summaries=by_group,
        aoi_level=aoi_level,
        caveat="Stability frequencies describe sensitivity to the specified perturbations; they are not probabilities that AOI assignments are scientifically true.",
    )

estimate_fixation_assignment_probability ¶

estimate_fixation_assignment_probability(assignments: Mapping[str, Sequence[Any]] | DataFrame, *, ids: Sequence[Any] | None = None, include_baseline: bool = True) -> pd.DataFrame

Estimate empirical assignment frequencies across a declared perturbation set.

Source code in src/eyeprocesspy/_aoi_assignment_core.py

def estimate_fixation_assignment_probability(
    assignments: Mapping[str, Sequence[Any]] | pd.DataFrame,
    *,
    ids: Sequence[Any] | None = None,
    include_baseline: bool = True,
) -> pd.DataFrame:
    """Estimate empirical assignment frequencies across a declared perturbation set."""
    if isinstance(assignments, pd.DataFrame):
        required = {"perturbation_id", "observation_id", "aoi_assignment"}
        if not required.issubset(assignments.columns):
            raise EyeProcessValidationError(
                "Assignment table must contain perturbation_id, observation_id, and aoi_assignment."
            )
        long = assignments.loc[:, list(required)].copy()
    else:
        rows = []
        lengths = {len(v) for v in assignments.values()}
        if len(lengths) != 1:
            raise EyeProcessValidationError(
                "All perturbation assignment vectors must have the same length."
            )
        n = next(iter(lengths), 0)
        obs = list(ids) if ids is not None else list(range(1, n + 1))
        if len(obs) != n:
            raise EyeProcessValidationError("`ids` must match the assignment vector length.")
        for pid, labels in assignments.items():
            if not include_baseline and str(pid) == "baseline":
                continue
            rows.append(
                pd.DataFrame(
                    {
                        "perturbation_id": str(pid),
                        "observation_id": obs,
                        "aoi_assignment": list(labels),
                    }
                )
            )
        long = (
            pd.concat(rows, ignore_index=True)
            if rows
            else pd.DataFrame(columns=["perturbation_id", "observation_id", "aoi_assignment"])
        )
    long = long.dropna(subset=["aoi_assignment"])
    if long.empty:
        return pd.DataFrame(
            columns=[
                "observation_id",
                "aoi",
                "assignment_count",
                "n_perturbations",
                "assignment_frequency",
            ]
        )
    counts = (
        long.groupby(["observation_id", "aoi_assignment"], dropna=False)
        .size()
        .rename("assignment_count")
        .reset_index()
    )
    denominators = (
        long.groupby("observation_id")["perturbation_id"]
        .nunique()
        .rename("n_perturbations")
        .reset_index()
    )
    out = counts.merge(denominators, on="observation_id", how="left")
    out["assignment_frequency"] = out["assignment_count"] / out["n_perturbations"]
    out = out.rename(columns={"aoi_assignment": "aoi"})
    out.attrs["caveat"] = (
        "Assignment frequency is a descriptive perturbation frequency, not a posterior probability of true AOI membership."
    )
    return out

recompute_aoi_features ¶

recompute_aoi_features(data: DataFrame, assignments: Sequence[Any], *, participant_col: str | None = None, trial_col: str | None = None, duration_col: str | None = None, time_col: str | None = None, perturbation_id: str | None = None, aoi_levels: Sequence[str] | None = None, observation_level: str = 'fixation') -> pd.DataFrame

Recompute AOI features without dropping observed zero-assignment cells.

A participant/trial × AOI cell receives a structural zero count only when the group contains at least one non-missing AOI assignment opportunity. Groups whose assignments are all missing retain NA counts/inspection. Dwell is NA when an assigned observation has missing duration, and first-fixation timing is never inferred from row order.

Source code in src/eyeprocesspy/_aoi_assignment_core.py

def recompute_aoi_features(
    data: pd.DataFrame,
    assignments: Sequence[Any],
    *,
    participant_col: str | None = None,
    trial_col: str | None = None,
    duration_col: str | None = None,
    time_col: str | None = None,
    perturbation_id: str | None = None,
    aoi_levels: Sequence[str] | None = None,
    observation_level: str = "fixation",
) -> pd.DataFrame:
    """Recompute AOI features without dropping observed zero-assignment cells.

    A participant/trial × AOI cell receives a structural zero count only when
    the group contains at least one non-missing AOI assignment opportunity.
    Groups whose assignments are all missing retain NA counts/inspection.
    Dwell is NA when an assigned observation has missing duration, and
    first-fixation timing is never inferred from row order.
    """
    frame = _frame(data, "data")
    observation_level = str(observation_level).lower()
    if observation_level not in {"fixation", "sample"}:
        raise EyeProcessValidationError("`observation_level` must be 'fixation' or 'sample'.")
    if len(assignments) != len(frame):
        raise EyeProcessValidationError("`assignments` must contain one value per data row.")
    frame = frame.copy()
    frame["aoi_assignment"] = list(assignments)
    group_cols = [c for c in (participant_col, trial_col) if c is not None]
    missing_groups = [c for c in group_cols if c not in frame.columns]
    if missing_groups:
        raise EyeProcessValidationError("Missing grouping column(s): " + ", ".join(missing_groups))
    if duration_col is not None and duration_col not in frame.columns:
        raise EyeProcessValidationError(f"`duration_col` `{duration_col}` is absent.")
    if time_col is not None and time_col not in frame.columns:
        raise EyeProcessValidationError(f"`time_col` `{time_col}` is absent.")
    if duration_col is None:
        warnings.warn(
            "No `duration_col` supplied; dwell is returned as NA rather than inferred.",
            RuntimeWarning,
            stacklevel=2,
        )
    if time_col is None:
        warnings.warn(
            "No `time_col` supplied; first_fixation is returned as NA rather than inferred from row order.",
            RuntimeWarning,
            stacklevel=2,
        )

    if aoi_levels is None:
        inferred = (
            frame.loc[
                frame["aoi_assignment"].notna()
                & ~frame["aoi_assignment"].isin([OUTSIDE, AMBIGUOUS]),
                "aoi_assignment",
            ]
            .astype(str)
            .drop_duplicates()
            .tolist()
        )
        levels = inferred
    else:
        levels = [str(v) for v in aoi_levels]
        if any(not v for v in levels) or len(set(levels)) != len(levels):
            raise EyeProcessValidationError(
                "`aoi_levels` must contain unique non-empty AOI identifiers."
            )

    columns = group_cols + [
        "aoi",
        "observation_level",
        "observation_count",
        "fixation_count",
        "sample_count",
        "dwell",
        "first_observation",
        "first_fixation",
        "inspected",
        "n_valid_observations",
        "n_missing_observations",
        "duration_complete",
        "time_complete",
        "perturbation_id",
    ]
    if not levels:
        return pd.DataFrame(columns=columns)

    grouped: Any
    if group_cols:
        grouper = group_cols[0] if len(group_cols) == 1 else group_cols
        grouped = frame.groupby(grouper, dropna=False, sort=True)
    else:
        grouped = [((), frame)]

    rows: list[dict[str, Any]] = []
    for key, group in grouped:
        if len(group_cols) == 1:
            key = (key,)
        elif not group_cols:
            key = ()
        group_values = {k: v for k, v in zip(group_cols, key)}
        valid_assignment = group["aoi_assignment"].notna()
        n_valid = int(valid_assignment.sum())
        n_missing = int((~valid_assignment).sum())

        for aoi in levels:
            row: dict[str, Any] = dict(group_values)
            row["aoi"] = aoi
            selected = group.loc[group["aoi_assignment"].astype("string").eq(aoi).fillna(False)]
            count = int(len(selected))

            row["observation_level"] = observation_level
            if n_valid == 0:
                row["observation_count"] = pd.NA
                row["fixation_count"] = pd.NA
                row["sample_count"] = pd.NA
                row["dwell"] = np.nan
                row["first_observation"] = np.nan
                row["first_fixation"] = np.nan
                row["inspected"] = pd.NA
                row["duration_complete"] = pd.NA
                row["time_complete"] = pd.NA
            else:
                row["observation_count"] = count
                row["fixation_count"] = count if observation_level == "fixation" else pd.NA
                row["sample_count"] = count if observation_level == "sample" else pd.NA
                row["inspected"] = bool(count > 0)
                if duration_col is None:
                    row["dwell"] = np.nan
                    row["duration_complete"] = pd.NA
                elif count == 0:
                    row["dwell"] = 0.0
                    row["duration_complete"] = True
                else:
                    durations = pd.to_numeric(selected[duration_col], errors="coerce")
                    duration_complete = bool(durations.notna().all())
                    row["duration_complete"] = duration_complete
                    row["dwell"] = float(durations.sum()) if duration_complete else np.nan

                if time_col is None:
                    row["first_observation"] = np.nan
                    row["first_fixation"] = np.nan
                    row["time_complete"] = pd.NA
                elif count == 0:
                    row["first_observation"] = np.nan
                    row["first_fixation"] = np.nan
                    row["time_complete"] = True
                else:
                    times = pd.to_numeric(selected[time_col], errors="coerce")
                    time_complete = bool(times.notna().all())
                    first_observation = float(times.min()) if times.notna().any() else np.nan
                    row["time_complete"] = time_complete
                    row["first_observation"] = first_observation
                    row["first_fixation"] = (
                        first_observation if observation_level == "fixation" else np.nan
                    )

            row["n_valid_observations"] = n_valid
            row["n_missing_observations"] = n_missing
            row["perturbation_id"] = perturbation_id
            rows.append(row)

    out = pd.DataFrame(rows, columns=columns)
    for count_col in ("observation_count", "fixation_count", "sample_count"):
        out[count_col] = out[count_col].astype("Int64")
    out["inspected"] = out["inspected"].astype("boolean")
    out["duration_complete"] = out["duration_complete"].astype("boolean")
    out["time_complete"] = out["time_complete"].astype("boolean")
    return out

_frame ¶

_frame(x: Any, name: str) -> pd.DataFrame

Source code in src/eyeprocesspy/_aoi_geometry_primitives.py

def _frame(x: Any, name: str) -> pd.DataFrame:
    if not isinstance(x, pd.DataFrame):
        raise EyeProcessValidationError(f"`{name}` must be a pandas DataFrame.")
    return x.copy()

_software_provenance ¶

_software_provenance() -> dict[str, Any]

Source code in src/eyeprocesspy/_aoi_geometry_primitives.py

def _software_provenance() -> dict[str, Any]:
    package_version: str | None

    try:
        from . import __version__ as package_version
    except Exception:  # pragma: no cover - defensive when module is executed standalone
        package_version = None

    return {
        "package": "eyeprocesspy",
        "version": package_version,
        "module": __name__,
    }

_stable_frame_hash ¶

_stable_frame_hash(frame: DataFrame) -> str

Source code in src/eyeprocesspy/_aoi_geometry_primitives.py

def _stable_frame_hash(frame: pd.DataFrame) -> str:
    def normalise(v: Any) -> Any:
        if isinstance(v, np.ndarray):
            return v.tolist()
        if isinstance(v, (list, tuple)):
            return [normalise(z) for z in v]
        if isinstance(v, Mapping):
            return {str(k): normalise(val) for k, val in sorted(v.items(), key=lambda z: str(z[0]))}
        if pd.isna(v) if np.isscalar(v) else False:
            return None
        if isinstance(v, (np.integer, np.floating)):
            return v.item()
        return v

    records = []
    for row in frame.to_dict(orient="records"):
        records.append({str(k): normalise(v) for k, v in sorted(row.items())})
    blob = json.dumps(records, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode(
        "utf-8"
    )
    return hashlib.sha256(blob).hexdigest()

_spec_value ¶

_spec_value(spec: Any, key: str) -> Any

Source code in src/eyeprocesspy/_aoi_geometry_units.py

def _spec_value(spec: Any, key: str) -> Any:
    if isinstance(spec, Mapping):
        return spec[key]
    try:
        return spec[key]
    except Exception as exc:
        raise EyeProcessValidationError(
            "Perturbation specifications must be mapping-like objects created by `aoi_perturbation_spec()`."
        ) from exc

aoi_perturbation_spec ¶

aoi_perturbation_spec(perturbation_id: str, operation: str = 'baseline', *, margin_x: float = 0.0, margin_y: float | None = None, translation_x: float = 0.0, translation_y: float = 0.0, unit: str = 'px', screen_width_px: int | None = None, screen_height_px: int | None = None, viewing_distance: float | None = None, physical_screen_size: Sequence[float] | None = None, degrees_per_pixel: Sequence[float] | None = None, seed: int | None = None, boundary_policy: str = 'warn') -> EyeResult

Create a validated, reproducible AOI perturbation specification.

Source code in src/eyeprocesspy/_aoi_geometry_units.py

def aoi_perturbation_spec(
    perturbation_id: str,
    operation: str = "baseline",
    *,
    margin_x: float = 0.0,
    margin_y: float | None = None,
    translation_x: float = 0.0,
    translation_y: float = 0.0,
    unit: str = "px",
    screen_width_px: int | None = None,
    screen_height_px: int | None = None,
    viewing_distance: float | None = None,
    physical_screen_size: Sequence[float] | None = None,
    degrees_per_pixel: Sequence[float] | None = None,
    seed: int | None = None,
    boundary_policy: str = "warn",
) -> EyeResult:
    """Create a validated, reproducible AOI perturbation specification."""
    if not isinstance(perturbation_id, str) or not perturbation_id.strip():
        raise EyeProcessValidationError("`perturbation_id` must be a non-empty string.")
    operation = str(operation).lower()
    operations = {"baseline", "dilation", "erosion", "translate", "jitter", "anisotropic_expansion"}
    if operation not in operations:
        raise EyeProcessValidationError(f"Unsupported perturbation operation `{operation}`.")
    unit = str(unit).lower()
    if unit not in {"px", "deg"}:
        raise EyeProcessValidationError("`unit` must be 'px' or 'deg'.")
    boundary_policy = str(boundary_policy).lower()
    if boundary_policy not in {"warn", "clip", "error", "allow"}:
        raise EyeProcessValidationError("`boundary_policy` must be warn, clip, error, or allow.")
    mx = _finite_scalar(margin_x, "margin_x")
    my = mx if margin_y is None else _finite_scalar(margin_y, "margin_y")
    tx = _finite_scalar(translation_x, "translation_x")
    ty = _finite_scalar(translation_y, "translation_y")
    if operation == "dilation" and (mx < 0 or my < 0):
        raise EyeProcessValidationError("Dilation margins must be non-negative.")
    if operation == "erosion" and (mx < 0 or my < 0):
        raise EyeProcessValidationError(
            "Erosion margins must be non-negative; erosion direction is implied by the operation."
        )
    if operation in {"dilation", "erosion"} and not math.isclose(mx, my, rel_tol=0, abs_tol=1e-12):
        raise EyeProcessValidationError(
            "Polygon-safe dilation/erosion uses a uniform margin; use anisotropic_expansion for different x/y margins."
        )
    resolved_dpp: tuple[float, float] | None = None
    if degrees_per_pixel is not None:
        resolved_dpp = _normalise_pair(degrees_per_pixel, "degrees_per_pixel", nonnegative=True)
        if min(resolved_dpp) <= 0:
            raise EyeProcessValidationError("`degrees_per_pixel` values must be positive.")
    elif all(
        v is not None
        for v in (screen_width_px, screen_height_px, viewing_distance, physical_screen_size)
    ):
        assert screen_width_px is not None
        assert screen_height_px is not None
        assert viewing_distance is not None
        assert physical_screen_size is not None

        _, _, dx, dy = _screen_geometry(
            screen_width_px=int(screen_width_px),
            screen_height_px=int(screen_height_px),
            viewing_distance=float(viewing_distance),
            physical_screen_size=physical_screen_size,
        )
        resolved_dpp = (dx, dy)
    if unit == "deg" and resolved_dpp is None:
        raise EyeProcessValidationError(
            "Degree-based perturbations require `degrees_per_pixel` or complete screen geometry and viewing distance."
        )
    if seed is not None:
        if isinstance(seed, bool) or int(seed) != seed or int(seed) < 0:
            raise EyeProcessValidationError("`seed` must be a non-negative integer or None.")
        seed = int(seed)
    return _result(
        "eye_aoi_perturbation_spec",
        perturbation_id=perturbation_id,
        operation=operation,
        margin_x=mx,
        margin_y=my,
        translation_x=tx,
        translation_y=ty,
        unit=unit,
        screen_width_px=None if screen_width_px is None else int(screen_width_px),
        screen_height_px=None if screen_height_px is None else int(screen_height_px),
        viewing_distance=viewing_distance,
        physical_screen_size=None
        if physical_screen_size is None
        else tuple(float(v) for v in physical_screen_size),
        degrees_per_pixel=resolved_dpp,
        seed=seed,
        boundary_policy=boundary_policy,
        software=_software_provenance(),
    )

convert_aoi_margin_to_degrees ¶

convert_aoi_margin_to_degrees(margin_pixels: Any, *, screen_width_px: int, screen_height_px: int, viewing_distance: float, physical_screen_size: Sequence[float]) -> tuple[float, float]

Convert pixel margins to degrees of visual angle on x and y axes.

Source code in src/eyeprocesspy/_aoi_geometry_units.py

def convert_aoi_margin_to_degrees(
    margin_pixels: Any,
    *,
    screen_width_px: int,
    screen_height_px: int,
    viewing_distance: float,
    physical_screen_size: Sequence[float],
) -> tuple[float, float]:
    """Convert pixel margins to degrees of visual angle on x and y axes."""
    mx, my = _normalise_pair(margin_pixels, "margin_pixels", nonnegative=False)
    _, _, dx, dy = _screen_geometry(
        screen_width_px=screen_width_px,
        screen_height_px=screen_height_px,
        viewing_distance=viewing_distance,
        physical_screen_size=physical_screen_size,
    )
    return (mx * dx, my * dy)

convert_aoi_margin_to_pixels ¶

convert_aoi_margin_to_pixels(margin_degrees: Any, *, screen_width_px: int, screen_height_px: int, viewing_distance: float, physical_screen_size: Sequence[float]) -> tuple[float, float]

Convert visual-angle margins to pixels on x and y axes.

Source code in src/eyeprocesspy/_aoi_geometry_units.py

def convert_aoi_margin_to_pixels(
    margin_degrees: Any,
    *,
    screen_width_px: int,
    screen_height_px: int,
    viewing_distance: float,
    physical_screen_size: Sequence[float],
) -> tuple[float, float]:
    """Convert visual-angle margins to pixels on x and y axes."""
    mx, my = _normalise_pair(margin_degrees, "margin_degrees", nonnegative=False)
    _, _, dx, dy = _screen_geometry(
        screen_width_px=screen_width_px,
        screen_height_px=screen_height_px,
        viewing_distance=viewing_distance,
        physical_screen_size=physical_screen_size,
    )
    return (mx / dx, my / dy)

validate_aoi_geometry ¶

validate_aoi_geometry(aois: DataFrame, *, allow_overlap: bool = True) -> EyeResult

Validate rectangular and polygonal AOI geometry without resolving overlap.

Source code in src/eyeprocesspy/_aoi_geometry_validation.py

def validate_aoi_geometry(aois: pd.DataFrame, *, allow_overlap: bool = True) -> EyeResult:
    """Validate rectangular and polygonal AOI geometry without resolving overlap."""
    frame = _frame(aois, "aois")
    if frame.empty:
        raise EyeProcessValidationError("`aois` must contain at least one AOI.")
    id_col = next((c for c in ("aoi_id", "aoi", "name", "label") if c in frame.columns), None)
    if id_col is None:
        raise EyeProcessValidationError(
            "`aois` must include an AOI identifier column such as `aoi_id`."
        )
    ids = frame[id_col].astype("string")
    if ids.isna().any() or (ids.str.len() == 0).any():
        raise EyeProcessValidationError("AOI identifiers must be non-missing and non-empty.")
    if ids.duplicated().any():
        raise EyeProcessValidationError(
            "AOI identifiers must be unique within an AOI specification."
        )

    rows: list[dict[str, Any]] = []
    normalized = frame.copy()
    normalized["aoi_id"] = ids.astype(str)
    for coordinate_column in ("xmin", "xmax", "ymin", "ymax"):
        if coordinate_column in normalized.columns:
            normalized[coordinate_column] = pd.to_numeric(
                normalized[coordinate_column], errors="coerce"
            ).astype(float)
    normalized["shape_type"] = [_infer_shape(row) for _, row in normalized.iterrows()]
    if "polygon" not in normalized.columns:
        normalized["polygon"] = [None] * len(normalized)

    for i, row in normalized.iterrows():
        aoi_id = row["aoi_id"]
        shape = row["shape_type"]
        if shape == "rectangle":
            required = ["xmin", "xmax", "ymin", "ymax"]
            missing = [c for c in required if c not in normalized.columns]
            if missing:
                raise EyeProcessValidationError(
                    "Rectangular AOIs require xmin, xmax, ymin, and ymax columns."
                )
            vals = {c: _finite_scalar(row[c], f"{aoi_id}.{c}") for c in required}
            if not vals["xmin"] < vals["xmax"] or not vals["ymin"] < vals["ymax"]:
                raise EyeProcessValidationError(
                    f"AOI `{aoi_id}` has zero or negative rectangle area."
                )
            area = (vals["xmax"] - vals["xmin"]) * (vals["ymax"] - vals["ymin"])
            normalized.loc[i, required] = [vals[c] for c in required]
            rows.append({"aoi_id": aoi_id, "shape_type": shape, "area": area, "valid": True})
        else:
            poly = _polygon_array(row["polygon"])
            if _polygon_self_intersects(poly):
                raise EyeProcessValidationError(f"AOI `{aoi_id}` polygon self-intersects.")
            area = abs(_signed_polygon_area(poly))
            if area <= 1e-12:
                raise EyeProcessValidationError(f"AOI `{aoi_id}` polygon has zero area.")
            normalized.at[i, "polygon"] = poly
            rows.append({"aoi_id": aoi_id, "shape_type": shape, "area": area, "valid": True})

    overlap = _pairwise_overlap(normalized)
    overlapping = overlap.loc[overlap["overlap"]]
    if len(overlapping) and not allow_overlap:
        pairs = ", ".join(f"{r.aoi_1}/{r.aoi_2}" for r in overlapping.itertuples())
        raise EyeProcessValidationError(f"AOIs overlap but `allow_overlap=False`: {pairs}.")

    return _result(
        "eye_aoi_geometry_validation",
        geometry=normalized.reset_index(drop=True),
        audit=pd.DataFrame(rows),
        overlap=overlap,
        overlap_present=bool(len(overlapping)),
        source_hash=_stable_frame_hash(normalized),
        software=_software_provenance(),
        status="valid_with_overlap" if len(overlapping) else "valid",
    )