Tsflex

https://predict-idlab.github.io/tsflex/
Admin
Tsflex is a flexible Python toolkit for time-series processing and feature extraction, designed for...
Tokens:144,476
Snippets:491
Trust Score:8.3
Update:1 year ago
Show doc for...
Tokens:
Raw
### Python: tsflex FeatureCollection example with tsfresh integration

Source: https://predict-idlab.github.io/tsflex/features/integrations

This example demonstrates how to use `tsfresh_settings_wrapper` to integrate tsfresh features into a tsflex `FeatureCollection`. It shows the instantiation of `MultipleFeatureDescriptors` with tsfresh features and the subsequent calculation of these features on sample data.

```python
from tsflex.features import FeatureCollection, MultipleFeatureDescriptors
from tsflex.features.integrations import tsfresh_settings_wrapper
from tsfresh.feature_extraction import MinimalFCParameters

minimal_tsfresh_feats = MultipleFeatureDescriptors(
    functions=tsfresh_settings_wrapper(MinimalFCParameters()),
    series_names=["sig_0", "sig_1"],  # list of signal names
    windows="15min", strides="2min",
)

fc = FeatureCollection(minimal_tsfresh_feats)
fc.calculate(data)  # calculate the features on your data

```

--------------------------------

### FuncWrapper Initialization and Usage Example

Source: https://predict-idlab.github.io/tsflex/features/index

This snippet demonstrates how to initialize and use the FuncWrapper class. It shows setting parameters like output_names, input_type, and vectorized, and then calling the wrapped function with input series.

```python
from tsflex.utils.classes import FuncWrapper
import numpy as np
import pandas as pd

# Example function to wrap
def sum_two_series(series1, series2):
    return series1 + series2

# Create a FuncWrapper instance
# Assuming 'sum_two_series' requires numpy arrays and produces one output named 'sum'
wrapper = FuncWrapper(func=sum_two_series, output_names='sum', input_type=np.ndarray, vectorized=False)

# Example input data
series_a = np.array([1, 2, 3, 4, 5])
series_b = np.array([5, 4, 3, 2, 1])

# Call the wrapped function
result = wrapper(series_a, series_b)
print(f"Result of calling FuncWrapper: {result}")

# Example with pandas Series and vectorized=True (requires careful setup)
# Note: For vectorized=True, input_type must be np.ndarray as per documentation.
# This example illustrates the structure but might need adjustments based on actual function requirements.

def vectorized_max(data):
    # data shape is (nb_segmented_windows, window_size)
    return np.max(data, axis=1)

# Example of a vectorized wrapper
# Note: The documentation states input_type should be np.ndarray when vectorized=True.
# The following is conceptual and might require specific data shaping for a real vectorized function.
# wrapper_vectorized = FuncWrapper(func=vectorized_max, output_names='max_val', input_type=np.ndarray, vectorized=True)

# Example data shaped for vectorized operation (conceptual)
# segmented_windows_data = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])
# result_vectorized = wrapper_vectorized(segmented_windows_data)
# print(f"Result of vectorized FuncWrapper: {result_vectorized}")

```

--------------------------------

### Data Validation and Logging Setup in tsflex

Source: https://predict-idlab.github.io/tsflex/features

This Python code snippet demonstrates the data validation logic within the tsflex library, ensuring input data is in the correct format (list of Series/DataFrames, or a single Series/DataFrame/GroupBy object). It also includes the setup for file logging, where existing handlers are cleared and a new handler is added if a logging file path is provided. This ensures that logs are written to the specified file without interference from previous configurations.

```python
        # Check valid data
        if isinstance(data, list):
            assert all(
                isinstance(d, (pd.Series, pd.DataFrame)) for d in data
            ), "All elements of the data list must be either a Series or a DataFrame!"
        else:
            assert isinstance(
                data, (pd.Series, pd.DataFrame, pd.core.groupby.DataFrameGroupBy)
            ), "The data must be either a Series, a DataFrame or a DataFrameGroupBy!"

        # check valid group_by
        assert group_by_all is None or group_by_consecutive is None, (
            "Only max one of the following parameters can be set: "
            + "`group_by_all` or `group_by_consecutive`"
        )
        assert not (
            (group_by_all is not None or group_by_consecutive is not None)
            and isinstance(data, pd.core.groupby.DataFrameGroupBy)
        ), (
            "Cannot use `group_by_all` or `group_by_consecutive` when `data` is"
            + " already a grouped DataFrame!"
        )

        # Delete other logging handlers
        delete_logging_handlers(logger)
        # Add logging handler (if path provided)
        f_handler = None
        if logging_file_path:
            f_handler = add_logging_handler(logger, logging_file_path)
```

--------------------------------

### Python Logger Setup and Utilities

Source: https://predict-idlab.github.io/tsflex/features/logger

Configures a package-specific logger for feature calculations and imports necessary utility functions. It also defines author information and imports core libraries such as logging, re, numpy, and pandas.

