Interfaces#

Abstract base classes and data structures that define the NPAP strategy pattern.

Data Classes#

class npap.interfaces.AggregationProfile[source]

Bases: object

Configure aggregation separating physical from statistical operations.

This profile distinguishes between:

Topology: How the graph structure is reduced
Physical: Electrical laws that must be preserved (coupled properties)
Statistical: Simple operations on independent properties

topology_strategy

Strategy for graph structure reduction (“simple”, “electrical”).

Type:: str

physical_strategy

Physical aggregation strategy (“kron_reduction”, “equivalent_impedance”).

Type:: str or None

physical_properties

Properties handled by physical strategy (e.g., [“reactance”, “resistance”]).

Type:: list[str]

physical_parameters

Additional parameters for physical strategies.

Type:: dict[str, Any]

node_properties

Mapping of node property names to aggregation strategies.

Type:: dict[str, str]

edge_properties

Mapping of edge property names to aggregation strategies.

Type:: dict[str, str]

edge_type_properties

Per-edge-type strategy overrides. Maps edge type values (from the edge_type_attribute attribute on edges) to per-type property strategy dicts. When populated, edges are aggregated separately per type and the result graph is a MultiDiGraph with one edge per type per cluster pair. Falls back to edge_properties for edges whose type is not listed here.

Type:: dict[str, dict[str, str]]

edge_type_attribute

Name of the edge attribute that stores the type label. Only used when edge_type_properties is populated. Defaults to "type".

Type:: str

default_node_strategy

Default strategy for unspecified node properties.

Type:: str

default_edge_strategy

Default strategy for unspecified edge properties.

Type:: str

warn_on_defaults

Whether to warn when using default strategies.

Type:: bool

mode

Indicator of which pre-defined mode is being used.

Type:: AggregationMode

Attributes:

physical_strategy

See npap.AggregationProfile for full member documentation.

topology_strategy: str = 'simple'

physical_strategy: str | None = None

physical_properties: list[str]

physical_parameters: dict[str, Any]

node_properties: dict[str, str]

edge_properties: dict[str, str]

edge_type_properties: dict[str, dict[str, str]]

edge_type_attribute: str = 'type'

default_node_strategy: str = 'average'

default_edge_strategy: str = 'sum'

warn_on_defaults: bool = True

mode: AggregationMode = 'custom'

__init__(topology_strategy='simple', physical_strategy=None, physical_properties=<factory>, physical_parameters=<factory>, node_properties=<factory>, edge_properties=<factory>, edge_type_properties=<factory>, edge_type_attribute='type', default_node_strategy='average', default_edge_strategy='sum', warn_on_defaults=True, mode=AggregationMode.CUSTOM)

Parameters:

topology_strategy (str)
physical_strategy (str | None)
physical_properties (list[str])
physical_parameters (dict[str, Any])
node_properties (dict[str, str])
edge_properties (dict[str, str])
edge_type_properties (dict[str, dict[str, str]])
edge_type_attribute (str)
default_node_strategy (str)
default_edge_strategy (str)
warn_on_defaults (bool)
mode (AggregationMode)

Return type:

None

class npap.interfaces.AggregationMode[source]

Bases: Enum

Define pre-defined aggregation modes for common use cases.

SIMPLE

Sum/average everything with simple topology.

Type:: str

GEOGRAPHICAL

Average coordinates, sum other properties.

Type:: str

DC_KRON

Kron reduction for DC networks.

Type:: str

CUSTOM

User-defined aggregation profile.

Type:: str

See npap.AggregationMode for full member documentation.

SIMPLE = 'simple'

GEOGRAPHICAL = 'geographical'

DC_KRON = 'dc_kron'

CUSTOM = 'custom'

class npap.interfaces.PartitionResult[source]

Bases: object

Store partition result with metadata for validation and tracking.

mapping

Dictionary mapping cluster_id to list of node_ids.

Type:: dict[int, list[Any]]

original_graph_hash

Hash of the original graph for compatibility validation.

Type:: str

strategy_name

Name of the partitioning strategy used.

Type:: str

strategy_metadata

Strategy-specific metadata and parameters.

Type:: dict[str, Any]

n_clusters

Number of clusters created.

Type:: int

See npap.PartitionResult for full member documentation.

mapping: dict[int, list[Any]]

original_graph_hash: str

strategy_name: str

strategy_metadata: dict[str, Any]

n_clusters: int

__init__(mapping, original_graph_hash, strategy_name, strategy_metadata, n_clusters)

Parameters:

mapping (dict[int, list[Any]])
original_graph_hash (str)
strategy_name (str)
strategy_metadata (dict[str, Any])
n_clusters (int)

Return type:

None

