delaynet package

Subpackages

• delaynet.connectivities package
  • Submodules
  • delaynet.connectivities.continuous_ordinal_patterns module
    • norm_window()
    • norm_windows()
    • pattern_distance()
    • pattern_transform()
    • pattern_transform_2d()
    • random_patterns()
  • delaynet.connectivities.granger module
    • gt_multi_lag()
    • gt_single_lag()
  • delaynet.connectivities.gravity module
    • grav()
    • gravity()
    • gravity_single()
  • delaynet.connectivities.linear_correlation module
    • linear_correlation()
  • delaynet.connectivities.mutual_information module
    • mutual_information()
  • delaynet.connectivities.rank_correlation module
    • rank_correlation()
  • delaynet.connectivities.transfer_entropy module
    • transfer_entropy()
  • Module contents
• delaynet.detrending_methods package
  • Submodules
  • delaynet.detrending_methods.delta module
    • delta()
  • delaynet.detrending_methods.identity module
    • identity()
  • delaynet.detrending_methods.second_difference module
    • second_difference()
  • delaynet.detrending_methods.z_score module
    • z_score()
  • Module contents
• delaynet.network_analysis package
  • Submodules
  • delaynet.network_analysis.metrics module
    • betweenness_centrality()
    • eigenvector_centrality()
    • global_efficiency()
    • isolated_nodes_inbound()
    • isolated_nodes_outbound()
    • link_density()
    • reciprocity()
    • transitivity()
  • delaynet.network_analysis.pruning module
    • statistical_pruning()
  • Module contents
• delaynet.preparation package
  • Submodules
  • delaynet.preparation.data_generator module
    • extract_airport_delay_time_series()
    • gen_delayed_causal_network()
    • gen_fmri()
    • gen_fmri_multiple()
    • gen_synthatdelays_independent_operations_with_trends()
    • gen_synthatdelays_random_connectivity()
  • Module contents
• delaynet.utils package
  • Submodules
  • delaynet.utils.bind_args module
    • bind_args()
  • delaynet.utils.dict_lookup module
    • dict_lookup()
  • delaynet.utils.lag_steps module
    • Connectivity
    • assure_lag_list()
    • find_optimal_lag()
  • delaynet.utils.logging module
  • Module contents

Submodules

delaynet.connectivity module

Module to provide a unified interface for all connectivity metrics.

delaynet.connectivity.connectivity(ts1: ndarray, ts2: ndarray, /, metric: str | Callable[[ndarray, ndarray, ...], tuple[float, int]], *args, lag_steps: int | list[int] | None = None, **kwargs) → tuple[float, int]

Calculate connectivity between two time series using a given metric.

Keyword arguments are forwarded to the metric function.

The metrics can be either a string or a function, implementing a connectivity metric. Find the metric string specifier using show_connectivity_metrics().

(Find all in submodule delaynet.connectivities, names are stored in delaynet.connectivities.__all_connectivity_metrics__)

If a callable is given, it should take two time series as input and return a tuple of float and int.

Parameters:

• ts1 (numpy.ndarray) – First time series. Positional only.

• ts2 (numpy.ndarray) – Second time series. Positional only.

• metric (str or Callable) – Metric to use.

• args (list) – Positional arguments forwarded to the connectivity function, see documentation.

• lag_steps (int | list[int] | None) – The number of lag steps to consider. Required. Can be integer for [1, …, num], or a list of integers.

• kwargs – Keyword arguments forwarded to the connectivity function, see documentation.

Returns:

Connectivity value and lag.

Return type:

tuple of float and int

Raises:

• ValueError – If the metric is unknown. Given as string.

• ValueError – If the metric returns an invalid value. Given a Callable.

• ValueError – If the metric is neither a string nor a Callable.

delaynet.connectivity.show_connectivity_metrics()

Pretty print all available connectivity metrics.

delaynet.decorators module

Decorators for the delaynet package.

delaynet.decorators.connectivity(*args)

Decorator for the connectivity functions.

As the connectivity functions take different arguments, this decorator is used to make sure that the connectivity functions are called with the correct positional and keyword arguments.

