flowstab.flow_stability

This sub-module provides tools to perform flow stability analysis on temporal networks using contact sequences. It enables loading temporal network data, computing Laplacian and inter-transition matrices, and extracting flow-based clusterings. It leverages a state-tracking system to ensure that computations are performed in the correct order and that all required data is available before each analysis step.

Key Features

  • State-tracked analysis pipeline for reproducibility and robustness.

  • Support for loading and representing temporal (contact sequence) networks.

  • Computation of Laplacian and inter-transition matrices for random walks.

  • Flow integral clustering and community detection (e.g., Louvain clustering).

  • Flexible time scale and direction support for forward/backward dynamics.

Examples

A typical usage workflow:

>>> from flowstab.flow_stability import FlowStability
>>> fs = FlowStability()
>>> fs.set_temporal_network(events_table="my_contacts.csv")
>>> fs.set_time_scale(10)
>>> fs.compute_laplacian_matrices()
>>> fs.compute_inter_transition_matrices()
>>> fs.time_direction = 0
>>> fs.set_flow_clustering()
>>> fs.find_louvain_clustering()
>>> print(fs.flow_clustering_forward)

Classes

FlowStability

Main class for performing flow stability analysis on a temporal network.

States

Enum of analysis states, used for internal state tracking.

ProcessException

Exception raised for errors in the flow stability process.

Attributes

logger

register

Exceptions

ProcessException

Common base class for all non-exit exceptions.

Classes

FlowStability

Conducts flow stability analysis using a temporal network.

States

Defines the stages of a flow stability analysis

Module Contents

exception flowstab.flow_stability.ProcessException[source]

Bases: Exception

Common base class for all non-exit exceptions.

Initialize self. See help(type(self)) for accurate signature.

class flowstab.flow_stability.FlowStability(*, temporal_network: tempnet.ContTempNetwork | None = None, time_scale: int | float | None = None, t_start: int | float | None = None, t_stop: int | float | None = None, time_direction: int | None = None, **kwargs: Any)[source]

Conducts flow stability analysis using a temporal network.

This class loads a temporal network (contact sequence) and provides methods for analyzing the stability of flows via Laplacian and inter-transition matrices, and clustering.

Raises:
  • ValueError – If inputs are not of valid types or parameters are inconsistent.

  • TypeError – If provided types are not as expected.

  • ProcessException – For errors during the processing steps.

Initialize a flow stability analysis instance.

Parameters:
  • temporal_network (ContTempNetwork or None, optional) – The temporal network data to analyze.

  • time_scale (int, float, or None, optional) – Characteristic time scale for the random walk’s transition rate.

  • t_start (int, float, or None, optional) – Start time for the analysis.

  • t_stop (int, float, or None, optional) – End time for the analysis.

  • time_direction (int or None, optional) – Direction of time for the analysis (-1, 0, or 1).

  • **kwargs (dict) – Additional keyword arguments for initializing the temporal network.

__repr__()[source]
compute_inter_transition_matrices(linear_approx=False, **kwargs)[source]

Compute inter-transition matrices for the temporal network.

Parameters:
  • linear_approx (bool, optional) – If True, use a linear approximation for the computation.

  • **kwargs (dict) – Additional arguments passed to the computation methods.

Returns:

self – The instance itself.

Return type:

FlowStability

compute_laplacian_matrices(**kwargs)[source]

Compute Laplacian matrices for the current temporal network.

Parameters:

**kwargs (dict) – Additional arguments passed to ContTempNetwork’s compute_laplacian_matrices method.

Returns:

self – The instance itself.

Return type:

FlowStability

find_louvain_clustering(**kwargs)[source]

This method finds the Louvain clusters for a flow integral clustering.

Parameters:

**kwargs (dict) – Arguments passed to FlowIntegralClustering’s find_louvain_clustering method.

Returns:

self – The instance itself.

Return type:

FlowStability

help(subject: str | None = None, verbose: bool = False)[source]
set_flow_clustering(**kwargs)[source]

Perform flow integral clustering analysis.

Parameters:

**kwargs (dict) – Arguments passed to FlowIntegralClustering.

Returns:

self – The instance itself.

Return type:

FlowStability

set_temporal_network(**kwargs)[source]

Set the temporal network for the flow stability analysis.

Parameters:

**kwargs (dict) – Arguments to initialize a ContTempNetwork instance. If no arguments are provided, the temporal network is set to None.

Returns:

self – The instance itself.

Return type:

FlowStability

set_time_scale(value: int | float | None = None, **kwargs)[source]

Set the time scale(s) for the analysis.

Parameters:
  • value (int, float, or None, optional) – Characteristic random walk inter-event time. If None and kwargs is empty, the median inter-event time will be used.

  • **kwargs (dict) – Arguments passed to numpy.logspace to generate multiple time scales.

Return type:

None

property flow_clustering_backward[source]

Get the dictionary of backward-time flow integral clustering results.

Returns:

Maps time_scale values to FlowIntegralClustering objects.

Return type:

dict

property flow_clustering_forward[source]

Get the dictionary of forward-time flow integral clustering results.

Returns:

Maps time_scale values to FlowIntegralClustering objects.

Return type:

dict

property lamda[source]

Iterator of lambda values, the inverse of the time scales.

Warning

This attribute is deprecated and will be removed in future versions.

Please use time_scale which is the inverse of lambda, instead.

Returns:

Each value is the inverse of a time scale, or None if time scale is None.

Return type:

iterator of float or None

property t_start[source]

Start time for Laplacian matrix calculation.

The laplacian matrices will be calculated form the first event time before or equal to this time-point.

Returns:

Start time for Laplacian matrices; events before this time are ignored.

Return type:

float or None

property t_stop[source]

Stop time for Laplacian matrix calculation.

The laplacian matrices will be calculated up to the first event time after or equal to this timepoint.

Returns:

End time for Laplacian matrices; events after this time are ignored.

Return type:

float or None

property temporal_network[source]

The temporal network used as basis for the analysis.

Returns:

The temporal network currently set for the analysis.

Return type:

ContTempNetwork or None

property time_direction[source]

Get the time direction for the analysis.

Returns:

Time direction: -1 (backward), 0 (both), or 1 (forward).

Return type:

int or None

property time_scale[source]

Time scales used for random walk transition rates.

Returns:

  • iterator – Iterator over the time scales. Each value determines the rate of the random walk’s transitions.

  • .. note:: – Single values are alos returned as an iterator.

class flowstab.flow_stability.States[source]

Bases: flowstab.state_tracking.OrderedEnum

Defines the stages of a flow stability analysis

CLUSTERING = 4[source]

Ready to compute the Louvain clusters.

FINAL = 5[source]

Computed all that there is.

INITIAL = 0[source]

Initiated an flow stability analysis with no, or incomplete data

INTER_T = 3[source]

Ready to compute the flow integral clustering

LAPLAC = 2[source]

Ready to calculate the inter transition matrices

TEMP_NW = 1[source]

Ready to calculate the Laplacian matrices

flowstab.flow_stability.logger[source]
flowstab.flow_stability.register[source]