Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/code-coverage.yml
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ jobs:
coverage xml -o coverage.xml

- name: "Save coverage report as artifact"
uses: actions/upload-artifact@v4
uses: actions/upload-artifact@v6
with:
name: coverage-report
path: coverage.xml
Expand Down
2 changes: 1 addition & 1 deletion .github/workflows/rossler_demo.yml
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ jobs:

# ---------- Upload validation artefacts ----------
- name: Upload validation artefacts
uses: actions/upload-artifact@v4
uses: actions/upload-artifact@v6
with:
name: surrogate-validation
path: |
Expand Down
2 changes: 1 addition & 1 deletion .github/workflows/scorecard.yml
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ jobs:
# Upload the results as artifacts (optional). Commenting out will disable
# uploads of run results in SARIF format to the repository Actions tab.
- name: "Upload artifact"
uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
with:
name: SARIF file
path: results.sarif
Expand Down
6 changes: 3 additions & 3 deletions SMdRQA/RQA2.py
Original file line number Diff line number Diff line change
Expand Up @@ -546,7 +546,7 @@
def load_results(self, filepath):
"""Load pre-computed RQA results from file."""
with open(filepath, 'rb') as f:
results = pickle.load(f)

Check warning on line 549 in SMdRQA/RQA2.py

View check run for this annotation

Precaution / Precaution Basic

PY013: Deserialization of Untrusted Data

Potential unsafe usage of 'pickle.load' that can allow instantiation of arbitrary objects.

self.data = results['data']
self._tau = results['tau']
Expand Down Expand Up @@ -1309,15 +1309,15 @@
with ProcessPoolExecutor(max_workers=self.max_workers) as executor:
futures = [
executor.submit(
self._generate_with_seed,
self._generate_with_seed,
kind, seed, **kwargs
)
for seed in seeds
]
for future in as_completed(futures):
surrogates.append(future.result())
return np.vstack(surrogates)

def _generate_with_seed(
self,
kind: Algorithm,
Expand All @@ -1331,7 +1331,7 @@
seed=seed,
max_workers=1 # Disable nested parallelism
)

# Dispatch to appropriate surrogate method
method_name = f"_{kind.lower()}"
method = getattr(temp_tests, method_name)
Expand Down
157 changes: 96 additions & 61 deletions docs/examples/rossler_attractor_analysis.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,18 +3,18 @@
# rossler_attractor_analysis.py - Corrected version based on actual RQA2.py implementation
# Set up matplotlib for non-interactive use

from tqdm import tqdm
from SMdRQA.RQA2 import RQA2, RQA2_simulators, RQA2_tests
import seaborn as sns
import os
from mpl_toolkits.mplot3d import Axes3D
import matplotlib.pyplot as plt
import numpy as np
import matplotlib
matplotlib.use('Agg') # Use non-interactive backend

import numpy as np
import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D
import os
import seaborn as sns

# Import from the actual RQA2 module structure
from SMdRQA.RQA2 import RQA2, RQA2_simulators, RQA2_tests
from tqdm import tqdm

print("Starting Rössler Attractor Analysis...")

Expand Down Expand Up @@ -42,16 +42,22 @@
# Chaotic attractor
ax1 = fig.add_subplot(121, projection='3d')
ax1.plot(x_chaos, y_chaos, z_chaos, c='maroon', lw=0.5, alpha=0.7)
ax1.set_title(f"Chaotic Rössler Attractor (a={A_CHAOS})", fontweight='bold', fontsize=14)
ax1.set_title(
f"Chaotic Rössler Attractor (a={A_CHAOS})",
fontweight='bold',
fontsize=14)
ax1.set_xlabel("X", fontsize=12)
ax1.set_ylabel("Y", fontsize=12)
ax1.set_ylabel("Y", fontsize=12)
ax1.set_zlabel("Z", fontsize=12)
ax1.view_init(30, -60)

# Periodic attractor
ax2 = fig.add_subplot(122, projection='3d')
ax2.plot(x_per, y_per, z_per, c='teal', lw=0.5, alpha=0.7)
ax2.set_title(f"Periodic Rössler Attractor (a={A_PER})", fontweight='bold', fontsize=14)
ax2.set_title(
f"Periodic Rössler Attractor (a={A_PER})",
fontweight='bold',
fontsize=14)
ax2.set_xlabel("X", fontsize=12)
ax2.set_ylabel("Y", fontsize=12)
ax2.set_zlabel("Z", fontsize=12)
Expand All @@ -67,17 +73,18 @@
methods = ["FT", "AAFT", "IAAFT", "IDFS", "WIAAFT", "PPS"]
metrics = ["lyapunov_exponent", "time_irreversibility"]