```python
"""Contains the used variables and functions to provide logging functionality.

See Also
--------
FeatureCollection: its `logging_file_path` of the `calculate` method.

"""

__author__ = "Jeroen Van Der Donckt"

import logging
import re

import numpy as np
import pandas as pd

from ..utils.argument_parsing import timedelta_to_str
from ..utils.logging import logging_file_to_df, remove_inner_brackets

# Package specific logger
logger = logging.getLogger("feature_calculation_logger")
logger.setLevel(logging.DEBUG)

```

--------------------------------

### tsflex SeriesPipeline and SeriesProcessor Components

Source: https://predict-idlab.github.io/tsflex/processing/index

Illustrates the fundamental usage of SeriesPipeline and SeriesProcessor classes in tsflex. It shows how to initialize a pipeline with processors and how to append additional processing steps, followed by applying the pipeline to data.

```python
import numpy as np
import scipy.signal as ss
from tsflex.processing import SeriesProcessor, SeriesPipeline

# The SeriesPipeline takes a List[SeriesProcessor] as input
processing_pipe = SeriesPipeline(processors=[
        SeriesProcessor(np.abs, ["series_a", "series_b"]),
        SeriesProcessor(ss.medfilt, "series_b", kernel_size=5) # (with kwargs)
    ]
)
# We can still append processing steps after instantiating.
processing_pipe.append(processor=SeriesProcessor(ss.detrend, "series_a"))

# Apply the processing steps
processing_pipe.process(...)
```

--------------------------------

### Consecutive Grouping Example

Source: https://predict-idlab.github.io/tsflex/features/index

Demonstrates how to use `group_by_consecutive` for grouping data by consecutive identical values. The output includes start and end timestamps for each group.

```python
        number_sold__sum__w=manual  store    __start      __end
    0          845                  0     2019-01-01 2019-01-01
    1          357                  3     2019-01-02 2019-01-02
    2          904                  6     2019-01-03 2019-01-03
    3          599                  3     2019-01-04 2019-01-05
    4          871                  0     2019-01-06 2019-01-06
    ...

```

--------------------------------

### Define StridedRolling with Explicit Segment Indices

Source: https://predict-idlab.github.io/tsflex/features/segmenter/index

This example shows how to use StridedRolling by explicitly providing segment start and end indices. This gives precise control over the window segments. The `approve_sparsity` parameter is set to `True` to acknowledge potential irregular data sampling.

```python
from tsflex.features.custom_features import StridedRolling
import pandas as pd
import numpy as np

# Sample data
data = pd.Series(np.random.randn(100))

# Explicit segment indices
segment_start_idxs = np.array([0, 10, 20, 30])
segment_end_idxs = np.array([5, 15, 25, 35])

# Initialize StridedRolling with explicit indices
strided_rolling_explicit = StridedRolling(
    data=data,
    window=5,  # Window size in terms of index
    segment_start_idxs=segment_start_idxs,
    segment_end_idxs=segment_end_idxs,
    approve_sparsity=True
)
```

--------------------------------

### TimeStridedRolling Get Output Index

Source: https://predict-idlab.github.io/tsflex/features/segmenter/strided_rolling

Generates the pandas Index for the output of time-based operations. It asserts that the input start and end indexes are of datetime64 type, ensuring chronological integrity. This method is crucial for aligning output features with their corresponding time periods.

```python
def _get_output_index(self, start_idxs: np.ndarray, end_idxs: np.ndarray, name: str) -> pd.Index:
    assert start_idxs.dtype.type == np.datetime64
```

--------------------------------

### tsflex FeatureCollection and FeatureDescriptor Usage

Source: https://predict-idlab.github.io/tsflex/features

Demonstrates the core components for feature extraction in tsflex: FeatureCollection and FeatureDescriptor. It shows how to initialize a FeatureCollection with a list of FeatureDescriptors and how to add more descriptors after instantiation. The example implies a subsequent call to calculate features.

```python
import numpy as np
import scipy.stats as ss
from tsflex.features import FeatureDescriptor, FeatureCollection

# The FeatureCollection takes a List[FeatureDescriptor] as input
# There is no need for using a FuncWrapper when dealing with simple feature functions
fc = FeatureCollection(feature_descriptors=[
        FeatureDescriptor(np.mean, "series_a", "1hour", "15min"),
        FeatureDescriptor(ss.skew, "series_b", "3hours", "5min")
    ]
)

# We can still add features after instantiating.
fc.add(features=[FeatureDescriptor(np.std, "series_a", "1hour", "15min")])

# Calculate the features
fc.calculate(...)

```

--------------------------------

### Consecutive Grouping Example for Feature Calculation

Source: https://predict-idlab.github.io/tsflex/features/feature_collection

Demonstrates the output format when using consecutive grouping for feature calculation. It includes columns for the aggregated feature (e.g., 'number_sold__sum__w'), the grouping key ('store'), and the start and end timestamps of each group ('__start', '__end'). NaN values in the grouping column are ignored.

```python
            number_sold__sum__w=manual  store    __start      __end
                0          845                  0     2019-01-01 2019-01-01
                1          357                  3     2019-01-02 2019-01-02
                2          904                  6     2019-01-03 2019-01-03
                3          599                  3     2019-01-04 2019-01-05
                4          871                  0     2019-01-06 2019-01-06
                ...

```

