data

feat.data

Py-FEAT Data classes.

Fex

Bases: DataFrame

Fex is a class to represent facial expression (Fex) data

Fex class is an enhanced pandas dataframe, with extra attributes and methods to help with facial expression data analysis.

Parameters:

Name	Type	Description	Default
filename		(str, optional) path to file	required
detector		(str, optional) name of software used to extract Fex. Currently only	required
sampling_freq	float	sampling rate of each row in Hz; defaults to None	required
features	Dataframe	features that correspond to each Fex row	required
sessions		Unique values indicating rows associated with a specific session (e.g., trial, subject, etc).Must be a 1D array of n_samples elements; defaults to None	required

Source code in feat/data.py

class Fex(DataFrame):
    """Fex is a class to represent facial expression (Fex) data

    Fex class is  an enhanced pandas dataframe, with extra attributes and methods to help with facial expression data analysis.

    Args:
        filename: (str, optional) path to file
        detector: (str, optional) name of software used to extract Fex. Currently only
        'Feat' is supported
        sampling_freq (float, optional): sampling rate of each row in Hz; defaults to None
        features (pd.Dataframe, optional): features that correspond to each Fex row
        sessions: Unique values indicating rows associated with a specific session (e.g., trial, subject, etc).Must be a 1D array of n_samples elements; defaults to None
    """

    _metadata = [
        "au_columns",
        "emotion_columns",
        "facebox_columns",
        "landmark_columns",
        "facepose_columns",
        "identity_columns",
        "gaze_columns",
        "blendshape_columns",
        "time_columns",
        "design_columns",
        "fex_columns",
        "filename",
        "sampling_freq",
        "features",
        "sessions",
        "detector",
        "face_model",
        "landmark_model",
        "au_model",
        "emotion_model",
        "facepose_model",
        "identity_model",
        "gaze_model",
        "verbose",
    ]

    def __finalize__(self, other, method=None, **kwargs):
        """Propagate metadata from other to self"""
        self = super().__finalize__(other, method=method, **kwargs)
        # merge operation: using metadata of the left object
        if method == "merge":
            for name in self._metadata:
                object.__setattr__(self, name, getattr(other.left, name, None))
        # concat operation: using metadata of the first object
        elif method == "concat":
            for name in self._metadata:
                object.__setattr__(self, name, getattr(other.objs[0], name, None))
        return self

    def __init__(self, *args, **kwargs):
        ### Columns ###
        self.au_columns = kwargs.pop("au_columns", None)
        self.emotion_columns = kwargs.pop("emotion_columns", None)
        self.facebox_columns = kwargs.pop("facebox_columns", None)
        self.landmark_columns = kwargs.pop("landmark_columns", None)
        self.facepose_columns = kwargs.pop("facepose_columns", None)
        self.identity_columns = kwargs.pop("identity_columns", None)
        self.gaze_columns = kwargs.pop("gaze_columns", None)
        self.blendshape_columns = kwargs.pop("blendshape_columns", None)
        self.time_columns = kwargs.pop("time_columns", None)
        self.design_columns = kwargs.pop("design_columns", None)

        ### Meta data ###
        self.filename = kwargs.pop("filename", None)
        self.sampling_freq = kwargs.pop("sampling_freq", None)
        self.detector = kwargs.pop("detector", None)
        self.face_model = kwargs.pop("face_model", None)
        self.landmark_model = kwargs.pop("landmark_model", None)
        self.au_model = kwargs.pop("au_model", None)
        self.emotion_model = kwargs.pop("emotion_model", None)
        self.facepose_model = kwargs.pop("facepose_model", None)
        self.identity_model = kwargs.pop("identity_model", None)
        self.gaze_model = kwargs.pop("gaze_model", None)
        self.features = kwargs.pop("features", None)
        self.sessions = kwargs.pop("sessions", None)

        self.verbose = kwargs.pop("verbose", False)

        super().__init__(*args, **kwargs)
        if self.sessions is not None:
            if not len(self.sessions) == len(self):
                raise ValueError("Make sure sessions is same length as data.")
            self.sessions = np.array(self.sessions)

    @property
    def _constructor(self):
        return Fex

    @property
    def _constructor_sliced(self):
        """
        Propagating custom metadata from sub-classed dfs to sub-classed series is not
        automatically handled. See: https://github.com/pandas-dev/pandas/issues/19850
        _constructor_sliced (which dataframes call when their return type is a series
        can only return a function definition or class definition. So to make sure we
        propagate attributes from Fex -> FexSeries we define another
        function that calls .__finalize__ on the returned FexSeries

        Inspired by how GeoPandas subclasses dataframes. See their _constructor_sliced
        here:
        https://github.com/geopandas/geopandas/blob/2eac5e212a7e2ebbca71f35707a2a196e4b09527/geopandas/geodataframe.py#L1460

        And their constructor function here: https://github.com/geopandas/geopandas/blob/2eac5e212a7e2ebbca71f35707a2a196e4b09527/geopandas/geoseries.py#L31
        """

        def _fexseries_constructor(*args, **kwargs):
            return FexSeries(*args, **kwargs).__finalize__(self)

        return _fexseries_constructor

    @property
    def aus(self):
        """Returns the Action Units data

        Returns Action Unit data using the columns set in fex.au_columns.

        Returns:
            DataFrame: Action Units data
        """
        return self[self.au_columns]

    @property
    def emotions(self):
        """Returns the emotion data

        Returns emotions data using the columns set in fex.emotion_columns.

        Returns:
            DataFrame: emotion data
        """
        return self[self.emotion_columns]

    @property
    def landmarks(self):
        """Returns the landmark data

        Returns landmark data using the columns set in fex.landmark_columns.

        Returns:
            DataFrame: landmark data
        """
        return self[self.landmark_columns]

    @property
    def landmark(self):
        """Returns the landmark data

        Returns landmark data using the columns set in fex.landmark_columns.

        Returns:
            DataFrame: landmark data
        """
        warnings.warn(
            "Fex.landmark has now been renamed to Fex.landmarks", DeprecationWarning
        )
        return self[self.landmark_columns]

    def landmarks_dlib68_xy(self, row=None):
        """Return (x, y) arrays in dlib-68 layout, regardless of source detector.

        Provides a uniform 2D-landmark interface across detectors so plotting
        and downstream consumers don't have to branch on the underlying topology:

        - **Detectorv1** (``Detectorv1(landmark_model="mobilefacenet")``) outputs
          136-d ``[x_0..x_67, y_0..y_67]`` (axis-major) — returned as-is.
        - **MPDetector** outputs the 478-vertex MediaPipe FaceMesh as 1434
          named columns ``x_i / y_i / z_i`` (i in 0..477), in image-space
          pixels. The 68 dlib-equivalent vertices are sampled by name via
          ``DLIB68_FROM_MP478`` (see ``feat.utils.blendshape_to_au``); z dropped.

        Args:
            row: optional pandas Series (single Fex row). If ``None``, operates
                on all rows of ``self`` and returns shape ``(n_rows, 68)`` per
                axis. Otherwise returns shape ``(68,)`` per axis.

        Returns:
            (x, y): two numpy arrays of dlib-68 layout. Shapes ``(68,)`` if
                ``row`` is given, else ``(n_rows, 68)``.
        """
        if self.landmark_columns is None:
            raise ValueError("landmark_columns is not set on this Fex")
        n_lm_cols = len(self.landmark_columns)
        source = self if row is None else row

        if n_lm_cols == 136:
            arr = source[self.landmark_columns].values.astype(np.float64)
            if row is None:
                return arr[:, :68], arr[:, 68:]
            return arr[:68], arr[68:]
        elif n_lm_cols == 1434:
            x_cols = [f"x_{i}" for i in DLIB68_FROM_MP478]
            y_cols = [f"y_{i}" for i in DLIB68_FROM_MP478]
            x = source[x_cols].values.astype(np.float64)
            y = source[y_cols].values.astype(np.float64)
            return x, y
        else:
            raise ValueError(
                f"Unexpected landmark_columns length ({n_lm_cols}); expected 136 "
                "(dlib-68) or 1434 (MP-478 mesh)."
            )

    @property
    def poses(self):
        """Returns the facepose data using the columns set in fex.facepose_columns

        Returns:
            DataFrame: facepose data
        """
        return self[self.facepose_columns]

    @property
    def gazes(self):
        """Returns the gaze data (pitch, yaw, optionally combined angle).

        Returns:
            DataFrame: gaze data populated by the active gaze model
            (L2CS by default in v0.7+). Raises AttributeError if the
            Detectorv1 was built with gaze_model=None.
        """
        if not self.gaze_columns:
            raise AttributeError(
                "No gaze columns are registered on this Fex. The active "
                "Detectorv1 may have been built with gaze_model=None."
            )
        return self[self.gaze_columns]

    @property
    def blendshapes(self):
        """Returns the 52 MediaPipe/ARKit blendshape coefficients in [0, 1].

        Raises AttributeError if no blendshape columns are registered (e.g. a
        pre-v2.5 multitask model, or DetectorV1, which has no blendshape head)."""
        if not self.blendshape_columns:
            raise AttributeError(
                "No blendshape columns are registered on this Fex. The active "
                "model may predate v2.5 (no blendshape head)."
            )
        return self[self.blendshape_columns]

    # DEPRECATE
    @property
    def facepose(self):
        """Returns the facepose data using the columns set in fex.facepose_columns

        Returns:
            DataFrame: facepose data
        """

        with warnings.catch_warnings():
            warnings.simplefilter("always", DeprecationWarning)
            warnings.warn(
                "Fex.facepose has now been renamed to Fex.poses", DeprecationWarning
            )

        return self[self.facepose_columns]

    @property
    def inputs(self):
        """Returns input column as string

        Returns input data in the "input" column.

        Returns:
            string: path to input image
        """
        return self["input"]

    # DEPRECATE
    @property
    def input(self):
        """Returns input column as string

        Returns input data in the "input" column.

        Returns:
            string: path to input image
        """

        with warnings.catch_warnings():
            warnings.simplefilter("always", DeprecationWarning)
            warnings.warn(
                "Fex.input has now been renamed to Fex.inputs", DeprecationWarning
            )

        return self["input"]

    @property
    def landmarks_x(self):
        """Returns the x landmarks.

        Returns:
            DataFrame: x landmarks.
        """
        x_cols = [col for col in self.landmark_columns if "x" in col]
        return self[x_cols]

    # DEPRECATE
    @property
    def landmark_x(self):
        """Returns the x landmarks.

        Returns:
            DataFrame: x landmarks.
        """

        with warnings.catch_warnings():
            warnings.simplefilter("always", DeprecationWarning)
            warnings.warn(
                "Fex.landmark_x has been renamed to Fex.landmarks_x", DeprecationWarning
            )
        x_cols = [col for col in self.landmark_columns if "x" in col]
        return self[x_cols]

    @property
    def landmarks_y(self):
        """Returns the y landmarks.

        Returns:
            DataFrame: y landmarks.
        """
        y_cols = [col for col in self.landmark_columns if "y" in col]
        return self[y_cols]

    # DEPRECATE
    @property
    def landmark_y(self):
        """Returns the y landmarks.

        Returns:
            DataFrame: y landmarks.
        """

        with warnings.catch_warnings():
            warnings.simplefilter("always", DeprecationWarning)
            warnings.warn(
                "Fex.landmark_y has been renamed to Fex.landmarks_y", DeprecationWarning
            )

        y_cols = [col for col in self.landmark_columns if "y" in col]
        return self[y_cols]

    @property
    def faceboxes(self):
        """Returns the facebox data

        Returns:
            DataFrame: facebox data
        """
        return self[self.facebox_columns]

    # DEPRECATE
    @property
    def facebox(self):
        """Returns the facebox data

        Returns:
            DataFrame: facebox data
        """

        with warnings.catch_warnings():
            warnings.simplefilter("always", DeprecationWarning)
            warnings.warn(
                "Fex.facebox has been renamed to Fex.faceboxes", DeprecationWarning
            )

        return self[self.facebox_columns]

    @property
    def identities(self):
        """Returns the identity labels

        Returns:
            DataFrame: identity data
        """
        return self.Identity

    @property
    def identity_embeddings(self):
        """Returns the identity embeddings

        Returns:
            DataFrame: identity data
        """
        # Detectorv1 populates identity_columns with the 512 embedding cols
        # (FEAT_IDENTITY_COLUMNS[1:] — the 'Identity' string-label column is
        # already stripped upstream in detector.py), so don't slice again.
        # The prior [1:] silently dropped Identity_1 and caused
        # cluster_identities to error on the resulting 511-col frame.
        return self[self.identity_columns]

    @property
    def time(self):
        """Returns the time data

        Returns the time information using fex.time_columns.

        Returns:
            DataFrame: time data
        """
        return self[self.time_columns]

    @property
    def design(self):
        """Returns the design data

        Returns the study design information using columns in fex.design_columns.

        Returns:
            DataFrame: time data
        """
        return self[self.design_columns]

    def read_file(self):
        """Loads file into FEX class

        Returns:
            DataFrame: Fex class
        """
        if self.detector == "OpenFace":
            return self.read_openface(self.filename)

        return self.read_feat(self.filename)

    @property
    def info(self):
        """Print all meta data of fex

        Loops through metadata set in self._metadata and prints out the information.
        """
        attr_list = []
        for name in self._metadata:
            attr_list.append(name + ": " + str(getattr(self, name, None)) + "\n")
        print(f"{self.__class__}\n" + "".join(attr_list))

    def _update_extracted_colnames(self, prefix=None, mode="replace"):
        cols2update = [
            "au_columns",
            "emotion_columns",
            "facebox_columns",
            "landmark_columns",
            "facepose_columns",
            "blendshape_columns",  # [0,1] coefficients aggregate like AUs
            # "gaze_columns" intentionally excluded — gaze isn't summarized
            # the same way the other metric groups are (no aggregation
            # makes sense over pitch/yaw radians per session).
            "time_columns",
        ]

        # Existing columns to handle different __init__s
        cols2update = list(
            filter(lambda col: getattr(self, col) is not None, cols2update)
        )
        original_vals = [getattr(self, c) for c in cols2update]

        # Ignore prefix and remove any existing
        if mode == "reset":
            new_vals = [
                list(map(lambda name: "".join(name.split("_")[1:]), names))
                for names in original_vals
            ]
            _ = [setattr(self, col, val) for col, val in zip(cols2update, new_vals)]
            return

        if not isinstance(prefix, list):
            prefix = [prefix]

        for i, p in enumerate(prefix):
            current_vals = [getattr(self, c) for c in cols2update]
            new_vals = [
                list(map(lambda name: f"{p}_{name}", names)) for names in original_vals
            ]
            if i == 0:
                update = new_vals
            else:
                update = [current + new for current, new in zip(current_vals, new_vals)]
            _ = [setattr(self, col, val) for col, val in zip(cols2update, update)]

    def _parse_features_labels(self, X, y):
        feature_groups = [
            "sessions",
            "emotions",
            "aus",
            "poses",
            "landmarks",
            "faceboxes",
        ]

        # String attribute access
        if isinstance(X, str) and any(map(lambda feature: feature in X, feature_groups)):
            X = X.split(",") if "," in X else [X]
            mX = []
            for x in X:
                mX.append(getattr(self, x))

            mX = pd.concat(mX, axis=1)
            if X == ["sessions"]:
                mX.columns = X

        elif isinstance(X, list):
            mX = self[X]
        else:
            mX = X

        if isinstance(y, str) and any(map(lambda feature: feature in y, feature_groups)):
            y = y.split(",") if "," in y else [y]
            my = []
            for yy in y:
                my.append(getattr(self, yy))

            my = pd.concat(my, axis=1)
            if y == ["sessions"]:
                my.columns = y

        elif isinstance(y, list):
            my = self[y]
        else:
            my = y

        return mX, my

    ###   Class Methods   ###
    def read_feat(self, filename=None, *args, **kwargs):
        """Reads facial expression detection results from Feat Detectorv1

        Args:
            filename (string, optional): Path to file. Defaults to None.

        Returns:
            Fex
        """
        # Check if filename exists in metadata.
        if filename is None:
            if self.filename:
                filename = self.filename
            else:
                raise ValueError("filename must be specified.")
        result = read_feat(filename, *args, **kwargs)
        return result

    def read_openface(self, filename=None, *args, **kwargs):
        """Reads facial expression detection results from OpenFace

        Args:
            filename (string, optional): Path to file. Defaults to None.

        Returns:
            Fex
        """
        if filename is None:
            if self.filename:
                filename = self.filename
            else:
                raise ValueError("filename must be specified.")
        result = read_openface(filename, *args, **kwargs)
        for name in self._metadata:
            attr_value = getattr(self, name, None)
            if attr_value and getattr(result, name, None) is None:
                setattr(result, name, attr_value)
        return result

    def itersessions(self):
        """Iterate over Fex sessions as (session, series) pairs.

        Returns:
            it: a generator that iterates over the sessions of the fex instance

        """
        for x in np.unique(self.sessions):
            yield x, self.loc[self.sessions == x, :]

    def append(self, data, session_id=None, axis=0):
        """Append a new Fex object to an existing object

        Args:
            data: (Fex) Fex instance to append
            session_id: session label
            axis: ([0,1]) Axis to append. Rows=0, Cols=1
        Returns:
            Fex instance
        """
        if not isinstance(data, self.__class__):
            raise ValueError("Make sure data is a Fex instance.")

        if self.empty:
            out = data.copy()
            if session_id is not None:
                out.sessions = np.repeat(session_id, len(data))
        else:
            if self.sampling_freq != data.sampling_freq:
                raise ValueError(
                    "Make sure Fex objects have the same " "sampling frequency"
                )
            if axis == 0:
                out = self.__class__(
                    pd.concat([self, data], axis=axis, ignore_index=True),
                    sampling_freq=self.sampling_freq,
                ).__finalize__(self)
                if session_id is not None:
                    out.sessions = np.hstack(
                        [self.sessions, np.repeat(session_id, len(data))]
                    )
                if self.features is not None:
                    if data.features is not None:
                        if self.features.shape[1] == data.features.shape[1]:
                            out.features = pd.concat(
                                [self.features, data.features], ignore_index=True
                            )
                        else:
                            raise ValueError(
                                "Different number of features in new dataset."
                            )
                    else:
                        out.features = self.features
                elif data.features is not None:
                    out.features = data.features
            elif axis == 1:
                out = self.__class__(
                    pd.concat([self, data], axis=axis), sampling_freq=self.sampling_freq
                ).__finalize__(self)
                if self.sessions is not None:
                    if data.sessions is not None:
                        if np.array_equal(self.sessions, data.sessions):
                            out.sessions = self.sessions
                        else:
                            raise ValueError("Both sessions must be identical.")
                    else:
                        out.sessions = self.sessions
                elif data.sessions is not None:
                    out.sessions = data.sessions
                if self.features is not None:
                    out.features = self.features
                    if data.features is not None:
                        out.features = pd.concat(
                            [self.features, data.features], axis=1
                        )
                elif data.features is not None:
                    out.features = data.features
            else:
                raise ValueError("Axis must be 1 or 0.")
        return out

    def regress(self, X, y, fit_intercept=True, *args, **kwargs):
        """Multiple OLS regression to predict Fex activity (y) from regressors (X).

        Args:
            X (list or str): Independent variable to predict.
            y (list or str): Dependent variable to be predicted.
            fit_intercept (bool): Whether to add intercept before fitting. Defaults to True.

        Returns:
            Dataframe of betas, ses, t-stats, p-values, df, residuals
        """

        mX, my = self._parse_features_labels(X, y)

        if fit_intercept:
            mX["intercept"] = 1

        b, se, t, p, df, res = regress(mX.to_numpy(), my.to_numpy(), *args, **kwargs)
        b_df = pd.DataFrame(b, index=mX.columns, columns=my.columns)
        se_df = pd.DataFrame(se, index=mX.columns, columns=my.columns)
        t_df = pd.DataFrame(t, index=mX.columns, columns=my.columns)
        p_df = pd.DataFrame(p, index=mX.columns, columns=my.columns)
        df_df = pd.DataFrame(
            np.full((len(mX.columns), len(my.columns)), df),
            index=mX.columns,
            columns=my.columns,
        )
        res_df = pd.DataFrame(res, columns=my.columns)
        return b_df, se_df, t_df, p_df, df_df, res_df

    def ttest_1samp(self, popmean=0):
        """Conducts 1 sample ttest.

        Uses scipy.stats.ttest_1samp to conduct 1 sample ttest

        Args:
            popmean (int, optional): Population mean to test against. Defaults to 0.
            threshold_dict ([type], optional): Dictionary for thresholding. Defaults to None. [NOT IMPLEMENTED]

        Returns:
            t, p: t-statistics and p-values
        """
        return ttest_1samp(self, popmean)

    def ttest_ind(self, col, sessions=None):
        """Conducts 2 sample ttest.

        Uses scipy.stats.ttest_ind to conduct 2 sample ttest on column col between sessions.

        Args:
            col (str): Column names to compare in a t-test between sessions
            session (array-like): session name to query Fex.sessions, otherwise uses the
            unique values in Fex.sessions.

        Returns:
            t, p: t-statistics and p-values
        """

        if sessions is None:
            sessions = pd.Series(self.sessions).unique()

        if len(sessions) != 2:
            raise ValueError(
                f"There must be exactly 2 session types to perform an independent t-test but {len(sessions)} were found."
            )

        sess1, sess2 = sessions
        a_mask = self.sessions == sess1
        a_mask = a_mask.values if isinstance(self.sessions, pd.Series) else a_mask
        b_mask = self.sessions == sess2
        b_mask = b_mask.values if isinstance(self.sessions, pd.Series) else b_mask
        a = self.loc[a_mask, col]
        b = self.loc[b_mask, col]

        return ttest_ind(a, b)

    def predict(self, X, y, model=LinearRegression, cv_kwargs={"cv": 5}, *args, **kwargs):
        """Predicts y from X using a sklearn model.

        Predict a variable of interest y using your model of choice from X, which can be a list of columns of the Fex instance or a dataframe.

        Args:
            X (list or DataFrame): List of column names or dataframe to be used as features for prediction
            y (string or array): y values to be predicted
            model (class, optional): Any sklearn model. Defaults to LinearRegression.
            args, kwargs: Model arguments

        Returns:
            model: Fit model instance.
        """

        mX, my = self._parse_features_labels(X, y)

        # user passes an uninitialized class, e.g. LogisticRegression
        if isinstance(model, type):
            clf = model(*args, **kwargs)
        else:
            # user passes an initialized estimator or pipeline, e.g. LogisticRegression()
            clf = model
        scores = cross_val_score(clf, mX, my, **cv_kwargs)
        _ = clf.fit(mX, my)
        return clf, scores

    def downsample(self, target, **kwargs):
        """Downsample Fex columns and return a Fex object.

        Args:
            target(float): target sampling frequency in Hz.
            kwargs: forwarded to feat.utils.stats.downsample.
        """
        df_ds = downsample(
            self, sampling_freq=self.sampling_freq, target=target, **kwargs
        ).__finalize__(self)
        df_ds.sampling_freq = target

        if self.features is not None:
            ds_features = downsample(
                self.features, sampling_freq=self.sampling_freq, target=target, **kwargs
            )
        else:
            ds_features = self.features
        df_ds.features = ds_features
        return df_ds

    def isc(self, col, index="frame", columns="input", method="pearson"):
        """[summary]

        Args:
            col (str]): Column name to compute the ISC for.
            index (str, optional): Column to be used in computing ISC. Usually this would be the column identifying the time such as the number of the frame. Defaults to "frame".
            columns (str, optional): Column to be used for ISC. Usually this would be the column identifying the video or subject. Defaults to "input".
            method (str, optional): Method to use for correlation pearson, kendall, or spearman. Defaults to "pearson".

        Returns:
            DataFrame: Correlation matrix with index as columns
        """
        if index is None:
            index = "frame"
        if columns is None:
            columns = "input"
        mat = pd.pivot_table(self, index=index, columns=columns, values=col).corr(
            method=method
        )
        return mat

    def upsample(self, target, target_type="hz", **kwargs):
        """Upsample Fex columns and return a Fex object.

        Args:
            target(float): upsampling target.
            target_type: 'hz' (default), 'samples', or 'seconds'.
            kwargs: forwarded to feat.utils.stats.upsample.
        """
        df_us = upsample(
            self,
            sampling_freq=self.sampling_freq,
            target=target,
            target_type=target_type,
            **kwargs,
        )
        if self.features is not None:
            us_features = upsample(
                self.features,
                sampling_freq=self.sampling_freq,
                target=target,
                target_type=target_type,
                **kwargs,
            )
        else:
            us_features = self.features
        return self.__class__(df_us, sampling_freq=target, features=us_features)

    def distance(self, method="euclidean", **kwargs):
        """Calculate distance between rows within a Fex() instance.

        Args:
            method: type of distance metric (can use any scikit learn or
                    sciypy metric)

        Returns:
            dist: Outputs a 2D distance matrix.

        """
        return pd.DataFrame(pairwise_distances(self, metric=method, **kwargs))

    def rectification(self, std=3):
        """Removes time points when the face position moved
        more than N standard deviations from the mean.

        Args:
            std (default 3): standard deviation from mean to remove outlier face locations
        Returns:
            data: cleaned FEX object

        """

        if self.facebox_columns and self.au_columns and self.emotion_columns:
            cleaned = deepcopy(self)
            face_columns = self.facebox_columns
            x_m = self.FaceRectX.mean()
            x_std = self.FaceRectX.std()
            y_m = self.FaceRectY.mean()
            y_std = self.FaceRectY.std()
            x_bool = (self.FaceRectX > std * x_std + x_m) | (
                self.FaceRectX < x_m - std * x_std
            )
            y_bool = (self.FaceRectY > std * y_std + y_m) | (
                self.FaceRectY < y_m - std * y_std
            )
            xy_bool = x_bool | y_bool
            cleaned.loc[
                xy_bool, face_columns + self.au_columns + self.emotion_columns
            ] = np.nan
            return cleaned
        else:
            raise ValueError("Facebox columns need to be defined.")

    def baseline(self, baseline="median", normalize=None, ignore_sessions=False):
        """Reference a Fex object to a baseline.

        Args:
            method: {'median', 'mean', 'begin', FexSeries instance}. Will subtract baseline from Fex object (e.g., mean, median).  If passing a Fex object, it will treat that as the baseline.
            normalize: (str). Can normalize results of baseline. Values can be [None, 'db','pct']; default None.
            ignore_sessions: (bool) If True, will ignore Fex.sessions information. Otherwise, method will be applied separately to each unique session.

        Returns:
            Fex object
        """
        if self.sessions is None or ignore_sessions:
            out = self.copy()
            if isinstance(baseline, str):
                if baseline == "median":
                    baseline_values = out.median()
                elif baseline == "mean":
                    baseline_values = out.mean()
                elif baseline == "begin":
                    baseline_values = out.iloc[0, :]
                else:
                    raise ValueError(
                        "%s is not implemented please use {mean, median, Fex}" % baseline
                    )
            elif isinstance(baseline, (Series, FexSeries)):
                baseline_values = baseline
            elif isinstance(baseline, (Fex, DataFrame)):
                raise ValueError("Must pass in a FexSeries not a FexSeries Instance.")

            if normalize == "db":
                out = 10 * np.log10(out - baseline_values) / baseline_values
            elif normalize == "pct":
                out = 100 * (out - baseline_values) / baseline_values
            else:
                out = out - baseline_values
        else:
            out = self.__class__(sampling_freq=self.sampling_freq)
            for k, v in self.itersessions():
                if isinstance(baseline, str):
                    if baseline == "median":
                        baseline_values = v.median()
                    elif baseline == "mean":
                        baseline_values = v.mean()
                    elif baseline == "begin":
                        baseline_values = v.iloc[0, :]
                    else:
                        raise ValueError(
                            "%s is not implemented please use {mean, median, Fex}"
                            % baseline
                        )
                elif isinstance(baseline, (Series, FexSeries)):
                    baseline_values = baseline
                elif isinstance(baseline, (Fex, DataFrame)):
                    raise ValueError("Must pass in a FexSeries not a FexSeries Instance.")

                if normalize == "db":
                    out = out.append(
                        10 * np.log10(v - baseline_values) / baseline_values,
                        session_id=k,
                    )
                elif normalize == "pct":
                    out = out.append(
                        100 * (v - baseline_values) / baseline_values, session_id=k
                    )
                else:
                    out = out.append(v - baseline_values, session_id=k)
        return out.__finalize__(self)

    def clean(
        self,
        detrend=True,
        standardize=True,
        confounds=None,
        low_pass=None,
        high_pass=None,
        ensure_finite=False,
        ignore_sessions=False,
    ):
        """Clean a time-series signal: detrend, regress confounds, filter, standardize.

        Operations applied in this order: detrend -> regress confounds ->
        Butterworth low/high-pass filter -> standardize -> ensure_finite. If
        ``Fex.sessions`` is set and ``ignore_sessions=False``, each session
        is cleaned independently.

        Args:
            detrend: subtract a linear trend from each column.
            standardize: rescale each column to zero-mean unit-variance.
            confounds: optional ``(T,)`` or ``(T, n_conf)`` confounds regressed
                out via OLS.
            low_pass: low-pass cutoff in Hz.
            high_pass: high-pass cutoff in Hz.
            ensure_finite: replace NaN/Inf with zero in the output.
            ignore_sessions: if True, treat all rows as one session.

        Returns:
            cleaned Fex instance.
        """
        if self.sessions is not None and not ignore_sessions:
            sessions = self.sessions
        else:
            sessions = None
        cleaned = clean_signal(
            self.values,
            detrend=detrend,
            standardize=standardize,
            confounds=confounds,
            low_pass=low_pass,
            high_pass=high_pass,
            ensure_finite=ensure_finite,
            sampling_freq=float(self.sampling_freq),
            runs=sessions,
        )
        return self.__class__(
            pd.DataFrame(cleaned, columns=self.columns),
            sampling_freq=self.sampling_freq,
            features=self.features,
            sessions=self.sessions,
        ).__finalize__(self)

    def decompose(self, algorithm="pca", axis=1, n_components=None, *args, **kwargs):
        """Decompose Fex instance

        Args:
            algorithm: (str) Algorithm to perform decomposition types=['pca','ica','nnmf','fa']
            axis: dimension to decompose [0,1]
            n_components: (int) number of components. If None then retain as many as possible.

        Returns:
            output: a dictionary of decomposition parameters
        """
        out = {}
        out["decomposition_object"] = set_decomposition_algorithm(
            algorithm=algorithm, n_components=n_components, *args, **kwargs
        )
        com_names = ["c%s" % str(x + 1) for x in range(n_components)]
        if axis == 0:
            out["decomposition_object"].fit(self.T)
            out["components"] = self.__class__(
                pd.DataFrame(
                    out["decomposition_object"].transform(self.T),
                    index=self.columns,
                    columns=com_names,
                ),
                sampling_freq=None,
            )
            out["weights"] = self.__class__(
                pd.DataFrame(
                    out["decomposition_object"].components_.T,
                    index=self.index,
                    columns=com_names,
                ),
                sampling_freq=self.sampling_freq,
                features=self.features,
                sessions=self.sessions,
            )
        if axis == 1:
            out["decomposition_object"].fit(self)
            out["components"] = self.__class__(
                pd.DataFrame(
                    out["decomposition_object"].transform(self), columns=com_names
                ),
                sampling_freq=self.sampling_freq,
                features=self.features,
                sessions=self.sessions,
            )
            out["weights"] = self.__class__(
                pd.DataFrame(
                    out["decomposition_object"].components_,
                    index=com_names,
                    columns=self.columns,
                ),
                sampling_freq=None,
            ).T
        return out

    def extract_mean(self, ignore_sessions=False):
        """Extract mean of each feature

        Args:
            ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

        Returns:
            Fex: mean values for each feature
        """

        prefix = "mean"
        if self.sessions is None or ignore_sessions:
            feats = pd.DataFrame(self.mean(numeric_only=True)).T
        else:
            feats = []
            for k, v in self.itersessions():
                feats.append(pd.Series(v.mean(numeric_only=True), name=k))
            feats = pd.concat(feats, axis=1).T
        feats = self.__class__(feats)
        feats.columns = f"{prefix}_" + feats.columns
        feats = feats.__finalize__(self)
        if ignore_sessions is False:
            feats.sessions = np.unique(self.sessions)
        feats._update_extracted_colnames(prefix)
        return feats

    def extract_std(self, ignore_sessions=False):
        """Extract std of each feature

        Args:
            ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

        Returns:
            Fex: mean values for each feature
        """

        prefix = "std"
        if self.sessions is None or ignore_sessions:
            feats = pd.DataFrame(self.std(numeric_only=True)).T
        else:
            feats = []
            for k, v in self.itersessions():
                feats.append(pd.Series(v.std(numeric_only=True), name=k))
            feats = pd.concat(feats, axis=1).T
        feats = self.__class__(feats)
        feats.columns = f"{prefix}_" + feats.columns
        feats = feats.__finalize__(self)
        if ignore_sessions is False:
            feats.sessions = np.unique(self.sessions)
        feats._update_extracted_colnames(prefix)
        return feats

    def extract_sem(self, ignore_sessions=False):
        """Extract std of each feature

        Args:
            ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

        Returns:
            Fex: mean values for each feature
        """

        prefix = "sem"
        if self.sessions is None or ignore_sessions:
            feats = pd.DataFrame(self.sem(numeric_only=True)).T
        else:
            feats = []
            for k, v in self.itersessions():
                feats.append(pd.Series(v.sem(numeric_only=True), name=k))
            feats = pd.concat(feats, axis=1).T
        feats = self.__class__(feats)
        feats.columns = f"{prefix}_" + feats.columns
        feats = feats.__finalize__(self)
        if ignore_sessions is False:
            feats.sessions = np.unique(self.sessions)
        feats._update_extracted_colnames(prefix)
        return feats

    def extract_min(self, ignore_sessions=False):
        """Extract minimum of each feature

        Args:
            ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

        Returns:
            Fex: (Fex) minimum values for each feature
        """

        prefix = "min"
        if self.sessions is None or ignore_sessions:
            feats = pd.DataFrame(self.min(numeric_only=True)).T
        else:
            feats = []
            for k, v in self.itersessions():
                feats.append(pd.Series(v.min(numeric_only=True), name=k))
            feats = pd.concat(feats, axis=1).T
        feats = self.__class__(feats)
        feats.columns = f"{prefix}_" + feats.columns
        feats = feats.__finalize__(self)
        if ignore_sessions is False:
            feats.sessions = np.unique(self.sessions)
        feats._update_extracted_colnames(prefix)
        return feats

    def extract_max(self, ignore_sessions=False):
        """Extract maximum of each feature

        Args:
            ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

        Returns:
            fex: (Fex) maximum values for each feature
        """
        prefix = "max"
        if self.sessions is None or ignore_sessions:
            feats = pd.DataFrame(self.max(numeric_only=True)).T
        else:
            feats = []
            for k, v in self.itersessions():
                feats.append(pd.Series(v.max(numeric_only=True), name=k))
            feats = pd.concat(feats, axis=1).T
        feats = self.__class__(feats)
        feats.columns = f"{prefix}_" + feats.columns
        feats = feats.__finalize__(self)
        if ignore_sessions is False:
            feats.sessions = np.unique(self.sessions)
        feats._update_extracted_colnames(prefix)
        return feats

    def extract_summary(
        self,
        mean=True,
        std=True,
        sem=True,
        max=True,
        min=True,
        ignore_sessions=False,
        *args,
        **kwargs,
    ):
        """Extract summary of multiple features

        Args:
            mean: (bool) extract mean of features
            std: (bool) extract std of features
            sem: (bool) extract sem of features
            max: (bool) extract max of features
            min: (bool) extract min of features
            ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

        Returns:
            fex: (Fex)
        """

        if mean is max is min is False:
            raise ValueError("At least one of min, max, mean must be True")

        out = self.__class__().__finalize__(self)
        if ignore_sessions is False:
            out.sessions = np.unique(self.sessions)

        col_updates = []

        if mean:
            new = self.extract_mean(ignore_sessions=ignore_sessions, *args, **kwargs)
            out = out.append(new, axis=1)
            col_updates.append("mean")
        if sem:
            new = self.extract_sem(ignore_sessions=ignore_sessions, *args, **kwargs)
            out = out.append(new, axis=1)
            col_updates.append("sem")
        if std:
            new = self.extract_std(ignore_sessions=ignore_sessions, *args, **kwargs)
            out = out.append(new, axis=1)
            col_updates.append("std")
        if max:
            new = self.extract_max(ignore_sessions=ignore_sessions, *args, **kwargs)
            out = out.append(new, axis=1)
            col_updates.append("max")
        if min:
            new = self.extract_min(ignore_sessions=ignore_sessions, *args, **kwargs)
            out = out.append(new, axis=1)
            col_updates.append("min")

        out._update_extracted_colnames(mode="reset")
        out._update_extracted_colnames(col_updates)

        return out

    def extract_wavelet(self, freq, num_cyc=3, mode="complex", ignore_sessions=False):
        """Perform feature extraction by convolving with a complex morlet wavelet

        Args:
            freq: (float) frequency to extract
            num_cyc: (float) number of cycles for wavelet
            mode: (str) feature to extract, e.g., 'complex','filtered','phase','magnitude','power']
            ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

        Returns:
            convolved: (Fex instance)

        """
        wav = wavelet(freq, sampling_freq=self.sampling_freq, num_cyc=num_cyc)
        if self.sessions is None or ignore_sessions:
            convolved = self.__class__(
                pd.DataFrame(
                    {x: convolve(y, wav, mode="same") for x, y in self.iteritems()}
                ),
                sampling_freq=self.sampling_freq,
            )
        else:
            convolved = self.__class__(sampling_freq=self.sampling_freq)
            for k, v in self.itersessions():
                session = self.__class__(
                    pd.DataFrame(
                        {x: convolve(y, wav, mode="same") for x, y in v.items()}
                    ),
                    sampling_freq=self.sampling_freq,
                )
                convolved = convolved.append(session, session_id=k)
        if mode == "complex":
            convolved = convolved
        elif mode == "filtered":
            convolved = np.real(convolved)
        elif mode == "phase":
            convolved = np.angle(convolved)
        elif mode == "magnitude":
            convolved = np.abs(convolved)
        elif mode == "power":
            convolved = np.abs(convolved) ** 2
        else:
            raise ValueError(
                "Mode must be ['complex','filtered','phase'," "'magnitude','power']"
            )
        convolved = self.__class__(
            convolved,
            sampling_freq=self.sampling_freq,
            features=self.features,
            sessions=self.sessions,
        )
        convolved.columns = "f" + "%s" % round(freq, 2) + "_" + mode + "_" + self.columns
        return convolved

    def extract_multi_wavelet(
        self, min_freq=0.06, max_freq=0.66, bank=8, *args, **kwargs
    ):
        """Convolve with a bank of morlet wavelets.

        Wavelets are equally spaced from min to max frequency. See extract_wavelet for more information and options.

        Args:
            min_freq: (float) minimum frequency to extract
            max_freq: (float) maximum frequency to extract
            bank: (int) size of wavelet bank
            num_cyc: (float) number of cycles for wavelet
            mode: (str) feature to extract, e.g., ['complex','filtered','phase','magnitude','power']
            ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

        Returns:
            convolved: (Fex instance)
        """
        out = []
        for f in np.geomspace(min_freq, max_freq, bank):
            out.append(self.extract_wavelet(f, *args, **kwargs))
        return self.__class__(
            pd.concat(out, axis=1),
            sampling_freq=self.sampling_freq,
            features=self.features,
            sessions=self.sessions,
        )

    def extract_boft(self, min_freq=0.06, max_freq=0.66, bank=8, *args, **kwargs):
        """Extract Bag of Temporal features

        Args:
            min_freq: maximum frequency of temporal filters
            max_freq: minimum frequency of temporal filters
            bank: number of temporal filter banks, filters are on exponential scale

        Returns:
            wavs: list of Morlet wavelets with corresponding freq
            hzs:  list of hzs for each Morlet wavelet
        """
        # First generate the wavelets
        target_hz = self.sampling_freq
        freqs = np.geomspace(min_freq, max_freq, bank)
        wavs, hzs = [], []
        for i, f in enumerate(freqs):
            wav = np.real(wavelet(f, sampling_freq=target_hz))
            wavs.append(wav)
            hzs.append(str(np.round(freqs[i], 2)))
        wavs = np.array(wavs)[::-1]
        hzs = np.array(hzs)[::-1]
        # # check asymptotes at lowest freq
        # asym = wavs[-1,:10].sum()
        # if asym > .001:
        #     print("Lowest frequency asymptotes at %2.8f " %(wavs[-1,:10].sum()))

        # Convolve data with wavelets
        Feats2Use = self.columns
        feats = pd.DataFrame()
        for feat in Feats2Use:
            _d = self[[feat]].T
            assert _d.isnull().sum().any() == 0, "Data contains NaNs. Cannot convolve. "
            for iw, cm in enumerate(wavs):
                convolved = np.apply_along_axis(
                    lambda m: np.convolve(m, cm, mode="full"), axis=1, arr=_d.values
                )
                # Extract bin features.
                out = pd.DataFrame(convolved.T).apply(calc_hist_auc, args=(None))
                # 6 bins hardcoded from calc_hist_auc
                colnames = [
                    "pos" + str(i) + "_hz_" + hzs[iw] + "_" + feat for i in range(6)
                ]
                colnames.extend(
                    ["neg" + str(i) + "_hz_" + hzs[iw] + "_" + feat for i in range(6)]
                )
                out = out.T
                out.columns = colnames
                feats = pd.concat([feats, out], axis=1)
        return self.__class__(
            feats, sampling_freq=self.sampling_freq, features=self.features
        )

    def _prepare_plot_aus(self, row, muscles, gaze):
        """
        Plot one or more faces based on their AU representation. This method is just a
        convenient wrapper for feat.plotting.plot_face. See that function for additional
        plotting args and kwargs.

        Args:
            force_separate_plot_per_detection (bool, optional): Whether to create a new
            figure for each detected face or plot to the same figure for multiple
            detections. Useful when you're know you're plotting multiple detections of a
           *single* face from multiple video frames. Default False

        """

        if self.au_model in ["svm", "xgb"]:
            au_lookup = "pyfeat"
            model = None
            feats = AU_LANDMARK_MAP["Feat"]
        else:
            au_lookup = self.au_model
            try:
                model = load_viz_model(f"{au_lookup}_aus_to_landmarks")
            except ValueError as _:
                raise NotImplementedError(
                    f"The AU model used for detection '{self.au_model}' has no corresponding AU visualization model. To fallback to plotting detections with facial landmarks, set faces='landmarks' in your call to .plot_detections. Otherwise, you can either use one of Py-Feat's custom AU detectors ('svm' or 'xgb') or train your own visualization model by following the tutorial at:\n\nhttps://py-feat.org/extra_tutorials/trainAUvisModel.html"
                )

            feats = AU_LANDMARK_MAP[au_lookup]

        rrow = row.copy()
        au = rrow[feats].to_numpy().squeeze()

        # If the caller passed gaze=True (or any truthy flag) and the Fex has
        # gaze_pitch / gaze_yaw columns, convert them to the 4-vector pupil
        # offset format the synthetic AU face renderer expects. Pre-formatted
        # 4-vectors are passed through.
        if isinstance(gaze, bool):
            if gaze and "gaze_pitch" in row.index and "gaze_yaw" in row.index:
                from feat.plotting import _gaze_to_pupil_offsets
                gaze = _gaze_to_pupil_offsets(row["gaze_pitch"], row["gaze_yaw"])
            else:
                gaze = None
        return au, gaze, muscles, model

    def plot_detections(
        self,
        faces="landmarks",
        faceboxes=True,
        muscles=False,
        poses=False,
        gazes=False,
        add_titles=True,
        au_barplot=True,
        emotion_barplot=True,
        plot_original_image=True,
    ):
        """
        Plots detection results by Feat. Can control plotting of face, AU barplot and
        Emotion barplot. The faces kwarg controls whether facial landmarks are draw on
        top of input images or whether faces are visualized using Py-Feat's AU
        visualization model using detected AUs. If detection was performed on a video an
        faces = 'landmarks', only an outline of the face will be draw without loading
        the underlying vidoe frame to save memory.


        Args:
            faces (str, optional): 'landmarks' to draw detected landmarks or 'aus' to
            generate a face from AU detections using Py-Feat's AU landmark model.
            Defaults to 'landmarks'.
            faceboxes (bool, optional): Whether to draw the bounding box around detected
            faces. Only applies if faces='landmarks'. Defaults to True.
            muscles (bool, optional): Whether to draw muscles from AU activity. Only
            applies if faces='aus'. Defaults to False.
            poses (bool, optional): Whether to draw facial poses. Only applies if
            faces='landmarks'. Defaults to False.
            gazes (bool, optional): Whether to draw gaze vectors. When
            faces='landmarks', a yellow arrow is drawn from each face's
            bbox center in the predicted gaze direction (using
            ``gaze_pitch``/``gaze_yaw`` columns from the L2CS gaze detector).
            When faces='aus', the gaze deflects the synthetic face's pupil
            positions. Defaults to False.
            add_titles (bool, optional): Whether to add the file name as a title above
            the face. Defaults to True.
            au_barplot (bool, optional): Whether to include a subplot for au detections. Defaults to True.
            emotion_barplot (bool, optional): Whether to include a subplot for emotion detections. Defaults to True.


        Returns:
            list: list of matplotlib figures
        """

        # Plotting logic, eventually refactor me!:
        # Possible detections:
        # 1. Single image - single-face
        # 2. Single image - multi-face
        # 3. Multi image - single-face per image
        # 4. Multi image - multi-face per image
        # 5. Multi image - single and multi-face mix per image
        # 6. Video - single-face for all frames
        # 7. Video - multi-face for all frames
        # 8. Video - single and multi-face mix across frames

        sns.set_context("paper", font_scale=2.0)

        num_subplots = bool(faces) + au_barplot + emotion_barplot

        all_figs = []
        for _, frame in enumerate(self.frame.unique()):
            # Determine figure width based on how many subplots we have
            f = plt.figure(figsize=(5 * num_subplots, 7))
            spec = f.add_gridspec(ncols=num_subplots, nrows=1)
            col_count = 0
            plot_data = self.query("frame == @frame")
            face_ax, au_ax, emo_ax = None, None, None
            if faces is not False:
                face_ax = f.add_subplot(spec[0, col_count])
                col_count += 1
            if au_barplot:
                au_ax = f.add_subplot(spec[0, col_count])
                col_count += 1
            if emotion_barplot:
                emo_ax = f.add_subplot(spec[0, col_count])
                col_count += 1

            for _face_i, (_, row) in enumerate(plot_data.iterrows()):
                # DRAW LANDMARKS ON IMAGE OR AU FACE
                if face_ax is not None:
                    facebox = row[self.facebox_columns].values

                    if not faces == "aus" and plot_original_image:
                        file_extension = os.path.basename(row["input"]).split(".")[-1]
                        if file_extension.lower() in [
                            "jpg",
                            "jpeg",
                            "png",
                            "bmp",
                            "tiff",
                            "pdf",
                        ]:
                            img = read_image(row["input"])
                        else:
                            img = decode_video(row["input"])[int(row["frame"])]
                        color = "w"
                        face_ax.imshow(img.permute([1, 2, 0]))
                    else:
                        color = "k"  # drawing lineface but not on photo

                    # Bbox / pose-axes / gaze-arrow overlays are anchored to
                    # the facebox in original-image coords. For faces='aus'
                    # the displayed face is the synthetic AU schematic in
                    # the viz model's own coord system, so the bbox/pose/
                    # gaze overlays don't align with it and just confuse
                    # the picture. Skip them in aus mode — gaze is still
                    # rendered on the synthetic face via plot_face's
                    # gaze= kwarg (handled by _prepare_plot_aus).
                    aus_mode = faces == "aus"

                    if faceboxes and not aus_mode:
                        rect = Rectangle(
                            (facebox[0], facebox[1]),
                            facebox[2],
                            facebox[3],
                            linewidth=2,
                            edgecolor="cyan",
                            fill=False,
                        )
                        face_ax.add_patch(rect)
                        # Label each box with its face number (1-based, in
                        # detection order) and identity if available, so
                        # multi-face frames are readable (#198). Single-face
                        # frames stay unlabelled.
                        if len(plot_data) > 1:
                            label = str(_face_i + 1)
                            if "Identity" in row.index and pd.notna(row["Identity"]):
                                label += f": {row['Identity']}"
                            face_ax.text(
                                facebox[0], facebox[1] - 4, label,
                                color="cyan", fontsize=12, fontweight="bold",
                                va="bottom", ha="left",
                            )

                    if poses and not aus_mode:
                        face_ax = draw_facepose(
                            pose=row[self.facepose_columns[:3]].values,
                            facebox=facebox,
                            ax=face_ax,
                        )

                    if gazes and not aus_mode and "gaze_pitch" in row.index and "gaze_yaw" in row.index:
                        face_ax = draw_facegaze(
                            pitch_rad=row["gaze_pitch"],
                            yaw_rad=row["gaze_yaw"],
                            facebox=facebox,
                            ax=face_ax,
                        )

                    if faces == "landmarks":
                        # Branch on landmark topology: dlib-68 (136 cols) returns
                        # as-is; MP-478 (1434 cols) samples the 68 dlib-equivalent
                        # vertices via DLIB68_FROM_MP478 so the same draw_lineface
                        # path works for both detectors.
                        currx, curry = self.landmarks_dlib68_xy(row=row)

                        # facelines
                        face_ax = draw_lineface(
                            currx, curry, ax=face_ax, color=color, linewidth=3
                        )
                        if not plot_original_image:
                            face_ax.invert_yaxis()

                    elif faces == "aus":
                        # Generate face from AU landmark model
                        if any(self.groupby("frame").size() > 1):
                            raise NotImplementedError(
                                "Plotting using AU landmark model is not currently supported for detections that contain multiple faces"
                            )
                        if muscles:
                            muscles = {"all": "heatmap"}
                        else:
                            muscles = {}
                        aus, gaze, muscles, model = self._prepare_plot_aus(
                            row, muscles=muscles, gaze=gazes
                        )
                        _ = row["input"] if add_titles else None
                        face_ax = plot_face(
                            model=model,
                            au=aus,
                            gaze=gaze,
                            ax=face_ax,
                            muscles=muscles,
                            title=None,
                        )
                    else:
                        raise ValueError(
                            f"faces={type(faces)} is not currently supported try ['False','landmarks','aus']."
                        )

                    if add_titles:
                        _ = face_ax.set_title(
                            "\n".join(wrap(os.path.basename(row["input"]))),
                            loc="center",
                            wrap=True,
                            fontsize=14,
                        )

                    face_ax.axes.get_xaxis().set_visible(False)
                    face_ax.axes.get_yaxis().set_visible(False)

            # DRAW AU BARPLOT
            if au_ax is not None:
                _ = plot_data.aus.T.plot(kind="barh", ax=au_ax)
                _ = au_ax.invert_yaxis()
                _ = au_ax.legend().remove()
                _ = au_ax.set(xlim=[0, 1.1], title="Action Units")

            # DRAW EMOTION BARPLOT
            if emo_ax is not None:
                _ = plot_data.emotions.T.plot(kind="barh", ax=emo_ax)
                _ = emo_ax.invert_yaxis()
                _ = emo_ax.legend().remove()
                _ = emo_ax.set(xlim=[0, 1.1], title="Emotions")

            f.tight_layout()
            all_figs.append(f)
        return all_figs

    def compute_identities(self, threshold=0.8, method="gallery",
                           min_cluster_size=2, inplace=False):
        """Compute Identities from face-identity embeddings.

        Args:
            threshold: cosine-similarity cutoff for the same person.
            method: ``"gallery"`` (default; single-pass, memory-efficient,
                scales to large videos), ``"hdbscan"`` (density-based, emits
                ``"Unknown"`` for outliers), or ``"connected"`` (legacy exact
                single-linkage — O(N²) memory, can OOM on long videos).
            min_cluster_size: minimum cluster size for ``method="hdbscan"``.
            inplace: write the ``Identity`` column in place instead of returning
                a copy.
        """
        target = self if inplace else self.copy()
        target["Identity"] = cluster_identities(
            target.identity_embeddings, threshold=threshold,
            method=method, min_cluster_size=min_cluster_size,
        )
        if not inplace:
            return target

    # TODO: turn this into a property using a @property and @sessions.settr decorators
    # Tried it but was running into maximum recursion depth errors. Maybe some
    # interaction with how pandas sub-classing works?? - ejolly
    def update_sessions(self, new_sessions):
        """
        Returns a copy of the Fex dataframe with a new sessions attribute after
        validation. `new_sessions` should be a dictionary mapping old to new names or an iterable with the same number of rows as the Fex dataframe

        Args:
            new_sessions (dict, Iterable): map or list of new session names

        Returns:
            Fex: self
        """

        out = deepcopy(self)

        if isinstance(new_sessions, dict):
            if not isinstance(out.sessions, pd.Series):
                out.sessions = pd.Series(out.sessions)

            out.sessions = out.sessions.map(new_sessions)

        elif isinstance(new_sessions, Iterable):
            if len(new_sessions) != out.shape[0]:
                raise ValueError(
                    f"When new_sessions are not a dictionary then they must but an iterable with length == the number of rows of this Fex dataframe {out.shape[0]}, but they have length {len(new_sessions)}."
                )

            out.sessions = new_sessions

        else:
            raise TypeError(
                f"new_sessions must be either be a dictionary mapping between old and new session values or an iterable with the same number of rows as this Fex dataframe {out.shape[0]}, but was type: {type(new_sessions)}"
            )

        return out

    def plot_singleframe_detections(
        self,
        bounding_boxes=False,
        landmarks=False,
        poses=False,
        emotions=False,
        aus=False,
        gazes=False,
        image_opacity=0.9,
        facebox_color="cyan",
        facebox_width=3,
        pose_width=2,
        landmark_color="white",
        landmark_width=2,
        gaze_color="yellow",
        gaze_width=3,
        emotions_position="right",
        emotions_opacity=1.0,
        emotions_color="white",
        emotions_size=12,
        au_heatmap_resolution=1000,
        au_opacity=0.9,
        au_cmap="Blues",
        *args,
        **kwargs,
    ):
        """
        Function to generate interactive plotly figure to interactively visualize py-feat detectors on a single image frame.

        Args:
            image_opacity (float): opacity of image overlay (default=.9)
            emotions_position (str): position around facebox to plot emotion annotations. default='right'
            emotions_opacity (float): opacity of emotion annotation text (default=1.)
            emotions_color (str): color of emotion annotation text (default='white')
            emotions_size (int): size of emotion annotations (default=14)
            frame_duration (int): duration in milliseconds to play each frame if plotting multiple frames (default=1000)
            facebox_color (str): color of facebox bounding box (default="cyan")
            facebox_width (int): line width of facebox bounding box (default=3)
            pose_width (int): line width of pose rotation plot (default=2)
            landmark_color (str): color of landmark detectors (default="white")
            landmark_width (int): line width of landmark detectors (default=2)
            au_cmap (str): colormap to use for AU heatmap (default='Blues')
            au_heatmap_resolution (int): resolution of heatmap values (default=1000)
            au_opacity (float): opacity of AU heatmaps (default=0.9)

        Returns:
            a plotly figure instance
        """

        n_frames = len(self["frame"].unique())

        if n_frames > 1:
            raise ValueError(
                "This function can only plot a single frame. Try using plot_multipleframe_detections() instead."
            )

        # Initialize Image
        frame_id = self["frame"].unique()[0]
        frame_fex = self.query("frame==@frame_id")
        frame_img = load_pil_img(frame_fex["input"].unique()[0], frame_id)
        img_width = frame_img.width
        img_height = frame_img.height

        # Create figure
        fig = go.Figure()

        # Add invisible scatter trace.
        # This trace is added to help the autoresize logic work.
        fig.add_trace(
            go.Scatter(
                x=[0, img_width],
                y=[0, img_height],
                mode="markers",
                marker_opacity=0,
            )
        )

        # Add image
        fig.add_layout_image(
            dict(
                x=0,
                sizex=img_width,
                y=img_height,
                sizey=img_height,
                xref="x",
                yref="y",
                opacity=image_opacity,
                layer="below",
                sizing="stretch",
                source=frame_img,
            )
        )

        # Add Face Bounding Box
        faceboxes_path = [
            dict(
                type="rect",
                x0=row["FaceRectX"],
                y0=img_height - row["FaceRectY"],
                x1=row["FaceRectX"] + row["FaceRectWidth"],
                y1=img_height - row["FaceRectY"] - row["FaceRectHeight"],
                line=dict(color=facebox_color, width=facebox_width),
            )
            for i, row in frame_fex.iterrows()
        ]

        # Add Landmarks
        landmarks_path = [
            draw_plotly_landmark(
                row,
                img_height,
                fig,
                line_color=landmark_color,
                line_width=landmark_width,
            )
            for i, row in frame_fex.iterrows()
        ]

        # Add Pose
        poses_path = flatten_list(
            [
                draw_plotly_pose(row, img_height, fig, line_width=pose_width)
                for i, row in frame_fex.iterrows()
            ]
        )

        # Add Emotions
        emotions_annotations = []
        for i, row in frame_fex.iterrows():
            # Use this Fex's own emotion columns rather than a hardcoded v1
            # list — Detectorv2 emits capitalized names (Neutral/Happy/...),
            # so hardcoding the lowercase v1 set KeyErrors on v2 detections.
            emotion_dict = (
                row[self.emotion_columns]
                .sort_values(ascending=False)
                .to_dict()
            )

            x_position, y_position, align, valign = emotion_annotation_position(
                row,
                img_height,
                img_width,
                emotions_size=emotions_size,
                emotions_position=emotions_position,
            )

            emotion_text = ""
            for emotion in emotion_dict:
                emotion_text += f"{emotion}: <i>{emotion_dict[emotion]:.2f}</i><br>"

            emotions_annotations.append(
                dict(
                    text=emotion_text,
                    x=x_position,
                    y=y_position,
                    opacity=emotions_opacity,
                    showarrow=False,
                    align=align,
                    valign=valign,
                    bgcolor="black",
                    font=dict(color=emotions_color, size=emotions_size),
                )
            )

        # Add AU Heatmaps
        aus_path = flatten_list(
            [
                draw_plotly_au(
                    row,
                    img_height,
                    fig,
                    cmap=au_cmap,
                    au_opacity=au_opacity,
                    heatmap_resolution=au_heatmap_resolution,
                )
                for i, row in frame_fex.iterrows()
            ]
        )

        # Add Gaze arrows
        gazes_path = flatten_list(
            [
                draw_plotly_gaze(
                    row,
                    img_height,
                    color=gaze_color,
                    line_width=gaze_width,
                )
                for i, row in frame_fex.iterrows()
            ]
        )

        # Configure other layout
        fig.update_layout(
            width=img_width,
            height=img_height,
            margin={"l": 0, "r": 0, "t": 0, "b": 0},
        )

        # Configure axes
        fig.update_xaxes(visible=False, range=[0, img_width])

        fig.update_yaxes(
            visible=False,
            range=[0, img_height],
            scaleanchor="x",  # the scaleanchor attribute ensures that the aspect ratio stays constant
        )

        # Add ALL overlays unconditionally with per-shape `visible` flags
        # set from the corresponding function arguments — that way each
        # overlay group can be toggled independently via the buttons below
        # (the prior approach of conditionally adding shapes meant clicking
        # one button replaced the entire `layout.shapes` array, so only
        # one overlay could be visible at a time).
        group_indices = {"bbox": [], "landmarks": [], "poses": [], "aus": [], "gazes": []}
        initial_vis = {
            "bbox": bool(bounding_boxes),
            "landmarks": bool(landmarks),
            "poses": bool(poses),
            "aus": bool(aus),
            "gazes": bool(gazes),
        }
        for group, shapes in (
            ("bbox", faceboxes_path),
            ("landmarks", landmarks_path),
            ("poses", poses_path),
            ("aus", aus_path),
            ("gazes", gazes_path),
        ):
            for s in shapes:
                idx = len(fig.layout.shapes)
                # add_shape() emits an immutable Shape; clone the dict so
                # we can include the `visible` field per our group flag.
                s = dict(s)
                s["visible"] = initial_vis[group]
                fig.add_shape(s)
                group_indices[group].append(idx)

        # Emotions are annotations (not shapes); track separately.
        emotion_indices = []
        for ann in emotions_annotations:
            idx = len(fig.layout.annotations)
            ann = dict(ann)
            ann["visible"] = bool(emotions)
            fig.add_annotation(ann)
            emotion_indices.append(idx)

        # Build toggle buttons. Each button's args (first click) sets the
        # group's visibility to the OPPOSITE of its initial state, and
        # args2 (second click) sets it back. Effectively per-group toggle,
        # independent across groups.
        def _shape_toggle_args(indices, target):
            return [{f"shapes[{i}].visible": target for i in indices}]

        def _annot_toggle_args(indices, target):
            return [{f"annotations[{i}].visible": target for i in indices}]

        buttons = []
        for label, group in (
            ("Bounding Box", "bbox"),
            ("Landmarks", "landmarks"),
            ("Poses", "poses"),
            ("AU", "aus"),
            ("Gaze", "gazes"),
        ):
            idxs = group_indices[group]
            if not idxs:
                continue
            init = initial_vis[group]
            buttons.append(dict(
                method="relayout",
                label=label,
                args=_shape_toggle_args(idxs, not init),
                args2=_shape_toggle_args(idxs, init),
            ))
        if emotion_indices:
            buttons.append(dict(
                method="relayout",
                label="Emotion",
                args=_annot_toggle_args(emotion_indices, not emotions),
                args2=_annot_toggle_args(emotion_indices, bool(emotions)),
            ))

        fig.update_layout(
            updatemenus=[
                dict(
                    type="buttons",
                    direction="left",
                    buttons=buttons,
                    pad={"r": 10, "t": 10},
                    showactive=False,
                    x=0.1,
                    xanchor="left",
                    y=1.12,
                    yanchor="top",
                )
            ]
        )

        return fig

    def plot_multipleframes_detections(
        self,
        bounding_boxes=False,
        landmarks=False,
        aus=False,
        poses=False,
        emotions=False,
        gazes=False,
        image_opacity=0.9,
        facebox_color="cyan",
        facebox_width=3,
        pose_width=2,
        landmark_color="white",
        landmark_width=2,
        gaze_color="yellow",
        gaze_width=3,
        emotions_position="right",
        emotions_opacity=1.0,
        emotions_color="white",
        emotions_size=12,
        au_cmap="Blues",
        au_heatmap_resolution=1000,
        au_opacity=0.9,
        frame_duration=1000,
        *args,
        **kwargs,
    ):
        """Plotly animation across the frames of a multi-frame (video) Fex.

        Each frame shows the underlying image with the *prespecified* overlays
        (unlike the single-frame plot, overlays can't be toggled live here — pass
        the ones you want). A play/pause control and a frame slider scrub through
        the detections. Overlay coordinates use the same y-flipped image space as
        ``plot_singleframe_detections``. Assumes all frames share one image size
        (true for a video).

        Returns:
            a plotly figure instance
        """
        frame_ids = sorted(self["frame"].unique())

        def _build(frame_id):
            ff = self.query("frame == @frame_id")
            img = load_pil_img(ff["input"].unique()[0], frame_id)
            w, h = img.width, img.height
            image = dict(
                x=0, sizex=w, y=h, sizey=h, xref="x", yref="y",
                opacity=image_opacity, layer="below", sizing="stretch", source=img,
            )
            shapes = []
            if bounding_boxes:
                shapes += [
                    dict(
                        type="rect",
                        x0=r["FaceRectX"], y0=h - r["FaceRectY"],
                        x1=r["FaceRectX"] + r["FaceRectWidth"],
                        y1=h - r["FaceRectY"] - r["FaceRectHeight"],
                        line=dict(color=facebox_color, width=facebox_width),
                    )
                    for _, r in ff.iterrows()
                ]
            if landmarks:
                shapes += [
                    draw_plotly_landmark(r, h, None, line_color=landmark_color,
                                         line_width=landmark_width)
                    for _, r in ff.iterrows()
                ]
            if poses:
                shapes += flatten_list(
                    [draw_plotly_pose(r, h, None, line_width=pose_width)
                     for _, r in ff.iterrows()]
                )
            if aus:
                shapes += flatten_list(
                    [draw_plotly_au(r, h, None, cmap=au_cmap, au_opacity=au_opacity,
                                    heatmap_resolution=au_heatmap_resolution)
                     for _, r in ff.iterrows()]
                )
            if gazes:
                shapes += flatten_list(
                    [draw_plotly_gaze(r, h, color=gaze_color, line_width=gaze_width)
                     for _, r in ff.iterrows()]
                )
            annotations = []
            if emotions:
                for _, r in ff.iterrows():
                    emo = r[self.emotion_columns].sort_values(ascending=False).to_dict()
                    xp, yp, align, valign = emotion_annotation_position(
                        r, h, w, emotions_size=emotions_size,
                        emotions_position=emotions_position,
                    )
                    text = "".join(f"{e}: <i>{v:.2f}</i><br>" for e, v in emo.items())
                    annotations.append(dict(
                        text=text, x=xp, y=yp, opacity=emotions_opacity,
                        showarrow=False, align=align, valign=valign, bgcolor="black",
                        font=dict(color=emotions_color, size=emotions_size),
                    ))
            return image, shapes, annotations, w, h

        image0, shapes0, annotations0, img_width, img_height = _build(frame_ids[0])

        fig = go.Figure()
        # Invisible trace anchors the axis range so layout images/shapes line up.
        fig.add_trace(go.Scatter(
            x=[0, img_width], y=[0, img_height], mode="markers", marker_opacity=0,
        ))
        fig.update_layout(
            images=[image0], shapes=shapes0, annotations=annotations0,
            width=img_width, height=img_height,
            margin={"l": 0, "r": 0, "t": 0, "b": 0},
        )

        plotly_frames = []
        for frame_id in frame_ids:
            image, shapes, annotations, _, _ = _build(frame_id)
            plotly_frames.append(go.Frame(
                name=str(int(frame_id)),
                layout=dict(images=[image], shapes=shapes, annotations=annotations),
            ))
        fig.frames = plotly_frames

        fig.update_layout(
            updatemenus=[dict(
                type="buttons", direction="left", showactive=False,
                x=0.1, xanchor="left", y=1.12, yanchor="top",
                pad={"r": 10, "t": 10},
                buttons=[
                    dict(label="Play", method="animate",
                         args=[None, {"frame": {"duration": frame_duration, "redraw": True},
                                      "fromcurrent": True}]),
                    dict(label="Pause", method="animate",
                         args=[[None], {"frame": {"duration": 0, "redraw": False},
                                        "mode": "immediate"}]),
                ],
            )],
            sliders=[dict(
                active=0, x=0.1, len=0.9, pad={"b": 10, "t": 10},
                steps=[dict(
                    method="animate", label=str(int(fid)),
                    args=[[str(int(fid))], {"frame": {"duration": 0, "redraw": True},
                                            "mode": "immediate"}],
                ) for fid in frame_ids],
            )],
        )
        fig.update_xaxes(visible=False, range=[0, img_width])
        fig.update_yaxes(visible=False, range=[0, img_height], scaleanchor="x")
        return fig

    def iplot_detections(
        self,
        bounding_boxes=False,
        landmarks=False,
        aus=False,
        poses=False,
        emotions=False,
        gazes=False,
        emotions_position="right",
        emotions_opacity=1.0,
        emotions_color="white",
        emotions_size=14,
        frame_duration=1000,
        facebox_color="cyan",
        facebox_width=3,
        pose_width=2,
        landmark_color="white",
        landmark_width=2,
        gaze_color="yellow",
        gaze_width=3,
        au_cmap="Blues",
        au_heatmap_resolution=1000,
        au_opacity=0.9,
        *args,
        **kwargs,
    ):
        """Plot Py-FEAT detection results using the plotly backend. There are
        two plot types. For a single frame, uses plot_singleframe_detections()
        to create an interactive plot where the different detector outputs can
        be toggled on or off. For multiple frames (a video), uses
        plot_multipleframes_detections() to create a plotly animation with a
        play/pause control and a frame slider; overlays can't be toggled live
        in the animation, so the detector outputs to draw must be prespecified.

        Args:
            bounding_boxes (bool): will include faceboxes when plotting detector output for multiple frames.
            landmarks (bool): will include face landmarks when plotting detector output for multiple frames.
            poses (bool): will include 3 axis line plot indicating x,y,z rotation information when plotting detector output for multiple frames.
            aus (bool): will include action unit heatmaps when plotting detector output for multiple frames.
            emotions (bool): will add text annotations indicating probability of discrete emotion when plotting detector output for multiple frames.
            emotions_position (str): position around facebox to plot emotion annotations. default='right'
            emotions_opacity (float): opacity of emotion annotation text (default=1.)
            emotions_color (str): color of emotion annotation text (default='white')
            emotions_size (int): size of emotion annotations (default=14)
            frame_duration (int): duration in milliseconds to play each frame if plotting multiple frames (default=1000)
            facebox_color (str): color of facebox bounding box (default="cyan")
            facebox_width (int): line width of facebox bounding box (default=3)
            pose_width (int): line width of pose rotation plot (default=2)
            landmark_color (str): color of landmark detectors (default="white")
            landmark_width (int): line width of landmark detectors (default=2)
            au_cmap (str): colormap to use for AU heatmap (default='Blues')
            au_heatmap_resolution (int): resolution of heatmap values (default=1000)
            au_opacity (float): opacity of AU heatmaps (default=0.9)

        Returns:
            a plotly figure instance
        """

        n_frames = len(self["frame"].unique())

        if n_frames == 1:
            fig = self.plot_singleframe_detections(
                bounding_boxes=bounding_boxes,
                landmarks=landmarks,
                poses=poses,
                emotions=emotions,
                aus=aus,
                gazes=gazes,
                facebox_color=facebox_color,
                facebox_width=facebox_width,
                pose_width=pose_width,
                landmark_color=landmark_color,
                landmark_width=landmark_width,
                gaze_color=gaze_color,
                gaze_width=gaze_width,
                emotions_position=emotions_position,
                emotions_opacity=emotions_opacity,
                emotions_color=emotions_color,
                emotions_size=emotions_size,
                au_cmap=au_cmap,
                au_heatmap_resolution=au_heatmap_resolution,
                au_opacity=au_opacity,
                *args,
                **kwargs,
            )
            return fig
        return self.plot_multipleframes_detections(
            bounding_boxes=bounding_boxes,
            landmarks=landmarks,
            aus=aus,
            poses=poses,
            emotions=emotions,
            gazes=gazes,
            image_opacity=kwargs.pop("image_opacity", 0.9),
            facebox_color=facebox_color,
            facebox_width=facebox_width,
            pose_width=pose_width,
            landmark_color=landmark_color,
            landmark_width=landmark_width,
            gaze_color=gaze_color,
            gaze_width=gaze_width,
            emotions_position=emotions_position,
            emotions_opacity=emotions_opacity,
            emotions_color=emotions_color,
            emotions_size=emotions_size,
            au_cmap=au_cmap,
            au_heatmap_resolution=au_heatmap_resolution,
            au_opacity=au_opacity,
            frame_duration=frame_duration,
        )