def compute_surrogate_metrics(signal, method):
"""Compute metrics for original signal and its surrogates."""
print(f" Processing {method} surrogates...")
tester = RQA2_tests(signal, seed=123, max_workers=2)

# Generate surrogates
surrogates = tester.generate(kind=method, n_surrogates=N_SURR)

# Compute metrics for original signal
orig_metrics = tester._calculate_all_metrics(signal)

# Compute metrics for all surrogates
surr_metrics = {m: [] for m in metrics}
for i, s in tqdm(enumerate(surrogates)):
Expand All @@ -87,17 +94,19 @@ def compute_surrogate_metrics(signal, method):
for m in metrics:
surr_metrics[m].append(s_vals[m])
print('Metric value:', s_vals[m])

return orig_metrics, surr_metrics


# Compute surrogate metrics for both regimes
results = {}
print("\nStarting surrogate analysis...")

for regime, signal in [("Chaotic", x_chaos), ("Periodic", x_per)]:
results[regime] = {}
print(f"\nProcessing {regime} regime (a={'0.1' if regime=='Chaotic' else '0.3'})...")

print(
f"\nProcessing {regime} regime (a={'0.1' if regime=='Chaotic' else '0.3'})...")

for method in methods:
try:
orig, surr = compute_surrogate_metrics(signal, method)
Expand All @@ -117,15 +126,15 @@ def compute_surrogate_metrics(signal, method):
for r, regime in enumerate(["Chaotic", "Periodic"]):
for m, metric in enumerate(metrics):
ax = axes[r, m]

# Plot histograms for each surrogate method
for i, method in enumerate(methods):
if method in results[regime]:
orig_val, surr_data = results[regime][method]

# Filter out NaN values
valid_surr = [x for x in surr_data[metric] if not np.isnan(x)]

if valid_surr:
sns.kdeplot(data=np.log(valid_surr),
ax=ax,
Expand All @@ -134,23 +143,25 @@ def compute_surrogate_metrics(signal, method):
alpha=0.6, # transparency
color=colors[i % len(colors)],
linewidth=2)

# Plot original value as vertical line
if not np.isnan(orig_val[metric]):
ax.axvline(np.log(orig_val[metric]), color=colors[i % len(colors)],
linestyle='--', linewidth=2, alpha=0.8)
ax.set_title(f"{regime}: {metric.replace('_', ' ').title()}",
fontweight='bold', fontsize=12)
ax.axvline(np.log(orig_val[metric]), color=colors[i % len(
colors)], linestyle='--', linewidth=2, alpha=0.8)

ax.set_title(f"{regime}: {metric.replace('_', ' ').title()}",
fontweight='bold', fontsize=12)
ax.set_xlabel("Log Value", fontsize=10)
ax.set_ylabel("Density", fontsize=10)
ax.grid(True, alpha=0.3)

# Add legend to the last subplot
axes[1, 1].legend(loc='upper right', framealpha=0.9, fontsize=9)

plt.suptitle("Surrogate Test Results: Original (dashed lines) vs Surrogate Distributions",
fontsize=14, fontweight='bold')
plt.suptitle(
"Surrogate Test Results: Original (dashed lines) vs Surrogate Distributions",
fontsize=14,
fontweight='bold')
plt.tight_layout()
plt.savefig('surrogate_results.png', dpi=300, bbox_inches='tight')
plt.close(fig)
Expand All @@ -159,28 +170,28 @@ def compute_surrogate_metrics(signal, method):
# Section 5: Recurrence Quantification Analysis
print("\nComputing RQA measures...")


def compute_rqa_analysis(signal, regime_name):
"""Compute RQA measures for a given signal."""
print(f" Processing {regime_name} regime...")

# Create RQA object with appropriate parameters
# Use the actual constructor parameters from RQA2.py
rq = RQA2(data=signal, normalize=True, reqrr=0.05, lmin=2)

# Compute RQA measures
measures = rq.compute_rqa_measures()

rq.plot_tau_mi_curve(save_path=regime_name + '_tau_mi_plot.png')

rq.plot_fnn_curve(save_path=regime_name + '_fnn_curve_plot.png')




# Get the recurrence plot
rp = rq.recurrence_plot

return rq, measures, rp