--------------------------------

### Create and Apply a tsflex Processing Pipeline

Source: https://predict-idlab.github.io/tsflex/processing/index

Demonstrates how to create a SeriesPipeline with multiple SeriesProcessor steps, including appending a new step, and then process data. This example uses pandas for data handling and numpy/scipy for signal processing functions.

```python
import pandas as pd
import scipy.signal as ss
import numpy as np
from tsflex.processing import SeriesProcessor, SeriesPipeline

# 1. -------- Get your time-indexed data --------
# Data contains 3 columns; ["ACC_x", "ACC_y", "ACC_z"]
url = "https://github.com/predict-idlab/tsflex/raw/main/examples/data/empatica/"
data = pd.read_parquet(url + "acc.parquet").set_index("timestamp")

# 2 -------- Construct your processing pipeline --------
processing_pipe = SeriesPipeline(
    processors=[
        SeriesProcessor(function=np.abs, series_names=["ACC_x", "ACC_y"]),
        SeriesProcessor(ss.medfilt, ["ACC_y", "ACC_z"], kernel_size=5) 
    ]
)
# -- 2.1. Append processing steps to your processing pipeline
processing_pipe.append(SeriesProcessor(ss.detrend, ["ACC_x", "ACC_z"]))

# 3 -------- Process the data --------
processing_pipe.process(data=data, return_df=True)
```

--------------------------------

### Determine Series Bounds - Python

Source: https://predict-idlab.github.io/tsflex/features/utils

Determines the start and end bounds of a list of pandas Series based on different methods ('inner', 'inner-outer', 'outer'). It returns the latest start and earliest stop ('inner'), the latest start and latest stop ('inner-outer'), or the earliest start and latest stop ('outer').

```python
def _determine_bounds(
    bound_method: str,
    series_list: List[pd.Series]
) -> Tuple[Any, Any]:
    """Determine the bounds of the passed series.

    Parameters
    ----------
    bound_method: str

    series_list : List[pd.Series]
        The list of series for which the bounds are determined.

    Returns
    -------
    Tuple[pd.Timestamp, pd.Timestamp]
        The start & end timestamp, respectively.

    """
    if bound_method == "inner":
        latest_start = series_list[0].index[0]
        earliest_stop = series_list[0].index[-1]
        for series in series_list[1:]:
            latest_start = max(latest_start, series.index[0])
            earliest_stop = min(earliest_stop, series.index[-1])
        return latest_start, earliest_stop
    elif bound_method == "inner-outer":
        latest_start = series_list[0].index[0]
        latest_stop = series_list[0].index[-1]
        for series in series_list[1:]:
            latest_start = max(latest_start, series.index[0])
            latest_stop = max(latest_stop, series.index[-1])
        return latest_start, latest_stop
    elif bound_method == "outer":
        earliest_start = series_list[0].index[0]
        latest_stop = series_list[0].index[-1]
        for series in series_list[1:]:
            earliest_start = min(earliest_start, series.index[0])
            latest_stop = max(latest_stop, series.index[-1])
        return earliest_start, latest_stop
    else:
        raise ValueError(f"invalid bound method string passed {bound_method}")
```

--------------------------------

### Instantiate FuncWrapper

Source: https://predict-idlab.github.io/tsflex/features/function_wrapper

Initializes a FuncWrapper instance. This constructor sets up the wrapped function, output names, expected input type, and vectorized execution flag. It also handles keyword arguments that will be passed to the wrapped function.

```python
def __init__(
    self,
    func: Callable,
    output_names: Optional[Union[List[str], str]] = None,
    input_type: Union[np.ndarray, pd.Series] = np.ndarray,
    vectorized: bool = False,
    **kwargs,
):
    """Create FuncWrapper instance."""
    self.func = func
    self.kwargs: dict = kwargs

    if isinstance(output_names, list):
        self.output_names = output_names
    elif isinstance(output_names, str):
        self.output_names = [output_names]
    elif not output_names:
        self.output_names = [_get_name(func)]
    else:
        raise TypeError(f"`output_names` is unexpected type {type(output_names)}")

    # for backwards compatibility
    input_type = np.ndarray if input_type is np.array else input_type
    assert input_type in SUPPORTED_STROLL_TYPES, "Invalid input_type!"
    assert not (
        vectorized & (input_type is not np.ndarray)
    ), "The input_type must be np.ndarray if vectorized is True!"
    self.input_type = input_type
    self.vectorized = vectorized

    self._freeze()
```

--------------------------------

### Update Start and End Indices for Time-Based Rolling

Source: https://predict-idlab.github.io/tsflex/features/segmenter/strided_rolling

This method updates the `start` and `end` attributes to reflect the sample-based indexing used in time-based rolling operations. It finds the positions of the current `start` and `end` times within the index of the first series and sets these as the new `start` and `end` sample indices.