The decorated function only needs to take the two ndarrays as the first positional arguments and can take any number of further arguments.

The decorator checks that all passed keyword arguments exist and that all positional arguments are passed.

Shape of the input time series must be equal.

Returns:

The decorated function.

Return type:

Callable

Raises:

TypeError – If mcb_kwargs is not None or a dict.

delaynet.decorators.detrending_method(*args, check_nan: bool = True, check_inf: bool = True, check_shape: bool = True)

Decorator for the detrending functions.

Input time series can be of any shape. For 1D arrays, the method is applied directly. For higher dimensional arrays, an ‘axis’ kwarg must be provided to specify along which axis to apply the detrending.

The decorator automatically detects if the detrending function has an ‘axis’ parameter in its signature. If it does, the axis is passed directly to the function. If not, apply_along_axis is used to apply the function along the specified axis.

Input must be non-empty. If check_shape is True, the shape of the output time series is checked. Otherwise, only the dimensionality must remain the same, so the length of the time series can change.

As the detrending methods take different arguments, this decorator is used to make sure that the detrending methods are called with the correct positional and keyword arguments.

The decorated function only needs to take the ndarray as the first positional argument and can take any number of further arguments.

The decorator checks that all passed keyword arguments exist and that all positional arguments are passed.

Parameters:

• check_nan (bool) – If True, check if the detrended time series contains NaNs.

• check_inf (bool) – If True, check if the detrended time series contains Infs.

• check_shape (bool) – If True, check if the shape of the detrended time series is equal to the shape of the input time series.

• detrend_func (Callable) – The detrending function to decorate.

Returns:

The decorated function.

Return type:

Callable

delaynet.detrending module

Module to provide unified interface for all detrending methods.

delaynet.detrending.detrend(ts: ndarray, /, method: str | Callable[[ndarray, ...], ndarray], *args, axis: int = None, **kwargs) → ndarray

Detrend time series using a given detrending method.

Keyword arguments are forwarded to the detrend function.

If check_kwargs is passed in kwargs with value False, the kwargs are not checked for availability. This is useful if you want to pass unused values in generic functions.

The detrending methods can be either a string or a function, implementing a detrending method. Find the method string specifier using show_detrending_methods().

(Find all in submodule delaynet.detrending_methods, names are stored in delaynet.detrending_methods.__all_detrending__)

If a callable is given, it should take a time series as input and return the Detrended time series.

Parameters:

• ts (numpy.ndarray) – Time series to detrend. 1D or 2D. Positional only.

• method (str or Callable) – Method to use for detrending.

• args (list) – Positional arguments forwarded to the detrending function, see documentation of the detrending methods.

• axis – Axis to detrend along. Needs to be passed for ts of higher dimensions. If 1D ts, axis is ignored.

• axis – int | None, optional

• kwargs (dict) – Keyword arguments forwarded to the detrending function, see documentation of the detrending methods.

Returns:

Detrended time series.

Return type:

numpy.ndarray

Raises:

• ValueError – If the method is unknown. Given as string.

• ValueError – If the method returns an invalid value. Given a Callable.

• ValueError – If the method is neither a string nor a Callable.

• ValueError – If the shape of the method output is not equal to the shape of the input time series.

delaynet.detrending.show_detrending_methods()

Pretty print all available detrending methods.

delaynet.evaluation module

Evaluation module to compare original and reconstructed networks.

delaynet.evaluation.roc_auc_rank_c(orig_net: ndarray, weight_mat: ndarray, rec_net: ndarray) → tuple[ndarray, float, float]

Get Receiver Operating Characteristic (roc) curve, Area Under Curve (auc) and Spearman rank correlation coefficient (rank_c) for a given reconstructed network.

Parameters:

• orig_net (np.ndarray( n_nodes, n_nodes )) – Original network

• weight_mat (np.ndarray( n_nodes, n_nodes )) – Weights of the original network

• rec_net (np.ndarray( n_nodes, n_nodes )) – Reconstructed network

Returns:

ROC curve, Area Under Curve, Spearman rank correlation coefficient

Return type:

tuple[ np.ndarray( n_nodes, 2 ), float, float ]

