npap.PartitionAggregatorManager#

class npap.PartitionAggregatorManager[source]#

Bases: object

Main orchestrator - the primary class users interact with.

Methods

`aggregate`([partition_result, profile, mode])	Aggregate using partition result and profile.
`aggregate_parallel_edges`([edge_properties, ...])	Convert current MultiDiGraph to simple DiGraph by aggregating parallel edges.
`full_workflow`(data_strategy, partition_strategy)	Execute complete workflow without storing intermediates.
`get_current_graph`()	Get the current graph.
`get_current_partition`()	Get the current partition result.
`group_by_voltage_levels`(target_levels[, ...])	Group bus voltage levels to predefined target values.
`load_data`(strategy, **kwargs)	Load data using specified strategy.
`partition`(strategy, **kwargs)	Partition current graph and store result.
`plot_network`([style, graph, show, config])	Plot the network on an interactive map.

__init__()[source]#

Methods

`__init__`()
`aggregate`([partition_result, profile, mode])	Aggregate using partition result and profile.
`aggregate_parallel_edges`([edge_properties, ...])	Convert current MultiDiGraph to simple DiGraph by aggregating parallel edges.
`full_workflow`(data_strategy, partition_strategy)	Execute complete workflow without storing intermediates.
`get_current_graph`()	Get the current graph.
`get_current_partition`()	Get the current partition result.
`group_by_voltage_levels`(target_levels[, ...])	Group bus voltage levels to predefined target values.
`load_data`(strategy, **kwargs)	Load data using specified strategy.
`partition`(strategy, **kwargs)	Partition current graph and store result.
`plot_network`([style, graph, show, config])	Plot the network on an interactive map.

__init__()[source]#

load_data(strategy, **kwargs)[source]#

Load data using specified strategy.

Return type:: DiGraph | MultiDiGraph
Parameters:: strategy (str)

partition(strategy, **kwargs)[source]#

Partition current graph and store result.

Return type:: PartitionResult
Parameters:: strategy (str)

aggregate(partition_result=None, profile=None, mode=None, **overrides)[source]#

Aggregate using partition result and profile.

Parameters:

partition_result (PartitionResult, optional) – Partition to use (or use stored partition).
profile (AggregationProfile, optional) – Aggregation profile (custom configuration).
mode (AggregationMode, optional) – Aggregation mode (pre-defined configuration).
**overrides (dict) – Override specific profile parameters when using mode.

Returns:

Aggregated graph. Returns a MultiDiGraph when profile.edge_type_properties is populated.

Return type:

nx.DiGraph or nx.MultiDiGraph

Notes

If both profile and mode are provided, profile takes precedence.

aggregate_parallel_edges(edge_properties=None, default_strategy='sum', warn_on_defaults=True)[source]#

Convert current MultiDiGraph to simple DiGraph by aggregating parallel edges.

This method aggregates all parallel edges between the same directed node pairs into single edges, using the specified aggregation strategies. The graph must be a MultiDiGraph for this operation to be meaningful.

For directed graphs, edges (A->B) and (B->A) are treated as separate edges.

Parameters:

edge_properties (dict[str, str], optional) – Dict mapping property names to aggregation strategies. Example: {"reactance": "average", "length": "sum"}.
default_strategy (str) – Strategy to use for properties not specified.
warn_on_defaults (bool) – Whether to warn when using default strategy.

Returns:

Simple DiGraph with parallel edges aggregated.

Return type:

nx.DiGraph

Raises:

ValueError – If no graph is loaded or graph is not a MultiDiGraph.

full_workflow(data_strategy, partition_strategy, aggregation_profile=None, aggregation_mode=None, **kwargs)[source]#

Execute complete workflow without storing intermediates.

If a MultiDiGraph is loaded, parallel edges will be automatically aggregated before partitioning. If a voltage-aware partitioning strategy is selected, voltage levels will be grouped first in 220kV and 380kV voltage levels.

Parameters:

data_strategy (str) – Data loading strategy name.
partition_strategy (str) – Partitioning strategy name.
aggregation_profile (AggregationProfile, optional) – Aggregation profile (custom).
aggregation_mode (AggregationMode, optional) – Aggregation mode (pre-defined).
**kwargs (dict) – Parameters for data loading, parallel edge aggregation, and partitioning.

Returns:

Aggregated graph.

Return type:

nx.DiGraph

Examples

>>> manager = PartitionAggregatorManager()
>>> result = manager.full_workflow(
...     data_strategy="csv_files",
...     partition_strategy="geographical_kmeans",
...     node_file="buses.csv",
...     edge_file="lines.csv",
...     n_clusters=10
... )

get_current_graph()[source]#

Get the current graph.

Returns:: Current graph, or None if no graph is loaded.
Return type:: nx.DiGraph or None

get_current_partition()[source]#

Get the current partition result.

Returns:: Current partition result, or None if not partitioned.
Return type:: PartitionResult or None

plot_network(style='simple', graph=None, show=True, config=None, **kwargs)[source]#

Plot the network on an interactive map.

Parameters:

style (str) –
Plot style:
- ’simple’: All edges same color (fast, minimal)
- ’voltage_aware’: Edges colored by type (line/trafo/dc_link)
- ’clustered’: Nodes colored by cluster assignment
graph (nx.DiGraph, optional) – Graph to plot (uses current graph if not provided).
show (bool) – Whether to display immediately.
config (PlotConfig, optional) – PlotConfig instance to override defaults. If provided, kwargs will further override values from this config.
**kwargs (dict) – Additional configuration options (see PlotConfig for full list).

Returns:

Plotly Figure object.

Return type:

go.Figure

Raises:

ValueError – If no graph is loaded or clustered style without partition.

Examples

>>> manager.plot_network(style="voltage_aware", title="My Network")
>>> manager.plot_network(style="clustered")  # After partitioning

group_by_voltage_levels(target_levels, voltage_attr='voltage', store_original=True, handle_missing='infer')[source]#

Group bus voltage levels to predefined target values.

This method reassigns each node’s voltage to the nearest target voltage level, creating clean voltage “islands” for subsequent voltage-aware partitioning.

Parameters:

target_levels (list[float]) – List of target voltage levels (kV) to harmonize to. Example: [220, 380] for European transmission grid.
voltage_attr (str) – Node attribute containing voltage level.
store_original (bool) – If True, stores original voltage in ‘original_{voltage_attr}’.
handle_missing (str) –
How to handle nodes without voltage data:
- ’infer’: Infer from connected neighbors
- ’nearest’: Assign to nearest target level
- ’error’: Raise an error
- ’skip’: Leave as None

Returns:

Summary dict with:

’total_nodes’: Total number of nodes processed
’reassignments’: Dict mapping original_voltage to (target_voltage, count)
’missing_handled’: Number of nodes with missing voltage
’voltage_distribution’: Dict mapping target_voltage to node_count

Return type:

dict[str, Any]

Raises:

ValueError – If no graph loaded, target_levels empty, or handle_missing=’error’ with missing voltages.

Examples

>>> manager.group_by_voltage_levels([220, 380])
{'total_nodes': 6000, 'voltage_distribution': {220: 3500, 380: 2500}, ...}