Aggregation#

Aggregation strategies for combining node and edge properties.

Basic Strategies#

npap.aggregation.basic_strategies.build_node_to_cluster_map(partition_map)[source]

Build a reverse mapping from node ID to cluster ID.

Args:: partition_map: Dict mapping cluster_id -> list of node IDs

Return type:: dict[Any, int]
Returns:: Dict mapping node_id -> cluster_id for O(1) lookups
Parameters:: partition_map (dict[int, list[Any]])

npap.aggregation.basic_strategies.build_cluster_edge_map(graph, node_to_cluster)[source]

Build a mapping from cluster pairs to their original edge data.

Single pass over all edges - O(E) complexity.

Args:: graph: Original NetworkX graph node_to_cluster: Mapping from node_id -> cluster_id

Return type:

dict[tuple[int, int], list[dict[str, Any]]]

Returns:

Dict mapping (source_cluster, target_cluster) -> list of edge attribute dicts

Parameters:

graph (DiGraph)
node_to_cluster (dict[Any, int])

npap.aggregation.basic_strategies.build_typed_cluster_edge_map(graph, node_to_cluster, type_attribute='type')[source]

Build cluster edge maps grouped by edge type.

Like build_cluster_edge_map but returns a nested dict keyed first by the value of the type_attribute on each edge, then by (source_cluster, target_cluster) pair. Edges without the type attribute are collected under the key "_untyped".

Single pass over all edges – O(E) complexity.

Parameters:

graph (nx.DiGraph) – Original NetworkX graph.
node_to_cluster (dict[Any, int]) – Mapping from node_id to cluster_id.
type_attribute (str) – Edge attribute that stores the type label (default "type").

Returns:

Nested mapping: edge_type -> (src_cluster, tgt_cluster) -> [edge_data].

Return type:

dict[str, dict[tuple[int, int], list[dict[str, Any]]]]

npap.aggregation.basic_strategies.build_cluster_connectivity_set(graph, node_to_cluster)[source]

Build a set of connected cluster pairs from original graph edges.

Single pass over all edges - O(E) complexity.

Args:: graph: Original NetworkX graph node_to_cluster: Mapping from node_id -> cluster_id

Return type:

set[tuple[int, int]]

Returns:

Set of (source_cluster, target_cluster) tuples where edges exist

Parameters:

graph (DiGraph)
node_to_cluster (dict[Any, int])

class npap.aggregation.basic_strategies.SimpleTopologyStrategy[source]

Bases: TopologyStrategy

Simple topology: one node per cluster, edges only where original connections exist.

This is the most basic topology strategy:

Creates one node per cluster
Creates a directed edge between two clusters only if there was at least one directed edge between nodes in those clusters in the original graph
Preserves edge direction from original graph
Does NOT create new edges

Attributes:

can_create_new_edges: Simple topology does not create new edges.

Methods

create_topology(graph, partition_map)

Create aggregated topology with basic node and edge mapping.

create_topology(graph, partition_map)[source]

Create aggregated topology with basic node and edge mapping.

Return type:

DiGraph

Parameters:

graph (DiGraph)
partition_map (dict[int, list[Any]])

property can_create_new_edges: bool: Simple topology does not create new edges.

class npap.aggregation.basic_strategies.ElectricalTopologyStrategy[source]

Bases: TopologyStrategy

Electrical topology: may create fully connected or partially connected topology for subsequent physical aggregation (e.g., Kron reduction).

This topology strategy is designed for electrical networks where: - Physical coupling may exist even without direct edges - Kron reduction or other physical methods will determine final connectivity - May start with fully connected graph and let physical strategy prune

Attributes:

can_create_new_edges: Electrical topology may create new edges depending on connectivity mode.

Methods

create_topology(graph, partition_map)

Create topology suitable for electrical aggregation.

__init__(initial_connectivity='existing')[source]

Initialize electrical topology strategy.

Args:

initial_connectivity: How to initialize edges

“existing”: Only edges where original connections exist (like simple)
“full”: Fully connected (all cluster pairs connected)

Parameters:: initial_connectivity (str)

create_topology(graph, partition_map)[source]

Create topology suitable for electrical aggregation.

Return type:

DiGraph

Parameters:

graph (DiGraph)
partition_map (dict[int, list[Any]])

property can_create_new_edges: bool: Electrical topology may create new edges depending on connectivity mode.

class npap.aggregation.basic_strategies.SumNodeStrategy[source]

Bases: NodePropertyStrategy

Sum numerical properties across nodes in a cluster.

Methods

aggregate_property(graph, nodes, property_name)

Sum property values across nodes using NumPy.

aggregate_property(graph, nodes, property_name)[source]

Sum property values across nodes using NumPy.

Return type:

Any

Parameters:

graph (DiGraph)
nodes (list[Any])
property_name (str)

class npap.aggregation.basic_strategies.AverageNodeStrategy[source]

Bases: NodePropertyStrategy

Average numerical properties across nodes in a cluster.

Methods

aggregate_property(graph, nodes, property_name)

Average property values across nodes using NumPy.

aggregate_property(graph, nodes, property_name)[source]

Average property values across nodes using NumPy.

Return type:

