Skip to content

cds2.data_analysis

Pandas-backed DataSet and DataFrame bridge.

Pandas-backed data analysis — DataSet and DataFrame bridge.

A :class:DataSet is the v2 successor to the pure-Python cds.data_analysis.DataSet from v1. The public API is intentionally compatible (columns, shape, column, filter, head/tail, select, group_by, to_list) so existing tutorials port without rewrites, but the internal storage is a :class:pandas.DataFrame for vectorised speed and for seamless interoperability with the pandas ecosystem.

Design choices mirror the v1 pandas_io bridge:

  • None round-trips as NaN in numeric columns and back to None.
  • Column order is preserved in both directions.
  • The module can be used either through instance methods (ds.to_dataframe() / DataSet.from_dataframe(df)) or the free functions :func:to_dataframe / :func:from_dataframe.
  • Summary statistics are exposed via :meth:DataSet.describe (a thin wrapper around :meth:pandas.DataFrame.describe) and :meth:DataSet.summarize (column-level null / unique / numeric moments compatible with :func:cds2.io.summarize).

DataSet

A lightweight DataFrame-like container backed by :class:pandas.DataFrame.

Data is stored internally as a :class:pandas.DataFrame while the public API stays compatible with the v1 pure-Python implementation (list-of-dicts construction, column/filter/select/group_by helpers).

