Timing Utilities

Performance measurement and timeout functionality for algorithm execution.

Timing Collection

Timing

Singleton class collects count, mean and max time for actions.

:cvar bool active: whether timing collection is active :cvar times dict: times collected {action1: {scale1: {count, total, max}, ...}, ...} :cvar set/None filter: only these actions will be recorded

Methods:

• __repr__ –

  Print out all timings in nice format

• add –

  Add an elapsed time without having called now() and record().

• now –

  Returns current time, generally for the start of an action

• off –

  Turn timing collection off, lose collected timings.

• on –

  Switching timing on and off

• record –

  Records time for specified action

• to_string –

  Print out timings in nice format, optionally only for specified

__repr__ classmethod

__repr__() -> str

Print out all timings in nice format

:return str: human friendly timing information

add classmethod

add(action: str, elapsed: float, scale: int = 100) -> None

Add an elapsed time without having called now() and record().

Parameters:

• action

  (str) –

  action being timed

• elapsed

  (float) –

  elapsed time to add

• scale

  (int, default: 100 ) –

  indication of scale of action e.g. num of nodes

:raises TypeError: if bad arg type :raises ValueError: if bad arg value

now classmethod

now() -> Optional[float]

Returns current time, generally for the start of an action

:returns float: epoch time in seconds

off classmethod

off() -> None

Turn timing collection off, lose collected timings.

on classmethod

on(active: bool, filter: Optional[set] = None) -> None

Switching timing on and off

:param bool active: whether timing should be on or off :param set/None filter: only these actions recorded

:raises TypeError: if bad arg type

record classmethod

record(action: str, scale: int, start: Optional[float]) -> Optional[float]

Records time for specified action

:param str action: action being timed :param int scale: indication of scale of action e.g. num of nodes :param int start: time at which action started

:raises TypeError: if bad arg type :raises ValueError: if bad arg value

:returns float: epoch time when this function called

to_string classmethod

to_string(filter: Optional[set] = None) -> str

Print out timings in nice format, optionally only for specified actions.

:param set/None filter: only return info about these actions.

:return str: human friendly timing information

Timeout Functions

run_with_timeout

run_with_timeout(
    func: Callable,
    args: tuple = (),
    kwargs: Optional[dict] = None,
    timeout_seconds: Optional[int] = None,
) -> Any

Run a function with a timeout using threading approach.

This is compatible with both subprocess-based (R, Java) and direct Python function calls (causal-learn). For subprocess calls, the timeout will terminate the subprocess if it exceeds the time limit.

:param func: Function to execute :param args: Positional arguments for the function :param kwargs: Keyword arguments for the function :param timeout_seconds: Maximum execution time in seconds, None for no timeout

:raises TimeoutError: if execution exceeds timeout_seconds :raises Exception: any exception raised by the wrapped function

:returns: Result from the wrapped function

with_timeout

with_timeout(timeout_seconds: Optional[int] = None) -> Callable

Decorator version of run_with_timeout.

:param timeout_seconds: Maximum execution time in seconds, None for no timeout

Usage

@with_timeout(300) # 5 minute timeout def my_algorithm(data): # algorithm implementation return result

TimeoutError

Raised when an operation times out

Usage Examples

Basic Timing Collection

from causaliq_core.utils.timing import Timing
from time import sleep

# Enable timing collection
Timing.on(True)

# Time an operation
start = Timing.now()
sleep(0.1)  # Simulate work
Timing.record("sleep_test", 1, start)

# View timing summary
print(Timing.summary())

# Turn off timing
Timing.off()

Filtered Timing

from causaliq_core.utils.timing import Timing

# Only collect timing for specific actions
Timing.on(True, filter={"critical_operation", "slow_function"})

# This will be recorded
start = Timing.now()
# ... do critical operation ...
Timing.record("critical_operation", 100, start)

# This will be ignored
start = Timing.now()
# ... do normal operation ...
Timing.record("normal_operation", 50, start)

Function Timeout Decorator

from causaliq_core.utils.timing import with_timeout, TimeoutError

@with_timeout(5)  # 5 second timeout
def slow_function():
    # This function will be interrupted if it takes > 5 seconds
    import time
    time.sleep(10)  # Would normally take 10 seconds

try:
    slow_function()
except TimeoutError:
    print("Function timed out!")

Direct Timeout Execution

from causaliq_core.utils.timing import run_with_timeout, TimeoutError

def potentially_slow_task(data):
    # Process data...
    return processed_data

try:
    result = run_with_timeout(
        potentially_slow_task,
        args=(my_data,),
        timeout_seconds=30
    )
    print(f"Completed: {result}")
except TimeoutError:
    print("Task took too long!")

Adding Pre-calculated Times

from causaliq_core.utils.timing import Timing

# Enable timing
Timing.on(True)

# Add a timing measurement without calling now()/record()
Timing.add("algorithm_run", elapsed_time=1.234, scale=500)

# View results
print(Timing.summary())

Features

• Singleton pattern: Global timing collection across application
• Filtering: Collect timing only for specified actions
• Scale tracking: Record scale indicators (e.g., number of nodes)
• Statistical summary: Automatic count, mean, and max calculations
• Thread-safe timeouts: Reliable timeout functionality for long-running operations
• Decorator support: Clean timeout decoration for functions
• Flexible timeout: Support both function decoration and direct execution