```python
def _update_start_end_indices_to_stroll_type(
        self,
        series_list: List[pd.Series]
    ) -> None:
        # update the start and end times to the sequence datatype
        self.start, self.end = np.searchsorted(
            series_list[0].index.values,
            [self.start.to_datetime64(), self.end.to_datetime64()],
            "left",
        )
```

--------------------------------

### Initialize FeatureDescriptor with Multiple Series Combinations

Source: https://predict-idlab.github.io/tsflex/features/feature

This Python code initializes a list of `FeatureDescriptor` objects by iterating over combinations of functions, series names, and window sizes. It handles different input types for `series_names` and ensures consistency in the number of input series required by functions. Dependencies include `FuncWrapper`, `Callable`, `Tuple`, `List`, `Optional`, `Union`, `pd.Timedelta`, `itertools`, and `FeatureDescriptor`.

```Python
def __init__(
        self,
        functions: Union[FuncWrapper, Callable, List[Union[FuncWrapper, Callable]]],
        series_names: Union[str, Tuple[str, ...], List[str], List[Tuple[str, ...]]],
        windows: Optional[
            Union[float, str, pd.Timedelta, List[Union[float, str, pd.Timedelta]]]
        ] = None,
        strides: Optional[
            Union[float, str, pd.Timedelta, List[Union[float, str, pd.Timedelta]]]
        ] = None,
    ):
        # Cast functions to FuncWrapper, this avoids creating multiple
        # FuncWrapper objects for the same function in the FeatureDescriptor
        def to_func_wrapper(f: Callable) -> FuncWrapper:
            return f if isinstance(f, FuncWrapper) else FuncWrapper(f)

        functions = [to_func_wrapper(f) for f in to_list(functions)]
        # Convert the series names to list of tuples
        series_names = [to_tuple(names) for names in to_list(series_names)]
        # Assert that function inputs (series) all have the same length
        assert all(
            len(series_names[0]) == len(series_name_tuple)
            for series_name_tuple in series_names
        )
        # Convert the other types to list
        windows = to_list(windows)

        self.feature_descriptions: List[FeatureDescriptor] = []
        # Iterate over all combinations
        combinations = [functions, series_names, windows]
        for function, series_name, window in itertools.product(*combinations):  # type: ignore[call-overload]
            self.feature_descriptions.append(
                FeatureDescriptor(function, series_name, window, strides)
            )
```

--------------------------------

### Check Start and End Indices - Python

Source: https://predict-idlab.github.io/tsflex/features/utils

Validates that start and end index arrays are of equal length and that all start indices are less than or equal to their corresponding end indices. This is crucial for ensuring valid segment definitions.

```python
def _check_start_end_array(start_idxs: np.ndarray, end_idxs: np.ndarray) -> None:
    """Check if the start and end indices are valid.

    These are valid if they are of the same length and if the start indices are smaller
    than the end indices.

    Parameters
    ----------
    start_idxs: np.ndarray
        The start indices.
    end_idxs: np.ndarray
        The end indices.
    """
    assert len(start_idxs) == len(end_idxs), "start_idxs and end_ixs must have equal length"
    assert np.all(start_idxs <= end_idxs), "for all corresponding values: segment_start_idxs <= segment_end_idxs"
```

--------------------------------

### Initialize FeatureCollection with FeatureDescriptors

Source: https://predict-idlab.github.io/tsflex/features/index

Demonstrates initializing a FeatureCollection with a list of FeatureDescriptor objects, each defining a feature with its function, series name, window, and stride. Simple feature functions like np.mean and ss.skew can be directly passed without FuncWrapper.

```python
import numpy as np
import scipy.stats as ss
from tsflex.features import FeatureDescriptor, FeatureCollection

# The FeatureCollection takes a List[FeatureDescriptor] as input
# There is no need for using a FuncWrapper when dealing with simple feature functions
fc = FeatureCollection(feature_descriptors=[
        FeatureDescriptor(np.mean, "series_a", "1hour", "15min"),
        FeatureDescriptor(ss.skew, "series_b", "3hours", "5min")
    ]
)

# We can still add features after instantiating.
fc.add(features=[FeatureDescriptor(np.std, "series_a", "1hour", "15min")])

# Calculate the features
fc.calculate(...)
```

--------------------------------

### Compute Start Indices for Stride using NumPy in Python

Source: https://predict-idlab.github.io/tsflex/features/segmenter/strided_rolling

Computes the starting indices for segments based on a given stride using efficient NumPy operations. It leverages pre-calculated start and stride values and the number of segments for the stride.

```python
def _get_np_start_idx_for_stride(self, stride: T) -> np.ndarray:
    """Compute the start index for the given single stride."""
    # ---------- Efficient numpy code ------- 
    np_start = self._get_np_value(self.start)
    np_stride = self._get_np_value(stride)
    # Compute the start times (these remain the same for each series)
    return np.arange(
        start=np_start,
        stop=np_start + self._calc_nb_segments_for_stride(stride) * np_stride,
        step=np_stride,
    )
```

--------------------------------

### Construct Start Indices (NumPy, Python)

Source: https://predict-idlab.github.io/tsflex/features/segmenter/index