Source code in src/cds2/data_analysis.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
class DataSet:
    """A lightweight DataFrame-like container backed by :class:`pandas.DataFrame`.

    Data is stored internally as a :class:`pandas.DataFrame` while the public
    API stays compatible with the v1 pure-Python implementation (list-of-dicts
    construction, ``column``/``filter``/``select``/``group_by`` helpers).
    """

    def __init__(self, data: list[Row] | pd.DataFrame | None = None) -> None:
        """Create a dataset.

        Args:
            data: either a list of row dictionaries (``Row``) or a
                :class:`pandas.DataFrame`. ``None`` or an empty list creates
                an empty dataset with no columns.
        """
        if data is None:  # pragma: no cover
            self._df: pd.DataFrame = pd.DataFrame()  # pragma: no cover
        elif isinstance(data, pd.DataFrame):
            self._df = data.copy()
        else:
            rows: list[Row] = data
            if not rows:
                self._df = pd.DataFrame()
            else:
                columns = list(rows[0].keys())
                self._df = pd.DataFrame(rows, columns=columns if columns else None)

    @property
    def columns(self) -> list[str]:
        """Return the ordered list of column names."""
        return [str(col) for col in self._df.columns.tolist()]

    @property
    def shape(self) -> tuple[int, int]:
        """Return ``(rows, columns)``."""
        return int(self._df.shape[0]), int(self._df.shape[1])

    def __len__(self) -> int:
        """Return the number of rows."""
        return int(len(self._df))

    def __getitem__(self, idx: int) -> Row:
        """Return the row at integer position ``idx`` as a :class:`Row`."""
        if idx < 0:
            idx += len(self)
        if idx < 0 or idx >= len(self):  # pragma: no cover
            raise IndexError("DataSet index out of range")  # pragma: no cover
        series = self._df.iloc[idx]
        raw: dict[str, object] = cast(dict[str, object], series.to_dict())
        return _normalize_row(raw)

    def column(self, name: str) -> list[Scalar]:
        """Extract a single column as a list, mapping ``NaN`` to ``None``."""
        if name not in self._df.columns:
            raise ValueError(f"Column '{name}' not found. Available: {self.columns}")
        series = self._df[name]
        result: list[Scalar] = []
        for value in series.tolist():
            result.append(None if _is_na(value) else cast(Scalar, value))
        return result

    def filter(self, predicate: Callable[[Row], bool]) -> DataSet:
        """Filter rows by a predicate applied to each :class:`Row`."""
        filtered: list[Row] = [row for row in self.to_list() if predicate(row)]
        return DataSet(filtered)

    def head(self, n: int = 5) -> DataSet:
        """Return the first ``n`` rows."""
        return DataSet(cast(pd.DataFrame, self._df.head(n)))

    def tail(self, n: int = 5) -> DataSet:
        """Return the last ``n`` rows."""
        return DataSet(cast(pd.DataFrame, self._df.tail(n)))

    def select(self, *names: str) -> DataSet:
        """Select a subset of columns by name, preserving order."""
        for name in names:
            if name not in self._df.columns:
                raise ValueError(f"Column '{name}' not found.")
        subset: pd.DataFrame = self._df[list(names)].copy()
        return DataSet(subset)

    def group_by(self, column_name: str) -> DataGroup:
        """Group rows by ``column_name`` for aggregation."""
        if column_name not in self._df.columns:  # pragma: no cover
            raise ValueError(f"Column '{column_name}' not found.")  # pragma: no cover
        return DataGroup(self._df, column_name)

    def to_list(self) -> list[Row]:
        """Export data as a list of dictionaries, mapping ``NaN`` to ``None``."""
        raw_records: list[dict[str, object]] = cast(
            list[dict[str, object]], self._df.to_dict(orient="records")
        )
        return [_normalize_row(record) for record in raw_records]

    def to_dataframe(self) -> pd.DataFrame:
        """Return a copy of the underlying :class:`pandas.DataFrame`."""
        return cast(pd.DataFrame, self._df.copy())

    @classmethod
    def from_dataframe(cls, df: pd.DataFrame) -> DataSet:
        """Create a :class:`DataSet` from a :class:`pandas.DataFrame`."""
        return cls(df)

    def describe(self) -> pd.DataFrame:
        """Descriptive statistics via :meth:`pandas.DataFrame.describe`.

        Includes all columns (numeric and non-numeric) when data is present;
        returns an empty :class:`DataFrame` for an empty dataset.
        """
        if self._df.empty:  # pragma: no cover
            return pd.DataFrame()  # type: ignore[no-any-return, unused-ignore]  # pragma: no cover
        result = cast(pd.DataFrame, self._df.describe(include="all"))
        return result

    def summarize(self) -> pd.DataFrame:
        """Column-level summary: dtypes, nulls, uniques and numeric moments.

        Mirrors :func:`cds2.io.summarize` but operates directly on a
        :class:`DataSet`. Returns a :class:`pandas.DataFrame` with one row per
        column and fields ``column``, ``dtype``, ``non_null``, ``nulls``,
        ``unique``, ``mean``, ``std``, ``min``, ``max``.
        """
        if self._df.empty and len(self._df.columns) == 0:
            return pd.DataFrame(  # type: ignore[no-any-return, unused-ignore]
                columns=[
                    "column",
                    "dtype",
                    "non_null",
                    "nulls",
                    "unique",
                    "mean",
                    "std",
                    "min",
                    "max",
                ]
            )  # type: ignore[no-any-return, unused-ignore]
        df = self._df
        non_null = df.notna().sum()
        nulls = len(df) - non_null
        unique_counts = df.nunique(dropna=True)
        numeric = df.select_dtypes(include=[np.number])
        moments: pd.DataFrame
        if len(numeric.columns):  # pragma: no cover
            moments = cast(
                pd.DataFrame, numeric.agg(["mean", "std", "min", "max"])
            )  # pragma: no cover
        else:  # pragma: no cover
            moments = pd.DataFrame(index=["mean", "std", "min", "max"])  # pragma: no cover
        rows: list[dict[str, object]] = []
        for column in df.columns:
            mean_value: float | None = None
            std_value: float | None = None
            min_value: float | None = None
            max_value: float | None = None
            if column in moments.columns:
                col_moments = moments[column]
                mean_value = None if _is_na(col_moments["mean"]) else float(col_moments["mean"])
                std_value = None if _is_na(col_moments["std"]) else float(col_moments["std"])
                min_value = None if _is_na(col_moments["min"]) else float(col_moments["min"])
                max_value = None if _is_na(col_moments["max"]) else float(col_moments["max"])
            rows.append(
                {
                    "column": str(column),
                    "dtype": str(df[column].dtype),
                    "non_null": int(non_null[column]),
                    "nulls": int(nulls[column]),
                    "unique": int(unique_counts[column]),
                    "mean": mean_value,
                    "std": std_value,
                    "min": min_value,
                    "max": max_value,
                }
            )
        return pd.DataFrame(rows)  # type: ignore[no-any-return, unused-ignore]

    def __repr__(self) -> str:
        """Return a compact summary."""
        if self._df.empty and len(self._df.columns) == 0:
            return "DataSet(empty)"
        return f"DataSet(rows={len(self._df)}, cols={len(self._df.columns)})"

