Warping Path Analysis

def warping_paths(s1, s2, window=None, max_dist=None, max_step=None,
                  max_length_diff=None, penalty=None, psi=None):
    """
    Compute DTW with full warping paths matrix.
    
    Returns both the optimal distance and the complete accumulated cost matrix,
    which can be used to extract alternative paths and analyze warping behavior.
    
    Parameters:
    - s1, s2: array-like, input sequences
    - window: int, warping window constraint
    - max_dist: float, early stopping threshold  
    - max_step: float, maximum step size
    - max_length_diff: int, maximum length difference
    - penalty: float, penalty for compression/expansion
    - psi: int, psi relaxation parameter
    
    Returns:
    tuple: (distance, paths_matrix)
        - distance: float, optimal DTW distance
        - paths_matrix: 2D array, accumulated cost matrix of shape (len(s1)+1, len(s2)+1)
    """

def warping_paths_fast(s1, s2, window=None, max_dist=None, max_step=None,
                       max_length_diff=None, penalty=None, psi=None):
    """
    Fast C version of warping paths computation.
    
    Same functionality as warping_paths() but uses optimized C implementation.
    
    Returns:
    tuple: (distance, paths_matrix)
    """

def warping_path(from_s, to_s, **kwargs):
    """
    Compute optimal warping path between two sequences.
    
    Returns the optimal sequence of index pairs that define how elements
    from the first sequence align with elements from the second sequence.
    
    Parameters:
    - from_s, to_s: array-like, input sequences
    - **kwargs: additional parameters passed to warping path computation
    
    Returns:
    list: sequence of (i, j) coordinate pairs representing optimal alignment
    """

def best_path(paths):
    """
    Compute optimal path from warping paths matrix.
    
    Traces back through the accumulated cost matrix to find the optimal
    warping path from end to beginning.
    
    Parameters:
    - paths: 2D array, warping paths matrix from warping_paths()
    
    Returns:
    list: sequence of (i, j) coordinate pairs representing optimal path
    """

def best_path2(paths):
    """
    Alternative optimal path computation with different implementation.
    
    Provides an alternative algorithm for path extraction that may be
    more suitable for certain matrix structures or constraints.
    
    Parameters:
    - paths: 2D array, warping paths matrix
    
    Returns:
    list: sequence of (i, j) coordinate pairs representing optimal path
    """

def warping_path_penalty(s1, s2, penalty_post=0, **kwargs):
    """
    DTW with post-calculation penalty based on warping path characteristics.
    
    Computes DTW and then applies additional penalties based on the
    resulting warping path properties (e.g., excessive compression or expansion).
    
    Parameters:
    - s1, s2: array-like, input sequences
    - penalty_post: float, post-calculation penalty factor
    - **kwargs: additional DTW parameters
    
    Returns:
    list: [distance, path, path_stepsize, DTW_matrix]
        - distance: float, DTW distance with penalty applied
        - path: list, optimal warping path coordinates
        - path_stepsize: array, step sizes along the path
        - DTW_matrix: 2D array, accumulated cost matrix
    """

def warping_amount(path):
    """
    Count compressions and expansions in warping path.
    
    Analyzes the warping path to quantify how much compression (multiple
    elements from one sequence aligned to single element) and expansion
    (single element aligned to multiple elements) occurs.
    
    Parameters:
    - path: list, sequence of (i, j) coordinate pairs from warping path
    
    Returns:
    int: total number of warping operations (compressions + expansions)
    """

def warp(from_s, to_s, **kwargs):
    """
    Warp one sequence to match another using optimal DTW alignment.
    
    Transforms the first sequence by stretching and compressing it according
    to the optimal warping path to best match the second sequence.
    
    Parameters:
    - from_s: array-like, source sequence to be warped
    - to_s: array-like, target sequence to warp towards
    - **kwargs: additional DTW parameters
    
    Returns:
    tuple: (warped_sequence, path)
        - warped_sequence: array, transformed version of from_s
        - path: list, warping path used for transformation
    """

from dtaidistance import dtw
import numpy as np

# Create two sequences
s1 = [1, 2, 3, 4, 3, 2, 1]
s2 = [1, 3, 4, 3, 1]

# Compute warping paths matrix
distance, paths = dtw.warping_paths(s1, s2)

print(f"DTW distance: {distance}")
print(f"Paths matrix shape: {paths.shape}")
print("Paths matrix:")
print(paths)

from dtaidistance import dtw

