rhythmic_segments.meta

Metadata-handling utilities for rhythmic segments.

Attributes

Aggregator

Functions

coerce_meta_frame(→ pandas.DataFrame)	Return meta as a pandas.DataFrame.
aggregate_meta(→ pandas.DataFrame)	Aggregate metadata windows aligned with numeric value blocks.
resolve_columns_and_names(→ Tuple[List[str], List[str]])
agg_copy(→ Dict[str, Any])	Return every column value keyed by its position.
agg_index(→ Dict[str, Any])	Return metadata from the row at index.
agg_first(→ Dict[str, Any])	Return metadata from the first interval.
agg_last(→ Dict[str, Any])	Return metadata from the last interval.
agg_join(→ Dict[str, Any])	Join values from each column.
agg_list(→ Dict[str, Any])	Return metadata as lists per column.
agg_mode(→ Dict[str, Any])	Return the most frequent value per column using first-occurrence tie breaks.
get_aggregator(→ Aggregator)	Return a named metadata aggregator.

Module Contents

rhythmic_segments.meta.Aggregator

rhythmic_segments.meta.coerce_meta_frame(meta: Any, expected_rows: int | None = None, missing_rows_message: str = 'meta must match the expected row count', *, columns: collections.abc.Iterable[str] | None = None, constants: collections.abc.Mapping[str, Any] | None = None) → pandas.DataFrame

Return meta as a pandas.DataFrame.

rhythmic_segments.meta.aggregate_meta(meta_blocks: collections.abc.Iterable[pandas.DataFrame], value_blocks: collections.abc.Iterable[numpy.ndarray], window_len: int, meta_agg: Aggregator, expected_records: int) → pandas.DataFrame

Aggregate metadata windows aligned with numeric value blocks.

rhythmic_segments.meta.resolve_columns_and_names(df: pandas.DataFrame, columns: collections.abc.Iterable[str] | None = None, names: collections.abc.Iterable[str] | None = None) → Tuple[List[str], List[str]]

rhythmic_segments.meta.agg_copy(df: pandas.DataFrame, *, columns: collections.abc.Iterable[str] | None = None, names: collections.abc.Iterable[str] | None = None) → Dict[str, Any]

Return every column value keyed by its position.

>>> import pandas as pd
>>> agg_copy(pd.DataFrame({"label": ["a", "b"]}))
{'label_1': 'a', 'label_2': 'b'}

rhythmic_segments.meta.agg_index(df: pandas.DataFrame, index: int, *, columns: collections.abc.Iterable[str] | None = None, names: collections.abc.Iterable[str] | None = None) → Dict[str, Any]

Return metadata from the row at index.

>>> import pandas as pd
>>> agg_index(pd.DataFrame({"label": ["a", "b"], "section": ["x", "y"]}), 1)
{'label': 'b', 'section': 'y'}
>>> agg_index(pd.DataFrame({"label": ["a", "b"], "section": ["x", "y"]}), 1, columns=["label"])
{'label': 'b'}

rhythmic_segments.meta.agg_first(df: pandas.DataFrame, *, columns: collections.abc.Iterable[str] | None = None, names: collections.abc.Iterable[str] | None = None) → Dict[str, Any]

Return metadata from the first interval.

>>> import pandas as pd
>>> agg_first(pd.DataFrame({"label": ["a", "b"]}))
{'label': 'a'}

rhythmic_segments.meta.agg_last(df: pandas.DataFrame, *, columns: collections.abc.Iterable[str] | None = None, names: collections.abc.Iterable[str] | None = None) → Dict[str, Any]

Return metadata from the last interval.

>>> import pandas as pd
>>> agg_last(pd.DataFrame({"label": ["a", "b"]}))
{'label': 'b'}

rhythmic_segments.meta.agg_join(df: pandas.DataFrame, *, separator: str = '-', columns: collections.abc.Iterable[str] | None = None, names: collections.abc.Iterable[str] | None = None) → Dict[str, Any]

Join values from each column.

>>> import pandas as pd
>>> agg_join(pd.DataFrame({"label": ["a", "b"], "section": ["x", "y"]}), separator="|")
{'label': 'a|b', 'section': 'x|y'}

rhythmic_segments.meta.agg_list(df: pandas.DataFrame, *, columns: collections.abc.Iterable[str] | None = None, names: collections.abc.Iterable[str] | None = None) → Dict[str, Any]

Return metadata as lists per column.

>>> import pandas as pd
>>> agg_list(pd.DataFrame({"label": ["a", "b"]}))
{'label': ['a', 'b']}

rhythmic_segments.meta.agg_mode(df: pandas.DataFrame, *, columns: collections.abc.Iterable[str] | None = None, names: collections.abc.Iterable[str] | None = None) → Dict[str, Any]

Return the most frequent value per column using first-occurrence tie breaks.

The most common value is reported for each selected column. When multiple values share the same top frequency, the value that appears first in the column is chosen so results stay deterministic.

>>> import pandas as pd
>>> agg_mode(pd.DataFrame({"label": ["a", "b", "a"], "section": ["x", "y", "y"]}))
{'label': 'a', 'section': 'y'}
>>> agg_mode(pd.DataFrame({"label": ["b", "a", "a", "b"]}))
{'label': 'b'}

rhythmic_segments.meta.get_aggregator(name: str, **kwargs: Any) → Aggregator

Return a named metadata aggregator.

Supported names are "copy", "index", "first", "last", "join", "list", and "mode". Additional keyword arguments are forwarded to the underlying aggregator.

>>> import pandas as pd
>>> agg = get_aggregator("first", columns=["label"])
>>> agg(pd.DataFrame({"label": ["a", "b"]}))
{'label': 'a'}
>>> join = get_aggregator("join", columns=["label"], separator="|")
>>> join(pd.DataFrame({"label": ["a", "b"]}))
{'label': 'a|b'}