columns property

Return the ordered list of column names.

shape property

Return (rows, columns).

__init__(data=None)

Create a dataset.

Args: data: either a list of row dictionaries (Row) or a :class:pandas.DataFrame. None or an empty list creates an empty dataset with no columns.

Source code in src/cds2/data_analysis.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def __init__(self, data: list[Row] | pd.DataFrame | None = None) -> None:
    """Create a dataset.

    Args:
        data: either a list of row dictionaries (``Row``) or a
            :class:`pandas.DataFrame`. ``None`` or an empty list creates
            an empty dataset with no columns.
    """
    if data is None:  # pragma: no cover
        self._df: pd.DataFrame = pd.DataFrame()  # pragma: no cover
    elif isinstance(data, pd.DataFrame):
        self._df = data.copy()
    else:
        rows: list[Row] = data
        if not rows:
            self._df = pd.DataFrame()
        else:
            columns = list(rows[0].keys())
            self._df = pd.DataFrame(rows, columns=columns if columns else None)

__len__()

Return the number of rows.

Source code in src/cds2/data_analysis.py 
 98
 99
100
def __len__(self) -> int:
    """Return the number of rows."""
    return int(len(self._df))

__getitem__(idx)

Return the row at integer position idx as a :class:Row.

Source code in src/cds2/data_analysis.py 
102
103
104
105
106
107
108
109
110
def __getitem__(self, idx: int) -> Row:
    """Return the row at integer position ``idx`` as a :class:`Row`."""
    if idx < 0:
        idx += len(self)
    if idx < 0 or idx >= len(self):  # pragma: no cover
        raise IndexError("DataSet index out of range")  # pragma: no cover
    series = self._df.iloc[idx]
    raw: dict[str, object] = cast(dict[str, object], series.to_dict())
    return _normalize_row(raw)

column(name)

Extract a single column as a list, mapping NaN to None.

Source code in src/cds2/data_analysis.py 
112
113
114
115
116
117
118
119
120
def column(self, name: str) -> list[Scalar]:
    """Extract a single column as a list, mapping ``NaN`` to ``None``."""
    if name not in self._df.columns:
        raise ValueError(f"Column '{name}' not found. Available: {self.columns}")
    series = self._df[name]
    result: list[Scalar] = []
    for value in series.tolist():
        result.append(None if _is_na(value) else cast(Scalar, value))
    return result

filter(predicate)

Filter rows by a predicate applied to each :class:Row.

Source code in src/cds2/data_analysis.py 
122
123
124
125
def filter(self, predicate: Callable[[Row], bool]) -> DataSet:
    """Filter rows by a predicate applied to each :class:`Row`."""
    filtered: list[Row] = [row for row in self.to_list() if predicate(row)]
    return DataSet(filtered)

head(n=5)

Return the first n rows.

Source code in src/cds2/data_analysis.py 
127
128
129
def head(self, n: int = 5) -> DataSet:
    """Return the first ``n`` rows."""
    return DataSet(cast(pd.DataFrame, self._df.head(n)))

