modus-py 0.1.0a1

Schema-driven, unit-aware Polars transform engine.

modus

Schema-driven, unit-aware Polars transform engine.

modus pairs ModusFrame — a Polars DataFrame with per-column Astropy unit metadata — with FrameTransform, a builder-pattern engine for expressing timeseries transform pipelines (grouping, cleaning, deriving, filtering, and aggregating) declaratively, while keeping unit metadata accurate throughout.

Installation

pip install modus-py

Quick start

import astropy.units as u
import polars
import modus

frame = modus.ModusFrame(
    data=polars.DataFrame({"hoist_pressure1": [101.3, 98.7]}),
    units={"hoist_pressure1": u.kPa},
)

velocity = frame.col("distance") / frame.col("time")  # UnitExpr; unit propagates
frame = frame.with_columns(velocity=velocity)

from datetime import timedelta
import modus

result = (
    modus.FrameTransform()
    .add_grouper(modus.DataGap("chunk", gap_seconds=60))
    .add_grouper(modus.EventSequence("body_up_event", source="dump_body_up_sts", value=1, within="chunk"))
    .add_cleaner(modus.Interpolate("hoist_pressure1_kpa", within="chunk"))
    .add_filter(modus.SuppressLeadingEvent("body_up_event", within_column="chunk"))
    .add_filter(modus.SuppressTrailingEvent("body_up_event"))
    .group_by("chunk", "body_up_event")
    .add_aggregation(modus.Mean("hoist_pressure1_kpa", output_name="mean_hoist_pressure1_kpa"))
    .add_aggregation(modus.Duration(output_name="body_up_duration_s"))
    .apply(frame)
)

labelled = result.labelled      # ModusFrame -- timeseries with grouper columns added
aggregated = result.aggregated  # ModusFrame -- one row per (chunk, body_up_event) group

Custom units

modus ships an extensible unit registry, in the same spirit as Astropy's own enabled-units model:

import astropy.units as u
import modus

modus.units.registry.register("kgcm2", u.def_unit("kgcm2", 98.0665 * u.kPa))

Custom derivations and aggregations

Both extension points are plugin registries keyed by name, so a schema or manifest layer built on top of modus can construct bespoke operations from declarative configuration without importing analytic-specific Python:

import modus

@modus.DerivationRegistry.register("state_inference")
class StateInferenceDerivation(modus.Derivation):
    ...

@modus.AggregationRegistry.register("weighted_percentile")
class WeightedPercentile(modus.Aggregation):
    ...

Development

pip install -e .[dev]
pytest

Licence

MIT — see LICENSE.