# Analyze both regimes
rqa_results = {}
for regime, signal in [("Chaotic", x_chaos), ("Periodic", x_per)]:
Expand All @@ -191,13 +202,17 @@ def compute_rqa_analysis(signal, regime_name):
'measures': measures,
'recurrence_plot': rp
}

# Print key measures
print(f" {regime} RQA measures:")
for key, value in measures.items():
if key in ['recurrence_rate', 'determinism', 'average_diagonal_length', 'max_diagonal_length']:
if key in [
'recurrence_rate',
'determinism',
'average_diagonal_length',
'max_diagonal_length']:
print(f" {key}: {value:.4f}")

except Exception as e:
print(f" Error computing RQA for {regime}: {e}")
rqa_results[regime] = None
Expand All @@ -206,29 +221,33 @@ def compute_rqa_analysis(signal, regime_name):
print("\nCreating recurrence plots...")
fig = plt.figure(figsize=(14, 6))

for i, (regime, signal) in enumerate([("Chaotic", x_chaos), ("Periodic", x_per)]):
ax = fig.add_subplot(1, 2, i+1)

for i, (regime, signal) in enumerate(
[("Chaotic", x_chaos), ("Periodic", x_per)]):
ax = fig.add_subplot(1, 2, i + 1)

if regime in rqa_results and rqa_results[regime] is not None:
rp = rqa_results[regime]['recurrence_plot']
measures = rqa_results[regime]['measures']

# Display recurrence plot
ax.imshow(rp, cmap='binary', origin='lower', aspect='equal')

# Create title with key measures
title = (f"{regime} Recurrence Plot\n"
f"RR={measures['recurrence_rate']:.3f}, "
f"DET={measures['determinism']:.3f}\n"
f"L={measures['average_diagonal_length']:.2f}, "
f"Lmax={measures['max_diagonal_length']:.0f}")
f"RR={measures['recurrence_rate']:.3f}, "
f"DET={measures['determinism']:.3f}\n"
f"L={measures['average_diagonal_length']:.2f}, "
f"Lmax={measures['max_diagonal_length']:.0f}")

ax.set_title(title, fontweight='bold', fontsize=11)
else:
ax.set_title(f"{regime} - Analysis Failed", fontweight='bold', fontsize=11)
ax.text(0.5, 0.5, "RQA computation failed", ha='center', va='center',
transform=ax.transAxes, fontsize=12)

ax.set_title(
f"{regime} - Analysis Failed",
fontweight='bold',
fontsize=11)
ax.text(0.5, 0.5, "RQA computation failed", ha='center', va='center',
transform=ax.transAxes, fontsize=12)

ax.set_xlabel("Time Index", fontsize=10)
ax.set_ylabel("Time Index", fontsize=10)

Expand All @@ -238,9 +257,9 @@ def compute_rqa_analysis(signal, regime_name):
print("Saved: recurrence_plots.png")

# Section 7: Summary Report
print("\n" + "="*60)
print("\n" + "=" * 60)
print("ANALYSIS SUMMARY")
print("="*60)
print("=" * 60)

print(f"\nSystem Parameters:")
print(f" Chaotic regime: a={A_CHAOS}, b={B}, c={C}")
Expand All @@ -255,18 +274,34 @@ def compute_rqa_analysis(signal, regime_name):
print(f"\nRQA Results Comparison:")
print(f"{'Measure':<25} {'Chaotic':<12} {'Periodic':<12}")
print("-" * 50)

key_measures = ['recurrence_rate', 'determinism', 'average_diagonal_length', 'max_diagonal_length']


key_measures = [
'recurrence_rate',
'determinism',
'average_diagonal_length',
'max_diagonal_length']

for measure in key_measures:
chaos_val = rqa_results.get('Chaotic', {}).get('measures', {}).get(measure, np.nan)
period_val = rqa_results.get('Periodic', {}).get('measures', {}).get(measure, np.nan)
chaos_val = rqa_results.get(
'Chaotic',
{}).get(
'measures',
{}).get(
measure,
np.nan)
period_val = rqa_results.get(
'Periodic',
{}).get(
'measures',
{}).get(
measure,
np.nan)
print(f"{measure:<25} {chaos_val:<12.4f} {period_val:<12.4f}")

print(f"\nFiles generated:")
print(" - rossler_3d_attractors.png")
print(" - surrogate_results.png")
print(" - surrogate_results.png")
print(" - recurrence_plots.png")

print(f"\nAnalysis completed successfully!")
print("="*60)
print("=" * 60)
Loading