Strategy Interfaces#

class npap.interfaces.DataLoadingStrategy[source]#

Bases: ABC

Define interface for data loading strategies.

Methods

`load`(**kwargs)	Load data and return a NetworkX directed graph.
`validate_inputs`(**kwargs)	Validate input parameters before loading.

abstractmethod load(**kwargs)[source]#

Load data and return a NetworkX directed graph.

Returns:: Loaded network graph.
Return type:: nx.DiGraph or nx.MultiDiGraph

abstractmethod validate_inputs(**kwargs)[source]#

Validate input parameters before loading.

Returns:: True if validation passes.
Return type:: bool
Raises:: DataLoadingError – If validation fails.

class npap.interfaces.PartitioningStrategy[source]#

Bases: ABC

Define interface for all partitioning algorithms.

Attributes:

required_attributes: Return required node/edge attributes.

Methods

partition(graph, **kwargs)

Partition nodes into clusters.

abstractmethod partition(graph, **kwargs)[source]#

Partition nodes into clusters.

Parameters:

graph (nx.DiGraph) – Input graph to partition.
**kwargs (dict) – Strategy-specific parameters.

Returns:

Mapping of cluster_id to list of node_ids.

Return type:

dict[int, list[Any]]

abstract property required_attributes: dict[str, list[str]]#

Return required node/edge attributes.

Returns:: Dictionary with ‘nodes’ and ‘edges’ keys listing required attributes.
Return type:: dict[str, list[str]]

class npap.interfaces.TopologyStrategy[source]#

Bases: ABC

Define interface for topology strategies.

Topology strategies create the skeleton of the aggregated graph:

Which nodes exist in the aggregated graph
Which edges connect them
NO property aggregation at this stage

Attributes:

can_create_new_edges: Check whether this strategy can create edges not in original graph.

Methods

create_topology(graph, partition_map)

Create aggregated graph structure without properties.

abstractmethod create_topology(graph, partition_map)[source]#

Create aggregated graph structure without properties.

Parameters:

graph (nx.DiGraph) – Original directed graph.
partition_map (dict[int, list[Any]]) – Mapping of cluster_id to list of original node ids.

Returns:

DiGraph with aggregated topology (nodes and edges, no attributes).

Return type:

nx.DiGraph

property can_create_new_edges: bool#

Check whether this strategy can create edges not in original graph.

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

class npap.interfaces.NodePropertyStrategy[source]#

Bases: ABC

Define interface for node property aggregation strategies.

Methods

aggregate_property(graph, nodes, property_name)

Aggregate a specific property across nodes.

abstractmethod aggregate_property(graph, nodes, property_name)[source]#

Aggregate a specific property across nodes.

Parameters:

graph (nx.DiGraph) – The graph containing node properties.
nodes (list[Any]) – List of node identifiers to aggregate.
property_name (str) – Name of the property to aggregate.

Returns:

Aggregated property value.

Return type:

Any

class npap.interfaces.EdgePropertyStrategy[source]#

Bases: ABC

Define interface for edge property aggregation strategies.

Attributes:

required_attributes: Return required attributes for this property strategy.

Methods

aggregate_property(original_edges, property_name)

Aggregate a specific property across edges.

abstractmethod aggregate_property(original_edges, property_name)[source]#

Aggregate a specific property across edges.

Parameters:

original_edges (list[dict[str, Any]]) – List of edge attribute dictionaries.
property_name (str) – Name of the property to aggregate.

Returns:

Aggregated property value.

Return type:

Any

property required_attributes: dict[str, list[str]]#

Return required attributes for this property strategy.

Returns:: Dictionary with ‘nodes’ and ‘edges’ keys listing required attributes.
Return type:: dict[str, list[str]]

class npap.interfaces.PhysicalAggregationStrategy[source]#

Bases: ABC

Define interface for physics-aware aggregation strategies.

These strategies operate on the entire graph and respect physical laws. They work on coupled properties (e.g., reactance and resistance together). They may create new edges based on electrical coupling.

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, ...)

Apply physical aggregation to the topology graph.

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

Apply physical aggregation to the topology graph.

Parameters:

original_graph (nx.DiGraph) – Full resolution directed graph with all properties.
partition_map (dict[int, list[Any]]) – Mapping of cluster_id to list of node_ids.
topology_graph (nx.DiGraph) – DiGraph with aggregated structure but no properties yet.
properties (list[str]) – Physical properties to aggregate (coupled).
parameters (dict[str, Any], optional) – Additional parameters for the strategy.

Returns:

DiGraph with physical properties correctly aggregated.

Return type:

nx.DiGraph

abstract property required_properties: list[str]#

Return properties this strategy requires.

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

abstract 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