Constructs a sorted array of unique start indices for all specified stride values. This function iterates through defined strides, calculates the start indices for each using NumPy, concatenates them, and then returns a unique, sorted array.

```python
def _construct_start_idxs(self) -> np.ndarray:
    """Construct the start indices of the segments (for all stride values).

    To realize this, we compute the start idxs for each stride and then merge them
    together (without duplicates) in a sorted array.
    """
    start_idxs = []
    for stride in self.strides:
        start_idxs += [self._get_np_start_idx_for_stride(stride)]
    # note - np.unique also sorts the array
    return np.unique(np.concatenate(start_idxs))
```

--------------------------------

### Python Data Preprocessing with tsflex

Source: https://predict-idlab.github.io/tsflex/processing/series_pipeline

This Python code snippet demonstrates the core data preprocessing logic within the tsflex library. It handles data conversion, logging setup, application of processing steps, and formatting of the output based on user-defined options like return_all_series and drop_keys. It also includes error handling for processing failures.

```python
from typing import Dict, List, Union, Set, Path
import pandas as pd
from tsflex.utils.logging import delete_logging_handlers, add_logging_handler
from tsflex.exceptions import _ProcessingError

def process_data(self, data: Union[List[pd.Series], pd.DataFrame], return_df: bool = False, return_all_series: bool = True, drop_keys: List[str] = None, copy: bool = False, logging_file_path: Union[str, Path] = None) -> Union[List[pd.Series], pd.DataFrame]:
    """
    Preprocesses the input time series data using a series of defined processing steps.

    Args:
        data : Dataframe or Series or list thereof, with all the required data for the
            processing steps. 
            **Remark**: each Series / DataFrame must have a ``pd.DatetimeIndex``.
            **Remark**: we assume that each name / column is unique.
        return_df : bool, optional
            Whether the output needs to be a series list or a DataFrame, by default
            False.
            If True the output series will be combined to a DataFrame with an outer
            merge.
        return_all_series : bool, optional
            Whether the output needs to return all the series, by default True.
            * If True the output will contain all series that were passed to this
            method.
            * If False the output will contain just the required series (see
            ``get_required_series``).
        drop_keys : List[str], optional
            Which keys should be dropped when returning the output, by default None.
        copy : bool, optional
            Whether the series in ``data`` should be copied, by default False.
        logging_file_path : Union[str, Path], optional
            The file path where the logged messages are stored, by default None.
            If ``None``, then no logging ``FileHandler`` will be used and the logging
            messages are only pushed to stdout. Otherwise, a logging ``FileHandler`` will
            write the logged messages to the given file path.

        Returns
        -------
        Union[List[pd.Series], pd.DataFrame]
            The preprocessed series.

        Notes
        -----
        * If a ``logging_file_path`` is provided, the execution (time) info can be
          retrieved by calling ``logger.get_processor_logs(logging_file_path)``. <br>
          Be aware that the ``logging_file_path`` gets cleared before the logger pushes
          logged messages. Hence, one should use a separate logging file for each
          constructed processing and feature instance with this library.
        * If a series processor its function output is a ``np.ndarray``, the input series
          dict (required dict for that function) must contain just 1 series! That series
          its name and index are used to return a series dict. When a user does not want
          a numpy array to replace its input series, it is his / her responsibility to
          create a new ``pd.Series`` (or ``pd.DataFrame``) of that numpy array with a
          different (column) name.
        * If ``func_output`` is a ``pd.Series``, keep in mind that the input series gets
          transformed (i.e., replaced) in the pipeline with the ``func_output`` when the
          series name is  equal.

        Raises
        ------
        _ProcessingError
            Error raised when a processing step fails.

        """
        # Delete other logging handlers
        delete_logging_handlers(logger)
        # Add logging handler (if path provided)
        if logging_file_path:
            f_handler = add_logging_handler(logger, logging_file_path)

        # Convert the data to a series_dict
        series_dict: Dict[str, pd.Series] = {}
        for s in to_series_list(data):
            # Assert the assumptions we make!
            if len(s):
                assert isinstance(s.index, pd.DatetimeIndex)
            # TODO: also check monotonic increasing?

            if s.name in self.get_required_series():
                series_dict[str(s.name)] = s.copy() if copy else s
            elif return_all_series:
                # If all the series have to be returned
                series_dict[str(s.name)] = s.copy() if copy else s

        output_keys: Set[str] = set()  # Maintain set of output series
        for processor in self.processing_steps:
            try:
                processed_dict = processor(series_dict)
                output_keys.update(processed_dict.keys())
                series_dict.update(processed_dict)
            except Exception as e:
                # Close the file handler (this avoids PermissionError: [WinError 32])
                if logging_file_path:
                    f_handler.close()
                    logger.removeHandler(f_handler)
                raise _ProcessingError(
                    "Error while processing function {}:\n {}".format(
                        processor.name, str(e)
                    )
                ) from e

        # Close the file handler (this avoids PermissionError: [WinError 32])
        if logging_file_path:
            f_handler.close()
            logger.removeHandler(f_handler)

        if not return_all_series:
            # Return just the output series
            output_dict = {key: series_dict[str(key)] for key in output_keys}
            series_dict = output_dict

        if drop_keys is not None:
            # Drop the keys that should not be included in the output
            output_dict = {
                key: series_dict[key]

```

