zero_order_hold_matching

ZeroOrderHold(features, name='aligned', *, drop=True)

Bases: Transform

Aligns multiple time series features using zero-order-hold.

Initialize the ZeroOrderHoldMatching transform.

Parameters:

Name	Type	Description	Default
features	list[str]	List of topics to align.	required
name	str	Name of the output time series feature.	'aligned'
drop	bool	Whether to drop the original features after alignment.	True

Source code in src/flowcean/polars/transforms/zero_order_hold_matching.py

def __init__(
    self,
    features: list[str],
    name: str = "aligned",
    *,
    drop: bool = True,
) -> None:
    """Initialize the ZeroOrderHoldMatching transform.

    Args:
        features: List of topics to align.
        name: Name of the output time series feature.
        drop: Whether to drop the original features after alignment.
    """
    super().__init__()
    self.features = features
    self.name = name
    self.drop = drop

zero_order_hold_align(data, columns, name)

Perform zero-order-hold alignment of multiple time series features.

Parameters:

Name	Type	Description	Default
data	LazyFrame	Input DataFrame containing struct-type time series columns.	required
columns	Iterable[str]	Names of struct columns to align using zero-order-hold.	required
name	str	Name of the output struct column.	required

Returns:

Type	Description
LazyFrame	zero-order-hold aligned time series

Source code in src/flowcean/polars/transforms/zero_order_hold_matching.py

def zero_order_hold_align(
    data: pl.LazyFrame,
    columns: Iterable[str],
    name: str,
) -> pl.LazyFrame:
    """Perform zero-order-hold alignment of multiple time series features.

    Args:
        data: Input DataFrame containing struct-type time series columns.
        columns: Names of struct columns to align using zero-order-hold.
        name: Name of the output struct column.

    Returns:
        zero-order-hold aligned time series
    """
    exploded = (
        data.with_row_index()
        .explode(column)
        .select(
            pl.col("index"),
            pl.col(column).struct.field("time"),
            pl.col(column)
            .struct.field("value")
            .name.prefix_fields(f"{column}/")
            .struct.unnest(),
        )
        for column in columns
    )

    return (
        pl.concat(exploded, how="align")
        .with_columns(pl.exclude("index", "time").forward_fill().over("index"))
        .drop_nulls()
        .select(
            pl.col("index"),
            pl.struct(
                pl.col("time"),
                pl.struct(
                    pl.exclude("index", "time"),
                ).alias("value"),
            ).alias(name),
        )
        .group_by("index", maintain_order=True)
        .agg(pl.all().implode())
        .drop("index")
    )