From 1f2511a9059ba8e21af8b37616b6e00fda83f762 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Thu, 14 Aug 2025 12:43:31 +0200 Subject: [PATCH 01/33] Groupby notes and ideas --- src/cyclebane/graph.py | 91 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 91 insertions(+) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 7b3a930..ea0f6a2 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -344,6 +344,97 @@ def _from_orig_key(self, key: Hashable) -> Hashable: def by_position(self, index_name: IndexName) -> PositionalIndexer: return PositionalIndexer(self, index_name) + def tmp(self): + graph = self.graph + for index_name, index in reversed(self.indices.items()): + graphs = [graph for i in index] + graph = nx.compose_all(graphs) + # say we have dims (x,y) + graphs_x = [] + graphs_x[0] = [graph for y in coords_y] + graphs_x[1] = [graph for y in coords_y] + graphs_xy = zip(graphs_x, coord_x) + # with grouping + # Pandas MultiIndex ~ binned variable, outer dim x, inner dim y + # coords_y = [[a, c], [b]] # dims=(x,) + + # sample=(sample,) mat=(sample,) mat_groups=(mat,) + # (sample,mat) # groupby + # (mat,) # reduce + + # TODO Can we have nodes that depend on (material,) but do not come from + # the grouping-reduce operation? How can we set those? Pass at same time + # to groupby (forwarding to map)? + + # Now + for index_name, index in reversed(self.indices.items()): + graphs = self._clone_graph(graph, index_name, index) + graph = nx.compose_all(graphs) + + # Then + for index_name, index in reversed(self.indices.items()): + if is_multi_index(index): + # Index looks like {'Si': [s1, s3], 'Ge': [s2]} + + # One graph per material, do not compose! + graphs = self._clone_graph(graph, index_name, index.keys()) + # IndexValues(axes=(mat,), values=(Si,)) + # IndexValues(axes=(mat,), values=(Ge,)) + + inner_index_name = index.inner_index # sample + # Si + # IndexValues(axes=(sample,), values=(s1,)) + # IndexValues(axes=(sample,), values=(s3,)) + # Ge + # IndexValues(axes=(sample,), values=(s2,)) + graphs = [ + self._clone_graph(graph_for_material, inner_index_name, inner_index) + for inner_index, graph_for_material in zip(index.values(), graphs) + ] + # No! We don't need grouped node, maybe? There is no compute for it! + # Final node names should be (at the grouped but not reduced node): + # Note: May want to flatten the list + # IndexValues(axes=(mat,sample), values=(Si,s1)) + # IndexValues(axes=(mat,sample), values=(Si,s3)) + # IndexValues(axes=(mat,sample), values=(Ge,s2)) + + graph = nx.compose_all(graphs) + else: + # Don't forget nodes not taking part in the grouping, which need + # IndexValues(axes=(sample,), values=(s1,)) + # IndexValues(axes=(sample,), values=(s2,)) + # IndexValues(axes=(sample,), values=(s3,)) + # Done by second loop iteration? ... but will set up all-to-all edges?! + # Can we delay the compose_all until after the loop? + graphs = self._clone_graph(graph, index_name, index) + graph = nx.compose_all(graphs) + + # 1. loop iteration + # index_name=mat # dim name + # index=mat_groups # list of materials + # => graph composed of graph copies, one for each material + # 1b. nested loop + # mat_groups = {'Si': [s1, s3], 'Ge': [s2]} + # + + # 2. loop iteration (important since there may be nodes mapped + # over sample that are not grouped by material) + # index_name=sample # dim name + # index=sample # list of samples + # naively this would put *all* samples into the subgraph for each material + + # 1. make small arrays, each different length + # 2. combine into ragged 2D array + graphs_x[0] = [graph for y in coords_y[0]] # len=2 + graphs_x[1] = [graph for y in coords_y[1]] # len=1 + graphs_xy = zip(graphs_x, coord_x) + + # 1. make 1D array + # 2. replace each value by array of different length + graphs_y = [graph for x in coords_x] + zip(graphs_y[0], coord_y[0]) # len=2 + zip(graphs_y[1], coord_y[1]) # len=1 + def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: """ Convert to a NetworkX graph, spelling out the internal array structures as From a3ffc8c8c6a413bd1cee9f9fd6ff7080d16c55b3 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Thu, 14 Aug 2025 14:15:25 +0200 Subject: [PATCH 02/33] More details --- src/cyclebane/graph.py | 9 +++++++++ src/cyclebane/node_values.py | 6 ++++++ 2 files changed, 15 insertions(+) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index ea0f6a2..667d85c 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -391,6 +391,15 @@ def tmp(self): self._clone_graph(graph_for_material, inner_index_name, inner_index) for inner_index, graph_for_material in zip(index.values(), graphs) ] + # graphs[Si]: [(e[Si],h[Si]), (e[Si],h[Si])] + # graphs[Ge]: [(e[Ge],h[Ge])] + + # graphs[Si]: [(d[s1],f[Si]), (d[s3],f[Si])] + # graphs[Ge]: [(d[s2],f[Ge])] + + # graphs[Si]: [(g[s1],j), (g[s3],j)] + # graphs[Ge]: [(g[s2],j)] + # No! We don't need grouped node, maybe? There is no compute for it! # Final node names should be (at the grouped but not reduced node): # Note: May want to flatten the list diff --git a/src/cyclebane/node_values.py b/src/cyclebane/node_values.py index 0830fdb..bdea979 100644 --- a/src/cyclebane/node_values.py +++ b/src/cyclebane/node_values.py @@ -427,6 +427,12 @@ def from_mapping( values: Mapping[Hashable, Sequence[Any]], axis_zero: int ) -> NodeValues: """Construct from a mapping of node names to value sequences.""" + # graph.map(param_table) + # {sample: [s1,s2,s3], material: [Si,Ge,Si], param: [p1,p2,p3]} + # graph.groupby(material) + # -> merge two indices into multi-index + # {material: {Si: {sample:[s1,s3]}, Ge: {sample:[s2]}} + # {material: {Si: {sample:[s1,s3], param:[p1,p3]}, Ge: {sample:[s2], param:[p2]}} value_arrays = { key: ValueArray.from_array_like(value, axis_zero=axis_zero) for key, value in values.items() From a226c14d2d510d1e0fd983de63238999c3fc4449 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Thu, 14 Aug 2025 17:15:22 +0200 Subject: [PATCH 03/33] More thoughts and problems --- src/cyclebane/graph.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 667d85c..227ce32 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -400,6 +400,8 @@ def tmp(self): # graphs[Si]: [(g[s1],j), (g[s3],j)] # graphs[Ge]: [(g[s2],j)] + # PROBLEM: Does not work for nodes that have two indices!? + # No! We don't need grouped node, maybe? There is no compute for it! # Final node names should be (at the grouped but not reduced node): # Note: May want to flatten the list @@ -455,8 +457,14 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: The name of the attribute on nodes that holds the array-like object. """ graph = self.graph + # What if we flatten all (dependent) indices into tuples? + # x(x) = [1,2,3] + # y(x) = [a,b,a] + # y(y) = [a,b] for index_name, index in reversed(self.indices.items()): # Find all nodes with this index + # nodes_by_index_name: dict[IndexName, list[NodeName]] = {} + # Some show up twice or more, if multiple map ops? nodes = [ node for node in graph.nodes() @@ -464,6 +472,7 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: in _node_indices(node.name if isinstance(node, NodeName) else node) ] # Make a copy for each index value + # single loop over flat index (list of tuples), but one rename per level graphs = [ _rename_successors( graph, successors=nodes, index=IndexValues((index_name,), (i,)) From 6501bc3f5e8176fdd4e80f25f8a037c9f8bc229b Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 18 Aug 2025 06:59:05 +0200 Subject: [PATCH 04/33] Split large file --- src/cyclebane/node_values.py | 362 +------------------------- src/cyclebane/value_array.py | 85 ++++++ src/cyclebane/value_array_adapters.py | 298 +++++++++++++++++++++ 3 files changed, 386 insertions(+), 359 deletions(-) create mode 100644 src/cyclebane/value_array.py create mode 100644 src/cyclebane/value_array_adapters.py diff --git a/src/cyclebane/node_values.py b/src/cyclebane/node_values.py index bdea979..525b3b3 100644 --- a/src/cyclebane/node_values.py +++ b/src/cyclebane/node_values.py @@ -2,16 +2,11 @@ # Copyright (c) 2024 Scipp contributors (https://github.com/scipp) from __future__ import annotations -from abc import ABC, abstractmethod from collections.abc import Hashable, Iterable, Iterator, Mapping, Sequence -from types import ModuleType -from typing import TYPE_CHECKING, Any, ClassVar, TypeVar +from typing import Any, TypeVar -if TYPE_CHECKING: - import numpy - import pandas - import scipp - import xarray +from . import value_array_adapters # noqa: F401 +from .value_array import ValueArray IndexName = Hashable IndexValue = Hashable @@ -19,357 +14,6 @@ T = TypeVar('T', bound='ValueArray') -class ValueArray(ABC): - """ - Abstract base class for a series of values with an index that can be sliced. - - Used by :py:class:`NodeValues` to store the values of a given node in a graph. The - abstraction allows for the use of different data structures to store the values of - nodes in a graph, such as pandas.DataFrame, xarray.DataArray, numpy.ndarray, or - simple Python iterables. - """ - - _registry: ClassVar = [] - - def __init_subclass__(cls) -> None: - super().__init_subclass__() - ValueArray._registry.append(cls) - - @staticmethod - def from_array_like(values: Any, *, axis_zero: int = 0) -> ValueArray: - # Reversed to ensure SequenceAdapter is tried last, as it is the most general - # SequenceAdapter is defined right after this class so it is registered first - for subclass in reversed(ValueArray._registry): - if (a := subclass.try_from(values, axis_zero=axis_zero)) is not None: - return a - raise ValueError(f'Cannot create ValueArray from {values}') - - @staticmethod - @abstractmethod - def try_from(obj: Any, *, axis_zero: int = 0) -> ValueArray | None: ... - - def __eq__(self, other: object) -> bool: - if type(self) is not type(other): - return NotImplemented - return self._equal(other) - - def __ne__(self, other: object) -> bool: - return not self == other - - @abstractmethod - def _equal(self: T, other: T) -> bool: ... - - @abstractmethod - def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: - """Return data by selecting from index with given name and index value.""" - - def loc(self, key: dict[IndexName, slice]) -> ValueArray: - if not all(isinstance(i, slice) for i in key.values()): - raise ValueError('ValueArray.loc only accepts slices, not integers') - if not set(key).issubset(set(self.index_names)): - raise ValueError( - f'ValueArray.loc got {key.keys()}, not a subset of {self.index_names}' - ) - return self[key] - - @abstractmethod - def __getitem__(self, key: dict[IndexName, slice]) -> ValueArray: - pass - - @property - @abstractmethod - def shape(self) -> tuple[int, ...]: - pass - - @property - @abstractmethod - def index_names(self) -> tuple[IndexName, ...]: - pass - - @property - @abstractmethod - def indices(self) -> dict[IndexName, Iterable[IndexValue]]: - pass - - -class SequenceAdapter(ValueArray): - def __init__( - self, - values: Sequence[Any], - *, - index: Iterable[IndexValue] | None = None, - axis_zero: int = 0, - ): - self._values = values - self._index = index or range(len(values)) - self._axis_zero = axis_zero - - @staticmethod - def try_from(obj: Any, *, axis_zero: int = 0) -> SequenceAdapter | None: - return SequenceAdapter(obj, axis_zero=axis_zero) - - def _equal(self, other: SequenceAdapter) -> bool: - return ( - self._values == other._values - and self._index == other._index - and self._axis_zero == other._axis_zero - ) - - def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: - if len(key) != 1: - raise ValueError('SequenceAdapter only supports single index') - _, i = key[0] - return self._values[self._index.index(i)] - - def __getitem__(self, key: dict[IndexName, slice]) -> SequenceAdapter: - _, i = next(iter(key.items())) - return SequenceAdapter( - self._values[i], index=self._index[i], axis_zero=self._axis_zero - ) - - @property - def shape(self) -> tuple[int, ...]: - return (len(self._values),) - - @property - def index_names(self) -> tuple[IndexName, ...]: - return (f'dim_{self._axis_zero}',) - - @property - def indices(self) -> dict[IndexName, Iterable[IndexValue]]: - return {f'dim_{self._axis_zero}': self._index} - - -class PandasSeriesAdapter(ValueArray): - def __init__(self, series: pandas.Series, *, axis_zero: int = 0): - self._series = series - self._axis_zero = axis_zero - - @staticmethod - def try_from(obj: Any, *, axis_zero: int = 0) -> PandasSeriesAdapter | None: - try: - import pandas - except ModuleNotFoundError: - return None - if isinstance(obj, pandas.Series): - return PandasSeriesAdapter(obj, axis_zero=axis_zero) - - def _equal(self, other: PandasSeriesAdapter) -> bool: - return ( - self._series.equals(other._series) and self._axis_zero == other._axis_zero - ) - - def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: - if len(key) != 1: - raise ValueError('PandasSeriesAdapter only supports single index') - index_name, i = key[0] - if index_name != self.index_names[0]: - raise ValueError( - f'Unexpected index name {index_name} for PandasSeriesAdapter with ' - f'index names {self.index_names}' - ) - return self._series.loc[i] - - def __getitem__(self, key: dict[IndexName, slice]) -> PandasSeriesAdapter: - _, i = next(iter(key.items())) - return PandasSeriesAdapter(self._series[i], axis_zero=self._axis_zero) - - @property - def shape(self) -> tuple[int, ...]: - return (len(self._series),) - - @property - def index_names(self) -> tuple[IndexName, ...]: - index_name = ( - self._series.index.name - if self._series.index.name is not None - else f'dim_{self._axis_zero}' - ) - return (index_name,) - - @property - def indices(self) -> dict[IndexName, Iterable[IndexValue]]: - return {self.index_names[0]: self._series.index} - - -class XarrayDataArrayAdapter(ValueArray): - def __init__( - self, - data_array: xarray.DataArray, - ): - default_indices = { - dim: range(size) - for dim, size in data_array.sizes.items() - if dim not in data_array.coords - } - self._data_array = data_array.assign_coords(default_indices) - - @staticmethod - def try_from(obj: Any, *, axis_zero: int = 0) -> XarrayDataArrayAdapter | None: - try: - import xarray - - if isinstance(obj, xarray.DataArray): - return XarrayDataArrayAdapter(obj) - except ModuleNotFoundError: - pass - - def _equal(self, other: XarrayDataArrayAdapter) -> bool: - return self._data_array.identical(other._data_array) - - def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: - return self._data_array.sel(dict(key)) - - def __getitem__(self, key: dict[IndexName, slice]) -> XarrayDataArrayAdapter: - return XarrayDataArrayAdapter(self._data_array.isel(key)) - - @property - def shape(self) -> tuple[int, ...]: - return self._data_array.shape - - @property - def index_names(self) -> tuple[IndexName, ...]: - return tuple(self._data_array.dims) - - @property - def indices(self) -> dict[IndexName, Iterable[IndexValue]]: - return { - dim: self._data_array.coords[dim].values for dim in self._data_array.dims - } - - -class ScippDataArrayAdapter(ValueArray): - def __init__(self, data_array: scipp.DataArray, scipp: ModuleType): - default_indices = { - dim: scipp.arange(dim, size, unit=None) - for dim, size in data_array.sizes.items() - if dim not in data_array.coords - } - self._data_array = data_array.assign_coords(default_indices) - self._scipp = scipp - - @staticmethod - def try_from(obj: Any, *, axis_zero: int = 0) -> ScippDataArrayAdapter | None: - try: - import scipp - - if isinstance(obj, scipp.Variable): - return ScippDataArrayAdapter(scipp.DataArray(obj), scipp=scipp) - if isinstance(obj, scipp.DataArray): - return ScippDataArrayAdapter(obj, scipp=scipp) - except ModuleNotFoundError: - pass - - def _equal(self, other: ScippDataArrayAdapter) -> bool: - return self._scipp.identical(self._data_array, other._data_array) - - def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: - values = self._data_array - for dim, value in key: - # Reconstruct label, to use label-based indexing instead of positional - if isinstance(value, tuple): - value, unit = value - else: - unit = None - label = self._scipp.scalar(value, unit=unit) - # Scipp indexing uses a comma to separate dimension label from the index, - # unlike Numpy and other libraries where it separates the indices for - # different axes. - values = values[dim, label] - return values - - def __getitem__(self, key: dict[IndexName, slice]) -> ScippDataArrayAdapter: - values = self._data_array - for dim, i in key: - values = values[dim, i] - return ScippDataArrayAdapter(values, scipp=self._scipp) - - @property - def shape(self) -> tuple[int, ...]: - return self._data_array.shape - - @property - def index_names(self) -> tuple[IndexName, ...]: - return tuple(self._data_array.dims) - - def _index_for_dim(self, dim: str) -> list[tuple[Any, scipp.Unit]]: - # Work around some NetworkX errors. Probably scipp.Variable lacks functionality. - # For now we return a list of tuples, where the first element is the value and - # the second is the unit. - coord = self._data_array.coords[dim] - unit = coord.unit - if unit is None: - return coord.values - unit = str(unit) - return [(value, unit) for value in coord.values] - - @property - def indices(self) -> dict[IndexName, Iterable[IndexValue]]: - return {dim: self._index_for_dim(dim) for dim in self._data_array.dims} - - -class NumpyArrayAdapter(ValueArray): - def __init__( - self, - array: numpy.ndarray, - *, - indices: dict[IndexName, Iterable[IndexValue]] | None = None, - axis_zero: int = 0, - ): - import numpy as np - - self._array = np.asarray(array) - if indices is None: - indices = { - f'dim_{i + axis_zero}': range(size) - for i, size in enumerate(self._array.shape) - } - self._indices = indices - self._axis_zero = axis_zero - - @staticmethod - def try_from(obj: Any, *, axis_zero: int = 0) -> NumpyArrayAdapter | None: - try: - import numpy - except ModuleNotFoundError: - return None - if isinstance(obj, numpy.ndarray): - return NumpyArrayAdapter(obj, axis_zero=axis_zero) - - def _equal(self, other: NumpyArrayAdapter) -> bool: - return ( - (self._array == other._array).all() - and self._indices == other._indices - and self._axis_zero == other._axis_zero - ) - - def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: - index_tuple = tuple(self._indices[k].index(i) for k, i in key) - return self._array[index_tuple] - - def __getitem__(self, key: dict[IndexName, slice]) -> NumpyArrayAdapter: - return NumpyArrayAdapter( - self._array[tuple(key.get(k, slice(None)) for k in self._indices)], - indices={ - index_name: (index_values[key.get(index_name, slice(None))]) - for index_name, index_values in self._indices.items() - }, - axis_zero=self._axis_zero, - ) - - @property - def shape(self) -> tuple[int, ...]: - return self._array.shape - - @property - def index_names(self) -> tuple[IndexName, ...]: - return tuple(self._indices) - - @property - def indices(self) -> dict[IndexName, Iterable[IndexValue]]: - return self._indices - - class NodeValues(Mapping[Hashable, ValueArray]): """ A collection of pandas.DataFrame-like objects with distinct indices. diff --git a/src/cyclebane/value_array.py b/src/cyclebane/value_array.py new file mode 100644 index 0000000..02bb78c --- /dev/null +++ b/src/cyclebane/value_array.py @@ -0,0 +1,85 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2025 Scipp contributors (https://github.com/scipp) +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections.abc import Hashable, Iterable +from typing import Any, ClassVar, TypeVar + +IndexName = Hashable +IndexValue = Hashable + +T = TypeVar('T', bound='ValueArray') + + +class ValueArray(ABC): + """ + Abstract base class for a series of values with an index that can be sliced. + + Used by :py:class:`NodeValues` to store the values of a given node in a graph. The + abstraction allows for the use of different data structures to store the values of + nodes in a graph, such as pandas.DataFrame, xarray.DataArray, numpy.ndarray, or + simple Python iterables. + """ + + _registry: ClassVar = [] + + def __init_subclass__(cls) -> None: + super().__init_subclass__() + ValueArray._registry.append(cls) + + @staticmethod + def from_array_like(values: Any, *, axis_zero: int = 0) -> ValueArray: + # Reversed to ensure SequenceAdapter is tried last, as it is the most general + # SequenceAdapter is defined right after this class so it is registered first + for subclass in reversed(ValueArray._registry): + if (a := subclass.try_from(values, axis_zero=axis_zero)) is not None: + return a + raise ValueError(f'Cannot create ValueArray from {values}') + + @staticmethod + @abstractmethod + def try_from(obj: Any, *, axis_zero: int = 0) -> ValueArray | None: ... + + def __eq__(self, other: object) -> bool: + if type(self) is not type(other): + return NotImplemented + return self._equal(other) + + def __ne__(self, other: object) -> bool: + return not self == other + + @abstractmethod + def _equal(self: T, other: T) -> bool: ... + + @abstractmethod + def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: + """Return data by selecting from index with given name and index value.""" + + def loc(self, key: dict[IndexName, slice]) -> ValueArray: + if not all(isinstance(i, slice) for i in key.values()): + raise ValueError('ValueArray.loc only accepts slices, not integers') + if not set(key).issubset(set(self.index_names)): + raise ValueError( + f'ValueArray.loc got {key.keys()}, not a subset of {self.index_names}' + ) + return self[key] + + @abstractmethod + def __getitem__(self, key: dict[IndexName, slice]) -> ValueArray: + pass + + @property + @abstractmethod + def shape(self) -> tuple[int, ...]: + pass + + @property + @abstractmethod + def index_names(self) -> tuple[IndexName, ...]: + pass + + @property + @abstractmethod + def indices(self) -> dict[IndexName, Iterable[IndexValue]]: + pass diff --git a/src/cyclebane/value_array_adapters.py b/src/cyclebane/value_array_adapters.py new file mode 100644 index 0000000..0d55094 --- /dev/null +++ b/src/cyclebane/value_array_adapters.py @@ -0,0 +1,298 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2024 Scipp contributors (https://github.com/scipp) +from __future__ import annotations + +from collections.abc import Hashable, Iterable, Sequence +from types import ModuleType +from typing import TYPE_CHECKING, Any, TypeVar + +from .value_array import ValueArray + +if TYPE_CHECKING: + import numpy + import pandas + import scipp + import xarray + +IndexName = Hashable +IndexValue = Hashable + +T = TypeVar('T', bound='ValueArray') + + +class SequenceAdapter(ValueArray): + def __init__( + self, + values: Sequence[Any], + *, + index: Iterable[IndexValue] | None = None, + axis_zero: int = 0, + ): + self._values = values + self._index = index or range(len(values)) + self._axis_zero = axis_zero + + @staticmethod + def try_from(obj: Any, *, axis_zero: int = 0) -> SequenceAdapter | None: + return SequenceAdapter(obj, axis_zero=axis_zero) + + def _equal(self, other: SequenceAdapter) -> bool: + return ( + self._values == other._values + and self._index == other._index + and self._axis_zero == other._axis_zero + ) + + def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: + if len(key) != 1: + raise ValueError('SequenceAdapter only supports single index') + _, i = key[0] + return self._values[self._index.index(i)] + + def __getitem__(self, key: dict[IndexName, slice]) -> SequenceAdapter: + _, i = next(iter(key.items())) + return SequenceAdapter( + self._values[i], index=self._index[i], axis_zero=self._axis_zero + ) + + @property + def shape(self) -> tuple[int, ...]: + return (len(self._values),) + + @property + def index_names(self) -> tuple[IndexName, ...]: + return (f'dim_{self._axis_zero}',) + + @property + def indices(self) -> dict[IndexName, Iterable[IndexValue]]: + return {f'dim_{self._axis_zero}': self._index} + + +class PandasSeriesAdapter(ValueArray): + def __init__(self, series: pandas.Series, *, axis_zero: int = 0): + self._series = series + self._axis_zero = axis_zero + + @staticmethod + def try_from(obj: Any, *, axis_zero: int = 0) -> PandasSeriesAdapter | None: + try: + import pandas + except ModuleNotFoundError: + return None + if isinstance(obj, pandas.Series): + return PandasSeriesAdapter(obj, axis_zero=axis_zero) + + def _equal(self, other: PandasSeriesAdapter) -> bool: + return ( + self._series.equals(other._series) and self._axis_zero == other._axis_zero + ) + + def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: + if len(key) != 1: + raise ValueError('PandasSeriesAdapter only supports single index') + index_name, i = key[0] + if index_name != self.index_names[0]: + raise ValueError( + f'Unexpected index name {index_name} for PandasSeriesAdapter with ' + f'index names {self.index_names}' + ) + return self._series.loc[i] + + def __getitem__(self, key: dict[IndexName, slice]) -> PandasSeriesAdapter: + _, i = next(iter(key.items())) + return PandasSeriesAdapter(self._series[i], axis_zero=self._axis_zero) + + @property + def shape(self) -> tuple[int, ...]: + return (len(self._series),) + + @property + def index_names(self) -> tuple[IndexName, ...]: + index_name = ( + self._series.index.name + if self._series.index.name is not None + else f'dim_{self._axis_zero}' + ) + return (index_name,) + + @property + def indices(self) -> dict[IndexName, Iterable[IndexValue]]: + return {self.index_names[0]: self._series.index} + + +class XarrayDataArrayAdapter(ValueArray): + def __init__( + self, + data_array: xarray.DataArray, + ): + default_indices = { + dim: range(size) + for dim, size in data_array.sizes.items() + if dim not in data_array.coords + } + self._data_array = data_array.assign_coords(default_indices) + + @staticmethod + def try_from(obj: Any, *, axis_zero: int = 0) -> XarrayDataArrayAdapter | None: + try: + import xarray + + if isinstance(obj, xarray.DataArray): + return XarrayDataArrayAdapter(obj) + except ModuleNotFoundError: + pass + + def _equal(self, other: XarrayDataArrayAdapter) -> bool: + return self._data_array.identical(other._data_array) + + def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: + return self._data_array.sel(dict(key)) + + def __getitem__(self, key: dict[IndexName, slice]) -> XarrayDataArrayAdapter: + return XarrayDataArrayAdapter(self._data_array.isel(key)) + + @property + def shape(self) -> tuple[int, ...]: + return self._data_array.shape + + @property + def index_names(self) -> tuple[IndexName, ...]: + return tuple(self._data_array.dims) + + @property + def indices(self) -> dict[IndexName, Iterable[IndexValue]]: + return { + dim: self._data_array.coords[dim].values for dim in self._data_array.dims + } + + +class ScippDataArrayAdapter(ValueArray): + def __init__(self, data_array: scipp.DataArray, scipp: ModuleType): + default_indices = { + dim: scipp.arange(dim, size, unit=None) + for dim, size in data_array.sizes.items() + if dim not in data_array.coords + } + self._data_array = data_array.assign_coords(default_indices) + self._scipp = scipp + + @staticmethod + def try_from(obj: Any, *, axis_zero: int = 0) -> ScippDataArrayAdapter | None: + try: + import scipp + + if isinstance(obj, scipp.Variable): + return ScippDataArrayAdapter(scipp.DataArray(obj), scipp=scipp) + if isinstance(obj, scipp.DataArray): + return ScippDataArrayAdapter(obj, scipp=scipp) + except ModuleNotFoundError: + pass + + def _equal(self, other: ScippDataArrayAdapter) -> bool: + return self._scipp.identical(self._data_array, other._data_array) + + def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: + values = self._data_array + for dim, value in key: + # Reconstruct label, to use label-based indexing instead of positional + if isinstance(value, tuple): + value, unit = value + else: + unit = None + label = self._scipp.scalar(value, unit=unit) + # Scipp indexing uses a comma to separate dimension label from the index, + # unlike Numpy and other libraries where it separates the indices for + # different axes. + values = values[dim, label] + return values + + def __getitem__(self, key: dict[IndexName, slice]) -> ScippDataArrayAdapter: + values = self._data_array + for dim, i in key: + values = values[dim, i] + return ScippDataArrayAdapter(values, scipp=self._scipp) + + @property + def shape(self) -> tuple[int, ...]: + return self._data_array.shape + + @property + def index_names(self) -> tuple[IndexName, ...]: + return tuple(self._data_array.dims) + + def _index_for_dim(self, dim: str) -> list[tuple[Any, scipp.Unit]]: + # Work around some NetworkX errors. Probably scipp.Variable lacks functionality. + # For now we return a list of tuples, where the first element is the value and + # the second is the unit. + coord = self._data_array.coords[dim] + unit = coord.unit + if unit is None: + return coord.values + unit = str(unit) + return [(value, unit) for value in coord.values] + + @property + def indices(self) -> dict[IndexName, Iterable[IndexValue]]: + return {dim: self._index_for_dim(dim) for dim in self._data_array.dims} + + +class NumpyArrayAdapter(ValueArray): + def __init__( + self, + array: numpy.ndarray, + *, + indices: dict[IndexName, Iterable[IndexValue]] | None = None, + axis_zero: int = 0, + ): + import numpy as np + + self._array = np.asarray(array) + if indices is None: + indices = { + f'dim_{i + axis_zero}': range(size) + for i, size in enumerate(self._array.shape) + } + self._indices = indices + self._axis_zero = axis_zero + + @staticmethod + def try_from(obj: Any, *, axis_zero: int = 0) -> NumpyArrayAdapter | None: + try: + import numpy + except ModuleNotFoundError: + return None + if isinstance(obj, numpy.ndarray): + return NumpyArrayAdapter(obj, axis_zero=axis_zero) + + def _equal(self, other: NumpyArrayAdapter) -> bool: + return ( + (self._array == other._array).all() + and self._indices == other._indices + and self._axis_zero == other._axis_zero + ) + + def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: + index_tuple = tuple(self._indices[k].index(i) for k, i in key) + return self._array[index_tuple] + + def __getitem__(self, key: dict[IndexName, slice]) -> NumpyArrayAdapter: + return NumpyArrayAdapter( + self._array[tuple(key.get(k, slice(None)) for k in self._indices)], + indices={ + index_name: (index_values[key.get(index_name, slice(None))]) + for index_name, index_values in self._indices.items() + }, + axis_zero=self._axis_zero, + ) + + @property + def shape(self) -> tuple[int, ...]: + return self._array.shape + + @property + def index_names(self) -> tuple[IndexName, ...]: + return tuple(self._indices) + + @property + def indices(self) -> dict[IndexName, Iterable[IndexValue]]: + return self._indices From 28322c57bb52a78c3861fe58a468514fb9fa5fe1 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 18 Aug 2025 11:17:12 +0200 Subject: [PATCH 05/33] Try supporting some MultiIndex series --- src/cyclebane/value_array_adapters.py | 32 +++++++++++++++++++++++- tests/pandas_series_adapter_test.py | 36 +++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 1 deletion(-) create mode 100644 tests/pandas_series_adapter_test.py diff --git a/src/cyclebane/value_array_adapters.py b/src/cyclebane/value_array_adapters.py index 0d55094..a9d0270 100644 --- a/src/cyclebane/value_array_adapters.py +++ b/src/cyclebane/value_array_adapters.py @@ -80,6 +80,7 @@ def try_from(obj: Any, *, axis_zero: int = 0) -> PandasSeriesAdapter | None: except ModuleNotFoundError: return None if isinstance(obj, pandas.Series): + # TODO Reject MultiIndex? return PandasSeriesAdapter(obj, axis_zero=axis_zero) def _equal(self, other: PandasSeriesAdapter) -> bool: @@ -91,23 +92,44 @@ def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: if len(key) != 1: raise ValueError('PandasSeriesAdapter only supports single index') index_name, i = key[0] - if index_name != self.index_names[0]: + if index_name not in self.index_names: raise ValueError( f'Unexpected index name {index_name} for PandasSeriesAdapter with ' f'index names {self.index_names}' ) + if self._get_multi_index() is not None: + s = self._series.xs(i, level=index_name) + if len(s) == 1: + return s.iloc[0] + is_constant = s.eq(s.iloc[0]).all() + if is_constant: + return s.iloc[0] + raise ValueError( + f'Cannot get single value from Pandas Series with index {index_name}' + f' and value {i}, as it is not constant.' + ) return self._series.loc[i] def __getitem__(self, key: dict[IndexName, slice]) -> PandasSeriesAdapter: _, i = next(iter(key.items())) return PandasSeriesAdapter(self._series[i], axis_zero=self._axis_zero) + def _get_multi_index(self) -> pandas.MultiIndex | None: + import pandas + + if isinstance(self._series.index, pandas.MultiIndex): + return self._series.index + @property def shape(self) -> tuple[int, ...]: + if (multi_index := self._get_multi_index()) is not None: + return tuple(len(level) for level in multi_index.levels) return (len(self._series),) @property def index_names(self) -> tuple[IndexName, ...]: + if (multi_index := self._get_multi_index()) is not None: + return tuple(multi_index.names) index_name = ( self._series.index.name if self._series.index.name is not None @@ -117,8 +139,16 @@ def index_names(self) -> tuple[IndexName, ...]: @property def indices(self) -> dict[IndexName, Iterable[IndexValue]]: + if (multi_index := self._get_multi_index()) is not None: + return dict(zip(multi_index.names, multi_index.levels, strict=True)) return {self.index_names[0]: self._series.index} + def group(self) -> PandasSeriesAdapter: + return PandasSeriesAdapter( + self._series.groupby(self._series).apply(lambda x: x, include_groups=False), + axis_zero=self._axis_zero, + ) + class XarrayDataArrayAdapter(ValueArray): def __init__( diff --git a/tests/pandas_series_adapter_test.py b/tests/pandas_series_adapter_test.py new file mode 100644 index 0000000..50beef4 --- /dev/null +++ b/tests/pandas_series_adapter_test.py @@ -0,0 +1,36 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2025 Scipp contributors (https://github.com/scipp) +import pandas as pd +import pytest + +from cyclebane.value_array_adapters import PandasSeriesAdapter + + +@pytest.fixture +def series() -> pd.Series: + df = pd.DataFrame( + { + 'material': ['A', 'A', 'B', 'A', 'C', 'C'], + 'sample': ['a', 'b', 'c', 'd', 'e', 'f'], + } + ).set_index('sample') + return df['material'] + + +class TestPandasSeriesAdapter: + def test_adapter_from_grouping_row(self, series): + adapter = PandasSeriesAdapter(series) + assert adapter.shape == (6,) + assert adapter.index_names == ('sample',) + assert adapter.sel((('sample', 'c'),)) == 'B' + + def test_group_returns_multi_index_like_series(self, series): + base_adapter = PandasSeriesAdapter(series) + adapter = base_adapter.group() + assert adapter.index_names == ('material', 'sample') + assert adapter.shape == (3, 6) + indices = adapter.indices + assert list(indices['material']) == ['A', 'B', 'C'] + assert list(indices['sample']) == ['a', 'b', 'c', 'd', 'e', 'f'] + assert adapter.sel((('sample', 'c'),)) == 'B' + assert adapter.sel((('material', 'A'),)) == 'A' From b8a81ee42c40ef1d1f333eff47b382a316ddeeda Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 18 Aug 2025 12:13:57 +0200 Subject: [PATCH 06/33] Begin adding groupby --- src/cyclebane/graph.py | 60 +++++++++++++++++++++++++++ src/cyclebane/node_values.py | 11 +++++ src/cyclebane/value_array.py | 11 +++++ src/cyclebane/value_array_adapters.py | 6 ++- tests/groupby_test.py | 49 ++++++++++++++++++++++ 5 files changed, 136 insertions(+), 1 deletion(-) create mode 100644 tests/groupby_test.py diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 227ce32..acd2759 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -265,6 +265,9 @@ def map(self, node_values: MappingToArrayLike) -> Graph: node_values=self._node_values.merge(new_values), ) + def groupby(self, node: Hashable) -> GroupbyGraph: + return GroupbyGraph(self.graph, node_values=self._node_values, node=node) + def reduce( self, key: None | Hashable = None, @@ -273,6 +276,7 @@ def reduce( axis: None | int = None, name: None | Hashable = None, attrs: None | dict[str, Any] = None, + _extra_index_name: None | IndexName = None, ) -> Graph: """ Reduce over the given index or axis previously created with :py:meth:`map`. @@ -314,6 +318,11 @@ def reduce( new_index = tuple(value for i, value in enumerate(indices) if i != axis) else: new_index = None + if _extra_index_name is not None: + if new_index is None: + new_index = (_extra_index_name,) + else: + new_index = (*new_index, _extra_index_name) if name in self.graph: raise ValueError(f"Node '{name}' already exists in the graph.") @@ -584,3 +593,54 @@ def __setitem__(self, branch: Hashable | slice, other: Graph) -> None: # Delay setting graph until we know no step fails self._node_values = self._node_values.merge(other._node_values) self.graph = graph + + +class GroupbyGraph: + """ + A graph that has been grouped by a specific index. + + This is a specialized graph that is used to represent the result of a groupby + operation on a Cyclebane graph. It allows for operations on the grouped data, + such as aggregation or summarization. + """ + + def __init__(self, graph: nx.DiGraph, node_values: NodeValues, node: Hashable): + graph = graph.copy() + node_values = node_values.copy() + groups = node_values[node].group() + self._group_index_name = node + self._index_name = groups.index_names[1] # Group-internal index to reduce over + del node_values[node] # Explicit since __setitem__ does not support replacing + node_values[node] = groups + # The grouping node becomes the new index name + # TODO Not strictly needed? + # graph.add_node(_node_with_indices(node, (node,))) + self._graph = Graph(graph, node_values=node_values) + + def reduce( + self, + key: None | Hashable = None, + *, + name: None | Hashable = None, + attrs: None | dict[str, Any] = None, + ) -> Graph: + """ + Reduce the grouped graph over the given index or axis. + + Parameters + ---------- + key: + The name of the source node to reduce. This is the original name prior to + mapping. If not given, tries to find a unique sink node. + name: + The name of the new node. If not given, a unique name is generated. + attrs: + Attributes to set on the new node(s). + """ + return self._graph.reduce( + key=key, + index=self._index_name, + name=name, + attrs=attrs, + _extra_index_name=self._group_index_name, + ) diff --git a/src/cyclebane/node_values.py b/src/cyclebane/node_values.py index 525b3b3..5221243 100644 --- a/src/cyclebane/node_values.py +++ b/src/cyclebane/node_values.py @@ -28,6 +28,10 @@ def __init__(self, values: Mapping[Any, ValueArray]): merged = self.merge(values) self._values = merged._values + def copy(self) -> NodeValues: + """Return a copy of the NodeValues.""" + return NodeValues(dict(self._values)) + def __len__(self) -> int: """Return the number of columns.""" return len(self._values) @@ -40,6 +44,13 @@ def __getitem__(self, key: Hashable) -> ValueArray: """Return the column with the given name.""" return self._values[key] + def __delitem__(self, key: Hashable) -> None: + """Remove the column with the given name.""" + if key in self._values: + del self._values[key] + else: + raise KeyError(f'Node "{key}" does not exist in NodeValues.') + def __setitem__(self, key: Hashable, value_array: ValueArray) -> None: """Add a single value array, checking for conflicts.""" # Check if the value array is identical to existing one diff --git a/src/cyclebane/value_array.py b/src/cyclebane/value_array.py index 02bb78c..159964b 100644 --- a/src/cyclebane/value_array.py +++ b/src/cyclebane/value_array.py @@ -83,3 +83,14 @@ def index_names(self) -> tuple[IndexName, ...]: @abstractmethod def indices(self) -> dict[IndexName, Iterable[IndexValue]]: pass + + def group(self) -> ValueArray: + """ + Group the values by their indices. + + This method is expected to return a new ValueArray that groups the values by + their indices, allowing for operations like aggregation or summarization. + """ + raise NotImplementedError( + 'ValueArray.group() is only implemented for Pandas series.' + ) diff --git a/src/cyclebane/value_array_adapters.py b/src/cyclebane/value_array_adapters.py index a9d0270..b68cfaa 100644 --- a/src/cyclebane/value_array_adapters.py +++ b/src/cyclebane/value_array_adapters.py @@ -144,8 +144,12 @@ def indices(self) -> dict[IndexName, Iterable[IndexValue]]: return {self.index_names[0]: self._series.index} def group(self) -> PandasSeriesAdapter: + if self._series.index.name is None: + series = self._series.rename_axis(f'dim_{self._axis_zero}') + else: + series = self._series return PandasSeriesAdapter( - self._series.groupby(self._series).apply(lambda x: x, include_groups=False), + series.groupby(series).apply(lambda x: x, include_groups=False), axis_zero=self._axis_zero, ) diff --git a/tests/groupby_test.py b/tests/groupby_test.py new file mode 100644 index 0000000..f69ce23 --- /dev/null +++ b/tests/groupby_test.py @@ -0,0 +1,49 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2025 Scipp contributors (https://github.com/scipp) + +import networkx as nx +import pandas as pd + +import cyclebane as cb + + +def idx( + name: str, *index: int, offset=None, dims: tuple[str, ...] = ('dim_0', 'dim_1') +) -> cb.graph.NodeName: + """Helper to create a NodeName with a tuple of indices.""" + return cb.graph.NodeName( + name, + cb.graph.IndexValues(dims[offset : len(index) + (offset or 0)], tuple(index)), + ) + + +def test_tmp() -> None: + g = nx.DiGraph() + g.add_edge('a', 'c') + g.add_edge('b', 'c') + df = pd.DataFrame({'a': [11, 22, 33], 'b': ['a', 'a', 'b']}) + + graph = cb.Graph(g) + mapped = graph.map(df) + grouped = mapped.groupby('b').reduce('c', name='d') + print(grouped.graph.nodes) + print(grouped.indices) + result = grouped.to_networkx() + for node in result.nodes: + print(node, result.nodes[node]) + + +def test_graphs_with_different_mapping_over_same_node_can_be_combined() -> None: + g = nx.DiGraph() + g.add_edge('a', 'b') + + graph = cb.Graph(g) + mapped = graph.map({'a': [1, 2, 3]}) + result = mapped.to_networkx() + + assert result.nodes[idx('a', 0)] == {'value': 1} + assert result.nodes[idx('a', 1)] == {'value': 2} + assert result.nodes[idx('a', 2)] == {'value': 3} + assert result.nodes[idx('b', 0)] == {} + assert result.nodes[idx('b', 1)] == {} + assert result.nodes[idx('b', 2)] == {} From dd745af0847bcc2bea43b0aaa8b5cf1ea956b84c Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 18 Aug 2025 12:23:23 +0200 Subject: [PATCH 07/33] Confirm that current to_networkx does not group correctly --- src/cyclebane/graph.py | 7 ------- tests/groupby_test.py | 23 ++++++++++++++++++----- 2 files changed, 18 insertions(+), 12 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index acd2759..adbac0f 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -466,14 +466,8 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: The name of the attribute on nodes that holds the array-like object. """ graph = self.graph - # What if we flatten all (dependent) indices into tuples? - # x(x) = [1,2,3] - # y(x) = [a,b,a] - # y(y) = [a,b] for index_name, index in reversed(self.indices.items()): # Find all nodes with this index - # nodes_by_index_name: dict[IndexName, list[NodeName]] = {} - # Some show up twice or more, if multiple map ops? nodes = [ node for node in graph.nodes() @@ -481,7 +475,6 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: in _node_indices(node.name if isinstance(node, NodeName) else node) ] # Make a copy for each index value - # single loop over flat index (list of tuples), but one rename per level graphs = [ _rename_successors( graph, successors=nodes, index=IndexValues((index_name,), (i,)) diff --git a/tests/groupby_test.py b/tests/groupby_test.py index f69ce23..2f4921b 100644 --- a/tests/groupby_test.py +++ b/tests/groupby_test.py @@ -1,5 +1,6 @@ # SPDX-License-Identifier: BSD-3-Clause # Copyright (c) 2025 Scipp contributors (https://github.com/scipp) +from collections.abc import Hashable import networkx as nx import pandas as pd @@ -8,7 +9,7 @@ def idx( - name: str, *index: int, offset=None, dims: tuple[str, ...] = ('dim_0', 'dim_1') + name: str, *index: Hashable, offset=None, dims: tuple[str, ...] = ('dim_0', 'dim_1') ) -> cb.graph.NodeName: """Helper to create a NodeName with a tuple of indices.""" return cb.graph.NodeName( @@ -26,11 +27,23 @@ def test_tmp() -> None: graph = cb.Graph(g) mapped = graph.map(df) grouped = mapped.groupby('b').reduce('c', name='d') - print(grouped.graph.nodes) - print(grouped.indices) result = grouped.to_networkx() - for node in result.nodes: - print(node, result.nodes[node]) + + # Nodes before grouping + assert result.nodes[idx('a', 0)] == {'value': 11} + assert result.nodes[idx('b', 0)] == {'value': 'a'} + assert result.nodes[idx('c', 0)] == {} + # Nodes after grouping + assert result.nodes[idx('d', 'a', dims=('b',))] == {} + + # Edges to grouped node + assert result.has_edge(idx('c', 0), idx('d', 'a', dims=('b',))) + assert result.has_edge(idx('c', 1), idx('d', 'a', dims=('b',))) + assert result.has_edge(idx('c', 2), idx('d', 'b', dims=('b',))) + # No cross-group edges + assert not result.has_edge(idx('c', 0), idx('d', 'b', dims=('b',))) + assert not result.has_edge(idx('c', 1), idx('d', 'b', dims=('b',))) + assert not result.has_edge(idx('c', 2), idx('d', 'a', dims=('b',))) def test_graphs_with_different_mapping_over_same_node_can_be_combined() -> None: From 25cf5b225bccc578d18332775f417edf5e50de9b Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 18 Aug 2025 12:25:14 +0200 Subject: [PATCH 08/33] Extract helper method --- src/cyclebane/graph.py | 34 ++++++++++++++++++++-------------- 1 file changed, 20 insertions(+), 14 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index adbac0f..18f4857 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -467,20 +467,7 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: """ graph = self.graph for index_name, index in reversed(self.indices.items()): - # Find all nodes with this index - nodes = [ - node - for node in graph.nodes() - if index_name - in _node_indices(node.name if isinstance(node, NodeName) else node) - ] - # Make a copy for each index value - graphs = [ - _rename_successors( - graph, successors=nodes, index=IndexValues((index_name,), (i,)) - ) - for i in index - ] + graphs = _clone_graph(graph, index_name, index) graph = nx.compose_all(graphs) # Replace all MappingNodes with their name new_names = { @@ -637,3 +624,22 @@ def reduce( attrs=attrs, _extra_index_name=self._group_index_name, ) + + +def _clone_graph( + graph: nx.DiGraph, index_name: IndexName, index: Iterable[IndexValue] +) -> list[nx.DiGraph]: + # Find all nodes with this index + nodes = [ + node + for node in graph.nodes() + if index_name + in _node_indices(node.name if isinstance(node, NodeName) else node) + ] + # Make a copy for each index value + return [ + _rename_successors( + graph, successors=nodes, index=IndexValues((index_name,), (i,)) + ) + for i in index + ] From d4f23c9f238e102d6b1517caca08a02ba3ce4b80 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 18 Aug 2025 12:50:43 +0200 Subject: [PATCH 09/33] First version passing tests --- src/cyclebane/graph.py | 19 ++++++++++++++++--- src/cyclebane/value_array_adapters.py | 23 ++++++++++++++++------- 2 files changed, 32 insertions(+), 10 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 18f4857..dcf549c 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -466,8 +466,19 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: The name of the attribute on nodes that holds the array-like object. """ graph = self.graph + grouped_index = 'dim_0' for index_name, index in reversed(self.indices.items()): + if index_name == grouped_index: + continue graphs = _clone_graph(graph, index_name, index) + if isinstance(index, dict): + inner_index_name = grouped_index + graphs = [ + _clone_graph(graph_for_material, inner_index_name, inner_index) + for inner_index, graph_for_material in zip(index.values(), graphs) + ] + # Flatten nested list of graphs + graphs = [g for sublist in graphs for g in sublist] graph = nx.compose_all(graphs) # Replace all MappingNodes with their name new_names = { @@ -587,9 +598,10 @@ class GroupbyGraph: def __init__(self, graph: nx.DiGraph, node_values: NodeValues, node: Hashable): graph = graph.copy() node_values = node_values.copy() - groups = node_values[node].group() + grouping = node_values[node] self._group_index_name = node - self._index_name = groups.index_names[1] # Group-internal index to reduce over + self._index_name = grouping.index_names[0] + groups = grouping.group() del node_values[node] # Explicit since __setitem__ does not support replacing node_values[node] = groups # The grouping node becomes the new index name @@ -617,13 +629,14 @@ def reduce( attrs: Attributes to set on the new node(s). """ - return self._graph.reduce( + graph = self._graph.reduce( key=key, index=self._index_name, name=name, attrs=attrs, _extra_index_name=self._group_index_name, ) + return graph def _clone_graph( diff --git a/src/cyclebane/value_array_adapters.py b/src/cyclebane/value_array_adapters.py index b68cfaa..2fb3d19 100644 --- a/src/cyclebane/value_array_adapters.py +++ b/src/cyclebane/value_array_adapters.py @@ -69,9 +69,18 @@ def indices(self) -> dict[IndexName, Iterable[IndexValue]]: class PandasSeriesAdapter(ValueArray): - def __init__(self, series: pandas.Series, *, axis_zero: int = 0): + def __init__( + self, + series: pandas.Series, + *, + axis_zero: int = 0, + _groups: dict[Hashable, pandas.Index[Any]] | None = None, + ): self._series = series + if self._get_multi_index() is None and self._series.index.name is None: + self._series = self._series.rename_axis(f'dim_{axis_zero}') self._axis_zero = axis_zero + self._groups = _groups @staticmethod def try_from(obj: Any, *, axis_zero: int = 0) -> PandasSeriesAdapter | None: @@ -140,17 +149,17 @@ def index_names(self) -> tuple[IndexName, ...]: @property def indices(self) -> dict[IndexName, Iterable[IndexValue]]: if (multi_index := self._get_multi_index()) is not None: - return dict(zip(multi_index.names, multi_index.levels, strict=True)) + base = dict(zip(multi_index.names, multi_index.levels, strict=True)) + base[multi_index.names[0]] = self._groups + return base return {self.index_names[0]: self._series.index} def group(self) -> PandasSeriesAdapter: - if self._series.index.name is None: - series = self._series.rename_axis(f'dim_{self._axis_zero}') - else: - series = self._series + groupby = self._series.groupby(self._series) return PandasSeriesAdapter( - series.groupby(series).apply(lambda x: x, include_groups=False), + groupby.apply(lambda x: x, include_groups=False), axis_zero=self._axis_zero, + _groups=groupby.groups, ) From 1426e966e6896d0094f250907aa3cc45721d6a75 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 18 Aug 2025 13:45:52 +0200 Subject: [PATCH 10/33] Fewer hacks? --- src/cyclebane/graph.py | 23 +++++++++++++---------- src/cyclebane/value_array_adapters.py | 3 +++ 2 files changed, 16 insertions(+), 10 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index dcf549c..d157ddf 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -466,20 +466,23 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: The name of the attribute on nodes that holds the array-like object. """ graph = self.graph - grouped_index = 'dim_0' + subindex: Iterable[Iterable[IndexValue]] | None = None for index_name, index in reversed(self.indices.items()): - if index_name == grouped_index: - continue - graphs = _clone_graph(graph, index_name, index) - if isinstance(index, dict): - inner_index_name = grouped_index + if subindex is None: + graphs = _clone_graph(graph, index_name, index) + else: graphs = [ - _clone_graph(graph_for_material, inner_index_name, inner_index) - for inner_index, graph_for_material in zip(index.values(), graphs) + _clone_graph(graph_for_group, index_name, inner_index) + for inner_index, graph_for_group in zip(subindex, graphs) ] # Flatten nested list of graphs graphs = [g for sublist in graphs for g in sublist] - graph = nx.compose_all(graphs) + if isinstance(index, dict): + subindex = index.values() + # No compose, delayed until we made clones within each group + else: + graph = nx.compose_all(graphs) + # Replace all MappingNodes with their name new_names = { node: NodeName(node.name.name, node.index) @@ -605,7 +608,7 @@ def __init__(self, graph: nx.DiGraph, node_values: NodeValues, node: Hashable): del node_values[node] # Explicit since __setitem__ does not support replacing node_values[node] = groups # The grouping node becomes the new index name - # TODO Not strictly needed? + # TODO Add node after grouping? But would need to set value? # graph.add_node(_node_with_indices(node, (node,))) self._graph = Graph(graph, node_values=node_values) diff --git a/src/cyclebane/value_array_adapters.py b/src/cyclebane/value_array_adapters.py index 2fb3d19..81f173f 100644 --- a/src/cyclebane/value_array_adapters.py +++ b/src/cyclebane/value_array_adapters.py @@ -148,7 +148,10 @@ def index_names(self) -> tuple[IndexName, ...]: @property def indices(self) -> dict[IndexName, Iterable[IndexValue]]: + # TODO Things are getting weird, maybe we don't want to reduce the groupby, + # just use a custom object? if (multi_index := self._get_multi_index()) is not None: + # return {multi_index.names: self._groups} base = dict(zip(multi_index.names, multi_index.levels, strict=True)) base[multi_index.names[0]] = self._groups return base From 7ada2093b47b51a6aa2eb0009fa452b1fe4fafd6 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 18 Aug 2025 13:59:45 +0200 Subject: [PATCH 11/33] Write down some better ideas --- src/cyclebane/graph.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index d157ddf..3bc0ae1 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -466,11 +466,20 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: The name of the attribute on nodes that holds the array-like object. """ graph = self.graph + # TODO Better idea: + # 1. Do not change PandasSeriesAdapter. + # 2. Instead use (node,index_name) as index in NodeValues (not just node). + # We can then have + # {(Material, sample_dim): [Si,Si,Ge], + # (Material, material_dim): [Si,Ge]} + # 3. Store grouping independently of NodeValues, they are unrelated. subindex: Iterable[Iterable[IndexValue]] | None = None for index_name, index in reversed(self.indices.items()): if subindex is None: graphs = _clone_graph(graph, index_name, index) else: + # Note how `index` is unused in this branch, we assume it is represented + # by the subindex (but split by group). graphs = [ _clone_graph(graph_for_group, index_name, inner_index) for inner_index, graph_for_group in zip(subindex, graphs) @@ -598,6 +607,7 @@ class GroupbyGraph: such as aggregation or summarization. """ + # TODO Should we support a custom new dim name here, instead of using `node`? def __init__(self, graph: nx.DiGraph, node_values: NodeValues, node: Hashable): graph = graph.copy() node_values = node_values.copy() From b20ee106d854758e4d0fe0ce91f4983dc506de8f Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Mon, 18 Aug 2025 14:44:42 +0200 Subject: [PATCH 12/33] Refactor again --- src/cyclebane/graph.py | 31 +++++++++++++++++++-------- src/cyclebane/value_array_adapters.py | 2 ++ 2 files changed, 24 insertions(+), 9 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 3bc0ae1..f90d2b0 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -473,22 +473,26 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: # {(Material, sample_dim): [Si,Si,Ge], # (Material, material_dim): [Si,Ge]} # 3. Store grouping independently of NodeValues, they are unrelated. - subindex: Iterable[Iterable[IndexValue]] | None = None + subindex_name: IndexName | None = None + subindex: Iterable[Iterable[IndexValue]] = [[]] + groupings = getattr(self, '_groups', {}) for index_name, index in reversed(self.indices.items()): - if subindex is None: + if index_name != subindex_name: graphs = _clone_graph(graph, index_name, index) else: # Note how `index` is unused in this branch, we assume it is represented # by the subindex (but split by group). graphs = [ _clone_graph(graph_for_group, index_name, inner_index) - for inner_index, graph_for_group in zip(subindex, graphs) + for inner_index, graph_for_group in zip( + subindex, graphs, strict=True + ) ] # Flatten nested list of graphs graphs = [g for sublist in graphs for g in sublist] - if isinstance(index, dict): - subindex = index.values() + if (grouping := groupings.get(index_name)) is not None: # No compose, delayed until we made clones within each group + subindex_name, subindex = grouping else: graph = nx.compose_all(graphs) @@ -610,13 +614,19 @@ class GroupbyGraph: # TODO Should we support a custom new dim name here, instead of using `node`? def __init__(self, graph: nx.DiGraph, node_values: NodeValues, node: Hashable): graph = graph.copy() - node_values = node_values.copy() grouping = node_values[node] self._group_index_name = node self._index_name = grouping.index_names[0] - groups = grouping.group() - del node_values[node] # Explicit since __setitem__ does not support replacing - node_values[node] = groups + groupby = grouping.group() + self._groups = groupby.groups + # Hack to get extra index + # Use tuple as key instead? + # If someone needs this, they can reduce explicitly? + # No, problem is merge with map over groups! + new_values = NodeValues.from_mapping({'xxxx': groupby.first()}, axis_zero=0) + node_values = node_values.merge(new_values) + # del node_values[node] # Explicit since __setitem__ does not support replacing + # node_values[node] = groups # The grouping node becomes the new index name # TODO Add node after grouping? But would need to set value? # graph.add_node(_node_with_indices(node, (node,))) @@ -649,6 +659,9 @@ def reduce( attrs=attrs, _extra_index_name=self._group_index_name, ) + graph._groups = { + self._group_index_name: (self._index_name, self._groups.values()) + } return graph diff --git a/src/cyclebane/value_array_adapters.py b/src/cyclebane/value_array_adapters.py index 81f173f..3dec0e0 100644 --- a/src/cyclebane/value_array_adapters.py +++ b/src/cyclebane/value_array_adapters.py @@ -159,6 +159,8 @@ def indices(self) -> dict[IndexName, Iterable[IndexValue]]: def group(self) -> PandasSeriesAdapter: groupby = self._series.groupby(self._series) + + return groupby return PandasSeriesAdapter( groupby.apply(lambda x: x, include_groups=False), axis_zero=self._axis_zero, From 5f71d1e814267104e866e5c0ad5a402468ac8471 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Tue, 19 Aug 2025 06:55:04 +0200 Subject: [PATCH 13/33] Refactor again --- src/cyclebane/graph.py | 78 +++++++++++++++++++++++---- src/cyclebane/value_array_adapters.py | 9 ++-- tests/groupby_test.py | 5 ++ 3 files changed, 79 insertions(+), 13 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index f90d2b0..a96a4a0 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -455,6 +455,18 @@ def tmp(self): zip(graphs_y[0], coord_y[0]) # len=2 zip(graphs_y[1], coord_y[1]) # len=1 + def _get_groupings( + self, + ) -> dict[IndexName, tuple[IndexName, Iterable[Iterable[IndexValue]]]]: + groupings = { + key.node: values._series + for key, values in self._node_values.items() + if isinstance(key, GroupingKey) + } + return { + key: (next(iter(values)).name, values) for key, values in groupings.items() + } + def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: """ Convert to a NetworkX graph, spelling out the internal array structures as @@ -473,10 +485,44 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: # {(Material, sample_dim): [Si,Si,Ge], # (Material, material_dim): [Si,Ge]} # 3. Store grouping independently of NodeValues, they are unrelated. - subindex_name: IndexName | None = None - subindex: Iterable[Iterable[IndexValue]] = [[]] - groupings = getattr(self, '_groups', {}) + # subindex_name: IndexName | None = None + # subindex: Iterable[Iterable[IndexValue]] = [[]] + + groupings = { + key.node: values + for key, values in self._node_values.items() + if isinstance(key, GroupingKey) + } + groupings = self._get_groupings() + is_subindex = [index_name for index_name, _ in groupings.values()] + # for index_name, index in reversed(self.indices.items()): + # if (grouping := groupings.get(index_name)) is not None: + # index_value = next(iter(index)) + # value = grouping.sel(((index_name, index_value),)) + # is_subindex.append(value.name) + print(f'{is_subindex=}') + for index_name, index in reversed(self.indices.items()): + # TODO Try sth. like this, looks cleaner. + if index_name in is_subindex: + continue + graphs = _clone_graph(graph, index_name, index) + if (grouping := groupings.get(index_name)) is not None: + # subindex = grouping._series + # subindex_name = next(iter(subindex)).name + subindex_name, subindex = grouping + is_subindex.append(subindex_name) + graphs = [ + _clone_graph(graph_for_group, subindex_name, inner_index) + for inner_index, graph_for_group in zip( + subindex, graphs, strict=True + ) + ] + # Flatten nested list of graphs + graphs = [g for sublist in graphs for g in sublist] + graph = nx.compose_all(graphs) + continue + if index_name != subindex_name: graphs = _clone_graph(graph, index_name, index) else: @@ -602,6 +648,11 @@ def __setitem__(self, branch: Hashable | slice, other: Graph) -> None: self.graph = graph +@dataclass(frozen=True, slots=True) +class GroupingKey: + node: Hashable + + class GroupbyGraph: """ A graph that has been grouped by a specific index. @@ -617,14 +668,21 @@ def __init__(self, graph: nx.DiGraph, node_values: NodeValues, node: Hashable): grouping = node_values[node] self._group_index_name = node self._index_name = grouping.index_names[0] - groupby = grouping.group() - self._groups = groupby.groups + groups = grouping.group(index_name=node) + # self._groups = groupby.groups # Hack to get extra index # Use tuple as key instead? # If someone needs this, they can reduce explicitly? # No, problem is merge with map over groups! - new_values = NodeValues.from_mapping({'xxxx': groupby.first()}, axis_zero=0) - node_values = node_values.merge(new_values) + # new_values = NodeValues.from_mapping({'xxxx': groupby.first()}, axis_zero=0) + + # Store grouping as a special node value. This has two reasons: + # 1. We want the resulting graph to have the grouping node's unique values as + # as a new index (and its name as the index name). + # 2. We need to store the grouping so we can perform the grouping operation when + # building the full graph in `Graph.to_networkx`. + node_values = node_values.merge({GroupingKey(node): groups}) + # del node_values[node] # Explicit since __setitem__ does not support replacing # node_values[node] = groups # The grouping node becomes the new index name @@ -659,9 +717,9 @@ def reduce( attrs=attrs, _extra_index_name=self._group_index_name, ) - graph._groups = { - self._group_index_name: (self._index_name, self._groups.values()) - } + # graph._groups = { + # self._group_index_name: (self._index_name, self._groups.values()) + # } return graph diff --git a/src/cyclebane/value_array_adapters.py b/src/cyclebane/value_array_adapters.py index 3dec0e0..a1eb4c4 100644 --- a/src/cyclebane/value_array_adapters.py +++ b/src/cyclebane/value_array_adapters.py @@ -157,10 +157,13 @@ def indices(self) -> dict[IndexName, Iterable[IndexValue]]: return base return {self.index_names[0]: self._series.index} - def group(self) -> PandasSeriesAdapter: - groupby = self._series.groupby(self._series) + def group(self, index_name: Hashable) -> PandasSeriesAdapter: + import pandas - return groupby + groupby = self._series.groupby(self._series) + groups = pandas.Series(groupby.groups) + groups.index.rename(index_name, inplace=True) + return PandasSeriesAdapter(groups) return PandasSeriesAdapter( groupby.apply(lambda x: x, include_groups=False), axis_zero=self._axis_zero, diff --git a/tests/groupby_test.py b/tests/groupby_test.py index 2f4921b..303e405 100644 --- a/tests/groupby_test.py +++ b/tests/groupby_test.py @@ -27,6 +27,11 @@ def test_tmp() -> None: graph = cb.Graph(g) mapped = graph.map(df) grouped = mapped.groupby('b').reduce('c', name='d') + + print('Start test') + print(grouped.indices) + print(grouped.graph.nodes) + result = grouped.to_networkx() # Nodes before grouping From 8e9dae897c359cf687f63e217d600aefc902d1fc Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Tue, 19 Aug 2025 09:29:35 +0200 Subject: [PATCH 14/33] Some cleanup --- src/cyclebane/graph.py | 157 +------------------------- src/cyclebane/value_array_adapters.py | 5 - 2 files changed, 3 insertions(+), 159 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index a96a4a0..a40733f 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -353,108 +353,6 @@ def _from_orig_key(self, key: Hashable) -> Hashable: def by_position(self, index_name: IndexName) -> PositionalIndexer: return PositionalIndexer(self, index_name) - def tmp(self): - graph = self.graph - for index_name, index in reversed(self.indices.items()): - graphs = [graph for i in index] - graph = nx.compose_all(graphs) - # say we have dims (x,y) - graphs_x = [] - graphs_x[0] = [graph for y in coords_y] - graphs_x[1] = [graph for y in coords_y] - graphs_xy = zip(graphs_x, coord_x) - # with grouping - # Pandas MultiIndex ~ binned variable, outer dim x, inner dim y - # coords_y = [[a, c], [b]] # dims=(x,) - - # sample=(sample,) mat=(sample,) mat_groups=(mat,) - # (sample,mat) # groupby - # (mat,) # reduce - - # TODO Can we have nodes that depend on (material,) but do not come from - # the grouping-reduce operation? How can we set those? Pass at same time - # to groupby (forwarding to map)? - - # Now - for index_name, index in reversed(self.indices.items()): - graphs = self._clone_graph(graph, index_name, index) - graph = nx.compose_all(graphs) - - # Then - for index_name, index in reversed(self.indices.items()): - if is_multi_index(index): - # Index looks like {'Si': [s1, s3], 'Ge': [s2]} - - # One graph per material, do not compose! - graphs = self._clone_graph(graph, index_name, index.keys()) - # IndexValues(axes=(mat,), values=(Si,)) - # IndexValues(axes=(mat,), values=(Ge,)) - - inner_index_name = index.inner_index # sample - # Si - # IndexValues(axes=(sample,), values=(s1,)) - # IndexValues(axes=(sample,), values=(s3,)) - # Ge - # IndexValues(axes=(sample,), values=(s2,)) - graphs = [ - self._clone_graph(graph_for_material, inner_index_name, inner_index) - for inner_index, graph_for_material in zip(index.values(), graphs) - ] - # graphs[Si]: [(e[Si],h[Si]), (e[Si],h[Si])] - # graphs[Ge]: [(e[Ge],h[Ge])] - - # graphs[Si]: [(d[s1],f[Si]), (d[s3],f[Si])] - # graphs[Ge]: [(d[s2],f[Ge])] - - # graphs[Si]: [(g[s1],j), (g[s3],j)] - # graphs[Ge]: [(g[s2],j)] - - # PROBLEM: Does not work for nodes that have two indices!? - - # No! We don't need grouped node, maybe? There is no compute for it! - # Final node names should be (at the grouped but not reduced node): - # Note: May want to flatten the list - # IndexValues(axes=(mat,sample), values=(Si,s1)) - # IndexValues(axes=(mat,sample), values=(Si,s3)) - # IndexValues(axes=(mat,sample), values=(Ge,s2)) - - graph = nx.compose_all(graphs) - else: - # Don't forget nodes not taking part in the grouping, which need - # IndexValues(axes=(sample,), values=(s1,)) - # IndexValues(axes=(sample,), values=(s2,)) - # IndexValues(axes=(sample,), values=(s3,)) - # Done by second loop iteration? ... but will set up all-to-all edges?! - # Can we delay the compose_all until after the loop? - graphs = self._clone_graph(graph, index_name, index) - graph = nx.compose_all(graphs) - - # 1. loop iteration - # index_name=mat # dim name - # index=mat_groups # list of materials - # => graph composed of graph copies, one for each material - # 1b. nested loop - # mat_groups = {'Si': [s1, s3], 'Ge': [s2]} - # - - # 2. loop iteration (important since there may be nodes mapped - # over sample that are not grouped by material) - # index_name=sample # dim name - # index=sample # list of samples - # naively this would put *all* samples into the subgraph for each material - - # 1. make small arrays, each different length - # 2. combine into ragged 2D array - graphs_x[0] = [graph for y in coords_y[0]] # len=2 - graphs_x[1] = [graph for y in coords_y[1]] # len=1 - graphs_xy = zip(graphs_x, coord_x) - - # 1. make 1D array - # 2. replace each value by array of different length - graphs_y = [graph for x in coords_x] - zip(graphs_y[0], coord_y[0]) # len=2 - zip(graphs_y[1], coord_y[1]) # len=1 - def _get_groupings( self, ) -> dict[IndexName, tuple[IndexName, Iterable[Iterable[IndexValue]]]]: @@ -478,69 +376,24 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: The name of the attribute on nodes that holds the array-like object. """ graph = self.graph - # TODO Better idea: - # 1. Do not change PandasSeriesAdapter. - # 2. Instead use (node,index_name) as index in NodeValues (not just node). - # We can then have - # {(Material, sample_dim): [Si,Si,Ge], - # (Material, material_dim): [Si,Ge]} - # 3. Store grouping independently of NodeValues, they are unrelated. - # subindex_name: IndexName | None = None - # subindex: Iterable[Iterable[IndexValue]] = [[]] - - groupings = { - key.node: values - for key, values in self._node_values.items() - if isinstance(key, GroupingKey) - } groupings = self._get_groupings() is_subindex = [index_name for index_name, _ in groupings.values()] - # for index_name, index in reversed(self.indices.items()): - # if (grouping := groupings.get(index_name)) is not None: - # index_value = next(iter(index)) - # value = grouping.sel(((index_name, index_value),)) - # is_subindex.append(value.name) - print(f'{is_subindex=}') for index_name, index in reversed(self.indices.items()): - # TODO Try sth. like this, looks cleaner. if index_name in is_subindex: continue graphs = _clone_graph(graph, index_name, index) if (grouping := groupings.get(index_name)) is not None: - # subindex = grouping._series - # subindex_name = next(iter(subindex)).name subindex_name, subindex = grouping - is_subindex.append(subindex_name) - graphs = [ + subgraphs = [ _clone_graph(graph_for_group, subindex_name, inner_index) for inner_index, graph_for_group in zip( subindex, graphs, strict=True ) ] # Flatten nested list of graphs - graphs = [g for sublist in graphs for g in sublist] + graphs = [g for sublist in subgraphs for g in sublist] graph = nx.compose_all(graphs) - continue - - if index_name != subindex_name: - graphs = _clone_graph(graph, index_name, index) - else: - # Note how `index` is unused in this branch, we assume it is represented - # by the subindex (but split by group). - graphs = [ - _clone_graph(graph_for_group, index_name, inner_index) - for inner_index, graph_for_group in zip( - subindex, graphs, strict=True - ) - ] - # Flatten nested list of graphs - graphs = [g for sublist in graphs for g in sublist] - if (grouping := groupings.get(index_name)) is not None: - # No compose, delayed until we made clones within each group - subindex_name, subindex = grouping - else: - graph = nx.compose_all(graphs) # Replace all MappingNodes with their name new_names = { @@ -710,17 +563,13 @@ def reduce( attrs: Attributes to set on the new node(s). """ - graph = self._graph.reduce( + return self._graph.reduce( key=key, index=self._index_name, name=name, attrs=attrs, _extra_index_name=self._group_index_name, ) - # graph._groups = { - # self._group_index_name: (self._index_name, self._groups.values()) - # } - return graph def _clone_graph( diff --git a/src/cyclebane/value_array_adapters.py b/src/cyclebane/value_array_adapters.py index a1eb4c4..54ae5f0 100644 --- a/src/cyclebane/value_array_adapters.py +++ b/src/cyclebane/value_array_adapters.py @@ -164,11 +164,6 @@ def group(self, index_name: Hashable) -> PandasSeriesAdapter: groups = pandas.Series(groupby.groups) groups.index.rename(index_name, inplace=True) return PandasSeriesAdapter(groups) - return PandasSeriesAdapter( - groupby.apply(lambda x: x, include_groups=False), - axis_zero=self._axis_zero, - _groups=groupby.groups, - ) class XarrayDataArrayAdapter(ValueArray): From 6229ebe0e4e00a18c57909c5439e0cd5b81ea1de Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Tue, 19 Aug 2025 09:57:07 +0200 Subject: [PATCH 15/33] Rename variables --- src/cyclebane/graph.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index a40733f..675ed2e 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -377,19 +377,17 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: """ graph = self.graph groupings = self._get_groupings() - is_subindex = [index_name for index_name, _ in groupings.values()] + is_grouped = [index_name for index_name, _ in groupings.values()] for index_name, index in reversed(self.indices.items()): - if index_name in is_subindex: + if index_name in is_grouped: continue graphs = _clone_graph(graph, index_name, index) if (grouping := groupings.get(index_name)) is not None: - subindex_name, subindex = grouping + subindex_name, subindices = grouping subgraphs = [ - _clone_graph(graph_for_group, subindex_name, inner_index) - for inner_index, graph_for_group in zip( - subindex, graphs, strict=True - ) + _clone_graph(group_graph, subindex_name, subindex) + for subindex, group_graph in zip(subindices, graphs, strict=True) ] # Flatten nested list of graphs graphs = [g for sublist in subgraphs for g in sublist] From fd4e32a96dfccb4da34783a4eca5130ce0a3c960 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Tue, 19 Aug 2025 10:04:30 +0200 Subject: [PATCH 16/33] Revert MultiIndex support --- src/cyclebane/value_array_adapters.py | 45 +++------------------------ 1 file changed, 4 insertions(+), 41 deletions(-) diff --git a/src/cyclebane/value_array_adapters.py b/src/cyclebane/value_array_adapters.py index 54ae5f0..d4047ba 100644 --- a/src/cyclebane/value_array_adapters.py +++ b/src/cyclebane/value_array_adapters.py @@ -69,18 +69,9 @@ def indices(self) -> dict[IndexName, Iterable[IndexValue]]: class PandasSeriesAdapter(ValueArray): - def __init__( - self, - series: pandas.Series, - *, - axis_zero: int = 0, - _groups: dict[Hashable, pandas.Index[Any]] | None = None, - ): + def __init__(self, series: pandas.Series, *, axis_zero: int = 0): self._series = series - if self._get_multi_index() is None and self._series.index.name is None: - self._series = self._series.rename_axis(f'dim_{axis_zero}') self._axis_zero = axis_zero - self._groups = _groups @staticmethod def try_from(obj: Any, *, axis_zero: int = 0) -> PandasSeriesAdapter | None: @@ -89,7 +80,6 @@ def try_from(obj: Any, *, axis_zero: int = 0) -> PandasSeriesAdapter | None: except ModuleNotFoundError: return None if isinstance(obj, pandas.Series): - # TODO Reject MultiIndex? return PandasSeriesAdapter(obj, axis_zero=axis_zero) def _equal(self, other: PandasSeriesAdapter) -> bool: @@ -101,44 +91,23 @@ def sel(self, key: tuple[tuple[IndexName, IndexValue], ...]) -> Any: if len(key) != 1: raise ValueError('PandasSeriesAdapter only supports single index') index_name, i = key[0] - if index_name not in self.index_names: + if index_name != self.index_names[0]: raise ValueError( f'Unexpected index name {index_name} for PandasSeriesAdapter with ' f'index names {self.index_names}' ) - if self._get_multi_index() is not None: - s = self._series.xs(i, level=index_name) - if len(s) == 1: - return s.iloc[0] - is_constant = s.eq(s.iloc[0]).all() - if is_constant: - return s.iloc[0] - raise ValueError( - f'Cannot get single value from Pandas Series with index {index_name}' - f' and value {i}, as it is not constant.' - ) return self._series.loc[i] def __getitem__(self, key: dict[IndexName, slice]) -> PandasSeriesAdapter: _, i = next(iter(key.items())) return PandasSeriesAdapter(self._series[i], axis_zero=self._axis_zero) - def _get_multi_index(self) -> pandas.MultiIndex | None: - import pandas - - if isinstance(self._series.index, pandas.MultiIndex): - return self._series.index - @property def shape(self) -> tuple[int, ...]: - if (multi_index := self._get_multi_index()) is not None: - return tuple(len(level) for level in multi_index.levels) return (len(self._series),) @property def index_names(self) -> tuple[IndexName, ...]: - if (multi_index := self._get_multi_index()) is not None: - return tuple(multi_index.names) index_name = ( self._series.index.name if self._series.index.name is not None @@ -148,19 +117,13 @@ def index_names(self) -> tuple[IndexName, ...]: @property def indices(self) -> dict[IndexName, Iterable[IndexValue]]: - # TODO Things are getting weird, maybe we don't want to reduce the groupby, - # just use a custom object? - if (multi_index := self._get_multi_index()) is not None: - # return {multi_index.names: self._groups} - base = dict(zip(multi_index.names, multi_index.levels, strict=True)) - base[multi_index.names[0]] = self._groups - return base return {self.index_names[0]: self._series.index} def group(self, index_name: Hashable) -> PandasSeriesAdapter: import pandas - groupby = self._series.groupby(self._series) + inner_index = self.index_names[0] + groupby = self._series.rename_axis(inner_index).groupby(self._series) groups = pandas.Series(groupby.groups) groups.index.rename(index_name, inplace=True) return PandasSeriesAdapter(groups) From 1d852edbb384c1e335cf9ce3a075f0666442cf0f Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Tue, 19 Aug 2025 11:33:18 +0200 Subject: [PATCH 17/33] Notes --- src/cyclebane/graph.py | 34 +++++++++++++++++++--------------- 1 file changed, 19 insertions(+), 15 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 675ed2e..ee848e5 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -379,6 +379,13 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: groupings = self._get_groupings() is_grouped = [index_name for index_name, _ in groupings.values()] + # for index_name, index in reversed(self.groupings.items()): + # # nested case + # for index_name, index in reversed(self.indices.items()): + # # regular case + # graphs = _clone_graph(graph, index_name, index) + # graph = nx.compose_all(graphs) + for index_name, index in reversed(self.indices.items()): if index_name in is_grouped: continue @@ -516,29 +523,26 @@ class GroupbyGraph: # TODO Should we support a custom new dim name here, instead of using `node`? def __init__(self, graph: nx.DiGraph, node_values: NodeValues, node: Hashable): graph = graph.copy() - grouping = node_values[node] + values_to_group_by = node_values[node] self._group_index_name = node - self._index_name = grouping.index_names[0] - groups = grouping.group(index_name=node) - # self._groups = groupby.groups - # Hack to get extra index - # Use tuple as key instead? - # If someone needs this, they can reduce explicitly? - # No, problem is merge with map over groups! - # new_values = NodeValues.from_mapping({'xxxx': groupby.first()}, axis_zero=0) + self._index_name = values_to_group_by.index_names[0] + groups = values_to_group_by.group(index_name=node) + + # 1. Store grouping dict + # 2. Store node (group index name) + # 3. Index name to reduce over + # groups = {material: {Si:(sample,[a,b]), Ge:(sample,[c])}} + # groups = {GroupingKey(material): {Si:(sample,[a,b]), Ge:(sample,[c])}} # Store grouping as a special node value. This has two reasons: # 1. We want the resulting graph to have the grouping node's unique values as # as a new index (and its name as the index name). # 2. We need to store the grouping so we can perform the grouping operation when # building the full graph in `Graph.to_networkx`. - node_values = node_values.merge({GroupingKey(node): groups}) - # del node_values[node] # Explicit since __setitem__ does not support replacing - # node_values[node] = groups - # The grouping node becomes the new index name - # TODO Add node after grouping? But would need to set value? - # graph.add_node(_node_with_indices(node, (node,))) + # Other option would be to store a special "value" on the reduce node?! + # TODO This is dropped by __getitem__! + node_values = node_values.merge({GroupingKey(node): groups}) self._graph = Graph(graph, node_values=node_values) def reduce( From 6e125a8e0856bf2f17b360d466d5dbf45f630457 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Tue, 19 Aug 2025 13:32:41 +0200 Subject: [PATCH 18/33] Add TODO --- src/cyclebane/graph.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index ee848e5..f97651a 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -545,6 +545,7 @@ def __init__(self, graph: nx.DiGraph, node_values: NodeValues, node: Hashable): node_values = node_values.merge({GroupingKey(node): groups}) self._graph = Graph(graph, node_values=node_values) + # TODO Require specifying index! def reduce( self, key: None | Hashable = None, From 366b43fd23fe7e8ee57bff01b830936f57224269 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Wed, 20 Aug 2025 12:33:47 +0200 Subject: [PATCH 19/33] Sketch out cleaner solution --- src/cyclebane/graph.py | 43 +++++++++++++++++---------- src/cyclebane/value_array_adapters.py | 7 +++++ 2 files changed, 34 insertions(+), 16 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index f97651a..1481be6 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -356,14 +356,17 @@ def by_position(self, index_name: IndexName) -> PositionalIndexer: def _get_groupings( self, ) -> dict[IndexName, tuple[IndexName, Iterable[Iterable[IndexValue]]]]: - groupings = { - key.node: values._series - for key, values in self._node_values.items() - if isinstance(key, GroupingKey) - } - return { - key: (next(iter(values)).name, values) for key, values in groupings.items() - } + groupings = {} + node_values = self._node_values.copy() + for key, values in self._node_values.items(): + if (get_grouping := getattr(values, 'get_grouping', None)) is not None: + grouping = get_grouping() + if grouping is not None: + del node_values[key] + group_index_name = values.index_names[0] + index_name = next(iter(grouping)).name + groupings[group_index_name] = (index_name, grouping) + return groupings, node_values def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: """ @@ -376,9 +379,12 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: The name of the attribute on nodes that holds the array-like object. """ graph = self.graph - groupings = self._get_groupings() + groupings, true_node_values = self._get_groupings() is_grouped = [index_name for index_name, _ in groupings.values()] + print(f'{groupings=}') + print(f'{is_grouped=}') + # for index_name, index in reversed(self.groupings.items()): # # nested case # for index_name, index in reversed(self.indices.items()): @@ -412,9 +418,9 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: for node in graph.nodes: if ( isinstance(node, NodeName) - and (node_values := self._node_values.get(node.name)) is not None + and (value_array := true_node_values.get(node.name)) is not None ): - graph.nodes[node][value_attr] = node_values.sel(node.index.to_tuple()) + graph.nodes[node][value_attr] = value_array.sel(node.index.to_tuple()) return graph @@ -522,11 +528,12 @@ class GroupbyGraph: # TODO Should we support a custom new dim name here, instead of using `node`? def __init__(self, graph: nx.DiGraph, node_values: NodeValues, node: Hashable): - graph = graph.copy() + self._graph = graph # .copy() + self._node_values = node_values values_to_group_by = node_values[node] self._group_index_name = node self._index_name = values_to_group_by.index_names[0] - groups = values_to_group_by.group(index_name=node) + self._groups = values_to_group_by.group(index_name=node) # 1. Store grouping dict # 2. Store node (group index name) @@ -542,8 +549,8 @@ def __init__(self, graph: nx.DiGraph, node_values: NodeValues, node: Hashable): # Other option would be to store a special "value" on the reduce node?! # TODO This is dropped by __getitem__! - node_values = node_values.merge({GroupingKey(node): groups}) - self._graph = Graph(graph, node_values=node_values) + # node_values = node_values.merge({GroupingKey(node): groups}) + # self._graph = Graph(graph, node_values=node_values) # TODO Require specifying index! def reduce( @@ -566,7 +573,11 @@ def reduce( attrs: Attributes to set on the new node(s). """ - return self._graph.reduce( + # Generate name here since we want to store grouping on the new "reduce" node. + name = name or _get_new_node_name(self._graph) + node_values = self._node_values.merge({name: self._groups}) + graph = Graph(self._graph, node_values=node_values) + return graph.reduce( key=key, index=self._index_name, name=name, diff --git a/src/cyclebane/value_array_adapters.py b/src/cyclebane/value_array_adapters.py index d4047ba..82d05d2 100644 --- a/src/cyclebane/value_array_adapters.py +++ b/src/cyclebane/value_array_adapters.py @@ -124,10 +124,17 @@ def group(self, index_name: Hashable) -> PandasSeriesAdapter: inner_index = self.index_names[0] groupby = self._series.rename_axis(inner_index).groupby(self._series) + # {Si:[a,b],Ge:[c]} groups = pandas.Series(groupby.groups) groups.index.rename(index_name, inplace=True) return PandasSeriesAdapter(groups) + def get_grouping(self) -> Iterable[Iterable[IndexValue]] | None: + import pandas + + if isinstance(self._series.iloc[0], pandas.Index): + return self._series + class XarrayDataArrayAdapter(ValueArray): def __init__( From 75283ada51d5168238c6e19a6772713f841b5178 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Wed, 20 Aug 2025 13:51:58 +0200 Subject: [PATCH 20/33] Sketch out cleaner approach --- src/cyclebane/graph.py | 68 +++++++++++++++++---------- src/cyclebane/value_array.py | 16 +++++++ src/cyclebane/value_array_adapters.py | 23 +++++---- 3 files changed, 73 insertions(+), 34 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 1481be6..41dda0d 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -10,6 +10,7 @@ import networkx as nx from .node_values import IndexName, IndexValue, NodeValues +from .value_array import Grouping def _get_unique_sink(graph: nx.DiGraph) -> Hashable: @@ -356,16 +357,15 @@ def by_position(self, index_name: IndexName) -> PositionalIndexer: def _get_groupings( self, ) -> dict[IndexName, tuple[IndexName, Iterable[Iterable[IndexValue]]]]: + # TODO return list, can have different groupings for some index names. groupings = {} node_values = self._node_values.copy() for key, values in self._node_values.items(): - if (get_grouping := getattr(values, 'get_grouping', None)) is not None: - grouping = get_grouping() - if grouping is not None: - del node_values[key] - group_index_name = values.index_names[0] - index_name = next(iter(grouping)).name - groupings[group_index_name] = (index_name, grouping) + if (grouping := values.get_grouping()) is not None: + del node_values[key] + group_index_name = values.index_names[0] + index_name = next(iter(grouping)).name + groupings[group_index_name] = (index_name, grouping) return groupings, node_values def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: @@ -379,11 +379,29 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: The name of the attribute on nodes that holds the array-like object. """ graph = self.graph - groupings, true_node_values = self._get_groupings() - is_grouped = [index_name for index_name, _ in groupings.values()] + # groupings, true_node_values = self._get_groupings() + # is_grouped = [index_name for index_name, _ in groupings.values()] - print(f'{groupings=}') - print(f'{is_grouped=}') + # print(f'{groupings=}') + # print(f'{is_grouped=}') + regular_indices = dict(reversed(self.indices.items())) + node_values = self._node_values.copy() + for key, values in self._node_values.items(): + if (grouping := values.get_grouping()) is not None: + del node_values[key] + del regular_indices[grouping.index_name] + index = regular_indices.pop(grouping.group_index_name) + graphs = _clone_graph(graph, grouping.group_index_name, index) + subgraphs = [ + _clone_graph(group_graph, grouping.index_name, idx) + for idx, group_graph in zip(grouping.indices, graphs, strict=True) + ] + # Flatten nested list of graphs + graphs = [g for sublist in subgraphs for g in sublist] + graph = nx.compose_all(graphs) + for index_name, index in regular_indices.items(): + graphs = _clone_graph(graph, index_name, index) + graph = nx.compose_all(graphs) # for index_name, index in reversed(self.groupings.items()): # # nested case @@ -392,19 +410,19 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: # graphs = _clone_graph(graph, index_name, index) # graph = nx.compose_all(graphs) - for index_name, index in reversed(self.indices.items()): - if index_name in is_grouped: - continue - graphs = _clone_graph(graph, index_name, index) - if (grouping := groupings.get(index_name)) is not None: - subindex_name, subindices = grouping - subgraphs = [ - _clone_graph(group_graph, subindex_name, subindex) - for subindex, group_graph in zip(subindices, graphs, strict=True) - ] - # Flatten nested list of graphs - graphs = [g for sublist in subgraphs for g in sublist] - graph = nx.compose_all(graphs) + # for index_name, index in reversed(self.indices.items()): + # if index_name in is_grouped: + # continue + # graphs = _clone_graph(graph, index_name, index) + # if (grouping := groupings.get(index_name)) is not None: + # subindex_name, subindices = grouping + # subgraphs = [ + # _clone_graph(group_graph, subindex_name, subindex) + # for subindex, group_graph in zip(subindices, graphs, strict=True) + # ] + # # Flatten nested list of graphs + # graphs = [g for sublist in subgraphs for g in sublist] + # graph = nx.compose_all(graphs) # Replace all MappingNodes with their name new_names = { @@ -418,7 +436,7 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: for node in graph.nodes: if ( isinstance(node, NodeName) - and (value_array := true_node_values.get(node.name)) is not None + and (value_array := node_values.get(node.name)) is not None ): graph.nodes[node][value_attr] = value_array.sel(node.index.to_tuple()) diff --git a/src/cyclebane/value_array.py b/src/cyclebane/value_array.py index 159964b..89f0588 100644 --- a/src/cyclebane/value_array.py +++ b/src/cyclebane/value_array.py @@ -4,6 +4,7 @@ from abc import ABC, abstractmethod from collections.abc import Hashable, Iterable +from dataclasses import dataclass from typing import Any, ClassVar, TypeVar IndexName = Hashable @@ -94,3 +95,18 @@ def group(self) -> ValueArray: raise NotImplementedError( 'ValueArray.group() is only implemented for Pandas series.' ) + + def get_grouping(self) -> Grouping | None: + """ + If the instance holds grouping information, return it. + + Meant to be overridden by subclasses that support grouping. + """ + return None + + +@dataclass +class Grouping: + indices: Iterable[Iterable[IndexValue]] + index_name: IndexName + group_index_name: IndexName diff --git a/src/cyclebane/value_array_adapters.py b/src/cyclebane/value_array_adapters.py index 82d05d2..753171a 100644 --- a/src/cyclebane/value_array_adapters.py +++ b/src/cyclebane/value_array_adapters.py @@ -6,7 +6,7 @@ from types import ModuleType from typing import TYPE_CHECKING, Any, TypeVar -from .value_array import ValueArray +from .value_array import Grouping, ValueArray if TYPE_CHECKING: import numpy @@ -69,9 +69,12 @@ def indices(self) -> dict[IndexName, Iterable[IndexValue]]: class PandasSeriesAdapter(ValueArray): - def __init__(self, series: pandas.Series, *, axis_zero: int = 0): + def __init__( + self, series: pandas.Series, *, axis_zero: int = 0, _is_grouping: bool = False + ): self._series = series self._axis_zero = axis_zero + self._is_grouping = _is_grouping @staticmethod def try_from(obj: Any, *, axis_zero: int = 0) -> PandasSeriesAdapter | None: @@ -127,13 +130,15 @@ def group(self, index_name: Hashable) -> PandasSeriesAdapter: # {Si:[a,b],Ge:[c]} groups = pandas.Series(groupby.groups) groups.index.rename(index_name, inplace=True) - return PandasSeriesAdapter(groups) - - def get_grouping(self) -> Iterable[Iterable[IndexValue]] | None: - import pandas - - if isinstance(self._series.iloc[0], pandas.Index): - return self._series + return PandasSeriesAdapter(groups, _is_grouping=True) + + def get_grouping(self) -> Grouping | None: + if self._is_grouping: + return Grouping( + indices=self._series, + group_index_name=self.index_names[0], + index_name=next(iter(self._series)).name, + ) class XarrayDataArrayAdapter(ValueArray): From 14b663daa8186985ec7e9adcb9a0d25bca092dcc Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Wed, 20 Aug 2025 14:02:54 +0200 Subject: [PATCH 21/33] Cleanup --- src/cyclebane/graph.py | 68 ++++----------------------- src/cyclebane/value_array.py | 2 +- src/cyclebane/value_array_adapters.py | 5 +- 3 files changed, 10 insertions(+), 65 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 41dda0d..0ee264f 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -10,7 +10,6 @@ import networkx as nx from .node_values import IndexName, IndexValue, NodeValues -from .value_array import Grouping def _get_unique_sink(graph: nx.DiGraph) -> Hashable: @@ -354,20 +353,6 @@ def _from_orig_key(self, key: Hashable) -> Hashable: def by_position(self, index_name: IndexName) -> PositionalIndexer: return PositionalIndexer(self, index_name) - def _get_groupings( - self, - ) -> dict[IndexName, tuple[IndexName, Iterable[Iterable[IndexValue]]]]: - # TODO return list, can have different groupings for some index names. - groupings = {} - node_values = self._node_values.copy() - for key, values in self._node_values.items(): - if (grouping := values.get_grouping()) is not None: - del node_values[key] - group_index_name = values.index_names[0] - index_name = next(iter(grouping)).name - groupings[group_index_name] = (index_name, grouping) - return groupings, node_values - def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: """ Convert to a NetworkX graph, spelling out the internal array structures as @@ -379,16 +364,14 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: The name of the attribute on nodes that holds the array-like object. """ graph = self.graph - # groupings, true_node_values = self._get_groupings() - # is_grouped = [index_name for index_name, _ in groupings.values()] - - # print(f'{groupings=}') - # print(f'{is_grouped=}') regular_indices = dict(reversed(self.indices.items())) node_values = self._node_values.copy() for key, values in self._node_values.items(): if (grouping := values.get_grouping()) is not None: del node_values[key] + # Note how this will raise if there are multiple groupings of the same + # index name, or into the same index name. We could support this if it + # is compatible, but the current graph building approach would not work. del regular_indices[grouping.index_name] index = regular_indices.pop(grouping.group_index_name) graphs = _clone_graph(graph, grouping.group_index_name, index) @@ -403,27 +386,6 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: graphs = _clone_graph(graph, index_name, index) graph = nx.compose_all(graphs) - # for index_name, index in reversed(self.groupings.items()): - # # nested case - # for index_name, index in reversed(self.indices.items()): - # # regular case - # graphs = _clone_graph(graph, index_name, index) - # graph = nx.compose_all(graphs) - - # for index_name, index in reversed(self.indices.items()): - # if index_name in is_grouped: - # continue - # graphs = _clone_graph(graph, index_name, index) - # if (grouping := groupings.get(index_name)) is not None: - # subindex_name, subindices = grouping - # subgraphs = [ - # _clone_graph(group_graph, subindex_name, subindex) - # for subindex, group_graph in zip(subindices, graphs, strict=True) - # ] - # # Flatten nested list of graphs - # graphs = [g for sublist in subgraphs for g in sublist] - # graph = nx.compose_all(graphs) - # Replace all MappingNodes with their name new_names = { node: NodeName(node.name.name, node.index) @@ -546,31 +508,14 @@ class GroupbyGraph: # TODO Should we support a custom new dim name here, instead of using `node`? def __init__(self, graph: nx.DiGraph, node_values: NodeValues, node: Hashable): - self._graph = graph # .copy() + self._graph = graph self._node_values = node_values values_to_group_by = node_values[node] self._group_index_name = node self._index_name = values_to_group_by.index_names[0] self._groups = values_to_group_by.group(index_name=node) - # 1. Store grouping dict - # 2. Store node (group index name) - # 3. Index name to reduce over - # groups = {material: {Si:(sample,[a,b]), Ge:(sample,[c])}} - # groups = {GroupingKey(material): {Si:(sample,[a,b]), Ge:(sample,[c])}} - - # Store grouping as a special node value. This has two reasons: - # 1. We want the resulting graph to have the grouping node's unique values as - # as a new index (and its name as the index name). - # 2. We need to store the grouping so we can perform the grouping operation when - # building the full graph in `Graph.to_networkx`. - - # Other option would be to store a special "value" on the reduce node?! - # TODO This is dropped by __getitem__! - # node_values = node_values.merge({GroupingKey(node): groups}) - # self._graph = Graph(graph, node_values=node_values) - - # TODO Require specifying index! + # TODO Require specifying index!? def reduce( self, key: None | Hashable = None, @@ -593,6 +538,9 @@ def reduce( """ # Generate name here since we want to store grouping on the new "reduce" node. name = name or _get_new_node_name(self._graph) + # Why do we store the grouping here? This works well with existing mechanisms, + # e.g., __getitem__, which needs to decided what subset of node values to keep + # when returning a subgraph. node_values = self._node_values.merge({name: self._groups}) graph = Graph(self._graph, node_values=node_values) return graph.reduce( diff --git a/src/cyclebane/value_array.py b/src/cyclebane/value_array.py index 89f0588..23647bb 100644 --- a/src/cyclebane/value_array.py +++ b/src/cyclebane/value_array.py @@ -85,7 +85,7 @@ def index_names(self) -> tuple[IndexName, ...]: def indices(self) -> dict[IndexName, Iterable[IndexValue]]: pass - def group(self) -> ValueArray: + def group(self, index_name: Hashable) -> ValueArray: """ Group the values by their indices. diff --git a/src/cyclebane/value_array_adapters.py b/src/cyclebane/value_array_adapters.py index 753171a..e6177b1 100644 --- a/src/cyclebane/value_array_adapters.py +++ b/src/cyclebane/value_array_adapters.py @@ -123,12 +123,9 @@ def indices(self) -> dict[IndexName, Iterable[IndexValue]]: return {self.index_names[0]: self._series.index} def group(self, index_name: Hashable) -> PandasSeriesAdapter: - import pandas - inner_index = self.index_names[0] groupby = self._series.rename_axis(inner_index).groupby(self._series) - # {Si:[a,b],Ge:[c]} - groups = pandas.Series(groupby.groups) + groups = type(self._series)(groupby.groups) groups.index.rename(index_name, inplace=True) return PandasSeriesAdapter(groups, _is_grouping=True) From 40331bc31d13d298e77428f966c00683380ee3fa Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Wed, 20 Aug 2025 14:07:36 +0200 Subject: [PATCH 22/33] More cleanup --- src/cyclebane/graph.py | 5 ---- src/cyclebane/node_values.py | 6 ----- src/cyclebane/value_array.py | 2 +- tests/pandas_series_adapter_test.py | 36 ----------------------------- 4 files changed, 1 insertion(+), 48 deletions(-) delete mode 100644 tests/pandas_series_adapter_test.py diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 0ee264f..21e63d1 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -492,11 +492,6 @@ def __setitem__(self, branch: Hashable | slice, other: Graph) -> None: self.graph = graph -@dataclass(frozen=True, slots=True) -class GroupingKey: - node: Hashable - - class GroupbyGraph: """ A graph that has been grouped by a specific index. diff --git a/src/cyclebane/node_values.py b/src/cyclebane/node_values.py index 5221243..2e83a4f 100644 --- a/src/cyclebane/node_values.py +++ b/src/cyclebane/node_values.py @@ -82,12 +82,6 @@ def from_mapping( values: Mapping[Hashable, Sequence[Any]], axis_zero: int ) -> NodeValues: """Construct from a mapping of node names to value sequences.""" - # graph.map(param_table) - # {sample: [s1,s2,s3], material: [Si,Ge,Si], param: [p1,p2,p3]} - # graph.groupby(material) - # -> merge two indices into multi-index - # {material: {Si: {sample:[s1,s3]}, Ge: {sample:[s2]}} - # {material: {Si: {sample:[s1,s3], param:[p1,p3]}, Ge: {sample:[s2], param:[p2]}} value_arrays = { key: ValueArray.from_array_like(value, axis_zero=axis_zero) for key, value in values.items() diff --git a/src/cyclebane/value_array.py b/src/cyclebane/value_array.py index 23647bb..62eff0a 100644 --- a/src/cyclebane/value_array.py +++ b/src/cyclebane/value_array.py @@ -105,7 +105,7 @@ def get_grouping(self) -> Grouping | None: return None -@dataclass +@dataclass(frozen=True, slots=True, kw_only=True) class Grouping: indices: Iterable[Iterable[IndexValue]] index_name: IndexName diff --git a/tests/pandas_series_adapter_test.py b/tests/pandas_series_adapter_test.py deleted file mode 100644 index 50beef4..0000000 --- a/tests/pandas_series_adapter_test.py +++ /dev/null @@ -1,36 +0,0 @@ -# SPDX-License-Identifier: BSD-3-Clause -# Copyright (c) 2025 Scipp contributors (https://github.com/scipp) -import pandas as pd -import pytest - -from cyclebane.value_array_adapters import PandasSeriesAdapter - - -@pytest.fixture -def series() -> pd.Series: - df = pd.DataFrame( - { - 'material': ['A', 'A', 'B', 'A', 'C', 'C'], - 'sample': ['a', 'b', 'c', 'd', 'e', 'f'], - } - ).set_index('sample') - return df['material'] - - -class TestPandasSeriesAdapter: - def test_adapter_from_grouping_row(self, series): - adapter = PandasSeriesAdapter(series) - assert adapter.shape == (6,) - assert adapter.index_names == ('sample',) - assert adapter.sel((('sample', 'c'),)) == 'B' - - def test_group_returns_multi_index_like_series(self, series): - base_adapter = PandasSeriesAdapter(series) - adapter = base_adapter.group() - assert adapter.index_names == ('material', 'sample') - assert adapter.shape == (3, 6) - indices = adapter.indices - assert list(indices['material']) == ['A', 'B', 'C'] - assert list(indices['sample']) == ['a', 'b', 'c', 'd', 'e', 'f'] - assert adapter.sel((('sample', 'c'),)) == 'B' - assert adapter.sel((('material', 'A'),)) == 'A' From fd43c92fd7572d2ecbefa6bed3e174aea7d368ed Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Wed, 20 Aug 2025 14:14:25 +0200 Subject: [PATCH 23/33] Cleanup test --- tests/groupby_test.py | 22 +--------------------- 1 file changed, 1 insertion(+), 21 deletions(-) diff --git a/tests/groupby_test.py b/tests/groupby_test.py index 303e405..d462972 100644 --- a/tests/groupby_test.py +++ b/tests/groupby_test.py @@ -18,7 +18,7 @@ def idx( ) -def test_tmp() -> None: +def test_basic_map_groupby_reduce_gives_correct_graph_structure() -> None: g = nx.DiGraph() g.add_edge('a', 'c') g.add_edge('b', 'c') @@ -28,10 +28,6 @@ def test_tmp() -> None: mapped = graph.map(df) grouped = mapped.groupby('b').reduce('c', name='d') - print('Start test') - print(grouped.indices) - print(grouped.graph.nodes) - result = grouped.to_networkx() # Nodes before grouping @@ -49,19 +45,3 @@ def test_tmp() -> None: assert not result.has_edge(idx('c', 0), idx('d', 'b', dims=('b',))) assert not result.has_edge(idx('c', 1), idx('d', 'b', dims=('b',))) assert not result.has_edge(idx('c', 2), idx('d', 'a', dims=('b',))) - - -def test_graphs_with_different_mapping_over_same_node_can_be_combined() -> None: - g = nx.DiGraph() - g.add_edge('a', 'b') - - graph = cb.Graph(g) - mapped = graph.map({'a': [1, 2, 3]}) - result = mapped.to_networkx() - - assert result.nodes[idx('a', 0)] == {'value': 1} - assert result.nodes[idx('a', 1)] == {'value': 2} - assert result.nodes[idx('a', 2)] == {'value': 3} - assert result.nodes[idx('b', 0)] == {} - assert result.nodes[idx('b', 1)] == {} - assert result.nodes[idx('b', 2)] == {} From a81600b612c93481d44f8e81b3bdce7e7c51bb54 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Wed, 20 Aug 2025 15:12:14 +0200 Subject: [PATCH 24/33] Check that multiple groupings in same graph currently not handled --- tests/groupby_test.py | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/tests/groupby_test.py b/tests/groupby_test.py index d462972..eaf3774 100644 --- a/tests/groupby_test.py +++ b/tests/groupby_test.py @@ -4,6 +4,7 @@ import networkx as nx import pandas as pd +import pytest import cyclebane as cb @@ -45,3 +46,29 @@ def test_basic_map_groupby_reduce_gives_correct_graph_structure() -> None: assert not result.has_edge(idx('c', 0), idx('d', 'b', dims=('b',))) assert not result.has_edge(idx('c', 1), idx('d', 'b', dims=('b',))) assert not result.has_edge(idx('c', 2), idx('d', 'a', dims=('b',))) + + +def test_group_in_different_ways() -> None: + g = nx.DiGraph() + g.add_edge('a', 'b') + g.add_edge('attach2', 'd') + df = pd.DataFrame( + {'a': [11, 22, 33], 'param1': ['a', 'a', 'b'], 'param2': ['x', 'y', 'x']} + ) + + graph = cb.Graph(g) + mapped = graph.map(df) + grouped = mapped.groupby('param1').reduce('b', name='grouped1') + grouped2 = mapped.groupby('param2').reduce('b', name='grouped2') + + # Map helper node over param2 so we have a place where we can attached the grouping + # by param2. + grouped = grouped.map( + pd.DataFrame({'attach2': [None, None], 'param2': ['x', 'y']}).set_index( + 'param2' + ) + ) + grouped['attach2'] = grouped2['grouped2'] + + with pytest.raises(KeyError, match='dim_0'): + grouped.to_networkx() From 6226448d318dff553fb6b070fcc3d095bfc1a2a5 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Wed, 20 Aug 2025 15:57:51 +0200 Subject: [PATCH 25/33] Hack that works in more cases, but not recursive groupby (yet?) --- src/cyclebane/graph.py | 26 +++++++++++++++++++++++--- src/cyclebane/node_values.py | 4 +++- tests/groupby_test.py | 4 ++-- 3 files changed, 28 insertions(+), 6 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 21e63d1..f03fcf2 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -366,25 +366,39 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: graph = self.graph regular_indices = dict(reversed(self.indices.items())) node_values = self._node_values.copy() + graphs_for_grouping = [] for key, values in self._node_values.items(): if (grouping := values.get_grouping()) is not None: del node_values[key] + # Subgraph of all ancestors and descendants + key = self._from_orig_key(key) + subgraph = graph.subgraph( + nx.ancestors(graph, key) | nx.descendants(graph, key) | {key} + ) # Note how this will raise if there are multiple groupings of the same # index name, or into the same index name. We could support this if it # is compatible, but the current graph building approach would not work. - del regular_indices[grouping.index_name] + regular_indices.pop(grouping.index_name, None) index = regular_indices.pop(grouping.group_index_name) - graphs = _clone_graph(graph, grouping.group_index_name, index) + graphs = _clone_graph(subgraph, grouping.group_index_name, index) subgraphs = [ _clone_graph(group_graph, grouping.index_name, idx) for idx, group_graph in zip(grouping.indices, graphs, strict=True) ] # Flatten nested list of graphs graphs = [g for sublist in subgraphs for g in sublist] - graph = nx.compose_all(graphs) + graphs_for_grouping.append(nx.compose_all(graphs)) + if graphs_for_grouping: + # If we have grouping, we need to merge the graphs for each grouping + graph = nx.compose_all(graphs_for_grouping) for index_name, index in regular_indices.items(): graphs = _clone_graph(graph, index_name, index) graph = nx.compose_all(graphs) + # Remove all nodes that are MappedNode + # graph = nx.subgraph_view( + # graph, + # filter_node=lambda node: not isinstance(node, MappedNode), + # ) # Replace all MappingNodes with their name new_names = { @@ -489,6 +503,12 @@ def __setitem__(self, branch: Hashable | slice, other: Graph) -> None: # Delay setting graph until we know no step fails self._node_values = self._node_values.merge(other._node_values) + + if sink.name in self._node_values: + node_values = self._node_values[sink.name] + del self._node_values[sink.name] + self._node_values[branch.name] = node_values + self.graph = graph diff --git a/src/cyclebane/node_values.py b/src/cyclebane/node_values.py index 2e83a4f..da0b7f4 100644 --- a/src/cyclebane/node_values.py +++ b/src/cyclebane/node_values.py @@ -59,7 +59,9 @@ def __setitem__(self, key: Hashable, value_array: ValueArray) -> None: if existing_value == value_array: return # No change needed else: - raise ValueError(f"Node '{key}' has already been mapped") + del self._values[key] # Remove existing value + # else: + # raise ValueError(f"Node '{key}' has already been mapped") # Check for index conflicts existing_indices = self.indices diff --git a/tests/groupby_test.py b/tests/groupby_test.py index eaf3774..75d35c3 100644 --- a/tests/groupby_test.py +++ b/tests/groupby_test.py @@ -70,5 +70,5 @@ def test_group_in_different_ways() -> None: ) grouped['attach2'] = grouped2['grouped2'] - with pytest.raises(KeyError, match='dim_0'): - grouped.to_networkx() + # with pytest.raises(KeyError, match='dim_0'): + grouped.to_networkx() From 5507f23a1673822ebbccd1d80de1e0ffacbeaa9e Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Thu, 21 Aug 2025 08:45:36 +0200 Subject: [PATCH 26/33] Clone grouping links separately, should work in more complex cases --- src/cyclebane/graph.py | 63 +++++++++++++++++++++++++++--------------- 1 file changed, 41 insertions(+), 22 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index f03fcf2..21310ba 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -363,37 +363,56 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: value_attr: The name of the attribute on nodes that holds the array-like object. """ - graph = self.graph + graph = self.graph.copy() + regular_indices = dict(reversed(self.indices.items())) node_values = self._node_values.copy() - graphs_for_grouping = [] + groupby_graphs = [] + print('-' * 80) + for edge in graph.edges: + print(edge) + print('-' * 80) for key, values in self._node_values.items(): if (grouping := values.get_grouping()) is not None: del node_values[key] - # Subgraph of all ancestors and descendants key = self._from_orig_key(key) - subgraph = graph.subgraph( - nx.ancestors(graph, key) | nx.descendants(graph, key) | {key} - ) - # Note how this will raise if there are multiple groupings of the same - # index name, or into the same index name. We could support this if it - # is compatible, but the current graph building approach would not work. - regular_indices.pop(grouping.index_name, None) - index = regular_indices.pop(grouping.group_index_name) - graphs = _clone_graph(subgraph, grouping.group_index_name, index) - subgraphs = [ - _clone_graph(group_graph, grouping.index_name, idx) - for idx, group_graph in zip(grouping.indices, graphs, strict=True) - ] - # Flatten nested list of graphs - graphs = [g for sublist in subgraphs for g in sublist] - graphs_for_grouping.append(nx.compose_all(graphs)) - if graphs_for_grouping: - # If we have grouping, we need to merge the graphs for each grouping - graph = nx.compose_all(graphs_for_grouping) + groupby_graph = graph.subgraph([*graph.predecessors(key), key]).copy() + graph.remove_edges_from(groupby_graph.edges) + for index_name, index in reversed(self.indices.items()): + if index_name == grouping.index_name: + continue + graphs = _clone_graph(groupby_graph, index_name, index) + if index_name == grouping.group_index_name: + subgraphs = [ + _clone_graph(group_graph, grouping.index_name, idx) + for idx, group_graph in zip( + grouping.indices, graphs, strict=True + ) + ] + # Flatten nested list of graphs + graphs = [g for sublist in subgraphs for g in sublist] + groupby_graph = nx.compose_all(graphs) + + # regular_indices.pop(grouping.index_name, None) + for node in groupby_graph.nodes: + print(node) + for edge in groupby_graph.edges: + print(edge) + groupby_graphs.append(groupby_graph) + + # if groupby_graphs: + # # If we have grouping, we need to merge the graphs for each grouping + # graph = nx.compose_all(groupby_graphs) + + print('-' * 80) + for node in graph.nodes: + print(node) + for edge in graph.edges: + print(edge) for index_name, index in regular_indices.items(): graphs = _clone_graph(graph, index_name, index) graph = nx.compose_all(graphs) + graph = nx.compose_all([*groupby_graphs, graph]) # Remove all nodes that are MappedNode # graph = nx.subgraph_view( # graph, From 4c7f5e2981269bb8999c5f2ca045d2f07858c700 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Thu, 21 Aug 2025 09:13:45 +0200 Subject: [PATCH 27/33] Cleanup --- src/cyclebane/graph.py | 70 ++++++++++++++++-------------------------- 1 file changed, 27 insertions(+), 43 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 21310ba..7778844 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -10,6 +10,7 @@ import networkx as nx from .node_values import IndexName, IndexValue, NodeValues +from .value_array import Grouping def _get_unique_sink(graph: nx.DiGraph) -> Hashable: @@ -365,59 +366,24 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: """ graph = self.graph.copy() - regular_indices = dict(reversed(self.indices.items())) + # Maintain a list of actual node values, without groupings, since we only want + # to set the former (user-provided) on (input) nodes. node_values = self._node_values.copy() groupby_graphs = [] - print('-' * 80) - for edge in graph.edges: - print(edge) - print('-' * 80) for key, values in self._node_values.items(): if (grouping := values.get_grouping()) is not None: del node_values[key] key = self._from_orig_key(key) groupby_graph = graph.subgraph([*graph.predecessors(key), key]).copy() graph.remove_edges_from(groupby_graph.edges) - for index_name, index in reversed(self.indices.items()): - if index_name == grouping.index_name: - continue - graphs = _clone_graph(groupby_graph, index_name, index) - if index_name == grouping.group_index_name: - subgraphs = [ - _clone_graph(group_graph, grouping.index_name, idx) - for idx, group_graph in zip( - grouping.indices, graphs, strict=True - ) - ] - # Flatten nested list of graphs - graphs = [g for sublist in subgraphs for g in sublist] - groupby_graph = nx.compose_all(graphs) - - # regular_indices.pop(grouping.index_name, None) - for node in groupby_graph.nodes: - print(node) - for edge in groupby_graph.edges: - print(edge) - groupby_graphs.append(groupby_graph) - - # if groupby_graphs: - # # If we have grouping, we need to merge the graphs for each grouping - # graph = nx.compose_all(groupby_graphs) - - print('-' * 80) - for node in graph.nodes: - print(node) - for edge in graph.edges: - print(edge) - for index_name, index in regular_indices.items(): + groupby_graphs.append(self._make_groupby_graph(grouping, groupby_graph)) + + for index_name, index in reversed(self.indices.items()): graphs = _clone_graph(graph, index_name, index) graph = nx.compose_all(graphs) - graph = nx.compose_all([*groupby_graphs, graph]) - # Remove all nodes that are MappedNode - # graph = nx.subgraph_view( - # graph, - # filter_node=lambda node: not isinstance(node, MappedNode), - # ) + + if groupby_graphs: + graph = nx.compose_all([*groupby_graphs, graph]) # Replace all MappingNodes with their name new_names = { @@ -437,6 +403,23 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: return graph + def _make_groupby_graph( + self, grouping: Grouping, groupby_graph: nx.DiGraph + ) -> nx.DiGraph: + for index_name, index in reversed(self.indices.items()): + if index_name == grouping.index_name: + continue + graphs = _clone_graph(groupby_graph, index_name, index) + if index_name == grouping.group_index_name: + subgraphs = [ + _clone_graph(group_graph, grouping.index_name, idx) + for idx, group_graph in zip(grouping.indices, graphs, strict=True) + ] + # Flatten nested list of graphs + graphs = [g for sublist in subgraphs for g in sublist] + groupby_graph = nx.compose_all(graphs) + return groupby_graph + def __getitem__(self, key: Hashable | slice) -> Graph: """ Get the branch of the graph rooted at the given node. @@ -523,6 +506,7 @@ def __setitem__(self, branch: Hashable | slice, other: Graph) -> None: # Delay setting graph until we know no step fails self._node_values = self._node_values.merge(other._node_values) + # TODO Need to update the key of node values if sink.name in self._node_values: node_values = self._node_values[sink.name] del self._node_values[sink.name] From 0b776054b23be19174ca681923166f4b9f3fe5c3 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Thu, 21 Aug 2025 09:35:33 +0200 Subject: [PATCH 28/33] Test mapped node data is preserved in setitem --- src/cyclebane/graph.py | 18 +++++++++++++----- src/cyclebane/node_values.py | 23 +++++++++++++++++------ tests/graph_test.py | 21 +++++++++++++++++++++ 3 files changed, 51 insertions(+), 11 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 7778844..ca4fb08 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -112,6 +112,12 @@ def _node_with_indices(node: Hashable, indices: tuple[IndexName, ...]) -> Mapped return MappedNode(name=node, indices=indices) +def _node_name(node: Hashable) -> Hashable: + if isinstance(node, MappedNode): + return node.name + return node + + def _node_indices(node: Hashable) -> tuple[IndexName, ...]: if isinstance(node, MappedNode): return node.indices @@ -506,11 +512,13 @@ def __setitem__(self, branch: Hashable | slice, other: Graph) -> None: # Delay setting graph until we know no step fails self._node_values = self._node_values.merge(other._node_values) - # TODO Need to update the key of node values - if sink.name in self._node_values: - node_values = self._node_values[sink.name] - del self._node_values[sink.name] - self._node_values[branch.name] = node_values + # Ensure we preserve the node values of the branch, if it exists. This step is + # necessary since __setitem__ effectively renames the sink node of the input + # graph to the branch name. + if _node_name(sink) in self._node_values: + node_values = self._node_values[_node_name(sink)] + del self._node_values[_node_name(sink)] + self._node_values[_node_name(branch)] = node_values self.graph = graph diff --git a/src/cyclebane/node_values.py b/src/cyclebane/node_values.py index da0b7f4..bd37eb2 100644 --- a/src/cyclebane/node_values.py +++ b/src/cyclebane/node_values.py @@ -54,14 +54,25 @@ def __delitem__(self, key: Hashable) -> None: def __setitem__(self, key: Hashable, value_array: ValueArray) -> None: """Add a single value array, checking for conflicts.""" # Check if the value array is identical to existing one - existing_value = self._values.get(key) - if existing_value is not None: - if existing_value == value_array: + old_value = self._values.get(key) + if old_value is not None: + if old_value == value_array: return # No change needed + elif old_value.index_names == value_array.index_names: + for old_index, new_index in zip( + old_value.indices.values(), + value_array.indices.values(), + strict=True, + ): + if any(i != j for i, j in zip(old_index, new_index, strict=True)): + raise ValueError( + f"Node '{key}' has already been mapped with different " + f"indices: existing {old_index} vs new {new_index}" + ) + # If indices match, we can replace the value + self._values[key] = value_array else: - del self._values[key] # Remove existing value - # else: - # raise ValueError(f"Node '{key}' has already been mapped") + raise ValueError(f"Node '{key}' has already been mapped") # Check for index conflicts existing_indices = self.indices diff --git a/tests/graph_test.py b/tests/graph_test.py index faac437..6ccd58c 100644 --- a/tests/graph_test.py +++ b/tests/graph_test.py @@ -696,6 +696,27 @@ def test_setitem_preserves_nodes_that_are_ancestors_of_unrelated_node() -> None: nx.utils.graphs_equal(graph.to_networkx(), g) +def test_setitem_preserves_node_values_of_sink_nodes() -> None: + g = nx.DiGraph() + g.add_edge('a', 'b') + g.add_edge('b', 'c') + + graph = cb.Graph(g) + mapped = graph.map({'a': [1, 2, 3]}) + # Special case: The graph we are setting has mapped node values associated with its + # sink node. The setitem effectively renames a to b in the graph, this ensures we + # are also renaming/preserving the associated node values. This is different from + # regular node attributes since mapped node values are not stored as node data in + # the underlying NetworkX graph, but in a separate data structure. + mapped['b'] = mapped['a'] + + result = mapped.to_networkx() + assert result.nodes[idx('b', 0)] == {'value': 1} + assert result.nodes[idx('b', 1)] == {'value': 2} + assert result.nodes[idx('b', 2)] == {'value': 3} + assert len(result.nodes) == 3 * 2 + + def test_getitem_returns_graph_containing_only_key_and_ancestors() -> None: g = nx.DiGraph() g.add_edge('a', 'b') From 42a14935eb31cd8e9e4d0ccf1390841e5cf1013b Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Thu, 21 Aug 2025 10:02:55 +0200 Subject: [PATCH 29/33] Update tests --- src/cyclebane/node_values.py | 4 +++- tests/graph_test.py | 10 ++++------ tests/node_values_test.py | 32 ++++++++++++++++++++++++++------ 3 files changed, 33 insertions(+), 13 deletions(-) diff --git a/src/cyclebane/node_values.py b/src/cyclebane/node_values.py index bd37eb2..fe32fc0 100644 --- a/src/cyclebane/node_values.py +++ b/src/cyclebane/node_values.py @@ -64,7 +64,9 @@ def __setitem__(self, key: Hashable, value_array: ValueArray) -> None: value_array.indices.values(), strict=True, ): - if any(i != j for i, j in zip(old_index, new_index, strict=True)): + if (len(old_index) != len(new_index)) or any( + i != j for i, j in zip(old_index, new_index, strict=True) + ): raise ValueError( f"Node '{key}' has already been mapped with different " f"indices: existing {old_index} vs new {new_index}" diff --git a/tests/graph_test.py b/tests/graph_test.py index 6ccd58c..b25448c 100644 --- a/tests/graph_test.py +++ b/tests/graph_test.py @@ -923,16 +923,14 @@ def test_setitem_allows_compatible_node_values(node_values) -> None: assert len(mapped.index_names) == 1 -def test_setitem_raises_if_node_values_equivalent_but_of_different_type() -> None: +def test_setitem_allows_changing_node_values() -> None: g = nx.DiGraph() g.add_edge('a', 'b') graph = cb.Graph(g) mapped1 = graph.map({'a': [1, 2]}).reduce('b', name='d') - mapped2 = graph.map({'a': np.array([1, 2])}).reduce('b', name='d') - # One could imagine treating this as equivalent, but we are strict in the - # comparison. - with pytest.raises(ValueError, match="Node 'a' has already been mapped"): - mapped1['x'] = mapped2['d'] + mapped2 = graph.map({'a': [1, 3]}).reduce('b', name='d') + mapped1['x'] = mapped2['d'] + assert len(mapped1.index_names) == 1 def test_setitem_raises_if_node_values_incompatible() -> None: diff --git a/tests/node_values_test.py b/tests/node_values_test.py index c65c82d..6d64eff 100644 --- a/tests/node_values_test.py +++ b/tests/node_values_test.py @@ -147,15 +147,16 @@ def test_merge_existing_node_equal_but_different_object(self): assert len(merged) == 1 assert merged is not node_values # Should return new object (copy) - def test_merge_existing_node_different_value_raises(self): + def test_merge_existing_node_different_value_works(self): """Re-adding existing node with different value.""" initial_values = {'a': ValueArray.from_array_like([1, 2, 3], axis_zero=0)} node_values = NodeValues(initial_values) new_values = {'a': ValueArray.from_array_like([4, 5, 6], axis_zero=0)} - with pytest.raises(ValueError, match="Node 'a' has already been mapped"): - node_values.merge(new_values) + merged = node_values.merge(new_values) + assert len(merged) == 1 + assert merged['a'] == new_values['a'] # Should update value def test_merge_empty_new_values(self): """Merging empty mapping.""" @@ -194,7 +195,7 @@ def test_merge_multiple_new_nodes_mixed_conflicts_raises(self): ): node_values.merge(new_values) - def test_merge_multiple_new_nodes_one_existing_raises(self): + def test_merge_multiple_new_nodes_one_existing_compatible_index(self): """Multiple new nodes where one already exists.""" initial_values = {'a': ValueArray.from_array_like([1, 2, 3], axis_zero=0)} node_values = NodeValues(initial_values) @@ -202,11 +203,30 @@ def test_merge_multiple_new_nodes_one_existing_raises(self): new_values = { 'a': ValueArray.from_array_like( [4, 5, 6], axis_zero=0 - ), # Exists but different + ), # Exists but different values 'b': ValueArray.from_array_like([7, 8, 9], axis_zero=1), # New } - with pytest.raises(ValueError, match="Node 'a' has already been mapped"): + merged = node_values.merge(new_values) + assert len(merged) == 2 + assert set(merged.keys()) == {'a', 'b'} + assert merged['a'].sel((('dim_0', 0),)) == 4 # Updated value + + def test_merge_multiple_new_nodes_one_existing_raises(self): + """Multiple new nodes where one already exists.""" + initial_values = {'a': ValueArray.from_array_like([1, 2, 3], axis_zero=0)} + node_values = NodeValues(initial_values) + + new_values = { + 'a': ValueArray.from_array_like([4, 5, 6, 8], axis_zero=0)[ + {'dim_0': slice(1, 4)} # Make a slice to obtain an index value conflict + ], # Exists but different + 'b': ValueArray.from_array_like([7, 8, 9], axis_zero=1), # New + } + + with pytest.raises( + ValueError, match="Node 'a' has already been mapped with different indices" + ): node_values.merge(new_values) def test_merge_partial_index_overlap_compatible(self): From 91cc000d5bf6186511da620803ee4a6826a8ab19 Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Thu, 21 Aug 2025 10:38:49 +0200 Subject: [PATCH 30/33] Test grouping twice --- tests/groupby_test.py | 108 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 107 insertions(+), 1 deletion(-) diff --git a/tests/groupby_test.py b/tests/groupby_test.py index 75d35c3..b11bd49 100644 --- a/tests/groupby_test.py +++ b/tests/groupby_test.py @@ -4,7 +4,6 @@ import networkx as nx import pandas as pd -import pytest import cyclebane as cb @@ -48,6 +47,113 @@ def test_basic_map_groupby_reduce_gives_correct_graph_structure() -> None: assert not result.has_edge(idx('c', 2), idx('d', 'a', dims=('b',))) +def test_group_twice_in_same_path() -> None: + g1 = nx.DiGraph() + g1.add_edge('a', 'c') + g1.add_edge('param1', 'c') + g1.add_edge('c', 'd') + + g2 = nx.DiGraph() + g2.add_edge('e', 'f') + g2.add_edge('param2', 'f') + + grouped = ( + cb.Graph(g1) + .map(pd.DataFrame({'a': [11, 22, 33, 44], 'param1': ['x', 'x', 'y', 'z']})) + .groupby('param1') + .reduce('d', name='grouped-d') + ) + mapped = cb.Graph(g2).map( + pd.DataFrame( + {'e': [1, 2, 3], 'param2': [0, 1, 1], 'param1': ['x', 'y', 'z']} + ).set_index('param1') + ) + + mapped['e'] = grouped + grouped_twice = mapped.groupby('param2').reduce('f', name='grouped-f') + + result = grouped_twice.to_networkx() + + # Nodes from second grouping (grouped-f) + assert result.nodes[idx('grouped-f', 0, dims=('param2',))] == {} + assert result.nodes[idx('grouped-f', 1, dims=('param2',))] == {} + + # Nodes from first grouping / mapping over param1 + assert result.nodes[idx('param2', 'x', dims=('param1',))] == {'value': 0} + assert result.nodes[idx('param2', 'y', dims=('param1',))] == {'value': 1} + assert result.nodes[idx('param2', 'z', dims=('param1',))] == {'value': 1} + assert result.nodes[idx('f', 'x', dims=('param1',))] == {} + assert result.nodes[idx('f', 'y', dims=('param1',))] == {} + assert result.nodes[idx('f', 'z', dims=('param1',))] == {} + # No value on 'e', was replaced by grouped-d links + assert result.nodes[idx('e', 'x', dims=('param1',))] == {} + assert result.nodes[idx('e', 'y', dims=('param1',))] == {} + assert result.nodes[idx('e', 'z', dims=('param1',))] == {} + assert idx('grouped-d', 'x', dims=('param1',)) not in result.nodes + assert idx('grouped-d', 'y', dims=('param1',)) not in result.nodes + assert idx('grouped-d', 'z', dims=('param1',)) not in result.nodes + + # Nodes from mapping over dim_0 + assert result.nodes[idx('a', 0)] == {'value': 11} + assert result.nodes[idx('a', 1)] == {'value': 22} + assert result.nodes[idx('a', 2)] == {'value': 33} + assert result.nodes[idx('a', 3)] == {'value': 44} + assert result.nodes[idx('param1', 0)] == {'value': 'x'} + assert result.nodes[idx('param1', 1)] == {'value': 'x'} + assert result.nodes[idx('param1', 2)] == {'value': 'y'} + assert result.nodes[idx('param1', 3)] == {'value': 'z'} + assert result.nodes[idx('c', 0)] == {} + assert result.nodes[idx('c', 1)] == {} + assert result.nodes[idx('c', 2)] == {} + assert result.nodes[idx('c', 3)] == {} + assert result.nodes[idx('d', 0)] == {} + assert result.nodes[idx('d', 1)] == {} + assert result.nodes[idx('d', 2)] == {} + assert result.nodes[idx('d', 3)] == {} + + # Edges within dim_0 (original graph structure) + assert result.has_edge(idx('a', 0), idx('c', 0)) + assert result.has_edge(idx('a', 1), idx('c', 1)) + assert result.has_edge(idx('a', 2), idx('c', 2)) + assert result.has_edge(idx('a', 3), idx('c', 3)) + assert result.has_edge(idx('param1', 0), idx('c', 0)) + assert result.has_edge(idx('param1', 1), idx('c', 1)) + assert result.has_edge(idx('param1', 2), idx('c', 2)) + assert result.has_edge(idx('param1', 3), idx('c', 3)) + assert result.has_edge(idx('c', 0), idx('d', 0)) + assert result.has_edge(idx('c', 1), idx('d', 1)) + assert result.has_edge(idx('c', 2), idx('d', 2)) + assert result.has_edge(idx('c', 3), idx('d', 3)) + + # Edges within param1 dimension (second graph structure) + assert result.has_edge( + idx('param2', 'x', dims=('param1',)), idx('f', 'x', dims=('param1',)) + ) + assert result.has_edge( + idx('param2', 'y', dims=('param1',)), idx('f', 'y', dims=('param1',)) + ) + assert result.has_edge( + idx('param2', 'z', dims=('param1',)), idx('f', 'z', dims=('param1',)) + ) + + # Edges from dim_0 to param1 grouping (first groupby) + assert result.has_edge(idx('d', 0), idx('e', 'x', dims=('param1',))) + assert result.has_edge(idx('d', 1), idx('e', 'x', dims=('param1',))) + assert result.has_edge(idx('d', 2), idx('e', 'y', dims=('param1',))) + assert result.has_edge(idx('d', 3), idx('e', 'z', dims=('param1',))) + + # Edges from param1 to param2 grouping (second groupby) + assert result.has_edge( + idx('f', 'x', dims=('param1',)), idx('grouped-f', 0, dims=('param2',)) + ) + assert result.has_edge( + idx('f', 'y', dims=('param1',)), idx('grouped-f', 1, dims=('param2',)) + ) + assert result.has_edge( + idx('f', 'z', dims=('param1',)), idx('grouped-f', 1, dims=('param2',)) + ) + + def test_group_in_different_ways() -> None: g = nx.DiGraph() g.add_edge('a', 'b') From 082d5865e4b36181ad74ff70717528b2ff82231e Mon Sep 17 00:00:00 2001 From: Simon Heybrock Date: Thu, 21 Aug 2025 10:46:20 +0200 Subject: [PATCH 31/33] Explain --- src/cyclebane/graph.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index ca4fb08..3f5585d 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -376,14 +376,24 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: # to set the former (user-provided) on (input) nodes. node_values = self._node_values.copy() groupby_graphs = [] + # Handle groupby/reduce operations. The regular iterative node duplication does + # not work in this case. We have to handle the graph edges that correspond to + # a particular groupby/reduce operation in isolation, or else we get broken + # result in the presence of multiple (chained or not) groupby operations. The + # resulting graphs that correspond to the grouping are later composed with the + # rest of the graph. for key, values in self._node_values.items(): if (grouping := values.get_grouping()) is not None: del node_values[key] key = self._from_orig_key(key) + # Note there should be only a single predecessor for the grouping node. groupby_graph = graph.subgraph([*graph.predecessors(key), key]).copy() + # Remove edges, or the loop for the regular map/reduce will add + # all-to-all edges between these nodes graph.remove_edges_from(groupby_graph.edges) groupby_graphs.append(self._make_groupby_graph(grouping, groupby_graph)) + # Handle regular map/reduce operations for index_name, index in reversed(self.indices.items()): graphs = _clone_graph(graph, index_name, index) graph = nx.compose_all(graphs) From c10be9fb740ee6e45828c94f23bb40984b79c10b Mon Sep 17 00:00:00 2001 From: Neil Vaytet Date: Mon, 8 Sep 2025 11:15:29 +0200 Subject: [PATCH 32/33] add match_index option to filter out unwanted matches --- src/cyclebane/graph.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 3f5585d..33c7ea2 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -340,7 +340,9 @@ def reduce( return Graph(graph, node_values=self._node_values) - def _from_orig_key(self, key: Hashable) -> Hashable: + def _from_orig_key( + self, key: Hashable, match_index: None | Hashable = None + ) -> Hashable: # Graph.map relabels nodes to include index names, which can be inconvenient # for the user. Is this convenience of finding the node by its original name # worth the complexity and a good idea? @@ -350,6 +352,8 @@ def _from_orig_key(self, key: Hashable) -> Hashable: for node in self.graph.nodes if isinstance(node, MappedNode) and node.name == key ] + if match_index is not None: + matches = [node for node in matches if match_index in node.indices] if len(matches) == 0: raise KeyError(f"Node '{key}' does not exist in the graph.") if len(matches) > 1: @@ -385,7 +389,7 @@ def to_networkx(self, value_attr: str = 'value') -> nx.DiGraph: for key, values in self._node_values.items(): if (grouping := values.get_grouping()) is not None: del node_values[key] - key = self._from_orig_key(key) + key = self._from_orig_key(key, match_index=grouping.group_index_name) # Note there should be only a single predecessor for the grouping node. groupby_graph = graph.subgraph([*graph.predecessors(key), key]).copy() # Remove edges, or the loop for the regular map/reduce will add From 8629a089b1cccc5fb13e47835173977caf39b47b Mon Sep 17 00:00:00 2001 From: Neil Vaytet Date: Mon, 8 Sep 2025 11:21:35 +0200 Subject: [PATCH 33/33] add contains method and remove node values of the branch if they exist --- src/cyclebane/graph.py | 3 +++ src/cyclebane/node_values.py | 4 ++++ 2 files changed, 7 insertions(+) diff --git a/src/cyclebane/graph.py b/src/cyclebane/graph.py index 33c7ea2..78a1656 100644 --- a/src/cyclebane/graph.py +++ b/src/cyclebane/graph.py @@ -525,6 +525,9 @@ def __setitem__(self, branch: Hashable | slice, other: Graph) -> None: # Delay setting graph until we know no step fails self._node_values = self._node_values.merge(other._node_values) + # Remove node values of the branch, if they exist + if _node_name(branch) in self._node_values: + del self._node_values[_node_name(branch)] # Ensure we preserve the node values of the branch, if it exists. This step is # necessary since __setitem__ effectively renames the sink node of the input diff --git a/src/cyclebane/node_values.py b/src/cyclebane/node_values.py index fe32fc0..274a81f 100644 --- a/src/cyclebane/node_values.py +++ b/src/cyclebane/node_values.py @@ -28,6 +28,10 @@ def __init__(self, values: Mapping[Any, ValueArray]): merged = self.merge(values) self._values = merged._values + def __contains__(self, key: Hashable) -> bool: + """Return True if the column with the given name exists.""" + return key in self._values + def copy(self) -> NodeValues: """Return a copy of the NodeValues.""" return NodeValues(dict(self._values))