pdmlabs.preprocessing.record_level.windowing

pdmlabs.preprocessing.record_level.windowing#

Sliding window preprocessing for transforming univariate time series into multivariate windows.

Windowing converts a 1D time series into sequences/windows, useful for: - Neural network inputs (LSTM, CNN expect sequences) - Learning temporal patterns - Converting univariate → multivariate representations

Example: Time series [1, 2, 3, 4, 5] with window_size=3 becomes:
[[1, 2, 3],

[2, 3, 4], [3, 4, 5]]

Each new row is a sliding window of past values (horizon = window_size).

Classes

Windowing(event_preferences[, ...])

Convert univariate time series into sliding windows for sequence-based methods.
class pdmlabs.preprocessing.record_level.windowing.Windowing(event_preferences: EventPreferences, slidingWindow=None, col_pos=0)#

Bases: RecordLevelPreProcessorInterface

Convert univariate time series into sliding windows for sequence-based methods.

This preprocessor transforms a single column (univariate) into a multivariate representation using sliding windows. Useful for neural network-based anomaly detectors (LSTM, CNN, etc.) that expect sequence inputs.

col_pos#

Column position (0-based index) of the univariate series to window. Other columns are kept as-is.

Type:

int

slidingWindow#

Window size (number of past timesteps to include). If None, automatically determines optimal window size. If < 2, auto-determines via utils.find_length().

Type:

int or None

Examples

>>> from pdmlabs.preprocessing.record_level.windowing import Windowing
>>> import pandas as pd
>>>
>>> # Univariate data (single sensor)
>>> df_train = pd.DataFrame({'sensor': [1.0, 1.1, 1.2, 1.3]})
>>> df_test = pd.DataFrame({'sensor': [1.4, 1.5, 1.6]})
>>>
>>> windower = Windowing(
...     event_preferences={'failure': [], 'reset': []},
...     slidingWindow=3,
...     col_pos=0
... )
>>> windower.fit([df_train], ['bearing_1'], events_df)
>>> df_test_windowed = windower.transform(df_test, 'bearing_1', events_df)
>>> # df_test_windowed now has columns s_0, s_1, s_2 (windows of size 3)
fit(historic_data: list, historic_sources: list[str], event_data: DataFrame, anomaly_ranges=None) → None#

Fit windowing (no-op, just placeholder).

Windowing is stateless, so fit() does nothing.

Parameters:

  • historic_data (list[pd.DataFrame]) – Ignored.

  • historic_sources (list[str]) – Ignored.

  • event_data (pd.DataFrame) – Ignored.

  • anomaly_ranges – Ignored.

get_params()#

Return hyperparameters.

Returns:

{‘col_pos’: column index, ‘slidingWindow’: window size or None}

Return type:

dict

transform(target_data: DataFrame, source: str, event_data: DataFrame) → DataFrame#

Apply sliding window transformation to univariate series.

Parameters:

  • target_data (pd.DataFrame) – Test data to window.

  • source (str) – Source identifier (unused by this preprocessor).

  • event_data (pd.DataFrame) – Event log (unused).

Returns:

Windowed data with columns s_0, s_1, …, plus other columns.

Return type:

pd.DataFrame

Examples

>>> df_windowed = windower.transform(df_test, 'bearing_1', events_df)
>>> print(df_windowed.columns)
Index(['s_0', 's_1', 's_2', ...], dtype='object')
transform_one(new_sample: Series, source: str, is_event: bool) → Series#

Transform a single sample (not fully implemented for streaming).

Parameters:

  • new_sample (pd.Series) – Single row.

  • source (str) – Source identifier.

  • is_event (bool) – Event flag.

Returns:

Placeholder (streaming windowing not yet implemented).

Return type:

pd.Series or None

Contents