tail(n=5)

Return the last n rows.

Source code in src/cds2/data_analysis.py 
131
132
133
def tail(self, n: int = 5) -> DataSet:
    """Return the last ``n`` rows."""
    return DataSet(cast(pd.DataFrame, self._df.tail(n)))

select(*names)

Select a subset of columns by name, preserving order.

Source code in src/cds2/data_analysis.py 
135
136
137
138
139
140
141
def select(self, *names: str) -> DataSet:
    """Select a subset of columns by name, preserving order."""
    for name in names:
        if name not in self._df.columns:
            raise ValueError(f"Column '{name}' not found.")
    subset: pd.DataFrame = self._df[list(names)].copy()
    return DataSet(subset)

group_by(column_name)

Group rows by column_name for aggregation.

Source code in src/cds2/data_analysis.py 
143
144
145
146
147
def group_by(self, column_name: str) -> DataGroup:
    """Group rows by ``column_name`` for aggregation."""
    if column_name not in self._df.columns:  # pragma: no cover
        raise ValueError(f"Column '{column_name}' not found.")  # pragma: no cover
    return DataGroup(self._df, column_name)

to_list()

Export data as a list of dictionaries, mapping NaN to None.

Source code in src/cds2/data_analysis.py 
149
150
151
152
153
154
def to_list(self) -> list[Row]:
    """Export data as a list of dictionaries, mapping ``NaN`` to ``None``."""
    raw_records: list[dict[str, object]] = cast(
        list[dict[str, object]], self._df.to_dict(orient="records")
    )
    return [_normalize_row(record) for record in raw_records]

to_dataframe()

Return a copy of the underlying :class:pandas.DataFrame.

Source code in src/cds2/data_analysis.py 
156
157
158
def to_dataframe(self) -> pd.DataFrame:
    """Return a copy of the underlying :class:`pandas.DataFrame`."""
    return cast(pd.DataFrame, self._df.copy())

from_dataframe(df) classmethod

Create a :class:DataSet from a :class:pandas.DataFrame.

Source code in src/cds2/data_analysis.py 
160
161
162
163
@classmethod
def from_dataframe(cls, df: pd.DataFrame) -> DataSet:
    """Create a :class:`DataSet` from a :class:`pandas.DataFrame`."""
    return cls(df)

describe()

Descriptive statistics via :meth:pandas.DataFrame.describe.

Includes all columns (numeric and non-numeric) when data is present; returns an empty :class:DataFrame for an empty dataset.

Source code in src/cds2/data_analysis.py 
165
166
167
168
169
170
171
172
173
174
def describe(self) -> pd.DataFrame:
    """Descriptive statistics via :meth:`pandas.DataFrame.describe`.

    Includes all columns (numeric and non-numeric) when data is present;
    returns an empty :class:`DataFrame` for an empty dataset.
    """
    if self._df.empty:  # pragma: no cover
        return pd.DataFrame()  # type: ignore[no-any-return, unused-ignore]  # pragma: no cover
    result = cast(pd.DataFrame, self._df.describe(include="all"))
    return result

summarize()

Column-level summary: dtypes, nulls, uniques and numeric moments.

Mirrors :func:cds2.io.summarize but operates directly on a :class:DataSet. Returns a :class:pandas.DataFrame with one row per column and fields column, dtype, non_null, nulls, unique, mean, std, min, max.