aus property

Returns the Action Units data

Returns Action Unit data using the columns set in fex.au_columns.

Returns:

Name	Type	Description
DataFrame		Action Units data

blendshapes property

Returns the 52 MediaPipe/ARKit blendshape coefficients in [0, 1].

Raises AttributeError if no blendshape columns are registered (e.g. a pre-v2.5 multitask model, or DetectorV1, which has no blendshape head).

design property

Returns the design data

Returns the study design information using columns in fex.design_columns.

Returns:

Name	Type	Description
DataFrame		time data

emotions property

Returns the emotion data

Returns emotions data using the columns set in fex.emotion_columns.

Returns:

Name	Type	Description
DataFrame		emotion data

facebox property

Returns the facebox data

Returns:

Name	Type	Description
DataFrame		facebox data

faceboxes property

Returns the facebox data

Returns:

Name	Type	Description
DataFrame		facebox data

facepose property

Returns the facepose data using the columns set in fex.facepose_columns

Returns:

Name	Type	Description
DataFrame		facepose data

gazes property

Returns the gaze data (pitch, yaw, optionally combined angle).

Returns:

Name	Type	Description
DataFrame		gaze data populated by the active gaze model
		(L2CS by default in v0.7+). Raises AttributeError if the
		Detectorv1 was built with gaze_model=None.