s1 = [0, 1, 2, 3, 2, 1, 0]
s2 = [1, 2, 3, 2, 1]

# Get the optimal warping path
path = dtw.warping_path(s1, s2)
print("Optimal warping path:")
for i, (x, y) in enumerate(path):
    print(f"Step {i}: s1[{x}]={s1[x] if x < len(s1) else 'end'} -> "
          f"s2[{y}]={s2[y] if y < len(s2) else 'end'}")

# Alternative: compute paths matrix first, then extract path
distance, paths_matrix = dtw.warping_paths(s1, s2)
optimal_path = dtw.best_path(paths_matrix)
print("Path from matrix:", optimal_path)

from dtaidistance import dtw

# Sequences with different temporal patterns
s1 = [1, 1, 2, 2, 3, 3, 2, 2, 1, 1]  # Slower pattern
s2 = [1, 2, 3, 2, 1]                   # Faster pattern

# Compute path and analyze warping
path = dtw.warping_path(s1, s2)
warping_ops = dtw.warping_amount(path)

print(f"Warping path length: {len(path)}")
print(f"Total warping operations: {warping_ops}")

# Analyze compression/expansion ratios
compressions = sum(1 for i in range(1, len(path)) 
                   if path[i][0] == path[i-1][0])
expansions = sum(1 for i in range(1, len(path)) 
                 if path[i][1] == path[i-1][1])

print(f"Compressions: {compressions}")
print(f"Expansions: {expansions}")

from dtaidistance import dtw
import matplotlib.pyplot as plt

# Original sequences
source = [1, 2, 4, 3, 1, 0, 1, 2]
target = [1, 3, 2, 1]

# Warp source to match target
warped_source, path = dtw.warp(source, target)

print("Original source:", source)
print("Target:", target)
print("Warped source:", warped_source)
print("Warping path:", path)

# Visualize the warping transformation
fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(10, 8))

ax1.plot(source, 'b-o', label='Original Source')
ax1.set_title('Original Source Sequence')
ax1.legend()

ax2.plot(target, 'r-o', label='Target')
ax2.set_title('Target Sequence')
ax2.legend()

ax3.plot(warped_source, 'g-o', label='Warped Source')
ax3.plot(target, 'r--', alpha=0.7, label='Target (reference)')
ax3.set_title('Warped Source vs Target')
ax3.legend()

plt.tight_layout()
plt.show()

from dtaidistance import dtw

s1 = [1, 2, 3, 4, 5]
s2 = [1, 1, 2, 2, 3, 3, 4, 4, 5, 5]  # More compressed version

# Regular DTW
regular_distance = dtw.distance(s1, s2)

# DTW with post-calculation penalty for excessive warping
results = dtw.warping_path_penalty(s1, s2, penalty_post=0.5)
penalized_distance, path, step_sizes, matrix = results

print(f"Regular DTW distance: {regular_distance}")
print(f"Penalized DTW distance: {penalized_distance}")
print(f"Penalty applied: {penalized_distance - regular_distance}")
print(f"Path step sizes: {step_sizes}")

from dtaidistance import dtw
import numpy as np
import time

# Generate large sequences
np.random.seed(42)
s1 = np.cumsum(np.random.randn(500))
s2 = np.cumsum(np.random.randn(450))

# Compare performance of different methods
methods = [
    ("Python warping_paths", lambda: dtw.warping_paths(s1, s2)),
    ("C warping_paths_fast", lambda: dtw.warping_paths_fast(s1, s2)),
    ("Path only", lambda: dtw.warping_path(s1, s2))
]

for name, method in methods:
    start_time = time.time()
    result = method()
    elapsed = time.time() - start_time
    
    if isinstance(result, tuple):
        distance = result[0]
        print(f"{name}: distance={distance:.4f}, time={elapsed:.4f}s")
    else:
        print(f"{name}: path_length={len(result)}, time={elapsed:.4f}s")

from dtaidistance import dtw
from dtaidistance.dtw_visualisation import plot_warpingpaths, plot_warping

s1 = [0, 1, 2, 1, 0, 1, 0, 0]
s2 = [0, 0, 1, 2, 1, 0, 0]

# Compute paths and optimal path
distance, paths = dtw.warping_paths(s1, s2)
path = dtw.best_path(paths)

# Visualize warping paths matrix
plot_warpingpaths(s1, s2, paths, path)

# Visualize optimal warping between sequences  
plot_warping(s1, s2, path)