Creating Custom Filters/Transformations in dFL¶

dFL allows you to add custom filters and transformations that can be used in the GUI for data processing. These include custom smoothing functions, normalization functions, and other data transformations. NOTE: at present all filters/transforamtions are only viewable in the normalizing* or smoothing** dropdown mensu in the Graph Configuration pane. This is however only cosmetic, any type of filter may be added, but it will only be accessible through one of those two menus.

Overview¶

Custom filters/transformations are added by:

Creating the transformation function in a utilities module
Registering the function in your data provider's configuration dictionaries

Function Signature¶

All custom transformation functions must follow this signature:

def my_custom_filter(
    record_name: str,
    signal_name: str,
    raw_signal: NDArray[Any],  # or np.ndarray
    times: NDArray[Any],        # or np.ndarray
    parameters: Dict[str, Any]
) -> NDArray[Any]:  # or np.ndarray
    """
    Apply custom transformation to a signal.

    Args:
        record_name (str): The name of the record being processed (can be unused)
        signal_name (str): The name of the signal being processed (can be unused)
        raw_signal (array-like): Input data array to transform
        times (array-like): Time array corresponding to the data (can be unused)
        parameters (dict): Dictionary of parameters for the transformation

    Returns:
        numpy.ndarray: Transformed data array (must be same length as input)
    """
    # Your transformation logic here
    return transformed_data

Important Requirements¶

All parameters are required, even if unused. Use _ prefix for unused parameters (e.g., _record_name).
Return type must be a numpy array with the same length as the input.
Parameter access: Use parameters.get("parameter_name", default_value) to safely access parameters.

Example: Simple Moving Average¶

Here's a complete example from the codebase:

import numpy as np
from typing import Dict, Any
from numpy.typing import NDArray