identities property

Returns the identity labels

Returns:

Name	Type	Description
DataFrame		identity data

identity_embeddings property

Returns the identity embeddings

Returns:

Name	Type	Description
DataFrame		identity data

info property

Print all meta data of fex

Loops through metadata set in self._metadata and prints out the information.

input property

Returns input column as string

Returns input data in the "input" column.

Returns:

Name	Type	Description
string		path to input image

inputs property

Returns input column as string

Returns input data in the "input" column.

Returns:

Name	Type	Description
string		path to input image

landmark property

Returns the landmark data

Returns landmark data using the columns set in fex.landmark_columns.

Returns:

Name	Type	Description
DataFrame		landmark data

landmark_x property

Returns the x landmarks.

Returns:

Name	Type	Description
DataFrame		x landmarks.

landmark_y property

Returns the y landmarks.

Returns:

Name	Type	Description
DataFrame		y landmarks.

landmarks property

Returns the landmark data

Returns landmark data using the columns set in fex.landmark_columns.

Returns:

Name	Type	Description
DataFrame		landmark data

landmarks_x property

Returns the x landmarks.

Returns:

Name	Type	Description
DataFrame		x landmarks.

landmarks_y property

Returns the y landmarks.

Returns:

Name	Type	Description
DataFrame		y landmarks.