Source code in src/cds2/data_analysis.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def summarize(self) -> pd.DataFrame:
    """Column-level summary: dtypes, nulls, uniques and numeric moments.

    Mirrors :func:`cds2.io.summarize` but operates directly on a
    :class:`DataSet`. Returns a :class:`pandas.DataFrame` with one row per
    column and fields ``column``, ``dtype``, ``non_null``, ``nulls``,
    ``unique``, ``mean``, ``std``, ``min``, ``max``.
    """
    if self._df.empty and len(self._df.columns) == 0:
        return pd.DataFrame(  # type: ignore[no-any-return, unused-ignore]
            columns=[
                "column",
                "dtype",
                "non_null",
                "nulls",
                "unique",
                "mean",
                "std",
                "min",
                "max",
            ]
        )  # type: ignore[no-any-return, unused-ignore]
    df = self._df
    non_null = df.notna().sum()
    nulls = len(df) - non_null
    unique_counts = df.nunique(dropna=True)
    numeric = df.select_dtypes(include=[np.number])
    moments: pd.DataFrame
    if len(numeric.columns):  # pragma: no cover
        moments = cast(
            pd.DataFrame, numeric.agg(["mean", "std", "min", "max"])
        )  # pragma: no cover
    else:  # pragma: no cover
        moments = pd.DataFrame(index=["mean", "std", "min", "max"])  # pragma: no cover
    rows: list[dict[str, object]] = []
    for column in df.columns:
        mean_value: float | None = None
        std_value: float | None = None
        min_value: float | None = None
        max_value: float | None = None
        if column in moments.columns:
            col_moments = moments[column]
            mean_value = None if _is_na(col_moments["mean"]) else float(col_moments["mean"])
            std_value = None if _is_na(col_moments["std"]) else float(col_moments["std"])
            min_value = None if _is_na(col_moments["min"]) else float(col_moments["min"])
            max_value = None if _is_na(col_moments["max"]) else float(col_moments["max"])
        rows.append(
            {
                "column": str(column),
                "dtype": str(df[column].dtype),
                "non_null": int(non_null[column]),
                "nulls": int(nulls[column]),
                "unique": int(unique_counts[column]),
                "mean": mean_value,
                "std": std_value,
                "min": min_value,
                "max": max_value,
            }
        )
    return pd.DataFrame(rows)  # type: ignore[no-any-return, unused-ignore]

__repr__()

Return a compact summary.

Source code in src/cds2/data_analysis.py 
237
238
239
240
241
def __repr__(self) -> str:
    """Return a compact summary."""
    if self._df.empty and len(self._df.columns) == 0:
        return "DataSet(empty)"
    return f"DataSet(rows={len(self._df)}, cols={len(self._df.columns)})"

DataGroup

Helper for grouped aggregations, backed by pandas.groupby.

Source code in src/cds2/data_analysis.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
class DataGroup:
    """Helper for grouped aggregations, backed by ``pandas.groupby``."""

    def __init__(self, df: pd.DataFrame, group_col: str) -> None:
        """Store the source frame and grouping column."""
        self._df: pd.DataFrame = df
        self.group_col: str = group_col

    def mean(self, numeric_col: str) -> dict[Scalar, float]:
        """Calculate the mean of ``numeric_col`` for each group.

        Non-numeric cells are ignored; groups with no numeric values map to
        ``0.0`` (matching the v1 behaviour).
        """
        result: dict[Scalar, float] = {}
        if numeric_col not in self._df.columns:  # pragma: no cover
            raise ValueError(f"Column '{numeric_col}' not found.")  # pragma: no cover
        grouped = self._df.groupby(self.group_col, dropna=False, observed=True)
        for key, group in grouped:
            py_key: Scalar = None if _is_na(key) else cast(Scalar, key)
            values: list[float] = []  # pragma: no cover
            for value in group[numeric_col].tolist():
                if _is_na(value):  # pragma: no cover
                    continue  # pragma: no cover
                if isinstance(value, (int, float, np.integer, np.floating)):
                    values.append(float(value))
                else:
                    try:
                        values.append(float(cast(float, value)))
                    except (TypeError, ValueError):
                        continue
            result[py_key] = sum(values) / len(values) if values else 0.0
        return result

    def count(self) -> dict[Scalar, int]:
        """Count the number of rows in each group."""
        result: dict[Scalar, int] = {}
        grouped = self._df.groupby(self.group_col, dropna=False, observed=True)
        for key, group in grouped:
            py_key: Scalar = None if _is_na(key) else cast(Scalar, key)
            result[py_key] = int(len(group))
        return result