Raises:

• TypeError – If any of the input parameters is not a numpy array.

• ValueError – If any of the input parameters is not a square matrix.

• ValueError – If the input parameters do not have the same number of nodes.

delaynet.network_reconstruction module

Network reconstruction module for delaynet.

This module provides functionality to reconstruct networks from time series data by applying connectivity measures to pairs of time series.

delaynet.network_reconstruction.format_time(seconds)

Format time in appropriate units (seconds, minutes, hours).

delaynet.network_reconstruction.is_sphinx_build()

Check if the code is running in a Sphinx build environment.

delaynet.network_reconstruction.print_progress(current, total, start_time, prefix='', sphinx_mode=None)

Print progress bar with percentage, counts, and estimated time remaining.

Parameters:

• current – Current progress value

• total – Total value to reach

• start_time – Start time for ETA calculation

• prefix – Prefix string for the progress bar

• sphinx_mode – How to handle progress in Sphinx documentation. If True, only print progress when current == total. If None, automatically detect Sphinx environment.

delaynet.network_reconstruction.reconstruct_network(time_series: ndarray, connectivity_measure: str | Callable[[ndarray, ndarray, ...], tuple[float, int]], lag_steps: int | list[int] | None = None, workers: int = None, **kwargs) → tuple[ndarray, ndarray]

Reconstruct a network from time series data.

This function applies a connectivity measure to all pairs of time series to construct a network represented by weight and lag matrices.

Parameters:

• time_series (numpy.ndarray) – Array of time series data with shape (n_time, n_nodes). Each column represents a time series for one node.

• connectivity_measure (str or Callable) – Connectivity measure to use. Can be either a string name of a built-in measure or a callable function. Available string measures can be found using delaynet.connectivity.show_connectivity_metrics(). If a callable is provided, it should take two time series as input and return a tuple of (float, int).

• lag_steps (int | list[int] | None) – The number of lag steps to consider. Required. Can be integer for [1, …, num], or a list of integers.

• workers (int | None) – Number of workers to use for parallel computation.

• kwargs (dict) – Additional keyword arguments passed to the connectivity measure.

Returns:

Tuple containing:

• weight_matrix: Matrix of p-values with shape (n_nodes, n_nodes). Lower p-values indicate stronger connections.

• lag_matrix: Matrix of optimal time lags with shape (n_nodes, n_nodes).

Return type:

tuple[numpy.ndarray, numpy.ndarray]

Raises:

• ValueError – If time_series has incorrect dimensions.

• ValueError – If connectivity_measure is unknown (when given as string).

• ValueError – If connectivity_measure returns invalid value (when given as callable).

• ValueError – If connectivity_measure is neither string nor callable.

Example:

>>> import numpy as np
>>> from delaynet.network_reconstruction import reconstruct_network
>>> # Generate sample data: 100 time points, 5 nodes
>>> data = np.random.randn(100, 5)
>>>
>>> # Using string metric
>>> weights, lags = reconstruct_network(data, "linear_correlation", lag_steps=5)
>>> weights.shape
(5, 5)
>>> lags.shape
(5, 5)
>>>
>>> # Using callable metric
>>> def custom_metric(ts1, ts2, lag_steps=None):
...     # Using numpy cov function
...     all_values = [np.cov(ts1[: -lag or None], ts2[lag:])[0,1] for lag in lag_steps]
...     idx_optimal = min(range(len(all_values)), key=all_values.__getitem__)
...     return all_values[idx_optimal], lag_steps[idx_optimal]
>>> weights, lags = reconstruct_network(data, custom_metric, lag_steps=5)
>>> weights.shape
(5, 5)

Note:

The diagonal elements of the weight matrix are set to 1.0 by default, indicating no significant self-connection.

delaynet.network_reconstruction.update_progress(counter, total, start_time, prefix, sphinx_mode=None)

Update progress from worker processes

Parameters:

• counter – Shared counter value

• total – Total number of items to process

• start_time – Start time for ETA calculation

• prefix – Prefix string for the progress bar

• sphinx_mode – How to handle progress in Sphinx documentation

Module contents

delaynet init.