poses property

Returns the facepose data using the columns set in fex.facepose_columns

Returns:

Name	Type	Description
DataFrame		facepose data

time property

Returns the time data

Returns the time information using fex.time_columns.

Returns:

Name	Type	Description
DataFrame		time data

__finalize__(other, method=None, **kwargs)

Propagate metadata from other to self

Source code in feat/data.py

def __finalize__(self, other, method=None, **kwargs):
    """Propagate metadata from other to self"""
    self = super().__finalize__(other, method=method, **kwargs)
    # merge operation: using metadata of the left object
    if method == "merge":
        for name in self._metadata:
            object.__setattr__(self, name, getattr(other.left, name, None))
    # concat operation: using metadata of the first object
    elif method == "concat":
        for name in self._metadata:
            object.__setattr__(self, name, getattr(other.objs[0], name, None))
    return self

append(data, session_id=None, axis=0)

Append a new Fex object to an existing object

Parameters:

Name	Type	Description	Default
data		(Fex) Fex instance to append	required
session_id		session label	None
axis		([0,1]) Axis to append. Rows=0, Cols=1	0

Returns: Fex instance

Source code in feat/data.py

def append(self, data, session_id=None, axis=0):
    """Append a new Fex object to an existing object

    Args:
        data: (Fex) Fex instance to append
        session_id: session label
        axis: ([0,1]) Axis to append. Rows=0, Cols=1
    Returns:
        Fex instance
    """
    if not isinstance(data, self.__class__):
        raise ValueError("Make sure data is a Fex instance.")

    if self.empty:
        out = data.copy()
        if session_id is not None:
            out.sessions = np.repeat(session_id, len(data))
    else:
        if self.sampling_freq != data.sampling_freq:
            raise ValueError(
                "Make sure Fex objects have the same " "sampling frequency"
            )
        if axis == 0:
            out = self.__class__(
                pd.concat([self, data], axis=axis, ignore_index=True),
                sampling_freq=self.sampling_freq,
            ).__finalize__(self)
            if session_id is not None:
                out.sessions = np.hstack(
                    [self.sessions, np.repeat(session_id, len(data))]
                )
            if self.features is not None:
                if data.features is not None:
                    if self.features.shape[1] == data.features.shape[1]:
                        out.features = pd.concat(
                            [self.features, data.features], ignore_index=True
                        )
                    else:
                        raise ValueError(
                            "Different number of features in new dataset."
                        )
                else:
                    out.features = self.features
            elif data.features is not None:
                out.features = data.features
        elif axis == 1:
            out = self.__class__(
                pd.concat([self, data], axis=axis), sampling_freq=self.sampling_freq
            ).__finalize__(self)
            if self.sessions is not None:
                if data.sessions is not None:
                    if np.array_equal(self.sessions, data.sessions):
                        out.sessions = self.sessions
                    else:
                        raise ValueError("Both sessions must be identical.")
                else:
                    out.sessions = self.sessions
            elif data.sessions is not None:
                out.sessions = data.sessions
            if self.features is not None:
                out.features = self.features
                if data.features is not None:
                    out.features = pd.concat(
                        [self.features, data.features], axis=1
                    )
            elif data.features is not None:
                out.features = data.features
        else:
            raise ValueError("Axis must be 1 or 0.")
    return out

baseline(baseline='median', normalize=None, ignore_sessions=False)

Reference a Fex object to a baseline.

Parameters:

Name	Type	Description	Default
method		{'median', 'mean', 'begin', FexSeries instance}. Will subtract baseline from Fex object (e.g., mean, median). If passing a Fex object, it will treat that as the baseline.	required
normalize		(str). Can normalize results of baseline. Values can be [None, 'db','pct']; default None.	None
ignore_sessions		(bool) If True, will ignore Fex.sessions information. Otherwise, method will be applied separately to each unique session.	False

Returns:

Type	Description
	Fex object

Source code in feat/data.py

def baseline(self, baseline="median", normalize=None, ignore_sessions=False):
    """Reference a Fex object to a baseline.

    Args:
        method: {'median', 'mean', 'begin', FexSeries instance}. Will subtract baseline from Fex object (e.g., mean, median).  If passing a Fex object, it will treat that as the baseline.
        normalize: (str). Can normalize results of baseline. Values can be [None, 'db','pct']; default None.
        ignore_sessions: (bool) If True, will ignore Fex.sessions information. Otherwise, method will be applied separately to each unique session.

    Returns:
        Fex object
    """
    if self.sessions is None or ignore_sessions:
        out = self.copy()
        if isinstance(baseline, str):
            if baseline == "median":
                baseline_values = out.median()
            elif baseline == "mean":
                baseline_values = out.mean()
            elif baseline == "begin":
                baseline_values = out.iloc[0, :]
            else:
                raise ValueError(
                    "%s is not implemented please use {mean, median, Fex}" % baseline
                )
        elif isinstance(baseline, (Series, FexSeries)):
            baseline_values = baseline
        elif isinstance(baseline, (Fex, DataFrame)):
            raise ValueError("Must pass in a FexSeries not a FexSeries Instance.")

        if normalize == "db":
            out = 10 * np.log10(out - baseline_values) / baseline_values
        elif normalize == "pct":
            out = 100 * (out - baseline_values) / baseline_values
        else:
            out = out - baseline_values
    else:
        out = self.__class__(sampling_freq=self.sampling_freq)
        for k, v in self.itersessions():
            if isinstance(baseline, str):
                if baseline == "median":
                    baseline_values = v.median()
                elif baseline == "mean":
                    baseline_values = v.mean()
                elif baseline == "begin":
                    baseline_values = v.iloc[0, :]
                else:
                    raise ValueError(
                        "%s is not implemented please use {mean, median, Fex}"
                        % baseline
                    )
            elif isinstance(baseline, (Series, FexSeries)):
                baseline_values = baseline
            elif isinstance(baseline, (Fex, DataFrame)):
                raise ValueError("Must pass in a FexSeries not a FexSeries Instance.")

            if normalize == "db":
                out = out.append(
                    10 * np.log10(v - baseline_values) / baseline_values,
                    session_id=k,
                )
            elif normalize == "pct":
                out = out.append(
                    100 * (v - baseline_values) / baseline_values, session_id=k
                )
            else:
                out = out.append(v - baseline_values, session_id=k)
    return out.__finalize__(self)

clean(detrend=True, standardize=True, confounds=None, low_pass=None, high_pass=None, ensure_finite=False, ignore_sessions=False)

Clean a time-series signal: detrend, regress confounds, filter, standardize.

Operations applied in this order: detrend -> regress confounds -> Butterworth low/high-pass filter -> standardize -> ensure_finite. If Fex.sessions is set and ignore_sessions=False, each session is cleaned independently.

Parameters:

Name	Type	Description	Default
detrend		subtract a linear trend from each column.	True
standardize		rescale each column to zero-mean unit-variance.	True
confounds		optional (T,) or (T, n_conf) confounds regressed out via OLS.	None
low_pass		low-pass cutoff in Hz.	None
high_pass		high-pass cutoff in Hz.	None
ensure_finite		replace NaN/Inf with zero in the output.	False
ignore_sessions		if True, treat all rows as one session.	False

Returns:

Type	Description
	cleaned Fex instance.

Source code in feat/data.py

def clean(
    self,
    detrend=True,
    standardize=True,
    confounds=None,
    low_pass=None,
    high_pass=None,
    ensure_finite=False,
    ignore_sessions=False,
):
    """Clean a time-series signal: detrend, regress confounds, filter, standardize.

    Operations applied in this order: detrend -> regress confounds ->
    Butterworth low/high-pass filter -> standardize -> ensure_finite. If
    ``Fex.sessions`` is set and ``ignore_sessions=False``, each session
    is cleaned independently.

    Args:
        detrend: subtract a linear trend from each column.
        standardize: rescale each column to zero-mean unit-variance.
        confounds: optional ``(T,)`` or ``(T, n_conf)`` confounds regressed
            out via OLS.
        low_pass: low-pass cutoff in Hz.
        high_pass: high-pass cutoff in Hz.
        ensure_finite: replace NaN/Inf with zero in the output.
        ignore_sessions: if True, treat all rows as one session.

    Returns:
        cleaned Fex instance.
    """
    if self.sessions is not None and not ignore_sessions:
        sessions = self.sessions
    else:
        sessions = None
    cleaned = clean_signal(
        self.values,
        detrend=detrend,
        standardize=standardize,
        confounds=confounds,
        low_pass=low_pass,
        high_pass=high_pass,
        ensure_finite=ensure_finite,
        sampling_freq=float(self.sampling_freq),
        runs=sessions,
    )
    return self.__class__(
        pd.DataFrame(cleaned, columns=self.columns),
        sampling_freq=self.sampling_freq,
        features=self.features,
        sessions=self.sessions,
    ).__finalize__(self)

compute_identities(threshold=0.8, method='gallery', min_cluster_size=2, inplace=False)

Compute Identities from face-identity embeddings.

Parameters:

Name	Type	Description	Default
threshold		cosine-similarity cutoff for the same person.	0.8
method		"gallery" (default; single-pass, memory-efficient, scales to large videos), "hdbscan" (density-based, emits "Unknown" for outliers), or "connected" (legacy exact single-linkage — O(N²) memory, can OOM on long videos).	'gallery'
min_cluster_size		minimum cluster size for method="hdbscan".	2
inplace		write the Identity column in place instead of returning a copy.	False

Source code in feat/data.py

def compute_identities(self, threshold=0.8, method="gallery",
                       min_cluster_size=2, inplace=False):
    """Compute Identities from face-identity embeddings.

    Args:
        threshold: cosine-similarity cutoff for the same person.
        method: ``"gallery"`` (default; single-pass, memory-efficient,
            scales to large videos), ``"hdbscan"`` (density-based, emits
            ``"Unknown"`` for outliers), or ``"connected"`` (legacy exact
            single-linkage — O(N²) memory, can OOM on long videos).
        min_cluster_size: minimum cluster size for ``method="hdbscan"``.
        inplace: write the ``Identity`` column in place instead of returning
            a copy.
    """
    target = self if inplace else self.copy()
    target["Identity"] = cluster_identities(
        target.identity_embeddings, threshold=threshold,
        method=method, min_cluster_size=min_cluster_size,
    )
    if not inplace:
        return target

decompose(algorithm='pca', axis=1, n_components=None, *args, **kwargs)

Decompose Fex instance

Parameters:

Name	Type	Description	Default
algorithm		(str) Algorithm to perform decomposition types=['pca','ica','nnmf','fa']	'pca'
axis		dimension to decompose [0,1]	1
n_components		(int) number of components. If None then retain as many as possible.	None

Returns:

Name	Type	Description
output		a dictionary of decomposition parameters

Source code in feat/data.py

def decompose(self, algorithm="pca", axis=1, n_components=None, *args, **kwargs):
    """Decompose Fex instance

    Args:
        algorithm: (str) Algorithm to perform decomposition types=['pca','ica','nnmf','fa']
        axis: dimension to decompose [0,1]
        n_components: (int) number of components. If None then retain as many as possible.

    Returns:
        output: a dictionary of decomposition parameters
    """
    out = {}
    out["decomposition_object"] = set_decomposition_algorithm(
        algorithm=algorithm, n_components=n_components, *args, **kwargs
    )
    com_names = ["c%s" % str(x + 1) for x in range(n_components)]
    if axis == 0:
        out["decomposition_object"].fit(self.T)
        out["components"] = self.__class__(
            pd.DataFrame(
                out["decomposition_object"].transform(self.T),
                index=self.columns,
                columns=com_names,
            ),
            sampling_freq=None,
        )
        out["weights"] = self.__class__(
            pd.DataFrame(
                out["decomposition_object"].components_.T,
                index=self.index,
                columns=com_names,
            ),
            sampling_freq=self.sampling_freq,
            features=self.features,
            sessions=self.sessions,
        )
    if axis == 1:
        out["decomposition_object"].fit(self)
        out["components"] = self.__class__(
            pd.DataFrame(
                out["decomposition_object"].transform(self), columns=com_names
            ),
            sampling_freq=self.sampling_freq,
            features=self.features,
            sessions=self.sessions,
        )
        out["weights"] = self.__class__(
            pd.DataFrame(
                out["decomposition_object"].components_,
                index=com_names,
                columns=self.columns,
            ),
            sampling_freq=None,
        ).T
    return out