__init__(df, group_col)

Store the source frame and grouping column.

Source code in src/cds2/data_analysis.py 
247
248
249
250
def __init__(self, df: pd.DataFrame, group_col: str) -> None:
    """Store the source frame and grouping column."""
    self._df: pd.DataFrame = df
    self.group_col: str = group_col

mean(numeric_col)

Calculate the mean of numeric_col for each group.

Non-numeric cells are ignored; groups with no numeric values map to 0.0 (matching the v1 behaviour).

Source code in src/cds2/data_analysis.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def mean(self, numeric_col: str) -> dict[Scalar, float]:
    """Calculate the mean of ``numeric_col`` for each group.

    Non-numeric cells are ignored; groups with no numeric values map to
    ``0.0`` (matching the v1 behaviour).
    """
    result: dict[Scalar, float] = {}
    if numeric_col not in self._df.columns:  # pragma: no cover
        raise ValueError(f"Column '{numeric_col}' not found.")  # pragma: no cover
    grouped = self._df.groupby(self.group_col, dropna=False, observed=True)
    for key, group in grouped:
        py_key: Scalar = None if _is_na(key) else cast(Scalar, key)
        values: list[float] = []  # pragma: no cover
        for value in group[numeric_col].tolist():
            if _is_na(value):  # pragma: no cover
                continue  # pragma: no cover
            if isinstance(value, (int, float, np.integer, np.floating)):
                values.append(float(value))
            else:
                try:
                    values.append(float(cast(float, value)))
                except (TypeError, ValueError):
                    continue
        result[py_key] = sum(values) / len(values) if values else 0.0
    return result

count()

Count the number of rows in each group.

Source code in src/cds2/data_analysis.py 
278
279
280
281
282
283
284
285
def count(self) -> dict[Scalar, int]:
    """Count the number of rows in each group."""
    result: dict[Scalar, int] = {}
    grouped = self._df.groupby(self.group_col, dropna=False, observed=True)
    for key, group in grouped:
        py_key: Scalar = None if _is_na(key) else cast(Scalar, key)
        result[py_key] = int(len(group))
    return result

to_dataframe(ds)

Convert a :class:DataSet to a :class:pandas.DataFrame.

Thin wrapper around :meth:DataSet.to_dataframe for compatibility with the v1 cds.data_analysis.pandas_io.to_dataframe bridge.

Source code in src/cds2/data_analysis.py 
288
289
290
291
292
293
294
def to_dataframe(ds: DataSet) -> pd.DataFrame:
    """Convert a :class:`DataSet` to a :class:`pandas.DataFrame`.

    Thin wrapper around :meth:`DataSet.to_dataframe` for compatibility with
    the v1 ``cds.data_analysis.pandas_io.to_dataframe`` bridge.
    """
    return ds.to_dataframe()

from_dataframe(df)

Convert a :class:pandas.DataFrame to a :class:DataSet.

Thin wrapper around :meth:DataSet.from_dataframe for compatibility with the v1 cds.data_analysis.pandas_io.from_dataframe bridge. NaN values are exposed as None through the :class:DataSet API, matching the v1 scalar contract.

Source code in src/cds2/data_analysis.py 
297
298
299
300
301
302
303
304
305
def from_dataframe(df: pd.DataFrame) -> DataSet:
    """Convert a :class:`pandas.DataFrame` to a :class:`DataSet`.

    Thin wrapper around :meth:`DataSet.from_dataframe` for compatibility with
    the v1 ``cds.data_analysis.pandas_io.from_dataframe`` bridge. ``NaN``
    values are exposed as ``None`` through the :class:`DataSet` API, matching
    the v1 scalar contract.
    """
    return DataSet.from_dataframe(df)