getml.pipeline.Features

def correlations(
    self, target_num: int = 0, sort: bool = True
) -> Tuple[NDArray[np.str_], NDArray[np.float_]]:
    """
    Returns the data for the feature correlations,
    as displayed in the getML Monitor.

    Args:
        target_num:
            Indicates for which target you want to view the
            importances.
            (Pipelines can have more than one target.)

        sort:
            Whether you want the results to be sorted.

    Returns:
        The first array contains the names of the features.
        The second array contains the correlations with the target.
    """

    cmd: Dict[str, Any] = {}

    cmd["type_"] = "Pipeline.feature_correlations"
    cmd["name_"] = self.pipeline

    cmd["target_num_"] = target_num

    with comm.send_and_get_socket(cmd) as sock:
        msg = comm.recv_string(sock)
        if msg != "Success!":
            comm.handle_engine_exception(msg)
        msg = comm.recv_string(sock)

    json_obj = json.loads(msg)

    names = np.asarray(json_obj["feature_names_"])
    correlations = np.asarray(json_obj["feature_correlations_"])

    assert len(correlations) <= len(names), "Correlations must be <= names"

    if hasattr(self, "data"):
        indices = np.asarray(
            [
                feature.index
                for feature in self.data
                if feature.target == self.targets[target_num]
                and feature.index < len(correlations)
            ]
        )

        names = names[indices]
        correlations = correlations[indices]

    if not sort:
        return names, correlations

    indices = np.argsort(np.abs(correlations))[::-1]

    return (names[indices], correlations[indices])

def filter(self, conditional: Callable[[Feature], bool]) -> Features:
    """
     Filters the Features container.

    Args:
        conditional:
            A callable that evaluates to a boolean for a given item.

    Returns:
            A container of filtered Features.

    ??? example
        ```python
        important_features = my_pipeline.features.filter(lambda feature: feature.importance > 0.1)
        correlated_features = my_pipeline.features.filter(lambda feature: feature.correlation > 0.3)
        ```
    """
    features_filtered = [feature for feature in self.data if conditional(feature)]
    return Features(self.pipeline, self.targets, data=features_filtered)

def importances(
    self, target_num: int = 0, sort: bool = True
) -> Tuple[NDArray[np.str_], NDArray[np.float_]]:
    """
    Returns the data for the feature importances,
    as displayed in the getML Monitor.

    Args:
        target_num:
            Indicates for which target you want to view the
            importances.
            (Pipelines can have more than one target.)

        sort:
            Whether you want the results to be sorted.

    Returns:
        The first array contains the names of the features.
        The second array contains their importances. By definition, all importances add up to 1.

    """

    cmd: Dict[str, Any] = {}

    cmd["type_"] = "Pipeline.feature_importances"
    cmd["name_"] = self.pipeline

    cmd["target_num_"] = target_num

    with comm.send_and_get_socket(cmd) as sock:
        msg = comm.recv_string(sock)
        if msg != "Success!":
            comm.handle_engine_exception(msg)
        msg = comm.recv_string(sock)

    json_obj = json.loads(msg)

    names = np.asarray(json_obj["feature_names_"])
    importances = np.asarray(json_obj["feature_importances_"])

    if hasattr(self, "data"):
        assert len(importances) <= len(names), "Importances must be <= names"

        indices = np.asarray(
            [
                feature.index
                for feature in self.data
                if feature.target == self.targets[target_num]
                and feature.index < len(importances)
            ]
        )

        names = names[indices]
        importances = importances[indices]

    if not sort:
        return names, importances

    assert len(importances) <= len(names), "Must have the same length"

    indices = np.argsort(importances)[::-1]

    return (names[indices], importances[indices])

def sort(
    self,
    by: Optional[str] = None,
    key: Optional[
        Callable[
            [Feature],
            Union[
                float,
                int,
                str,
            ],
        ]
    ] = None,
    descending: Optional[bool] = None,
) -> Features:
    """
    Sorts the Features container. If no arguments are provided the
    container is sorted by target and name.

    Args:
        by:
            The name of field to sort by. Possible fields:
                - name(s)
                - correlation(s)
                - importances(s)
        key:
            A callable that evaluates to a sort key for a given item.
        descending:
            Whether to sort in descending order.

    Returns:
            A container of sorted Features.

    ??? example
        ```python
        by_correlation = my_pipeline.features.sort(by="correlation")

        by_importance = my_pipeline.features.sort(key=lambda feature: feature.importance)
        ```
    """

    reverse = False if descending is None else descending

    if (by is not None) and (key is not None):
        raise ValueError("Only one of `by` and `key` can be provided.")

    if key is not None:
        features_sorted = sorted(self.data, key=key, reverse=reverse)
        return self._make_features(features_sorted)

    else:
        if by is None:
            features_sorted = sorted(
                self.data, key=lambda feature: feature.index, reverse=reverse
            )
            features_sorted.sort(key=lambda feature: feature.target)
            return self._make_features(features_sorted)

        if re.match(pattern="names?$", string=by):
            features_sorted = sorted(
                self.data, key=lambda feature: feature.name, reverse=reverse
            )
            return self._make_features(features_sorted)

        if re.match(pattern="correlations?$", string=by):
            reverse = True if descending is None else descending
            features_sorted = sorted(
                self.data,
                key=lambda feature: abs(feature.correlation),
                reverse=reverse,
            )
            return self._make_features(features_sorted)

        if re.match(pattern="importances?$", string=by):
            reverse = True if descending is None else descending
            features_sorted = sorted(
                self.data,
                key=lambda feature: feature.importance,
                reverse=reverse,
            )
            return self._make_features(features_sorted)

        raise ValueError(f"Cannot sort by: {by}.")