distance(method='euclidean', **kwargs)

Calculate distance between rows within a Fex() instance.

Parameters:

Name	Type	Description	Default
method		type of distance metric (can use any scikit learn or sciypy metric)	'euclidean'

Returns:

Name	Type	Description
dist		Outputs a 2D distance matrix.

Source code in feat/data.py

def distance(self, method="euclidean", **kwargs):
    """Calculate distance between rows within a Fex() instance.

    Args:
        method: type of distance metric (can use any scikit learn or
                sciypy metric)

    Returns:
        dist: Outputs a 2D distance matrix.

    """
    return pd.DataFrame(pairwise_distances(self, metric=method, **kwargs))

downsample(target, **kwargs)

Downsample Fex columns and return a Fex object.

Parameters:

Name	Type	Description	Default
target	float	target sampling frequency in Hz.	required
kwargs		forwarded to feat.utils.stats.downsample.	{}

Source code in feat/data.py

def downsample(self, target, **kwargs):
    """Downsample Fex columns and return a Fex object.

    Args:
        target(float): target sampling frequency in Hz.
        kwargs: forwarded to feat.utils.stats.downsample.
    """
    df_ds = downsample(
        self, sampling_freq=self.sampling_freq, target=target, **kwargs
    ).__finalize__(self)
    df_ds.sampling_freq = target

    if self.features is not None:
        ds_features = downsample(
            self.features, sampling_freq=self.sampling_freq, target=target, **kwargs
        )
    else:
        ds_features = self.features
    df_ds.features = ds_features
    return df_ds

extract_boft(min_freq=0.06, max_freq=0.66, bank=8, *args, **kwargs)

Extract Bag of Temporal features

Parameters:

Name	Type	Description	Default
min_freq		maximum frequency of temporal filters	0.06
max_freq		minimum frequency of temporal filters	0.66
bank		number of temporal filter banks, filters are on exponential scale	8

Returns:

Name	Type	Description
wavs		list of Morlet wavelets with corresponding freq
hzs		list of hzs for each Morlet wavelet

Source code in feat/data.py

def extract_boft(self, min_freq=0.06, max_freq=0.66, bank=8, *args, **kwargs):
    """Extract Bag of Temporal features

    Args:
        min_freq: maximum frequency of temporal filters
        max_freq: minimum frequency of temporal filters
        bank: number of temporal filter banks, filters are on exponential scale

    Returns:
        wavs: list of Morlet wavelets with corresponding freq
        hzs:  list of hzs for each Morlet wavelet
    """
    # First generate the wavelets
    target_hz = self.sampling_freq
    freqs = np.geomspace(min_freq, max_freq, bank)
    wavs, hzs = [], []
    for i, f in enumerate(freqs):
        wav = np.real(wavelet(f, sampling_freq=target_hz))
        wavs.append(wav)
        hzs.append(str(np.round(freqs[i], 2)))
    wavs = np.array(wavs)[::-1]
    hzs = np.array(hzs)[::-1]
    # # check asymptotes at lowest freq
    # asym = wavs[-1,:10].sum()
    # if asym > .001:
    #     print("Lowest frequency asymptotes at %2.8f " %(wavs[-1,:10].sum()))

    # Convolve data with wavelets
    Feats2Use = self.columns
    feats = pd.DataFrame()
    for feat in Feats2Use:
        _d = self[[feat]].T
        assert _d.isnull().sum().any() == 0, "Data contains NaNs. Cannot convolve. "
        for iw, cm in enumerate(wavs):
            convolved = np.apply_along_axis(
                lambda m: np.convolve(m, cm, mode="full"), axis=1, arr=_d.values
            )
            # Extract bin features.
            out = pd.DataFrame(convolved.T).apply(calc_hist_auc, args=(None))
            # 6 bins hardcoded from calc_hist_auc
            colnames = [
                "pos" + str(i) + "_hz_" + hzs[iw] + "_" + feat for i in range(6)
            ]
            colnames.extend(
                ["neg" + str(i) + "_hz_" + hzs[iw] + "_" + feat for i in range(6)]
            )
            out = out.T
            out.columns = colnames
            feats = pd.concat([feats, out], axis=1)
    return self.__class__(
        feats, sampling_freq=self.sampling_freq, features=self.features
    )

extract_max(ignore_sessions=False)

Extract maximum of each feature

Parameters:

Name	Type	Description	Default
ignore_sessions		(bool) ignore sessions or extract separately by sessions if available.	False

Returns:

Name	Type	Description
fex		(Fex) maximum values for each feature

Source code in feat/data.py

def extract_max(self, ignore_sessions=False):
    """Extract maximum of each feature

    Args:
        ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

    Returns:
        fex: (Fex) maximum values for each feature
    """
    prefix = "max"
    if self.sessions is None or ignore_sessions:
        feats = pd.DataFrame(self.max(numeric_only=True)).T
    else:
        feats = []
        for k, v in self.itersessions():
            feats.append(pd.Series(v.max(numeric_only=True), name=k))
        feats = pd.concat(feats, axis=1).T
    feats = self.__class__(feats)
    feats.columns = f"{prefix}_" + feats.columns
    feats = feats.__finalize__(self)
    if ignore_sessions is False:
        feats.sessions = np.unique(self.sessions)
    feats._update_extracted_colnames(prefix)
    return feats

extract_mean(ignore_sessions=False)

Extract mean of each feature

Parameters:

Name	Type	Description	Default
ignore_sessions		(bool) ignore sessions or extract separately by sessions if available.	False

Returns:

Name	Type	Description
Fex		mean values for each feature

Source code in feat/data.py

def extract_mean(self, ignore_sessions=False):
    """Extract mean of each feature

    Args:
        ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

    Returns:
        Fex: mean values for each feature
    """

    prefix = "mean"
    if self.sessions is None or ignore_sessions:
        feats = pd.DataFrame(self.mean(numeric_only=True)).T
    else:
        feats = []
        for k, v in self.itersessions():
            feats.append(pd.Series(v.mean(numeric_only=True), name=k))
        feats = pd.concat(feats, axis=1).T
    feats = self.__class__(feats)
    feats.columns = f"{prefix}_" + feats.columns
    feats = feats.__finalize__(self)
    if ignore_sessions is False:
        feats.sessions = np.unique(self.sessions)
    feats._update_extracted_colnames(prefix)
    return feats

extract_min(ignore_sessions=False)

Extract minimum of each feature

Parameters:

Name	Type	Description	Default
ignore_sessions		(bool) ignore sessions or extract separately by sessions if available.	False

Returns:

Name	Type	Description
Fex		(Fex) minimum values for each feature

Source code in feat/data.py

def extract_min(self, ignore_sessions=False):
    """Extract minimum of each feature

    Args:
        ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

    Returns:
        Fex: (Fex) minimum values for each feature
    """

    prefix = "min"
    if self.sessions is None or ignore_sessions:
        feats = pd.DataFrame(self.min(numeric_only=True)).T
    else:
        feats = []
        for k, v in self.itersessions():
            feats.append(pd.Series(v.min(numeric_only=True), name=k))
        feats = pd.concat(feats, axis=1).T
    feats = self.__class__(feats)
    feats.columns = f"{prefix}_" + feats.columns
    feats = feats.__finalize__(self)
    if ignore_sessions is False:
        feats.sessions = np.unique(self.sessions)
    feats._update_extracted_colnames(prefix)
    return feats

extract_multi_wavelet(min_freq=0.06, max_freq=0.66, bank=8, *args, **kwargs)

Convolve with a bank of morlet wavelets.

Wavelets are equally spaced from min to max frequency. See extract_wavelet for more information and options.

Parameters:

Name	Type	Description	Default
min_freq		(float) minimum frequency to extract	0.06
max_freq		(float) maximum frequency to extract	0.66
bank		(int) size of wavelet bank	8
num_cyc		(float) number of cycles for wavelet	required
mode		(str) feature to extract, e.g., ['complex','filtered','phase','magnitude','power']	required
ignore_sessions		(bool) ignore sessions or extract separately by sessions if available.	required

Returns:

Name	Type	Description
convolved		(Fex instance)

Source code in feat/data.py

def extract_multi_wavelet(
    self, min_freq=0.06, max_freq=0.66, bank=8, *args, **kwargs
):
    """Convolve with a bank of morlet wavelets.

    Wavelets are equally spaced from min to max frequency. See extract_wavelet for more information and options.

    Args:
        min_freq: (float) minimum frequency to extract
        max_freq: (float) maximum frequency to extract
        bank: (int) size of wavelet bank
        num_cyc: (float) number of cycles for wavelet
        mode: (str) feature to extract, e.g., ['complex','filtered','phase','magnitude','power']
        ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

    Returns:
        convolved: (Fex instance)
    """
    out = []
    for f in np.geomspace(min_freq, max_freq, bank):
        out.append(self.extract_wavelet(f, *args, **kwargs))
    return self.__class__(
        pd.concat(out, axis=1),
        sampling_freq=self.sampling_freq,
        features=self.features,
        sessions=self.sessions,
    )

extract_sem(ignore_sessions=False)

Extract std of each feature

Parameters:

Name	Type	Description	Default
ignore_sessions		(bool) ignore sessions or extract separately by sessions if available.	False

Returns:

Name	Type	Description
Fex		mean values for each feature

Source code in feat/data.py

def extract_sem(self, ignore_sessions=False):
    """Extract std of each feature

    Args:
        ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

    Returns:
        Fex: mean values for each feature
    """

    prefix = "sem"
    if self.sessions is None or ignore_sessions:
        feats = pd.DataFrame(self.sem(numeric_only=True)).T
    else:
        feats = []
        for k, v in self.itersessions():
            feats.append(pd.Series(v.sem(numeric_only=True), name=k))
        feats = pd.concat(feats, axis=1).T
    feats = self.__class__(feats)
    feats.columns = f"{prefix}_" + feats.columns
    feats = feats.__finalize__(self)
    if ignore_sessions is False:
        feats.sessions = np.unique(self.sessions)
    feats._update_extracted_colnames(prefix)
    return feats

extract_std(ignore_sessions=False)

Extract std of each feature

Parameters:

Name	Type	Description	Default
ignore_sessions		(bool) ignore sessions or extract separately by sessions if available.	False

Returns:

Name	Type	Description
Fex		mean values for each feature

Source code in feat/data.py

def extract_std(self, ignore_sessions=False):
    """Extract std of each feature

    Args:
        ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

    Returns:
        Fex: mean values for each feature
    """

    prefix = "std"
    if self.sessions is None or ignore_sessions:
        feats = pd.DataFrame(self.std(numeric_only=True)).T
    else:
        feats = []
        for k, v in self.itersessions():
            feats.append(pd.Series(v.std(numeric_only=True), name=k))
        feats = pd.concat(feats, axis=1).T
    feats = self.__class__(feats)
    feats.columns = f"{prefix}_" + feats.columns
    feats = feats.__finalize__(self)
    if ignore_sessions is False:
        feats.sessions = np.unique(self.sessions)
    feats._update_extracted_colnames(prefix)
    return feats

extract_summary(mean=True, std=True, sem=True, max=True, min=True, ignore_sessions=False, *args, **kwargs)

Extract summary of multiple features

Parameters:

Name	Type	Description	Default
mean		(bool) extract mean of features	True
std		(bool) extract std of features	True
sem		(bool) extract sem of features	True
max		(bool) extract max of features	True
min		(bool) extract min of features	True
ignore_sessions		(bool) ignore sessions or extract separately by sessions if available.	False

Returns:

Name	Type	Description
fex		(Fex)

Source code in feat/data.py

def extract_summary(
    self,
    mean=True,
    std=True,
    sem=True,
    max=True,
    min=True,
    ignore_sessions=False,
    *args,
    **kwargs,
):
    """Extract summary of multiple features

    Args:
        mean: (bool) extract mean of features
        std: (bool) extract std of features
        sem: (bool) extract sem of features
        max: (bool) extract max of features
        min: (bool) extract min of features
        ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

    Returns:
        fex: (Fex)
    """

    if mean is max is min is False:
        raise ValueError("At least one of min, max, mean must be True")

    out = self.__class__().__finalize__(self)
    if ignore_sessions is False:
        out.sessions = np.unique(self.sessions)

    col_updates = []

    if mean:
        new = self.extract_mean(ignore_sessions=ignore_sessions, *args, **kwargs)
        out = out.append(new, axis=1)
        col_updates.append("mean")
    if sem:
        new = self.extract_sem(ignore_sessions=ignore_sessions, *args, **kwargs)
        out = out.append(new, axis=1)
        col_updates.append("sem")
    if std:
        new = self.extract_std(ignore_sessions=ignore_sessions, *args, **kwargs)
        out = out.append(new, axis=1)
        col_updates.append("std")
    if max:
        new = self.extract_max(ignore_sessions=ignore_sessions, *args, **kwargs)
        out = out.append(new, axis=1)
        col_updates.append("max")
    if min:
        new = self.extract_min(ignore_sessions=ignore_sessions, *args, **kwargs)
        out = out.append(new, axis=1)
        col_updates.append("min")

    out._update_extracted_colnames(mode="reset")
    out._update_extracted_colnames(col_updates)

    return out

extract_wavelet(freq, num_cyc=3, mode='complex', ignore_sessions=False)

Perform feature extraction by convolving with a complex morlet wavelet

Parameters:

Name	Type	Description	Default
freq		(float) frequency to extract	required
num_cyc		(float) number of cycles for wavelet	3
mode		(str) feature to extract, e.g., 'complex','filtered','phase','magnitude','power']	'complex'
ignore_sessions		(bool) ignore sessions or extract separately by sessions if available.	False

Returns:

Name	Type	Description
convolved		(Fex instance)

Source code in feat/data.py

def extract_wavelet(self, freq, num_cyc=3, mode="complex", ignore_sessions=False):
    """Perform feature extraction by convolving with a complex morlet wavelet

    Args:
        freq: (float) frequency to extract
        num_cyc: (float) number of cycles for wavelet
        mode: (str) feature to extract, e.g., 'complex','filtered','phase','magnitude','power']
        ignore_sessions: (bool) ignore sessions or extract separately by sessions if available.

    Returns:
        convolved: (Fex instance)

    """
    wav = wavelet(freq, sampling_freq=self.sampling_freq, num_cyc=num_cyc)
    if self.sessions is None or ignore_sessions:
        convolved = self.__class__(
            pd.DataFrame(
                {x: convolve(y, wav, mode="same") for x, y in self.iteritems()}
            ),
            sampling_freq=self.sampling_freq,
        )
    else:
        convolved = self.__class__(sampling_freq=self.sampling_freq)
        for k, v in self.itersessions():
            session = self.__class__(
                pd.DataFrame(
                    {x: convolve(y, wav, mode="same") for x, y in v.items()}
                ),
                sampling_freq=self.sampling_freq,
            )
            convolved = convolved.append(session, session_id=k)
    if mode == "complex":
        convolved = convolved
    elif mode == "filtered":
        convolved = np.real(convolved)
    elif mode == "phase":
        convolved = np.angle(convolved)
    elif mode == "magnitude":
        convolved = np.abs(convolved)
    elif mode == "power":
        convolved = np.abs(convolved) ** 2
    else:
        raise ValueError(
            "Mode must be ['complex','filtered','phase'," "'magnitude','power']"
        )
    convolved = self.__class__(
        convolved,
        sampling_freq=self.sampling_freq,
        features=self.features,
        sessions=self.sessions,
    )
    convolved.columns = "f" + "%s" % round(freq, 2) + "_" + mode + "_" + self.columns
    return convolved

iplot_detections(bounding_boxes=False, landmarks=False, aus=False, poses=False, emotions=False, gazes=False, emotions_position='right', emotions_opacity=1.0, emotions_color='white', emotions_size=14, frame_duration=1000, facebox_color='cyan', facebox_width=3, pose_width=2, landmark_color='white', landmark_width=2, gaze_color='yellow', gaze_width=3, au_cmap='Blues', au_heatmap_resolution=1000, au_opacity=0.9, *args, **kwargs)

Plot Py-FEAT detection results using the plotly backend. There are two plot types. For a single frame, uses plot_singleframe_detections() to create an interactive plot where the different detector outputs can be toggled on or off. For multiple frames (a video), uses plot_multipleframes_detections() to create a plotly animation with a play/pause control and a frame slider; overlays can't be toggled live in the animation, so the detector outputs to draw must be prespecified.

Parameters:

Name	Type	Description	Default
bounding_boxes	bool	will include faceboxes when plotting detector output for multiple frames.	False
landmarks	bool	will include face landmarks when plotting detector output for multiple frames.	False
poses	bool	will include 3 axis line plot indicating x,y,z rotation information when plotting detector output for multiple frames.	False
aus	bool	will include action unit heatmaps when plotting detector output for multiple frames.	False
emotions	bool	will add text annotations indicating probability of discrete emotion when plotting detector output for multiple frames.	False
emotions_position	str	position around facebox to plot emotion annotations. default='right'	'right'
emotions_opacity	float	opacity of emotion annotation text (default=1.)	1.0
emotions_color	str	color of emotion annotation text (default='white')	'white'
emotions_size	int	size of emotion annotations (default=14)	14
frame_duration	int	duration in milliseconds to play each frame if plotting multiple frames (default=1000)	1000
facebox_color	str	color of facebox bounding box (default="cyan")	'cyan'
facebox_width	int	line width of facebox bounding box (default=3)	3
pose_width	int	line width of pose rotation plot (default=2)	2
landmark_color	str	color of landmark detectors (default="white")	'white'
landmark_width	int	line width of landmark detectors (default=2)	2
au_cmap	str	colormap to use for AU heatmap (default='Blues')	'Blues'
au_heatmap_resolution	int	resolution of heatmap values (default=1000)	1000
au_opacity	float	opacity of AU heatmaps (default=0.9)	0.9

Returns:

Type	Description
	a plotly figure instance

Source code in feat/data.py

def iplot_detections(
    self,
    bounding_boxes=False,
    landmarks=False,
    aus=False,
    poses=False,
    emotions=False,
    gazes=False,
    emotions_position="right",
    emotions_opacity=1.0,
    emotions_color="white",
    emotions_size=14,
    frame_duration=1000,
    facebox_color="cyan",
    facebox_width=3,
    pose_width=2,
    landmark_color="white",
    landmark_width=2,
    gaze_color="yellow",
    gaze_width=3,
    au_cmap="Blues",
    au_heatmap_resolution=1000,
    au_opacity=0.9,
    *args,
    **kwargs,
):
    """Plot Py-FEAT detection results using the plotly backend. There are
    two plot types. For a single frame, uses plot_singleframe_detections()
    to create an interactive plot where the different detector outputs can
    be toggled on or off. For multiple frames (a video), uses
    plot_multipleframes_detections() to create a plotly animation with a
    play/pause control and a frame slider; overlays can't be toggled live
    in the animation, so the detector outputs to draw must be prespecified.

    Args:
        bounding_boxes (bool): will include faceboxes when plotting detector output for multiple frames.
        landmarks (bool): will include face landmarks when plotting detector output for multiple frames.
        poses (bool): will include 3 axis line plot indicating x,y,z rotation information when plotting detector output for multiple frames.
        aus (bool): will include action unit heatmaps when plotting detector output for multiple frames.
        emotions (bool): will add text annotations indicating probability of discrete emotion when plotting detector output for multiple frames.
        emotions_position (str): position around facebox to plot emotion annotations. default='right'
        emotions_opacity (float): opacity of emotion annotation text (default=1.)
        emotions_color (str): color of emotion annotation text (default='white')
        emotions_size (int): size of emotion annotations (default=14)
        frame_duration (int): duration in milliseconds to play each frame if plotting multiple frames (default=1000)
        facebox_color (str): color of facebox bounding box (default="cyan")
        facebox_width (int): line width of facebox bounding box (default=3)
        pose_width (int): line width of pose rotation plot (default=2)
        landmark_color (str): color of landmark detectors (default="white")
        landmark_width (int): line width of landmark detectors (default=2)
        au_cmap (str): colormap to use for AU heatmap (default='Blues')
        au_heatmap_resolution (int): resolution of heatmap values (default=1000)
        au_opacity (float): opacity of AU heatmaps (default=0.9)

    Returns:
        a plotly figure instance
    """

    n_frames = len(self["frame"].unique())

    if n_frames == 1:
        fig = self.plot_singleframe_detections(
            bounding_boxes=bounding_boxes,
            landmarks=landmarks,
            poses=poses,
            emotions=emotions,
            aus=aus,
            gazes=gazes,
            facebox_color=facebox_color,
            facebox_width=facebox_width,
            pose_width=pose_width,
            landmark_color=landmark_color,
            landmark_width=landmark_width,
            gaze_color=gaze_color,
            gaze_width=gaze_width,
            emotions_position=emotions_position,
            emotions_opacity=emotions_opacity,
            emotions_color=emotions_color,
            emotions_size=emotions_size,
            au_cmap=au_cmap,
            au_heatmap_resolution=au_heatmap_resolution,
            au_opacity=au_opacity,
            *args,
            **kwargs,
        )
        return fig
    return self.plot_multipleframes_detections(
        bounding_boxes=bounding_boxes,
        landmarks=landmarks,
        aus=aus,
        poses=poses,
        emotions=emotions,
        gazes=gazes,
        image_opacity=kwargs.pop("image_opacity", 0.9),
        facebox_color=facebox_color,
        facebox_width=facebox_width,
        pose_width=pose_width,
        landmark_color=landmark_color,
        landmark_width=landmark_width,
        gaze_color=gaze_color,
        gaze_width=gaze_width,
        emotions_position=emotions_position,
        emotions_opacity=emotions_opacity,
        emotions_color=emotions_color,
        emotions_size=emotions_size,
        au_cmap=au_cmap,
        au_heatmap_resolution=au_heatmap_resolution,
        au_opacity=au_opacity,
        frame_duration=frame_duration,
    )