--------------------------------

### Compute Start Indices for Stride (NumPy, Python)

Source: https://predict-idlab.github.io/tsflex/features/segmenter/index

Computes the starting indices for segments given a specific stride. This function leverages NumPy for efficient array operations, calculating the sequence of start times for all possible windows based on the stride length and series boundaries.

```python
def _get_np_start_idx_for_stride(self, stride: T) -> np.ndarray:
    """Compute the start index for the given single stride."""
    # ---------- Efficient numpy code -------
    np_start = self._get_np_value(self.start)
    np_stride = self._get_np_value(stride)
    # Compute the start times (these remain the same for each series)
    return np.arange(
        start=np_start,
        stop=np_start + self._calc_nb_segments_for_stride(stride) * np_stride,
        step=np_stride,
    )
```

--------------------------------

### Instantiate FuncWrapper with Parameters

Source: https://predict-idlab.github.io/tsflex/features

This snippet demonstrates how to create an instance of FuncWrapper. It takes a function, optional output names, input type specification, a flag for vectorized execution, and additional keyword arguments.

```python
from tsflex.utils.classes import FuncWrapper
import numpy as np

def my_function(data):
    return np.sum(data)

# Example 1: Basic instantiation
wrapper_basic = FuncWrapper(func=my_function)

# Example 2: With output names and input type
wrapper_with_params = FuncWrapper(
    func=my_function,
    output_names="sum_output",
    input_type=np.ndarray
)

# Example 3: With vectorized execution
vectorized_wrapper = FuncWrapper(
    func=np.max, 
    vectorized=True, 
    axis=1
)

# Example 4: With keyword arguments
def multiply_by(data, factor):
    return data * factor

wrapper_with_kwargs = FuncWrapper(
    func=multiply_by,
    kwargs={'factor': 2}
)
```

--------------------------------

### Initialize FeatureDescriptor with Function, Series, Window, and Stride

Source: https://predict-idlab.github.io/tsflex/features/feature

Initializes a FeatureDescriptor object. It takes a function (Callable or FuncWrapper), series names, an optional window size, and an optional stride. The window and stride can be specified as floats, strings, or pandas Timedelta objects. The constructor includes type checking for the function, window, and stride to ensure compatibility.

```python
def __init__(
    self,
    function: Union[FuncWrapper, Callable],
    series_name: Union[str, Tuple[str, ...]],
    window: Optional[Union[float, str, pd.Timedelta]] = None,
    stride: Optional[
        Union[float, str, pd.Timedelta, List[Union[float, str, pd.Timedelta]]]
    ] = None,
):
    strides = sorted(set(to_list(stride)))  # omit duplicate stride values
    if window is None:
        assert strides == [None], "stride must be None if window is None"
    self.series_name: Tuple[str, ...] = to_tuple(series_name)
    self.window = parse_time_arg(window) if isinstance(window, str) else window
    if strides == [None]:
        self.stride = None
    else:
        self.stride = [
            parse_time_arg(s) if isinstance(s, str) else s for s in strides
        ]

    # Verify whether window and stride are either both sequence or time based
    dtype_set = set(
        AttributeParser.determine_type(v)
        for v in [self.window] + to_list(self.stride)
    ).difference([DataType.UNDEFINED])
    if len(dtype_set) > 1:
        raise TypeError(
            f"a combination of window ({self.window} type={type(self.window)}) and"
            f" stride ({self.stride}) is not supported!"
        )

    # Order of if statements is important (as FuncWrapper also is a Callable)!
    if isinstance(function, FuncWrapper):
        self.function: FuncWrapper = function
    elif isinstance(function, Callable):  # type: ignore[arg-type]
        self.function: FuncWrapper = FuncWrapper(function)  # type: ignore[no-redef]
    else:
        raise TypeError(
            "Expected feature function to be `Callable` or `FuncWrapper` but is a"
            f" {type(function)}."
        )

    # Construct a function-string
    f_name = str(self.function)
    self._func_str: str = f"{self.__class__.__name__} - func: {f_name}"

    self._freeze()

```

--------------------------------

### Segmentation with Start and End Indices

Source: https://predict-idlab.github.io/tsflex/features/feature_collection

Allows for custom segmentation of data by providing explicit start and end indices. When used, these indices define the segments, overriding any window or stride parameters. This method supports variable-length segments. Ensure start indices are less than or equal to their corresponding end indices.

```python
segment_start_idxs: list[int]
segment_end_idxs: list[int]
```

--------------------------------

### Initialize SeriesPipeline with Processors (Python)

Source: https://predict-idlab.github.io/tsflex/processing/index

Initializes a SeriesPipeline with a list of SeriesProcessor or SeriesPipeline instances. These processors are applied sequentially. If a SeriesPipeline is provided, its internal processors are flattened and added.

