flowstab

Flow Stability: Temporal Network Flow-Based Clustering and Analysis

This package provides tools for stability analysis and clustering of temporal networks using flow-based methods. The core component is the FlowStability class, which implements a reproducible, state-tracked analytical pipeline for temporal (contact sequence) data. The package also includes logging utilities to facilitate debugging and reproducible research.

Main Features

• Load and process temporal networks (contact sequences) from files or arrays.

• Compute Laplacian and inter-transition matrices for random walks on temporal networks.

• Extract flow-based clusterings using methods such as the Louvain algorithm.

• Track analysis progress and enforce correct computational order through a robust state machine.

• Configurable logging for both development and production.

Logging

The package sets up a default logger on import. You can adjust the logging level:

>>> import flowstab
>>> flowstab.set_log_level("DEBUG")

Usage Example

A typical workflow for analyzing a temporal network and extracting clusters:

>>> from flowstab import FlowStability
>>> fs = FlowStability()
>>> fs.set_temporal_network(event_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)

This workflow ensures that all computational steps are performed in the required order; the state machine will warn or prevent you from skipping prerequisites.

If you are unsure what method needs to be run next and/or what parameter needs to be set in the current state run fs.state.next. This will return a tuple providing a list of parameters to set, and the name of the next method to run:

>>> fs.state.next
>>> ['t_start', ..], 'set_temporal_network'

To learn more about a parameter or method you might use fs.state.info or fs.state.howto:

>>> print(fs.state.info['time_scale'])
>>> Time scales used for random walk transition rates.
>>>
>>> Returns
>>> -------
>>> iterator
>>>     Iterator over the time scales. Each value determines the rate of the
>>>     random walks transitions.
>>>
>>> .. note::
>>>     Single values are alos returned as an iterator.

>>> print(fs.state.howto['time_scale'])
>>> Set the time scale(s) for the random walks transition rate.
>>>
>>> .. note::
>>>     You might also use `set_time_scale` to directly create a range of
>>>     time scales.
>>>
>>> Parameters
>>> ----------
>>> time_scale : None, int, float, or iterator of float
>>>     If None, a default value is used.
>>>     If an int or float, a single time scale is set.
>>>     If an iterator, it must yield float or int values.
>>>
>>> Raises
>>> ------
>>> TypeError
>>>     If the input is not None, int, float, or an iterator of numbers.

Module Contents

• FlowStability

  Main class for performing flow stability analysis and clustering.

• set_log_level

  Function to set the global logging level for the package.

Author

Alexandre Bovet <alexandre.bovet@maths.ox.ac.uk>

Contributors

• Jonas I. Liechti <j-i-l@t4d.ch>

License

GNU Lesser General Public License v3 or later (LGPLv3+).

Submodules

• flowstab.flow_stability
• flowstab.helpers
• flowstab.logger
• flowstab.network_clustering
• flowstab.parallel_clustering
• flowstab.scripts
• flowstab.state_tracking

Attributes

__version__

Classes

FlowStability

Conducts flow stability analysis using a temporal network.

Functions

get_logger()	Return the logger instance.
set_log_level(level)	Set the logging level for the package.
setup_logger([log_level])	Set up the logger for the package.

Package Contents

class flowstab.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)

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__()

compute_inter_transition_matrices(linear_approx=False, **kwargs)

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)

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)

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)

set_flow_clustering(**kwargs)

Perform flow integral clustering analysis.

Parameters:

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

Returns:

self – The instance itself.

Return type:

FlowStability

set_temporal_network(**kwargs)

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)

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

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

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

Returns:

Maps time_scale values to FlowIntegralClustering objects.

Return type:

dict

property lamda

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

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

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

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

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

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.

flowstab.get_logger()

Return the logger instance.

flowstab.set_log_level(level)

Set the logging level for the package.

Parameters:

level (str) – The logging level as a string (e.g., ‘DEBUG’, ‘INFO’).

flowstab.setup_logger(log_level=logging.INFO)

Set up the logger for the package.

Parameters:

log_level (int) – The logging level (e.g., logging.DEBUG, logging.INFO).

flowstab.__version__ = 'unknown'