isc(col, index='frame', columns='input', method='pearson')

[summary]

Parameters:

Name	Type	Description	Default
col	str]	Column name to compute the ISC for.	required
index	str	Column to be used in computing ISC. Usually this would be the column identifying the time such as the number of the frame. Defaults to "frame".	'frame'
columns	str	Column to be used for ISC. Usually this would be the column identifying the video or subject. Defaults to "input".	'input'
method	str	Method to use for correlation pearson, kendall, or spearman. Defaults to "pearson".	'pearson'

Returns:

Name	Type	Description
DataFrame		Correlation matrix with index as columns

Source code in feat/data.py

def isc(self, col, index="frame", columns="input", method="pearson"):
    """[summary]

    Args:
        col (str]): Column name to compute the ISC for.
        index (str, optional): Column to be used in computing ISC. Usually this would be the column identifying the time such as the number of the frame. Defaults to "frame".
        columns (str, optional): Column to be used for ISC. Usually this would be the column identifying the video or subject. Defaults to "input".
        method (str, optional): Method to use for correlation pearson, kendall, or spearman. Defaults to "pearson".

    Returns:
        DataFrame: Correlation matrix with index as columns
    """
    if index is None:
        index = "frame"
    if columns is None:
        columns = "input"
    mat = pd.pivot_table(self, index=index, columns=columns, values=col).corr(
        method=method
    )
    return mat

itersessions()

Iterate over Fex sessions as (session, series) pairs.

Returns:

Name	Type	Description
it		a generator that iterates over the sessions of the fex instance

Source code in feat/data.py

def itersessions(self):
    """Iterate over Fex sessions as (session, series) pairs.

    Returns:
        it: a generator that iterates over the sessions of the fex instance

    """
    for x in np.unique(self.sessions):
        yield x, self.loc[self.sessions == x, :]

landmarks_dlib68_xy(row=None)

Return (x, y) arrays in dlib-68 layout, regardless of source detector.

Provides a uniform 2D-landmark interface across detectors so plotting and downstream consumers don't have to branch on the underlying topology:

• Detectorv1 (Detectorv1(landmark_model="mobilefacenet")) outputs 136-d [x_0..x_67, y_0..y_67] (axis-major) — returned as-is.
• MPDetector outputs the 478-vertex MediaPipe FaceMesh as 1434 named columns x_i / y_i / z_i (i in 0..477), in image-space pixels. The 68 dlib-equivalent vertices are sampled by name via DLIB68_FROM_MP478 (see feat.utils.blendshape_to_au); z dropped.

Parameters:

Name	Type	Description	Default
row		optional pandas Series (single Fex row). If None, operates on all rows of self and returns shape (n_rows, 68) per axis. Otherwise returns shape (68,) per axis.	None

Returns:

Type	Description
(x, y)	two numpy arrays of dlib-68 layout. Shapes (68,) if row is given, else (n_rows, 68).

Source code in feat/data.py

def landmarks_dlib68_xy(self, row=None):
    """Return (x, y) arrays in dlib-68 layout, regardless of source detector.

    Provides a uniform 2D-landmark interface across detectors so plotting
    and downstream consumers don't have to branch on the underlying topology:

    - **Detectorv1** (``Detectorv1(landmark_model="mobilefacenet")``) outputs
      136-d ``[x_0..x_67, y_0..y_67]`` (axis-major) — returned as-is.
    - **MPDetector** outputs the 478-vertex MediaPipe FaceMesh as 1434
      named columns ``x_i / y_i / z_i`` (i in 0..477), in image-space
      pixels. The 68 dlib-equivalent vertices are sampled by name via
      ``DLIB68_FROM_MP478`` (see ``feat.utils.blendshape_to_au``); z dropped.

    Args:
        row: optional pandas Series (single Fex row). If ``None``, operates
            on all rows of ``self`` and returns shape ``(n_rows, 68)`` per
            axis. Otherwise returns shape ``(68,)`` per axis.

    Returns:
        (x, y): two numpy arrays of dlib-68 layout. Shapes ``(68,)`` if
            ``row`` is given, else ``(n_rows, 68)``.
    """
    if self.landmark_columns is None:
        raise ValueError("landmark_columns is not set on this Fex")
    n_lm_cols = len(self.landmark_columns)
    source = self if row is None else row

    if n_lm_cols == 136:
        arr = source[self.landmark_columns].values.astype(np.float64)
        if row is None:
            return arr[:, :68], arr[:, 68:]
        return arr[:68], arr[68:]
    elif n_lm_cols == 1434:
        x_cols = [f"x_{i}" for i in DLIB68_FROM_MP478]
        y_cols = [f"y_{i}" for i in DLIB68_FROM_MP478]
        x = source[x_cols].values.astype(np.float64)
        y = source[y_cols].values.astype(np.float64)
        return x, y
    else:
        raise ValueError(
            f"Unexpected landmark_columns length ({n_lm_cols}); expected 136 "
            "(dlib-68) or 1434 (MP-478 mesh)."
        )

plot_detections(faces='landmarks', faceboxes=True, muscles=False, poses=False, gazes=False, add_titles=True, au_barplot=True, emotion_barplot=True, plot_original_image=True)

Plots detection results by Feat. Can control plotting of face, AU barplot and Emotion barplot. The faces kwarg controls whether facial landmarks are draw on top of input images or whether faces are visualized using Py-Feat's AU visualization model using detected AUs. If detection was performed on a video an faces = 'landmarks', only an outline of the face will be draw without loading the underlying vidoe frame to save memory.

Parameters:

Name	Type	Description	Default
faces	str	'landmarks' to draw detected landmarks or 'aus' to	'landmarks'
faceboxes	bool	Whether to draw the bounding box around detected	True
muscles	bool	Whether to draw muscles from AU activity. Only	False
poses	bool	Whether to draw facial poses. Only applies if	False
gazes	bool	Whether to draw gaze vectors. When	False
add_titles	bool	Whether to add the file name as a title above	True
au_barplot	bool	Whether to include a subplot for au detections. Defaults to True.	True
emotion_barplot	bool	Whether to include a subplot for emotion detections. Defaults to True.	True

Returns:

Name	Type	Description
list		list of matplotlib figures

Source code in feat/data.py

def plot_detections(
    self,
    faces="landmarks",
    faceboxes=True,
    muscles=False,
    poses=False,
    gazes=False,
    add_titles=True,
    au_barplot=True,
    emotion_barplot=True,
    plot_original_image=True,
):
    """
    Plots detection results by Feat. Can control plotting of face, AU barplot and
    Emotion barplot. The faces kwarg controls whether facial landmarks are draw on
    top of input images or whether faces are visualized using Py-Feat's AU
    visualization model using detected AUs. If detection was performed on a video an
    faces = 'landmarks', only an outline of the face will be draw without loading
    the underlying vidoe frame to save memory.


    Args:
        faces (str, optional): 'landmarks' to draw detected landmarks or 'aus' to
        generate a face from AU detections using Py-Feat's AU landmark model.
        Defaults to 'landmarks'.
        faceboxes (bool, optional): Whether to draw the bounding box around detected
        faces. Only applies if faces='landmarks'. Defaults to True.
        muscles (bool, optional): Whether to draw muscles from AU activity. Only
        applies if faces='aus'. Defaults to False.
        poses (bool, optional): Whether to draw facial poses. Only applies if
        faces='landmarks'. Defaults to False.
        gazes (bool, optional): Whether to draw gaze vectors. When
        faces='landmarks', a yellow arrow is drawn from each face's
        bbox center in the predicted gaze direction (using
        ``gaze_pitch``/``gaze_yaw`` columns from the L2CS gaze detector).
        When faces='aus', the gaze deflects the synthetic face's pupil
        positions. Defaults to False.
        add_titles (bool, optional): Whether to add the file name as a title above
        the face. Defaults to True.
        au_barplot (bool, optional): Whether to include a subplot for au detections. Defaults to True.
        emotion_barplot (bool, optional): Whether to include a subplot for emotion detections. Defaults to True.


    Returns:
        list: list of matplotlib figures
    """

    # Plotting logic, eventually refactor me!:
    # Possible detections:
    # 1. Single image - single-face
    # 2. Single image - multi-face
    # 3. Multi image - single-face per image
    # 4. Multi image - multi-face per image
    # 5. Multi image - single and multi-face mix per image
    # 6. Video - single-face for all frames
    # 7. Video - multi-face for all frames
    # 8. Video - single and multi-face mix across frames

    sns.set_context("paper", font_scale=2.0)

    num_subplots = bool(faces) + au_barplot + emotion_barplot

    all_figs = []
    for _, frame in enumerate(self.frame.unique()):
        # Determine figure width based on how many subplots we have
        f = plt.figure(figsize=(5 * num_subplots, 7))
        spec = f.add_gridspec(ncols=num_subplots, nrows=1)
        col_count = 0
        plot_data = self.query("frame == @frame")
        face_ax, au_ax, emo_ax = None, None, None
        if faces is not False:
            face_ax = f.add_subplot(spec[0, col_count])
            col_count += 1
        if au_barplot:
            au_ax = f.add_subplot(spec[0, col_count])
            col_count += 1
        if emotion_barplot:
            emo_ax = f.add_subplot(spec[0, col_count])
            col_count += 1

        for _face_i, (_, row) in enumerate(plot_data.iterrows()):
            # DRAW LANDMARKS ON IMAGE OR AU FACE
            if face_ax is not None:
                facebox = row[self.facebox_columns].values

                if not faces == "aus" and plot_original_image:
                    file_extension = os.path.basename(row["input"]).split(".")[-1]
                    if file_extension.lower() in [
                        "jpg",
                        "jpeg",
                        "png",
                        "bmp",
                        "tiff",
                        "pdf",
                    ]:
                        img = read_image(row["input"])
                    else:
                        img = decode_video(row["input"])[int(row["frame"])]
                    color = "w"
                    face_ax.imshow(img.permute([1, 2, 0]))
                else:
                    color = "k"  # drawing lineface but not on photo

                # Bbox / pose-axes / gaze-arrow overlays are anchored to
                # the facebox in original-image coords. For faces='aus'
                # the displayed face is the synthetic AU schematic in
                # the viz model's own coord system, so the bbox/pose/
                # gaze overlays don't align with it and just confuse
                # the picture. Skip them in aus mode — gaze is still
                # rendered on the synthetic face via plot_face's
                # gaze= kwarg (handled by _prepare_plot_aus).
                aus_mode = faces == "aus"

                if faceboxes and not aus_mode:
                    rect = Rectangle(
                        (facebox[0], facebox[1]),
                        facebox[2],
                        facebox[3],
                        linewidth=2,
                        edgecolor="cyan",
                        fill=False,
                    )
                    face_ax.add_patch(rect)
                    # Label each box with its face number (1-based, in
                    # detection order) and identity if available, so
                    # multi-face frames are readable (#198). Single-face
                    # frames stay unlabelled.
                    if len(plot_data) > 1:
                        label = str(_face_i + 1)
                        if "Identity" in row.index and pd.notna(row["Identity"]):
                            label += f": {row['Identity']}"
                        face_ax.text(
                            facebox[0], facebox[1] - 4, label,
                            color="cyan", fontsize=12, fontweight="bold",
                            va="bottom", ha="left",
                        )

                if poses and not aus_mode:
                    face_ax = draw_facepose(
                        pose=row[self.facepose_columns[:3]].values,
                        facebox=facebox,
                        ax=face_ax,
                    )

                if gazes and not aus_mode and "gaze_pitch" in row.index and "gaze_yaw" in row.index:
                    face_ax = draw_facegaze(
                        pitch_rad=row["gaze_pitch"],
                        yaw_rad=row["gaze_yaw"],
                        facebox=facebox,
                        ax=face_ax,
                    )

                if faces == "landmarks":
                    # Branch on landmark topology: dlib-68 (136 cols) returns
                    # as-is; MP-478 (1434 cols) samples the 68 dlib-equivalent
                    # vertices via DLIB68_FROM_MP478 so the same draw_lineface
                    # path works for both detectors.
                    currx, curry = self.landmarks_dlib68_xy(row=row)

                    # facelines
                    face_ax = draw_lineface(
                        currx, curry, ax=face_ax, color=color, linewidth=3
                    )
                    if not plot_original_image:
                        face_ax.invert_yaxis()

                elif faces == "aus":
                    # Generate face from AU landmark model
                    if any(self.groupby("frame").size() > 1):
                        raise NotImplementedError(
                            "Plotting using AU landmark model is not currently supported for detections that contain multiple faces"
                        )
                    if muscles:
                        muscles = {"all": "heatmap"}
                    else:
                        muscles = {}
                    aus, gaze, muscles, model = self._prepare_plot_aus(
                        row, muscles=muscles, gaze=gazes
                    )
                    _ = row["input"] if add_titles else None
                    face_ax = plot_face(
                        model=model,
                        au=aus,
                        gaze=gaze,
                        ax=face_ax,
                        muscles=muscles,
                        title=None,
                    )
                else:
                    raise ValueError(
                        f"faces={type(faces)} is not currently supported try ['False','landmarks','aus']."
                    )

                if add_titles:
                    _ = face_ax.set_title(
                        "\n".join(wrap(os.path.basename(row["input"]))),
                        loc="center",
                        wrap=True,
                        fontsize=14,
                    )

                face_ax.axes.get_xaxis().set_visible(False)
                face_ax.axes.get_yaxis().set_visible(False)

        # DRAW AU BARPLOT
        if au_ax is not None:
            _ = plot_data.aus.T.plot(kind="barh", ax=au_ax)
            _ = au_ax.invert_yaxis()
            _ = au_ax.legend().remove()
            _ = au_ax.set(xlim=[0, 1.1], title="Action Units")

        # DRAW EMOTION BARPLOT
        if emo_ax is not None:
            _ = plot_data.emotions.T.plot(kind="barh", ax=emo_ax)
            _ = emo_ax.invert_yaxis()
            _ = emo_ax.legend().remove()
            _ = emo_ax.set(xlim=[0, 1.1], title="Emotions")

        f.tight_layout()
        all_figs.append(f)
    return all_figs

plot_multipleframes_detections(bounding_boxes=False, landmarks=False, aus=False, poses=False, emotions=False, gazes=False, image_opacity=0.9, facebox_color='cyan', facebox_width=3, pose_width=2, landmark_color='white', landmark_width=2, gaze_color='yellow', gaze_width=3, emotions_position='right', emotions_opacity=1.0, emotions_color='white', emotions_size=12, au_cmap='Blues', au_heatmap_resolution=1000, au_opacity=0.9, frame_duration=1000, *args, **kwargs)

Plotly animation across the frames of a multi-frame (video) Fex.