def simple_moving_average(
    _record_name: str,
    _signal_name: str,
    raw_signal: NDArray[Any],
    _times: NDArray[Any],
    parameters: Dict[str, Any]
) -> NDArray[Any]:
    """
    Apply simple moving average smoothing to a 1D array.

    Args:
        _record_name: Unused
        _signal_name: Unused
        raw_signal: Input data to smooth
        _times: Unused
        parameters: Dictionary containing window_size parameter

    Returns:
        Smoothed data array
    """
    data = np.array(raw_signal)

    # Access parameter with fallback naming conventions
    window_size = parameters.get(
        "simple_moving_average_moving_average_window_size",
        parameters.get("moving_average_window_size", 1)
    )
    window_size = int(window_size)

    # Validate window size
    if window_size < 1:
        window_size = 1
    if window_size > data.shape[0]:
        window_size = data.shape[0]

    # Apply moving average
    weights = np.ones(window_size) / window_size
    smoothed = np.convolve(data, weights, mode="valid")

    # Pad edges to maintain original length
    padding = window_size - 1
    left_pad = np.full(padding // 2, smoothed[0])
    right_pad = np.full(padding - padding // 2, smoothed[-1])

    return np.concatenate([left_pad, smoothed, right_pad])

Example: Robust Scaling (Normalization)¶

import numpy as np
from typing import Dict, Any
from numpy.typing import NDArray

def robust_scaling(
    record_name: str,
    signal_name: str,
    raw_signal: NDArray[Any],
    times: NDArray[Any],
    parameters: Dict[str, Any]
) -> NDArray[Any]:
    """
    Apply robust scaling using median and IQR.

    Formula: (x - median) / IQR
    """
    data = np.asarray(raw_signal)

    # Handle empty input
    if data.size == 0:
        return np.array([])

    # Calculate robust statistics
    median = np.median(data)
    q1, q3 = np.percentile(data, [25, 75])
    iqr = q3 - q1

    # Handle edge cases
    if iqr == 0:
        if np.all(data == data[0]):
            return np.zeros_like(data)
        iqr = np.std(data)
        if iqr == 0:
            iqr = 1.0

    # Apply scaling
    scaled_data = (data - median) / iqr

    # Clip extreme values
    return np.clip(scaled_data, -1e6, 1e6)

Registering Your Filter¶

After creating your function, register it in your data provider's configuration dictionaries.

For Smoothing Functions¶

In your data provider file (e.g., your_data_provider.py), add to custom_smoothing_options:

from your_utilities import simple_moving_average, your_custom_filter

custom_smoothing_options = {
    "simple_moving_average": {
        "display_name": "Simple Moving Average",
        "parameters": {
            "moving_average_window_size": {
                "default": 1,
                "min": 0,
                "max": None,  # None means no maximum
                "display_name": "Window Size"
            }
        },
        "function": simple_moving_average,
    },
    "your_custom_filter": {
        "display_name": "Your Custom Filter Name",
        "parameters": {
            "your_custom_filter_parameter_name": {
                "default": 0.5,
                "min": 0.0,
                "max": 1.0,
                "display_name": "Parameter Display Name"
            }
        },
        "function": your_custom_filter,
    }
}

For Normalization Functions¶

Add to custom_normalizing_options:

from your_utilities import robust_scaling, box_cox_and_zscore_normalize

custom_normalizing_options = {
    "robust_scaling": {
        "display_name": "Robust Scaling",
        "parameters": None,  # No parameters needed
        "function": robust_scaling
    },
    "box_cox_and_zscore_normalize": {
        "display_name": "Box-Cox and Z-score Normalize",
        "parameters": {
            "box_cox_and_zscore_normalize_lambda_value": {
                "default": 2.0,
                "min": -5.0,
                "max": 5.0,
                "display_name": "Lambda Value"
            },
            "box_cox_and_zscore_normalize_shift_value": {
                "default": 3.0,
                "min": 0.0,
                "max": None,
                "display_name": "Shift Value"
            }
        },
        "function": box_cox_and_zscore_normalize
    }
}

Complete Integration¶

Add these dictionaries to your get_provider() return value:

def get_provider():
    # ... other setup code ...

    data_coordinator_info = {
        "fetch_data": fetch_data,
        "dataset_id": "your_dataset",
        # ... other fields ...
        "custom_smoothing_options": custom_smoothing_options,
        "custom_normalizing_options": custom_normalizing_options,
        # ... rest of configuration ...
    }
    return data_coordinator_info

Parameter Naming Convention¶

When accessing parameters in your function, use this naming convention:

Full name: {function_name}_{parameter_name}
Short name: {parameter_name} (for backward compatibility)

Example: - Function: simple_moving_average - Parameter in GUI: moving_average_window_size - Access in code: parameters.get("simple_moving_average_moving_average_window_size", parameters.get("moving_average_window_size", 1))

Parameter Types¶

When defining parameters in the dictionary, you can specify:

{
    "parameter_name": {
        "default": default_value,           # Default value
        "min": minimum_value,              # Optional: minimum value (float/int)
        "max": maximum_value,               # Optional: maximum value (None = no limit)
        "display_name": "Display Name",     # Name shown in GUI
        "options": {                        # Optional: dropdown options
            "value1": "Display 1",
            "value2": "Display 2"
        }
    }
}

Best Practices¶

Handle edge cases: Empty arrays, single values, constant values, zero division
Validate parameters: Check ranges and types before using
Maintain array length: Output should have the same length as input
Use numpy arrays: Always convert to/return numpy arrays
Document thoroughly: Include docstrings explaining parameters and behavior
Test with real data: Verify your function works with actual signal data

Examples from the Codebase¶

Several working examples can be found in:

weather_app/weather_utilities.py: simple_moving_average, robust_scaling, trim_data
ga_offline_app/ga_offline_utilities.py: simple_moving_average, box_cox_and_zscore_normalize, robust_scaling
additive_manufacturing/am_utilities.py: Simple implementations
cme_app/cme_utilities.py: Timezone-aware transformations

These examples demonstrate various patterns for different use cases (numeric vs. datetime trimming, parameter handling, etc.).

Using LLM Prompt Engineering¶

Instead of manually editing the data provider, you can also use LLM prompt engineering to automatically generate and integrate custom filters/transformations into your data provider.