```python
class SeriesPipeline:
    """Pipeline for applying ``SeriesProcessor`` objects sequentially.

    Parameters
    ----------
    processors : List[Union[SeriesProcessor, SeriesPipeline]], optional
        List of ``SeriesProcessor`` or ``SeriesPipeline`` instances that will be applied
        sequentially to the internal series dict, by default None.
        **The processing steps will be executed in the same order as passed in this
        list.**

    """

    def __init__(n
```

--------------------------------

### Python: Many-to-One Processing Function Example

Source: https://predict-idlab.github.io/tsflex/processing/index

Example of a 'many-to-one' processing function in Python using pandas Series. This function takes multiple Series as input and outputs a single Series representing the absolute difference between two input Series.

```python
def abs_diff(s1: pd.Series, s2: pd.Series) -> pd.Series:
    return pd.Series(np.abs(s1-s2), name=f"abs_diff_{s1.name}-{s2.name}")
```

--------------------------------

### Initialize Series Container with Data and Windowing Parameters

Source: https://predict-idlab.github.io/tsflex/features/segmenter/index

Initializes a SeriesContainer object, handling various input data types (Series, DataFrame, lists) and windowing configurations. It standardizes input data, determines index bounds, constructs segment indices based on provided or calculated start/end times, and prepares series containers for processing. Dependencies include pandas, numpy, and internal tsflex utility functions.

```python
def __init__(
        self,
        data: Union[pd.Series, pd.DataFrame, List[Union[pd.Series, pd.DataFrame]]],
        window: Optional[T],
        strides: Optional[Union[T, List[T]]] = None,
        segment_start_idxs: Optional[np.ndarray] = None,
        segment_end_idxs: Optional[np.ndarray] = None,
        start_idx: Optional[T] = None,
        end_idx: Optional[T] = None,
        func_data_type: Union[np.ndarray, pd.Series] = np.ndarray,
        window_idx: str = "end",
        include_final_window: bool = False,
        approve_sparsity: bool = False,
    ):
        if strides is not None:
            strides = to_list(strides)

        # Check the passed segment indices
        if segment_start_idxs is not None and segment_end_idxs is not None:
            _check_start_end_array(segment_start_idxs, segment_end_idxs)

        if window is not None:
            assert AttributeParser.check_expected_type(
                [window] + ([] if strides is None else strides), self.win_str_type
            )

        self.window = window  # type: ignore[var-annotated]
        self.strides = strides  # type: ignore[var-annotated]

        self.window_idx = window_idx
        self.include_final_window = include_final_window
        self.approve_sparsity = approve_sparsity

        assert func_data_type in SUPPORTED_STROLL_TYPES
        self.data_type = func_data_type

        # 0. Standardize the input
        series_list: List[pd.Series] = to_series_list(data)
        self.series_dtype = AttributeParser.determine_type(series_list)
        self.series_key: Tuple[str, ...] = tuple([str(s.name) for s in series_list])

        # 1. Determine the start index
        self.start, self.end = start_idx, end_idx  # type: ignore[var-annotated]
        if self.start is None or self.end is None:
            # We always pass start_idx and end_idx from the FeatureCollection.calculate
            # Hence, this code is only useful for testing purposes
            start, end = _determine_bounds("inner", series_list)

            # update self.start & self.end if it was not passed
            self.start = start if self.start is None else self.start
            self.end = end if self.end is None else self.end

        # Especially useful when the index dtype differs from the win-stride-dtype
        # e.g. -> performing a int-based stroll on time-indexed data
        # Note: this is very niche and thus requires advanced knowledge
        # TODO: this code can be omitted if we remove TimeIndexSampleStridedRolling
        self._update_start_end_indices_to_stroll_type(series_list)

        # 2. Construct the index ranges
        # Either use the passed segment indices or compute the start or end times of the
        # segments. The segment indices have precedence over the stride (and window) for
        # index computation.
        if segment_start_idxs is not None or segment_end_idxs is not None:
            self.strides = None
            if segment_start_idxs is not None and segment_end_idxs is not None:
                # When both the start and end points are passed, the window does not
                # matter.
                self.window = None
                np_start_times = self._parse_segment_idxs(segment_start_idxs)
                np_end_times = self._parse_segment_idxs(segment_end_idxs)
            elif segment_start_idxs is not None:  # segment_end_idxs is None
                np_start_times = self._parse_segment_idxs(segment_start_idxs)
                np_end_times = np_start_times + self._get_np_value(self.window)
            else:  # segment_end_idxs is not None and segment_start_idxs is None
                np_end_times = self._parse_segment_idxs(segment_end_idxs)  # type: ignore[arg-type]
                np_start_times = np_end_times - self._get_np_value(self.window)
        else:
            np_start_times = self._construct_start_idxs()
            np_end_times = np_start_times + self._get_np_value(self.window)

        # Check the numpy start and end indices
        _check_start_end_array(np_start_times, np_end_times)

        # 3. Create a new-index which will be used for DataFrame reconstruction
        # Note: the index-name of the first passed series will be re-used as index-name
        self.index = self._get_output_index(
            np_start_times, np_end_times, name=series_list[0].index.name
        )

        # 4. Store the series containers
        self.series_containers = self._construct_series_containers(
            series_list, np_start_times, np_end_times
        )

        # 5. Check the sparsity assumption

```