Each frame shows the underlying image with the prespecified overlays (unlike the single-frame plot, overlays can't be toggled live here — pass the ones you want). A play/pause control and a frame slider scrub through the detections. Overlay coordinates use the same y-flipped image space as plot_singleframe_detections. Assumes all frames share one image size (true for a video).

Returns:

Type	Description
	a plotly figure instance

Source code in feat/data.py

def plot_multipleframes_detections(
    self,
    bounding_boxes=False,
    landmarks=False,
    aus=False,
    poses=False,
    emotions=False,
    gazes=False,
    image_opacity=0.9,
    facebox_color="cyan",
    facebox_width=3,
    pose_width=2,
    landmark_color="white",
    landmark_width=2,
    gaze_color="yellow",
    gaze_width=3,
    emotions_position="right",
    emotions_opacity=1.0,
    emotions_color="white",
    emotions_size=12,
    au_cmap="Blues",
    au_heatmap_resolution=1000,
    au_opacity=0.9,
    frame_duration=1000,
    *args,
    **kwargs,
):
    """Plotly animation across the frames of a multi-frame (video) Fex.

    Each frame shows the underlying image with the *prespecified* overlays
    (unlike the single-frame plot, overlays can't be toggled live here — pass
    the ones you want). A play/pause control and a frame slider scrub through
    the detections. Overlay coordinates use the same y-flipped image space as
    ``plot_singleframe_detections``. Assumes all frames share one image size
    (true for a video).

    Returns:
        a plotly figure instance
    """
    frame_ids = sorted(self["frame"].unique())

    def _build(frame_id):
        ff = self.query("frame == @frame_id")
        img = load_pil_img(ff["input"].unique()[0], frame_id)
        w, h = img.width, img.height
        image = dict(
            x=0, sizex=w, y=h, sizey=h, xref="x", yref="y",
            opacity=image_opacity, layer="below", sizing="stretch", source=img,
        )
        shapes = []
        if bounding_boxes:
            shapes += [
                dict(
                    type="rect",
                    x0=r["FaceRectX"], y0=h - r["FaceRectY"],
                    x1=r["FaceRectX"] + r["FaceRectWidth"],
                    y1=h - r["FaceRectY"] - r["FaceRectHeight"],
                    line=dict(color=facebox_color, width=facebox_width),
                )
                for _, r in ff.iterrows()
            ]
        if landmarks:
            shapes += [
                draw_plotly_landmark(r, h, None, line_color=landmark_color,
                                     line_width=landmark_width)
                for _, r in ff.iterrows()
            ]
        if poses:
            shapes += flatten_list(
                [draw_plotly_pose(r, h, None, line_width=pose_width)
                 for _, r in ff.iterrows()]
            )
        if aus:
            shapes += flatten_list(
                [draw_plotly_au(r, h, None, cmap=au_cmap, au_opacity=au_opacity,
                                heatmap_resolution=au_heatmap_resolution)
                 for _, r in ff.iterrows()]
            )
        if gazes:
            shapes += flatten_list(
                [draw_plotly_gaze(r, h, color=gaze_color, line_width=gaze_width)
                 for _, r in ff.iterrows()]
            )
        annotations = []
        if emotions:
            for _, r in ff.iterrows():
                emo = r[self.emotion_columns].sort_values(ascending=False).to_dict()
                xp, yp, align, valign = emotion_annotation_position(
                    r, h, w, emotions_size=emotions_size,
                    emotions_position=emotions_position,
                )
                text = "".join(f"{e}: <i>{v:.2f}</i><br>" for e, v in emo.items())
                annotations.append(dict(
                    text=text, x=xp, y=yp, opacity=emotions_opacity,
                    showarrow=False, align=align, valign=valign, bgcolor="black",
                    font=dict(color=emotions_color, size=emotions_size),
                ))
        return image, shapes, annotations, w, h

    image0, shapes0, annotations0, img_width, img_height = _build(frame_ids[0])

    fig = go.Figure()
    # Invisible trace anchors the axis range so layout images/shapes line up.
    fig.add_trace(go.Scatter(
        x=[0, img_width], y=[0, img_height], mode="markers", marker_opacity=0,
    ))
    fig.update_layout(
        images=[image0], shapes=shapes0, annotations=annotations0,
        width=img_width, height=img_height,
        margin={"l": 0, "r": 0, "t": 0, "b": 0},
    )

    plotly_frames = []
    for frame_id in frame_ids:
        image, shapes, annotations, _, _ = _build(frame_id)
        plotly_frames.append(go.Frame(
            name=str(int(frame_id)),
            layout=dict(images=[image], shapes=shapes, annotations=annotations),
        ))
    fig.frames = plotly_frames

    fig.update_layout(
        updatemenus=[dict(
            type="buttons", direction="left", showactive=False,
            x=0.1, xanchor="left", y=1.12, yanchor="top",
            pad={"r": 10, "t": 10},
            buttons=[
                dict(label="Play", method="animate",
                     args=[None, {"frame": {"duration": frame_duration, "redraw": True},
                                  "fromcurrent": True}]),
                dict(label="Pause", method="animate",
                     args=[[None], {"frame": {"duration": 0, "redraw": False},
                                    "mode": "immediate"}]),
            ],
        )],
        sliders=[dict(
            active=0, x=0.1, len=0.9, pad={"b": 10, "t": 10},
            steps=[dict(
                method="animate", label=str(int(fid)),
                args=[[str(int(fid))], {"frame": {"duration": 0, "redraw": True},
                                        "mode": "immediate"}],
            ) for fid in frame_ids],
        )],
    )
    fig.update_xaxes(visible=False, range=[0, img_width])
    fig.update_yaxes(visible=False, range=[0, img_height], scaleanchor="x")
    return fig

plot_singleframe_detections(bounding_boxes=False, landmarks=False, poses=False, emotions=False, aus=False, gazes=False, image_opacity=0.9, facebox_color='cyan', facebox_width=3, pose_width=2, landmark_color='white', landmark_width=2, gaze_color='yellow', gaze_width=3, emotions_position='right', emotions_opacity=1.0, emotions_color='white', emotions_size=12, au_heatmap_resolution=1000, au_opacity=0.9, au_cmap='Blues', *args, **kwargs)

Function to generate interactive plotly figure to interactively visualize py-feat detectors on a single image frame.

Parameters:

Name	Type	Description	Default
image_opacity	float	opacity of image overlay (default=.9)	0.9
emotions_position	str	position around facebox to plot emotion annotations. default='right'	'right'
emotions_opacity	float	opacity of emotion annotation text (default=1.)	1.0
emotions_color	str	color of emotion annotation text (default='white')	'white'
emotions_size	int	size of emotion annotations (default=14)	12
frame_duration	int	duration in milliseconds to play each frame if plotting multiple frames (default=1000)	required
facebox_color	str	color of facebox bounding box (default="cyan")	'cyan'
facebox_width	int	line width of facebox bounding box (default=3)	3
pose_width	int	line width of pose rotation plot (default=2)	2
landmark_color	str	color of landmark detectors (default="white")	'white'
landmark_width	int	line width of landmark detectors (default=2)	2
au_cmap	str	colormap to use for AU heatmap (default='Blues')	'Blues'
au_heatmap_resolution	int	resolution of heatmap values (default=1000)	1000
au_opacity	float	opacity of AU heatmaps (default=0.9)	0.9

Returns:

Type	Description
	a plotly figure instance

Source code in feat/data.py

def plot_singleframe_detections(
    self,
    bounding_boxes=False,
    landmarks=False,
    poses=False,
    emotions=False,
    aus=False,
    gazes=False,
    image_opacity=0.9,
    facebox_color="cyan",
    facebox_width=3,
    pose_width=2,
    landmark_color="white",
    landmark_width=2,
    gaze_color="yellow",
    gaze_width=3,
    emotions_position="right",
    emotions_opacity=1.0,
    emotions_color="white",
    emotions_size=12,
    au_heatmap_resolution=1000,
    au_opacity=0.9,
    au_cmap="Blues",
    *args,
    **kwargs,
):
    """
    Function to generate interactive plotly figure to interactively visualize py-feat detectors on a single image frame.

    Args:
        image_opacity (float): opacity of image overlay (default=.9)
        emotions_position (str): position around facebox to plot emotion annotations. default='right'
        emotions_opacity (float): opacity of emotion annotation text (default=1.)
        emotions_color (str): color of emotion annotation text (default='white')
        emotions_size (int): size of emotion annotations (default=14)
        frame_duration (int): duration in milliseconds to play each frame if plotting multiple frames (default=1000)
        facebox_color (str): color of facebox bounding box (default="cyan")
        facebox_width (int): line width of facebox bounding box (default=3)
        pose_width (int): line width of pose rotation plot (default=2)
        landmark_color (str): color of landmark detectors (default="white")
        landmark_width (int): line width of landmark detectors (default=2)
        au_cmap (str): colormap to use for AU heatmap (default='Blues')
        au_heatmap_resolution (int): resolution of heatmap values (default=1000)
        au_opacity (float): opacity of AU heatmaps (default=0.9)

    Returns:
        a plotly figure instance
    """

    n_frames = len(self["frame"].unique())

    if n_frames > 1:
        raise ValueError(
            "This function can only plot a single frame. Try using plot_multipleframe_detections() instead."
        )

    # Initialize Image
    frame_id = self["frame"].unique()[0]
    frame_fex = self.query("frame==@frame_id")
    frame_img = load_pil_img(frame_fex["input"].unique()[0], frame_id)
    img_width = frame_img.width
    img_height = frame_img.height

    # Create figure
    fig = go.Figure()

    # Add invisible scatter trace.
    # This trace is added to help the autoresize logic work.
    fig.add_trace(
        go.Scatter(
            x=[0, img_width],
            y=[0, img_height],
            mode="markers",
            marker_opacity=0,
        )
    )

    # Add image
    fig.add_layout_image(
        dict(
            x=0,
            sizex=img_width,
            y=img_height,
            sizey=img_height,
            xref="x",
            yref="y",
            opacity=image_opacity,
            layer="below",
            sizing="stretch",
            source=frame_img,
        )
    )

    # Add Face Bounding Box
    faceboxes_path = [
        dict(
            type="rect",
            x0=row["FaceRectX"],
            y0=img_height - row["FaceRectY"],
            x1=row["FaceRectX"] + row["FaceRectWidth"],
            y1=img_height - row["FaceRectY"] - row["FaceRectHeight"],
            line=dict(color=facebox_color, width=facebox_width),
        )
        for i, row in frame_fex.iterrows()
    ]

    # Add Landmarks
    landmarks_path = [
        draw_plotly_landmark(
            row,
            img_height,
            fig,
            line_color=landmark_color,
            line_width=landmark_width,
        )
        for i, row in frame_fex.iterrows()
    ]

    # Add Pose
    poses_path = flatten_list(
        [
            draw_plotly_pose(row, img_height, fig, line_width=pose_width)
            for i, row in frame_fex.iterrows()
        ]
    )

    # Add Emotions
    emotions_annotations = []
    for i, row in frame_fex.iterrows():
        # Use this Fex's own emotion columns rather than a hardcoded v1
        # list — Detectorv2 emits capitalized names (Neutral/Happy/...),
        # so hardcoding the lowercase v1 set KeyErrors on v2 detections.
        emotion_dict = (
            row[self.emotion_columns]
            .sort_values(ascending=False)
            .to_dict()
        )

        x_position, y_position, align, valign = emotion_annotation_position(
            row,
            img_height,
            img_width,
            emotions_size=emotions_size,
            emotions_position=emotions_position,
        )

        emotion_text = ""
        for emotion in emotion_dict:
            emotion_text += f"{emotion}: <i>{emotion_dict[emotion]:.2f}</i><br>"

        emotions_annotations.append(
            dict(
                text=emotion_text,
                x=x_position,
                y=y_position,
                opacity=emotions_opacity,
                showarrow=False,
                align=align,
                valign=valign,
                bgcolor="black",
                font=dict(color=emotions_color, size=emotions_size),
            )
        )

    # Add AU Heatmaps
    aus_path = flatten_list(
        [
            draw_plotly_au(
                row,
                img_height,
                fig,
                cmap=au_cmap,
                au_opacity=au_opacity,
                heatmap_resolution=au_heatmap_resolution,
            )
            for i, row in frame_fex.iterrows()
        ]
    )

    # Add Gaze arrows
    gazes_path = flatten_list(
        [
            draw_plotly_gaze(
                row,
                img_height,
                color=gaze_color,
                line_width=gaze_width,
            )
            for i, row in frame_fex.iterrows()
        ]
    )

    # Configure other layout
    fig.update_layout(
        width=img_width,
        height=img_height,
        margin={"l": 0, "r": 0, "t": 0, "b": 0},
    )

    # Configure axes
    fig.update_xaxes(visible=False, range=[0, img_width])

    fig.update_yaxes(
        visible=False,
        range=[0, img_height],
        scaleanchor="x",  # the scaleanchor attribute ensures that the aspect ratio stays constant
    )

    # Add ALL overlays unconditionally with per-shape `visible` flags
    # set from the corresponding function arguments — that way each
    # overlay group can be toggled independently via the buttons below
    # (the prior approach of conditionally adding shapes meant clicking
    # one button replaced the entire `layout.shapes` array, so only
    # one overlay could be visible at a time).
    group_indices = {"bbox": [], "landmarks": [], "poses": [], "aus": [], "gazes": []}
    initial_vis = {
        "bbox": bool(bounding_boxes),
        "landmarks": bool(landmarks),
        "poses": bool(poses),
        "aus": bool(aus),
        "gazes": bool(gazes),
    }
    for group, shapes in (
        ("bbox", faceboxes_path),
        ("landmarks", landmarks_path),
        ("poses", poses_path),
        ("aus", aus_path),
        ("gazes", gazes_path),
    ):
        for s in shapes:
            idx = len(fig.layout.shapes)
            # add_shape() emits an immutable Shape; clone the dict so
            # we can include the `visible` field per our group flag.
            s = dict(s)
            s["visible"] = initial_vis[group]
            fig.add_shape(s)
            group_indices[group].append(idx)

    # Emotions are annotations (not shapes); track separately.
    emotion_indices = []
    for ann in emotions_annotations:
        idx = len(fig.layout.annotations)
        ann = dict(ann)
        ann["visible"] = bool(emotions)
        fig.add_annotation(ann)
        emotion_indices.append(idx)

    # Build toggle buttons. Each button's args (first click) sets the
    # group's visibility to the OPPOSITE of its initial state, and
    # args2 (second click) sets it back. Effectively per-group toggle,
    # independent across groups.
    def _shape_toggle_args(indices, target):
        return [{f"shapes[{i}].visible": target for i in indices}]

    def _annot_toggle_args(indices, target):
        return [{f"annotations[{i}].visible": target for i in indices}]

    buttons = []
    for label, group in (
        ("Bounding Box", "bbox"),
        ("Landmarks", "landmarks"),
        ("Poses", "poses"),
        ("AU", "aus"),
        ("Gaze", "gazes"),
    ):
        idxs = group_indices[group]
        if not idxs:
            continue
        init = initial_vis[group]
        buttons.append(dict(
            method="relayout",
            label=label,
            args=_shape_toggle_args(idxs, not init),
            args2=_shape_toggle_args(idxs, init),
        ))
    if emotion_indices:
        buttons.append(dict(
            method="relayout",
            label="Emotion",
            args=_annot_toggle_args(emotion_indices, not emotions),
            args2=_annot_toggle_args(emotion_indices, bool(emotions)),
        ))

    fig.update_layout(
        updatemenus=[
            dict(
                type="buttons",
                direction="left",
                buttons=buttons,
                pad={"r": 10, "t": 10},
                showactive=False,
                x=0.1,
                xanchor="left",
                y=1.12,
                yanchor="top",
            )
        ]
    )

    return fig

predict(X, y, model=LinearRegression, cv_kwargs={'cv': 5}, *args, **kwargs)

Predicts y from X using a sklearn model.

Predict a variable of interest y using your model of choice from X, which can be a list of columns of the Fex instance or a dataframe.

Parameters:

Name	Type	Description	Default
X	list or DataFrame	List of column names or dataframe to be used as features for prediction	required
y	string or array	y values to be predicted	required
model	class	Any sklearn model. Defaults to LinearRegression.	LinearRegression
args, kwargs		Model arguments	required

Returns:

Name	Type	Description
model		Fit model instance.

Source code in feat/data.py

def predict(self, X, y, model=LinearRegression, cv_kwargs={"cv": 5}, *args, **kwargs):
    """Predicts y from X using a sklearn model.

    Predict a variable of interest y using your model of choice from X, which can be a list of columns of the Fex instance or a dataframe.

    Args:
        X (list or DataFrame): List of column names or dataframe to be used as features for prediction
        y (string or array): y values to be predicted
        model (class, optional): Any sklearn model. Defaults to LinearRegression.
        args, kwargs: Model arguments

    Returns:
        model: Fit model instance.
    """

    mX, my = self._parse_features_labels(X, y)

    # user passes an uninitialized class, e.g. LogisticRegression
    if isinstance(model, type):
        clf = model(*args, **kwargs)
    else:
        # user passes an initialized estimator or pipeline, e.g. LogisticRegression()
        clf = model
    scores = cross_val_score(clf, mX, my, **cv_kwargs)
    _ = clf.fit(mX, my)
    return clf, scores

read_feat(filename=None, *args, **kwargs)

Reads facial expression detection results from Feat Detectorv1

Parameters:

Name	Type	Description	Default
filename	string	Path to file. Defaults to None.	None

Returns:

Type	Description
	Fex

Source code in feat/data.py

def read_feat(self, filename=None, *args, **kwargs):
    """Reads facial expression detection results from Feat Detectorv1

    Args:
        filename (string, optional): Path to file. Defaults to None.

    Returns:
        Fex
    """
    # Check if filename exists in metadata.
    if filename is None:
        if self.filename:
            filename = self.filename
        else:
            raise ValueError("filename must be specified.")
    result = read_feat(filename, *args, **kwargs)
    return result

read_file()

Loads file into FEX class

Returns:

Name	Type	Description
DataFrame		Fex class

Source code in feat/data.py

def read_file(self):
    """Loads file into FEX class

    Returns:
        DataFrame: Fex class
    """
    if self.detector == "OpenFace":
        return self.read_openface(self.filename)

    return self.read_feat(self.filename)

read_openface(filename=None, *args, **kwargs)

Reads facial expression detection results from OpenFace

Parameters:

Name	Type	Description	Default
filename	string	Path to file. Defaults to None.	None

Returns:

Type	Description
	Fex

Source code in feat/data.py

def read_openface(self, filename=None, *args, **kwargs):
    """Reads facial expression detection results from OpenFace

    Args:
        filename (string, optional): Path to file. Defaults to None.

    Returns:
        Fex
    """
    if filename is None:
        if self.filename:
            filename = self.filename
        else:
            raise ValueError("filename must be specified.")
    result = read_openface(filename, *args, **kwargs)
    for name in self._metadata:
        attr_value = getattr(self, name, None)
        if attr_value and getattr(result, name, None) is None:
            setattr(result, name, attr_value)
    return result

rectification(std=3)

Removes time points when the face position moved more than N standard deviations from the mean.

Parameters:

Name	Type	Description	Default
std	default 3	standard deviation from mean to remove outlier face locations	3

Returns: data: cleaned FEX object

Source code in feat/data.py

def rectification(self, std=3):
    """Removes time points when the face position moved
    more than N standard deviations from the mean.

    Args:
        std (default 3): standard deviation from mean to remove outlier face locations
    Returns:
        data: cleaned FEX object

    """

    if self.facebox_columns and self.au_columns and self.emotion_columns:
        cleaned = deepcopy(self)
        face_columns = self.facebox_columns
        x_m = self.FaceRectX.mean()
        x_std = self.FaceRectX.std()
        y_m = self.FaceRectY.mean()
        y_std = self.FaceRectY.std()
        x_bool = (self.FaceRectX > std * x_std + x_m) | (
            self.FaceRectX < x_m - std * x_std
        )
        y_bool = (self.FaceRectY > std * y_std + y_m) | (
            self.FaceRectY < y_m - std * y_std
        )
        xy_bool = x_bool | y_bool
        cleaned.loc[
            xy_bool, face_columns + self.au_columns + self.emotion_columns
        ] = np.nan
        return cleaned
    else:
        raise ValueError("Facebox columns need to be defined.")

regress(X, y, fit_intercept=True, *args, **kwargs)

Multiple OLS regression to predict Fex activity (y) from regressors (X).

Parameters:

Name	Type	Description	Default
X	list or str	Independent variable to predict.	required
y	list or str	Dependent variable to be predicted.	required
fit_intercept	bool	Whether to add intercept before fitting. Defaults to True.	True

Returns:

Type	Description
	Dataframe of betas, ses, t-stats, p-values, df, residuals

Source code in feat/data.py

def regress(self, X, y, fit_intercept=True, *args, **kwargs):
    """Multiple OLS regression to predict Fex activity (y) from regressors (X).

    Args:
        X (list or str): Independent variable to predict.
        y (list or str): Dependent variable to be predicted.
        fit_intercept (bool): Whether to add intercept before fitting. Defaults to True.

    Returns:
        Dataframe of betas, ses, t-stats, p-values, df, residuals
    """

    mX, my = self._parse_features_labels(X, y)

    if fit_intercept:
        mX["intercept"] = 1

    b, se, t, p, df, res = regress(mX.to_numpy(), my.to_numpy(), *args, **kwargs)
    b_df = pd.DataFrame(b, index=mX.columns, columns=my.columns)
    se_df = pd.DataFrame(se, index=mX.columns, columns=my.columns)
    t_df = pd.DataFrame(t, index=mX.columns, columns=my.columns)
    p_df = pd.DataFrame(p, index=mX.columns, columns=my.columns)
    df_df = pd.DataFrame(
        np.full((len(mX.columns), len(my.columns)), df),
        index=mX.columns,
        columns=my.columns,
    )
    res_df = pd.DataFrame(res, columns=my.columns)
    return b_df, se_df, t_df, p_df, df_df, res_df

ttest_1samp(popmean=0)

Conducts 1 sample ttest.

Uses scipy.stats.ttest_1samp to conduct 1 sample ttest

Parameters:

Name	Type	Description	Default
popmean	int	Population mean to test against. Defaults to 0.	0
threshold_dict	[type]	Dictionary for thresholding. Defaults to None. [NOT IMPLEMENTED]	required

Returns:

Type	Description
	t, p: t-statistics and p-values

Source code in feat/data.py

def ttest_1samp(self, popmean=0):
    """Conducts 1 sample ttest.

    Uses scipy.stats.ttest_1samp to conduct 1 sample ttest

    Args:
        popmean (int, optional): Population mean to test against. Defaults to 0.
        threshold_dict ([type], optional): Dictionary for thresholding. Defaults to None. [NOT IMPLEMENTED]

    Returns:
        t, p: t-statistics and p-values
    """
    return ttest_1samp(self, popmean)

ttest_ind(col, sessions=None)

Conducts 2 sample ttest.

Uses scipy.stats.ttest_ind to conduct 2 sample ttest on column col between sessions.

Parameters:

Name	Type	Description	Default
col	str	Column names to compare in a t-test between sessions	required
session	array - like	session name to query Fex.sessions, otherwise uses the	required

Returns:

Type	Description
	t, p: t-statistics and p-values

Source code in feat/data.py

def ttest_ind(self, col, sessions=None):
    """Conducts 2 sample ttest.

    Uses scipy.stats.ttest_ind to conduct 2 sample ttest on column col between sessions.

    Args:
        col (str): Column names to compare in a t-test between sessions
        session (array-like): session name to query Fex.sessions, otherwise uses the
        unique values in Fex.sessions.

    Returns:
        t, p: t-statistics and p-values
    """

    if sessions is None:
        sessions = pd.Series(self.sessions).unique()

    if len(sessions) != 2:
        raise ValueError(
            f"There must be exactly 2 session types to perform an independent t-test but {len(sessions)} were found."
        )

    sess1, sess2 = sessions
    a_mask = self.sessions == sess1
    a_mask = a_mask.values if isinstance(self.sessions, pd.Series) else a_mask
    b_mask = self.sessions == sess2
    b_mask = b_mask.values if isinstance(self.sessions, pd.Series) else b_mask
    a = self.loc[a_mask, col]
    b = self.loc[b_mask, col]

    return ttest_ind(a, b)

update_sessions(new_sessions)

Returns a copy of the Fex dataframe with a new sessions attribute after validation. new_sessions should be a dictionary mapping old to new names or an iterable with the same number of rows as the Fex dataframe

Parameters:

Name	Type	Description	Default
new_sessions	(dict, Iterable)	map or list of new session names	required

Returns:

Name	Type	Description
Fex		self

Source code in feat/data.py

def update_sessions(self, new_sessions):
    """
    Returns a copy of the Fex dataframe with a new sessions attribute after
    validation. `new_sessions` should be a dictionary mapping old to new names or an iterable with the same number of rows as the Fex dataframe

    Args:
        new_sessions (dict, Iterable): map or list of new session names

    Returns:
        Fex: self
    """

    out = deepcopy(self)

    if isinstance(new_sessions, dict):
        if not isinstance(out.sessions, pd.Series):
            out.sessions = pd.Series(out.sessions)

        out.sessions = out.sessions.map(new_sessions)

    elif isinstance(new_sessions, Iterable):
        if len(new_sessions) != out.shape[0]:
            raise ValueError(
                f"When new_sessions are not a dictionary then they must but an iterable with length == the number of rows of this Fex dataframe {out.shape[0]}, but they have length {len(new_sessions)}."
            )

        out.sessions = new_sessions

    else:
        raise TypeError(
            f"new_sessions must be either be a dictionary mapping between old and new session values or an iterable with the same number of rows as this Fex dataframe {out.shape[0]}, but was type: {type(new_sessions)}"
        )

    return out

upsample(target, target_type='hz', **kwargs)

Upsample Fex columns and return a Fex object.

Parameters:

Name	Type	Description	Default
target	float	upsampling target.	required
target_type		'hz' (default), 'samples', or 'seconds'.	'hz'
kwargs		forwarded to feat.utils.stats.upsample.	{}

Source code in feat/data.py

def upsample(self, target, target_type="hz", **kwargs):
    """Upsample Fex columns and return a Fex object.

    Args:
        target(float): upsampling target.
        target_type: 'hz' (default), 'samples', or 'seconds'.
        kwargs: forwarded to feat.utils.stats.upsample.
    """
    df_us = upsample(
        self,
        sampling_freq=self.sampling_freq,
        target=target,
        target_type=target_type,
        **kwargs,
    )
    if self.features is not None:
        us_features = upsample(
            self.features,
            sampling_freq=self.sampling_freq,
            target=target,
            target_type=target_type,
            **kwargs,
        )
    else:
        us_features = self.features
    return self.__class__(df_us, sampling_freq=target, features=us_features)

FexSeries

Bases: Series

This is a sub-class of pandas series. While not having additional methods of it's own required to retain normal slicing functionality for the Fex class, i.e. how slicing is typically handled in pandas. All methods should be called on Fex below.

Source code in feat/data.py

class FexSeries(Series):
    """
    This is a sub-class of pandas series. While not having additional methods
    of it's own required to retain normal slicing functionality for the
    Fex class, i.e. how slicing is typically handled in pandas.
    All methods should be called on Fex below.
    """

    def __init__(self, *args, **kwargs):
        ### Columns ###
        self.au_columns = kwargs.pop("au_columns", None)
        self.emotion_columns = kwargs.pop("emotion_columns", None)
        self.facebox_columns = kwargs.pop("facebox_columns", None)
        self.landmark_columns = kwargs.pop("landmark_columns", None)
        self.facepose_columns = kwargs.pop("facepose_columns", None)
        self.identity_columns = kwargs.pop("identity_columns", None)
        self.gaze_columns = kwargs.pop("gaze_columns", None)
        self.blendshape_columns = kwargs.pop("blendshape_columns", None)
        self.time_columns = kwargs.pop(
            "time_columns", ["Timestamp", "MediaTime", "FrameNo", "FrameTime"]
        )
        self.design_columns = kwargs.pop("design_columns", None)

        ### Meta data ###
        self.filename = kwargs.pop("filename", None)
        self.sampling_freq = kwargs.pop("sampling_freq", None)
        self.detector = kwargs.pop("detector", None)
        self.face_model = kwargs.pop("face_model", None)
        self.landmark_model = kwargs.pop("landmark_model", None)
        self.au_model = kwargs.pop("au_model", None)
        self.emotion_model = kwargs.pop("emotion_model", None)
        self.facepose_model = kwargs.pop("facepose_model", None)
        self.identity_model = kwargs.pop("identity_model", None)
        self.gaze_model = kwargs.pop("gaze_model", None)
        self.features = kwargs.pop("features", None)
        self.sessions = kwargs.pop("sessions", None)
        super().__init__(*args, **kwargs)

    _metadata = [
        "au_columns",
        "emotion_columns",
        "facebox_columns",
        "landmark_columns",
        "facepose_columns",
        "identity_columns",
        "gaze_columns",
        "blendshape_columns",
        "time_columns",
        "design_columns",
        "fex_columns",
        "filename",
        "sampling_freq",
        "features",
        "sessions",
        "detector",
        "face_model",
        "landmark_model",
        "au_model",
        "emotion_model",
        "facepose_model",
        "identity_model",
        "gaze_model",
        "verbose",
    ]

    @property
    def _constructor(self):
        return FexSeries

    @property
    def _constructor_expanddim(self):
        return Fex

    def __finalize__(self, other, method=None, **kwargs):
        """Propagate metadata from other to self"""
        for name in self._metadata:
            object.__setattr__(self, name, getattr(other, name, None))
        return self

    @property
    def aus(self):
        """Returns the Action Units data

        Returns:
            DataFrame: Action Units data
        """
        return self[self.au_columns]

    @property
    def emotions(self):
        """Returns the emotion data

        Returns:
            DataFrame: emotion data
        """
        return self[self.emotion_columns]

    @property
    def landmarks(self):
        """Returns the landmark data

        Returns:
            DataFrame: landmark data
        """
        return self[self.landmark_columns]

    # DEPRECATE
    @property
    def landmark(self):
        """Returns the landmark data

        Returns:
            DataFrame: landmark data
        """
        warnings.warn(
            "Fex.landmark has now been renamed to Fex.landmarks", DeprecationWarning
        )
        return self[self.landmark_columns]

    @property
    def poses(self):
        """Returns the facepose data

        Returns:
            DataFrame: facepose data
        """

        return self[self.facepose_columns]

    @property
    def gazes(self):
        """Returns the gaze data (pitch, yaw, optionally combined angle).

        Returns:
            DataFrame: gaze data populated by the active gaze model
            (L2CS by default in v0.7+). Raises AttributeError if the
            Detectorv1 was built with gaze_model=None.
        """
        if not self.gaze_columns:
            raise AttributeError(
                "No gaze columns are registered on this Fex. The active "
                "Detectorv1 may have been built with gaze_model=None."
            )
        return self[self.gaze_columns]

    @property
    def blendshapes(self):
        """Returns the 52 MediaPipe/ARKit blendshape coefficients in [0, 1].

        Raises AttributeError if no blendshape columns are registered (e.g. a
        pre-v2.5 multitask model, or DetectorV1, which has no blendshape head)."""
        if not self.blendshape_columns:
            raise AttributeError(
                "No blendshape columns are registered on this Fex. The active "
                "model may predate v2.5 (no blendshape head)."
            )
        return self[self.blendshape_columns]

    # DEPRECATE
    @property
    def facepose(self):
        """Returns the facepose data

        Returns:
            DataFrame: facepose data
        """

        warnings.warn(
            "Fex.facepose has now been renamed to Fex.poses", DeprecationWarning
        )
        return self[self.facepose_columns]

    @property
    def inputs(self):
        """Returns input column as string

        Returns:
            string: path to input image
        """
        return self["input"]

    # DEPRECATE
    @property
    def input(self):
        """Returns input column as string

        Returns:
            string: path to input image
        """
        warnings.warn("Fex.input has now been renamed to Fex.inputs", DeprecationWarning)
        return self["input"]

    @property
    def landmarks_x(self):
        """Returns the x landmarks.

        Returns:
            DataFrame: x landmarks.
        """
        x_cols = [col for col in self.landmark_columns if "x" in col]
        return self[x_cols]

    # DEPRECATE
    @property
    def landmark_x(self):
        """Returns the x landmarks.

        Returns:
            DataFrame: x landmarks.
        """

        with warnings.catch_warnings():
            warnings.simplefilter("always", DeprecationWarning)
            warnings.warn(
                "Fex.landmark_x has been renamed to Fex.landmarks_x", DeprecationWarning
            )

        x_cols = [col for col in self.landmark_columns if "x" in col]
        return self[x_cols]

    @property
    def landmarks_y(self):
        """Returns the y landmarks.

        Returns:
            DataFrame: y landmarks.
        """
        y_cols = [col for col in self.landmark_columns if "y" in col]
        return self[y_cols]

    # DEPRECATE
    @property
    def landmark_y(self):
        """Returns the y landmarks.

        Returns:
            DataFrame: y landmarks.
        """

        with warnings.catch_warnings():
            warnings.simplefilter("always", DeprecationWarning)
            warnings.warn(
                "Fex.landmark_y has been renamed to Fex.landmarks_y", DeprecationWarning
            )

        y_cols = [col for col in self.landmark_columns if "y" in col]
        return self[y_cols]

    @property
    def faceboxes(self):
        """Returns the facebox data

        Returns:
            DataFrame: facebox data
        """
        return self[self.facebox_columns]

    # DEPRECATE
    @property
    def facebox(self):
        """Returns the facebox data

        Returns:
            DataFrame: facebox data
        """

        with warnings.catch_warnings():
            warnings.simplefilter("always", DeprecationWarning)
            warnings.warn(
                "Fex.facebox has been renamed to Fex.faceboxes", DeprecationWarning
            )

        return self[self.facebox_columns]

    @property
    def identity(self):
        """Returns the identity data

        Returns:
            DataFrame: identity data
        """
        return self[self.identity_columns]

    @property
    def time(self):
        """Returns the time data

        Returns:
            DataFrame: time data
        """
        return self[self.time_columns]

    @property
    def design(self):
        """Returns the design data

        Returns:
            DataFrame: time data
        """
        return self[self.design_columns]

    @property
    def info(self):
        """Print class meta data."""
        attr_list = []
        for name in self._metadata:
            attr_list.append(name + ": " + str(getattr(self, name, None)) + "\n")
        print(f"{self.__class__}\n" + "".join(attr_list))

    def plot_detections(self, *args, **kwargs):
        """Alias for Fex.plot_detections"""
        return Fex(self).T.__finalize__(self).plot_detections(*args, **kwargs)

aus property

Returns the Action Units data

Returns:

Name	Type	Description
DataFrame		Action Units data

blendshapes property

Returns the 52 MediaPipe/ARKit blendshape coefficients in [0, 1].

Raises AttributeError if no blendshape columns are registered (e.g. a pre-v2.5 multitask model, or DetectorV1, which has no blendshape head).

design property

Returns the design data

Returns:

Name	Type	Description
DataFrame		time data

emotions property

Returns the emotion data

Returns:

Name	Type	Description
DataFrame		emotion data

facebox property

Returns the facebox data

Returns:

Name	Type	Description
DataFrame		facebox data

faceboxes property

Returns the facebox data

Returns:

Name	Type	Description
DataFrame		facebox data

facepose property

Returns the facepose data

Returns:

Name	Type	Description
DataFrame		facepose data

gazes property

Returns the gaze data (pitch, yaw, optionally combined angle).

Returns:

Name	Type	Description
DataFrame		gaze data populated by the active gaze model
		(L2CS by default in v0.7+). Raises AttributeError if the
		Detectorv1 was built with gaze_model=None.

identity property

Returns the identity data

Returns:

Name	Type	Description
DataFrame		identity data

info property

Print class meta data.

input property

Returns input column as string

Returns:

Name	Type	Description
string		path to input image

inputs property

Returns input column as string

Returns:

Name	Type	Description
string		path to input image

landmark property

Returns the landmark data

Returns:

Name	Type	Description
DataFrame		landmark data

landmark_x property

Returns the x landmarks.

Returns:

Name	Type	Description
DataFrame		x landmarks.

landmark_y property

Returns the y landmarks.

Returns:

Name	Type	Description
DataFrame		y landmarks.

landmarks property

Returns the landmark data

Returns:

Name	Type	Description
DataFrame		landmark data

landmarks_x property

Returns the x landmarks.

Returns:

Name	Type	Description
DataFrame		x landmarks.

landmarks_y property

Returns the y landmarks.

Returns:

Name	Type	Description
DataFrame		y landmarks.

poses property

Returns the facepose data

Returns:

Name	Type	Description
DataFrame		facepose data

time property

Returns the time data

Returns:

Name	Type	Description
DataFrame		time data

__finalize__(other, method=None, **kwargs)

Propagate metadata from other to self

Source code in feat/data.py

def __finalize__(self, other, method=None, **kwargs):
    """Propagate metadata from other to self"""
    for name in self._metadata:
        object.__setattr__(self, name, getattr(other, name, None))
    return self

plot_detections(*args, **kwargs)

Alias for Fex.plot_detections

Source code in feat/data.py

def plot_detections(self, *args, **kwargs):
    """Alias for Fex.plot_detections"""
    return Fex(self).T.__finalize__(self).plot_detections(*args, **kwargs)

ImageDataset

Bases: Dataset

Torch Image Dataset

Parameters:

Name	Type	Description	Default
output_size	tuple or int	Desired output size. If tuple, output is matched to	None
preserve_aspect_ratio	bool	Output size is matched to preserve aspect ratio. Note that longest edge of output size is preserved, but actual output may differ from intended output_size.	True
padding	bool	Transform image to exact output_size. If tuple, will preserve	False

Returns:

Name	Type	Description
Dataset		dataset of [batch, channels, height, width] that can be passed to DataLoader

Source code in feat/data.py

class ImageDataset(Dataset):
    """Torch Image Dataset

    Args:
        output_size (tuple or int): Desired output size. If tuple, output is matched to
        output_size. If int, will set largest edge to output_size if target size is
        bigger, or smallest edge if target size is smaller to keep aspect ratio the
        same.
        preserve_aspect_ratio (bool): Output size is matched to preserve aspect ratio. Note that longest edge of output size is preserved, but actual output may differ from intended output_size.
        padding (bool): Transform image to exact output_size. If tuple, will preserve
        aspect ratio by adding padding. If int, will set both sides to the same size.

    Returns:
        Dataset: dataset of [batch, channels, height, width] that can be passed to DataLoader
    """

    def __init__(
        self, images, output_size=None, preserve_aspect_ratio=True, padding=False
    ):
        if isinstance(images, str):
            images = [images]
        self.images = images
        self.output_size = output_size
        self.preserve_aspect_ratio = preserve_aspect_ratio
        self.padding = padding

    def __len__(self):
        return len(self.images)

    def __getitem__(self, idx):
        # Dimensions are [channels, height, width]
        try:
            img = read_image(self.images[idx])
        except Exception:
            img = Image.open(self.images[idx])
            img = transforms.PILToTensor()(img)

        # Drop alpha channel
        if img.shape[0] == 4:
            img = img[:3, ...]

        if img.shape[0] == 1:
            img = torch.cat([img, img, img], dim=0)

        if self.output_size is not None:
            logging.info(
                f"ImageDataSet: RESCALING WARNING: from {img.shape} to output_size={self.output_size}"
            )
            transform = Compose(
                [
                    Rescale(
                        self.output_size,
                        preserve_aspect_ratio=self.preserve_aspect_ratio,
                        padding=self.padding,
                    )
                ]
            )
            transformed_img = transform(img)
            return {
                "Image": transformed_img["Image"],
                "Scale": transformed_img["Scale"],
                "Padding": transformed_img["Padding"],
                "FileName": self.images[idx],
            }

        else:
            return {
                "Image": img,
                "Scale": 1.0,
                "Padding": {"Left": 0, "Top": 0, "Right": 0, "Bottom": 0},
                "FileName": self.images[idx],
            }

VideoDataset

Bases: Dataset

Torch Video Dataset

Parameters:

Name	Type	Description	Default
skip_frames	int	number of frames to skip	None

Returns:

Name	Type	Description
Dataset		dataset of [batch, channels, height, width] that can be passed to DataLoader

Source code in feat/data.py

class VideoDataset(Dataset):
    """Torch Video Dataset

    Args:
        skip_frames (int): number of frames to skip

    Returns:
        Dataset: dataset of [batch, channels, height, width] that can be passed to DataLoader
    """

    def __init__(self, video_file, skip_frames=None, output_size=None):
        self.file_name = video_file
        self.skip_frames = skip_frames
        self.output_size = output_size
        self._decoder = None
        self._decoder_pid = None
        self.get_video_metadata(video_file)
        # This is the list of frame ids used to slice the video not video_frames
        self.video_frames = np.arange(
            0, self.metadata["num_frames"], 1 if skip_frames is None else skip_frames
        )

    def __len__(self):
        # Number of frames respective skip_frames
        return len(self.video_frames)

    def __getitem__(self, idx):
        # Get the frame data and frame number respective skip_frames.
        # torchcodec returns [C, H, W] uint8 tensors directly - matches
        # the layout read_image produces, so no axis swap is needed.
        frame_data, frame_idx = self.load_frame(idx)

        # Rescale if needed like in ImageDataset
        if self.output_size is not None:
            logging.info(
                f"VideoDataset: RESCALING WARNING: from {self.metadata['shape']} to output_size={self.output_size}"
            )
            transform = Compose(
                [Rescale(self.output_size, preserve_aspect_ratio=True, padding=False)]
            )
            transformed_frame_data = transform(frame_data)

            return {
                "Image": transformed_frame_data["Image"],
                "Frame": frame_idx,
                "FileName": self.file_name,
                "Scale": transformed_frame_data["Scale"],
                "Padding": transformed_frame_data["Padding"],
            }
        else:
            return {
                "Image": frame_data,
                "Frame": frame_idx,
                "FileName": self.file_name,
                "Scale": 1.0,
                "Padding": {"Left": 0, "Top": 0, "Right": 0, "Bottom": 0},
            }

    def _get_decoder(self):
        """Lazy-construct the torchcodec decoder.

        Holding the decoder on `self` and reusing it across __getitem__
        calls avoids re-opening the video for every frame. The decoder
        is dropped on pickling (see ``__getstate__``) so each DataLoader
        worker reopens its own; the underlying torchcodec C++ stream
        registration does not survive a serialization round-trip.

        Also reopen when the owning PID changes: Linux DataLoader workers
        are *forked*, not pickled, so ``__getstate__`` never runs and the
        parent's live decoder would otherwise be shared across workers —
        corrupting its ffmpeg stream state ("Invalid data found when
        processing input"). The PID guard forces a per-process decoder.
        """
        pid = os.getpid()
        if self._decoder is None or self._decoder_pid != pid:
            self._decoder = VideoDecoder(self.file_name)
            self._decoder_pid = pid
        return self._decoder

    def __getstate__(self):
        # The torchcodec VideoDecoder pickles successfully but deserializes
        # into a state where `validateActiveStream` fails on first frame
        # access. Drop it from the pickled payload so DataLoader workers
        # re-open their own decoder via _get_decoder() lazy construction.
        state = self.__dict__.copy()
        state["_decoder"] = None
        return state

    def get_video_metadata(self, video_file):
        m = self._get_decoder().metadata
        self.metadata = {
            "fps": float(m.average_fps),
            "height": m.height,
            "width": m.width,
            "num_frames": m.num_frames,
            "shape": (m.height, m.width),
        }

    def load_frame(self, idx):
        """Load a single frame from the video by index.

        Returns the frame as a `[C, H, W]` uint8 tensor (torchcodec's
        native layout, matching torchvision.io.read_image).
        """
        frame_idx = int(self.video_frames[idx])
        frame_data = self._get_decoder()[frame_idx]
        return frame_data, frame_idx

    def calc_approx_frame_time(self, idx):
        """Calculate the approximate time of a frame in a video

        Args:
            frame_idx (int): frame number

        Returns:
            float: time in seconds
        """
        frame_time = idx / self.metadata["fps"]
        total_time = self.metadata["num_frames"] / self.metadata["fps"]
        time = total_time if idx >= self.metadata["num_frames"] else frame_time
        return self.convert_sec_to_min_sec(time)

    @staticmethod
    def convert_sec_to_min_sec(duration):
        minutes = int(duration // 60)
        seconds = int(duration % 60)
        return f"{minutes:02d}:{seconds:02d}"

calc_approx_frame_time(idx)

Calculate the approximate time of a frame in a video

Parameters:

Name	Type	Description	Default
frame_idx	int	frame number	required

Returns:

Name	Type	Description
float		time in seconds

Source code in feat/data.py

def calc_approx_frame_time(self, idx):
    """Calculate the approximate time of a frame in a video

    Args:
        frame_idx (int): frame number

    Returns:
        float: time in seconds
    """
    frame_time = idx / self.metadata["fps"]
    total_time = self.metadata["num_frames"] / self.metadata["fps"]
    time = total_time if idx >= self.metadata["num_frames"] else frame_time
    return self.convert_sec_to_min_sec(time)

load_frame(idx)

Load a single frame from the video by index.

Returns the frame as a [C, H, W] uint8 tensor (torchcodec's native layout, matching torchvision.io.read_image).

Source code in feat/data.py

def load_frame(self, idx):
    """Load a single frame from the video by index.

    Returns the frame as a `[C, H, W]` uint8 tensor (torchcodec's
    native layout, matching torchvision.io.read_image).
    """
    frame_idx = int(self.video_frames[idx])
    frame_data = self._get_decoder()[frame_idx]
    return frame_data, frame_idx

imageLoader_DISFAPlus

Bases: ImageDataset

Loading images from DISFA dataset. Assuming that the user has just unzipped the downloaded DISFAPlus data

Source code in feat/data.py

class imageLoader_DISFAPlus(ImageDataset):
    """
    Loading images from DISFA dataset. Assuming that the user has just unzipped the downloaded DISFAPlus data
    """

    def __init__(
        self,
        data_dir="/Storage/Data/DISFAPlusDataset/",
        output_size=None,
        preserve_aspect_ratio=True,
        padding=False,
        sample=None,
    ):
        super().__init__(
            images=None,
            output_size=output_size,
            preserve_aspect_ratio=preserve_aspect_ratio,
            padding=padding,
        )

        # Load all dir info for DISFA+
        self.avail_AUs = [
            "AU1",
            "AU2",
            "AU4",
            "AU5",
            "AU6",
            "AU9",
            "AU12",
            "AU15",
            "AU17",
            "AU20",
            "AU25",
            "AU26",
        ]

        self.data_dir = data_dir
        self.sample = sample
        self.main_file = self._load_data()

        self.output_size = output_size
        self.preserve_aspect_ratio = preserve_aspect_ratio
        self.padding = padding

    def _load_data(self):
        print("data loading in progress")
        all_subjects = os.listdir(os.path.join(self.data_dir, "Labels"))
        if self.sample:
            all_subjects = np.random.choice(
                all_subjects, size=int(self.sample * len(all_subjects)), replace=False
            )

        sessions = [
            os.listdir(os.path.join(self.data_dir, "Labels", subj))
            for subj in all_subjects
        ]
        # all image directory
        dfs = []
        for i, subj in enumerate(all_subjects):
            for sess in sessions[i]:
                AU_f = []
                for au_added in self.avail_AUs:
                    AU_file = pd.read_csv(
                        os.path.join(self.data_dir, "Labels", subj, sess)
                        + "/"
                        + au_added
                        + ".txt",
                        skiprows=2,
                        header=None,
                        names=["intensity"],
                        index_col=0,
                        sep=r"\s{2,}",
                    )
                    AU_file.rename({"intensity": f"{au_added}"}, axis=1, inplace=True)
                    AU_f.append(AU_file)
                AU_pd = pd.concat(AU_f, axis=1)
                AU_pd = AU_pd.reset_index(level=0)
                AU_pd["session"] = sess
                AU_pd["subject"] = subj
                dfs.append(AU_pd)
        df = pd.concat(dfs, ignore_index=True)
        df["image_path"] = [
            os.path.join(self.data_dir, "Images", df["subject"][i], df["session"][i])
            + "/"
            + df["index"][i]
            for i in range(df.shape[0])
        ]
        return df

    def __len__(self):
        return self.main_file.shape[0]

    def __getitem__(self, idx):
        # Dimensions are [channels, height, width]
        img = read_image(self.main_file["image_path"].iloc[idx])
        label = self.main_file.loc[idx, self.avail_AUs].to_numpy().astype(np.int16)

        if self.output_size is not None:
            logging.info(
                f"imageLoader_DISFAPlus: RESCALING WARNING: from {img.shape} to output_size={self.output_size}"
            )
            transform = Compose(
                [
                    Rescale(
                        self.output_size,
                        preserve_aspect_ratio=self.preserve_aspect_ratio,
                        padding=self.padding,
                    )
                ]
            )
            transformed_img = transform(img)
            return {
                "Image": transformed_img["Image"],
                "label": torch.from_numpy(label),
                "Scale": transformed_img["Scale"],
                "Padding": transformed_img["Padding"],
                "FileName": self.main_file["image_path"][idx],
            }

        else:
            return {
                "Image": img,
                "label": torch.from_numpy(label),
                "Scale": 1.0,
                "Padding": {"Left": 0, "Top": 0, "Right": 0, "Bottom": 0},
            }