"""
Base class for partially temporal networks
"""
import numpy as np
import pandas as pd
from phasik.classes.TemporalNetwork import TemporalNetwork, _process_input_tedges
__all__ = ["PartiallyTemporalNetwork"]
[docs]class PartiallyTemporalNetwork(TemporalNetwork):
"""Base class for partially temporal networks
Partially temporal networks are temporal networks for which we do not have
temporal information about all edges.
Attributes
----------
nodes : list of (str or int)
Sorted list of node names. Node names can be either strings or integers,
but they all need to be of the same type.
times : list of (int or float)
Sorted list of times for which we have temporal information
tedges : pandas.DataFrame
Dataframe containing tedges (potentially weighted).
Columns are ['i', 'j', 't', ('weight')] and each row represents a tedge.
snapshots : numpy array
Array of shape (T, N, N) storing the instantaneous values of
the adjacency matrix A_{ij}(t).
temporal_nodes : list of (str or int)
List of nodes that are part of a temporal edge
temporal_edges : list of tuples
List of edges for which we have temporal information
"""
def __init__(self):
super().__init__()
self._temporal_nodes = []
self._temporal_edges = []
@property
def nodes(self):
return super().nodes
@property
def tedges(self):
return super().tedges
@property
def snapshots(self):
return super().snapshots
@property
def temporal_nodes(self):
return self._temporal_nodes
@temporal_nodes.setter
def temporal_nodes(self, nodes):
self._temporal_nodes = nodes
@property
def temporal_edges(self):
return self._temporal_edges
@temporal_edges.setter
def temporal_edges(self, edges):
self._temporal_edges = edges
[docs] def temporal_neighbors(self):
"""Returns a dict of neighbors in the aggregated network that are temporal nodes"""
neighbors = super().neighbors()
return {
node: [u for u in value if u in self.temporal_nodes]
for node, value in neighbors.items()
}
[docs] def number_of_temporal_edges(self):
"""Returns the number of temporal edges in the temporal network"""
return len(self._temporal_edges)
[docs] def number_of_temporal_nodes(self):
"""Returns the number of temporal nodes in the temporal network"""
return len(self._temporal_nodes)
[docs] def fraction_of_temporal_nodes(self):
"""Returns the fraction of temporal edges in the temporal network"""
return self.number_of_temporal_nodes() / self.N()
[docs] def fraction_of_temporal_edges(self):
"""Returns the fraction of temporal edges in the temporal network"""
return self.number_of_temporal_edges() / self.number_of_edges()
@classmethod
def from_tedges(
cls, tedges, temporal_nodes=None, temporal_edges=None, normalise="max"
):
"""Creates a PartiallyTemporalNetwork from a dataframe of tedges
Parameters
----------
tedges : pandas.DataFrame or list of tuples
List of tedges with 'i', 'j', 't', and optionally 'weight'
If DataFrame, these are the name of the columns, and each row contains a tedge
temporal_nodes : list of (str or int)
List of temporal nodes
temporal_edges : list of tuples
List of temporal edges
normalise : {'max', 'minmax'}
Choice of normalsation of the edge timeseries
Returns
-------
TN : PartiallyTemporalNetwork
"""
TN = super().from_tedges(tedges, normalise=normalise)
if temporal_nodes is None:
TN._temporal_nodes = TN.nodes
else:
TN._temporal_nodes = temporal_nodes
if temporal_edges is None:
TN._temporal_edges = TN.edges_aggregated()
else:
TN._temporal_edges = temporal_edges
return TN
@classmethod
def from_edge_timeseries(
cls,
edge_timeseries,
temporal_nodes=None,
temporal_edges=None,
normalise="max",
):
"""Creates a PartiallyTemporalNetwork from a DataFrame of edge timeseries
All edges in the network are those of the timeseries, and nodes are extracted from edge names
Parameters
----------
edge_timeseries : pandas.DataFrame
Dataframe where each row is a timeseries, with index as edge names and columns as times
temporal_nodes : list of (str or int)
List of temporal nodes
temporal_edges : list of tuples
List of temporal edges
normalise : {'max', 'minmax'}
Choice of normalsation of the edge timeseries
Returns
-------
PartiallyTemporalNetwork
"""
TN = super().from_edge_timeseries(edge_timeseries, normalise=normalise)
if temporal_nodes is None:
TN._temporal_nodes = TN.nodes
else:
TN._temporal_nodes = temporal_nodes
if temporal_edges is None:
TN._temporal_edges = TN.edges_aggregated()
else:
TN._temporal_edges = temporal_edges
return TN
@classmethod
def from_node_timeseries(cls, node_timeseries, normalise="max"):
"""Creates a partially temporal network by combining node timeseries into edge timeseries.
By construction, the underlying static network created is always fully connected.
Parameters
----------
node_timeseries : pandas.DataFrame
Timeseries of nodes, indexed by node name and times as columns
normalise : {'max', 'minmax'}
Choice of normalsation of the edge timeseries
Returns
-------
PartiallyTemporalNetwork
"""
TN = super().from_node_timeseries(node_timeseries, normalise=normalise)
TN._temporal_nodes = TN.nodes
TN._temporal_edges = TN.edges_aggregated()
return TN
@classmethod
def from_static_network_and_tedges(
cls,
static_network,
tedges,
static_edge_default_weight=None,
normalise="max",
):
"""Creates a partially temporal network by combining a static network with tedges
Parameters
----------
static_network : networkx.Graph
Static network into which to integrate the temporal information
tedges : pandas.DataFrame or list of tuples
Tedges must be of the form (i, j, t, weight)
static_edge_default_weight : float, optional
Weight to use for edges that have no temporal information
normalise : {'max', 'minmax'}
Choice of normalsation of the edge timeseries
Returns
-------
PartiallyTemporalNetwork
"""
tedges = _process_input_tedges(tedges)
if "weight" not in tedges.columns:
tedges["weight"] = 1 # add column with weight 1
# convert static network's edges to DataFrame
static_network_edges = pd.DataFrame(static_network.edges)
static_network_edges.columns = ["static_i", "static_j"]
# sort nodes in each row, for undirected edges
static_network_edges[["static_i", "static_j"]] = np.sort(
static_network_edges[["static_i", "static_j"]], axis=1
)
tedges[["i", "j"]] = np.sort(tedges[["i", "j"]], axis=1)
# check that all static network edges have temporal info
edges_aggregated = set(tedges[["i", "j"]].itertuples(index=False, name=None))
missing_edges = list(
set(static_network.edges).difference(edges_aggregated)
) # edges with no temporal information
if not missing_edges:
print("INFO: all edges have temporal information. This could be a TempNet.")
# Keep all edges present in the static network, and only those
# Add all time edges corresponding to those
# For edges that have no temporal information (no corresponding tedge), sets t and weight as Nan
tedges_merged = pd.merge(
static_network_edges,
tedges,
how="left",
left_on=["static_i", "static_j"],
right_on=["i", "j"],
)
if (
static_edge_default_weight is None
): # remove all edges without temporal information
tedges_final = tedges_merged.dropna()
temporal_edges = [
edge for edge in edges_aggregated if edge not in missing_edges
]
tedges_final = tedges_final[["i", "j", "t", "weight"]]
else: # add default weight across all timepoints for edges without temporal information
if missing_edges:
times = (
tedges["t"]
.drop_duplicates()
.to_frame()
.assign(weight=static_edge_default_weight)
)
tedges_missing = tedges_merged[tedges_merged["t"].isnull()]
tedges_static = tedges_missing[["static_i", "static_j"]].assign(
weight=static_edge_default_weight
)
tedges_static = tedges_static.merge(times, on="weight")[
["static_i", "static_j", "t", "weight"]
]
tedges_static.columns = ["i", "j", "t", "weight"]
tedges_temporal = tedges_merged.dropna()[
["static_i", "static_j", "t", "weight"]
]
tedges_temporal.columns = ["i", "j", "t", "weight"]
tedges_final = pd.concat([tedges_temporal, tedges_static])
temporal_edges = sorted(
set(tedges_temporal[["i", "j"]].itertuples(index=False, name=None))
)
else: # all edges have temporal information
tedges_final = tedges
temporal_edges = edges_aggregated
temporal_nodes = list(set([node for edge in temporal_edges for node in edge]))
TN = cls.from_tedges(
tedges_final, temporal_nodes, temporal_edges, normalise=normalise
)
return TN