def to_pandas(self) -> pd.DataFrame:
    """
    Returns all information related to the features in a pandas data frame.

    Returns:
        A pandas data frame containing the features' names, importances, correlations, and SQL code.
    """

    return self._to_pandas()

def to_sql(
    self,
    targets: bool = True,
    subfeatures: bool = True,
    dialect: str = sqlite3,
    schema: Optional[str] = None,
    nchar_categorical: int = 128,
    nchar_join_key: int = 128,
    nchar_text: int = 4096,
    size_threshold: Optional[int] = 50000,
) -> SQLCode:
    """
    Returns SQL statements visualizing the features.

    Args:
        targets:
            Whether you want to include the target columns
            in the main table.

        subfeatures:
            Whether you want to include the code for the
            subfeatures of a snowflake schema.

        dialect:
            The SQL dialect to use. Must be from
            [`dialect`][getml.pipeline.dialect]. Please
            note that not all dialects are supported
            in the getML Community edition.

        schema:
            The schema in which to wrap all generated tables and
            indices. None for no schema. Not applicable to all dialects.
            For the BigQuery and MySQL dialects, the schema is identical
            to the database ID.

        nchar_categorical:
            The maximum number of characters used in the
            VARCHAR for categorical columns. Not applicable
            to all dialects.

        nchar_join_key:
            The maximum number of characters used in the
            VARCHAR for join keys. Not applicable
            to all dialects.

        nchar_text:
            The maximum number of characters used in the
            VARCHAR for text columns. Not applicable
            to all dialects.

        size_threshold:
            The maximum number of characters to display
            in a single feature. Displaying extremely
            complicated features can crash your iPython
            notebook or lead to unexpectedly high memory
            consumption, which is why a reasonable
            upper limit is advantageous. Set to None
            for no upper limit.

    Returns:
            Object representing the features.

    ??? example
        ```python
        my_pipeline.features.to_sql()
        ```

    Note:
        Only fitted pipelines
        ([`fit`][getml.Pipeline.fit]) can hold trained
        features which can be returned as SQL statements.

    Note:
        The getML Community edition only supports
        transpilation to human-readable SQL. Passing
        'sqlite3' will also produce human-readable SQL.

    """

    if not isinstance(targets, bool):
        raise TypeError("'targets' must be a bool!")

    if not isinstance(subfeatures, bool):
        raise TypeError("'subfeatures' must be a bool!")

    if not isinstance(dialect, str):
        raise TypeError("'dialect' must be a string!")

    if not isinstance(nchar_categorical, int):
        raise TypeError("'nchar_categorical' must be an int!")

    if not isinstance(nchar_join_key, int):
        raise TypeError("'nchar_join_key' must be an int!")

    if not isinstance(nchar_text, int):
        raise TypeError("'nchar_text' must be an int!")

    if dialect not in _all_dialects:
        raise ValueError(
            "'dialect' must from getml.pipeline.dialect, "
            + "meaning that is must be one of the following: "
            + str(_all_dialects)
            + "."
        )

    if size_threshold is not None and not isinstance(size_threshold, int):
        raise TypeError("'size_threshold' must be an int or None!")

    if size_threshold is not None and size_threshold <= 0:
        raise ValueError("'size_threshold' must be a positive number!")

    cmd: Dict[str, Any] = {}

    cmd["type_"] = "Pipeline.to_sql"
    cmd["name_"] = self.pipeline

    cmd["targets_"] = targets
    cmd["subfeatures_"] = subfeatures
    cmd["dialect_"] = dialect
    cmd["schema_"] = schema or ""
    cmd["nchar_categorical_"] = nchar_categorical
    cmd["nchar_join_key_"] = nchar_join_key
    cmd["nchar_text_"] = nchar_text

    if size_threshold is not None:
        cmd["size_threshold_"] = size_threshold

    with comm.send_and_get_socket(cmd) as sock:
        msg = comm.recv_string(sock)
        if msg != "Found!":
            comm.handle_engine_exception(msg)
        sql = comm.recv_string(sock)

    return SQLCode(sql.split("\n\n\n"), dialect)