Any

Parameters:

graph (DiGraph)
nodes (list[Any])
property_name (str)

class npap.aggregation.basic_strategies.FirstNodeStrategy[source]

Bases: NodePropertyStrategy

Take the first available value for non-numerical properties.

Methods

aggregate_property(graph, nodes, property_name)

Take first available property value.

aggregate_property(graph, nodes, property_name)[source]

Take first available property value.

Return type:

Any

Parameters:

graph (DiGraph)
nodes (list[Any])
property_name (str)

class npap.aggregation.basic_strategies.SumEdgeStrategy[source]

Bases: EdgePropertyStrategy

Sum numerical properties across edges.

Attributes:

required_attributes: Return required attributes for this property strategy.

Methods

aggregate_property(original_edges, property_name)

Sum property values across edges using NumPy.

aggregate_property(original_edges, property_name)[source]

Sum property values across edges using NumPy.

Return type:

Any

Parameters:

original_edges (list[dict[str, Any]])
property_name (str)

class npap.aggregation.basic_strategies.AverageEdgeStrategy[source]

Bases: EdgePropertyStrategy

Average numerical properties across edges.

Attributes:

required_attributes: Return required attributes for this property strategy.

Methods

aggregate_property(original_edges, property_name)

Average property values across edges using NumPy.

aggregate_property(original_edges, property_name)[source]

Average property values across edges using NumPy.

Return type:

Any

Parameters:

original_edges (list[dict[str, Any]])
property_name (str)

class npap.aggregation.basic_strategies.FirstEdgeStrategy[source]

Bases: EdgePropertyStrategy

Take the first available value for non-numerical properties.

Attributes:

required_attributes: Return required attributes for this property strategy.

Methods

aggregate_property(original_edges, property_name)

Take first available property value.

aggregate_property(original_edges, property_name)[source]

Take first available property value.

Return type:

Any

Parameters:

original_edges (list[dict[str, Any]])
property_name (str)

class npap.aggregation.basic_strategies.EquivalentReactanceStrategy[source]

Bases: EdgePropertyStrategy

Aggregates reactance for a set of parallel edges.

This strategy calculates the equivalent reactance by first converting each edge’s reactance (x) to susceptance (b = 1/x), summing the susceptances, and then converting the total susceptance back to an equivalent reactance (x_eq = 1 / b_eq).

This correctly models the physics of parallel lines as: b_eq = b_1 + b_2 + … x_eq = 1 / b_eq

Attributes:

required_attributes: Return required attributes for this property strategy.

Methods

aggregate_property(original_edges, property_name)

Calculate the equivalent reactance for the given parallel edges.

aggregate_property(original_edges, property_name)[source]

Calculate the equivalent reactance for the given parallel edges.

Uses NumPy for vectorized susceptance calculation.

Return type:

Any

Returns:

Equivalent reactance value, or float(‘inf’) if non-numeric values present

Raises:

AggregationError – If no edges have the property at all (entirely missing attribute).

Parameters:

original_edges (list[dict[str, Any]])
property_name (str)

Aggregation Modes#

Pre-defined aggregation modes for common use cases

Each mode provides a validated combination of: - Topology strategy - Physical aggregation strategy (if applicable) - Statistical property aggregation defaults

npap.aggregation.modes.get_mode_profile(mode, **overrides)[source]

Get pre-configured aggregation profile for a given mode.

Parameters:

mode (AggregationMode) – Aggregation mode enum.
**overrides – Override any profile parameters.

Returns:

AggregationProfile configured for the mode.

Return type:

AggregationProfile

Examples

>>> profile = get_mode_profile(AggregationMode.GEOGRAPHICAL)

Physical Strategies#

class npap.aggregation.physical_strategies.KronReductionStrategy[source]

Bases: PhysicalAggregationStrategy

Kron reduction for DC power flow networks

TODO: Implementation pending. This is a placeholder.

Attributes:

can_create_edges: Check whether this strategy can create new edges.
modifies_properties: Return properties modified by this physical strategy.
required_properties: Return properties this strategy requires.
required_topology: Return required topology strategy for this physical aggregation.

Methods

aggregate(original_graph, partition_map, ...)

Kron reduction - TO BE IMPLEMENTED

property required_properties: list[str]

Return properties this strategy requires.

Returns:: List of required property names (e.g., [‘reactance’, ‘resistance’]).
Return type:: list[str]

property modifies_properties: list[str]

Return properties modified by this physical strategy.

These properties should NOT be aggregated statistically afterward, as they are already handled by the physical strategy.

Returns:: List of property names modified by this strategy.
Return type:: list[str]

property can_create_edges: bool

Check whether this strategy can create new edges.

Returns:: True if strategy can create new edges.
Return type:: bool

property required_topology: str

Return required topology strategy for this physical aggregation.

Returns:: Name of the required topology strategy.
Return type:: str

aggregate(original_graph, partition_map, topology_graph, properties, parameters=None)[source]

Kron reduction - TO BE IMPLEMENTED

Return type:

Graph

Parameters:

original_graph (Graph)
partition_map (dict[int, list[Any]])
topology_graph (Graph)
properties (list[str])
parameters (dict[str, Any])