--------------------------------

### Python: Many-to-Many Processing Function Example

Source: https://predict-idlab.github.io/tsflex/processing/index

Example of a 'many-to-many' processing function in Python using pandas Series. This function takes two Series as input and outputs a list of two Series: one for the absolute difference and one for the squared difference between the input Series.

```python
def abs_square_diff(s1: pd.Series, s2: pd.Series) -> List[pd.Series]:
    s_abs = pd.Series(np.abs(s1-s2), name=f"abs_{s1.name}-{s2.name}")
    s_square = pd.Series(np.square(s1-s2), name=f"square_{s1.name}-{s2.name}")
    return [s_abs, s_square]
```

--------------------------------

### Python: One-to-Many Processing Function Example

Source: https://predict-idlab.github.io/tsflex/processing/index

Example of a 'one-to-many' processing function in Python using pandas Series. This function takes a single Series as input and outputs a list of two Series: one for the absolute values and one for the squared values of the input Series.

```python
def abs_square(s1: pd.Series) -> List[pd.Series]:
    s1_abs = pd.Series(np.abs(s1), name=f"abs_{s1.name}")
    s1_square = pd.Series(np.square(s1), name=f"square_{s1.name}")
    return [s1_abs, s1_square]
```

--------------------------------

### Create and Calculate FeatureCollection with Pandas and SciPy

Source: https://predict-idlab.github.io/tsflex/features/index

An executable example demonstrating how to create a FeatureCollection with multiple FeatureDescriptors, calculate features from Pandas DataFrame, and specifying window and stride. It uses pandas for data handling, scipy.stats for statistical functions, and tsflex for feature extraction.

```python
import pandas as pd
import scipy.stats as ss
import numpy as np
from tsflex.features import FeatureDescriptor, FeatureCollection, FuncWrapper

# 1. -------- Get your time-indexed data --------
# Data contains 1 column; ["TMP"]
url = "https://github.com/predict-idlab/tsflex/raw/main/examples/data/empatica/"
data = pd.read_parquet(url + "tmp.parquet").set_index("timestamp")

# 2 -------- Construct your feature collection --------
fc = FeatureCollection(
    feature_descriptors=[
        FeatureDescriptor(
            function=FuncWrapper(func=ss.skew, output_names="skew"),
            series_name="TMP", 
            window="5min", stride="2.5min",
        )
    ]
)
# -- 2.1. Add features to your feature collection
# NOTE: tsflex allows features to have different windows and strides
fc.add(FeatureDescriptor(np.min, "TMP", '2.5min', '2.5min'))

# 3 -------- Calculate features --------
fc.calculate(data=data, return_df=True)
```

--------------------------------

### Initialize MultipleFeatureDescriptors

Source: https://predict-idlab.github.io/tsflex/features/feature

Creates a MultipleFeatureDescriptors object, which generates a list of FeatureDescriptor objects by combining all possible variations of the provided functions, series names, windows, and strides. The total number of created FeatureDescriptor objects is the product of the lengths of these input lists.

```python
class MultipleFeatureDescriptors:
    """Create a MultipleFeatureDescriptors object.

    Create a list of features from **all** combinations of the given parameter
    lists. Total number of created `FeatureDescriptor`s will be:

        len(func_inputs)*len(functions)*len(windows)*len(strides).

    Parameters
    ----------
    functions : Union[FuncWrapper, Callable, List[Union[FuncWrapper, Callable]]]
        The functions, can be either of both types (even in a single array).
    series_names : Union[str, Tuple[str, ...], List[str], List[Tuple[str, ...]]]
        The names of the series on which the feature function should be applied.

        * If `series_names` is a (list of) string (or tuple of a single string),
          then each `function` should require just one series as input.
    """
    # ... (implementation details would follow)
```

--------------------------------

### Parse Segment Indices with Bounds Checking

Source: https://predict-idlab.github.io/tsflex/features/segmenter/strided_rolling

This function parses segment indices, ensuring they are of type datetime64. It checks if the segment indices fall within the defined start and end bounds of the data. If any indices are outside these bounds, a RuntimeWarning is issued. The function expects a NumPy array of segment indices and the start and end bounds of the data, potentially with timezone information.

```python
def _parse_segment_idxs(self, segment_idxs: np.ndarray) -> np.ndarray:
        segment_idxs = segment_idxs.astype("datetime64")
        start_, end_ = self.start, self.end
        if start_.tz is not None:
            # Convert to UTC (allowing comparison with the segment_idxs)
            assert end_.tz is not None
            start_ = start_.tz_convert(None)
            end_ = end_.tz_convert(None)
        if any((segment_idxs < start_) | (segment_idxs > end_)):
            warnings.warn(self.OUTSIDE_DATA_BOUNDS_WARNING, RuntimeWarning)
        return segment_idxs
```
Tokens:
Raw