Check all nodes of a nipype 1.0 workflow and extract node name and duration

check_node_duration.py

"""Check node duration in a Nipype workflow."""
from pathlib import path
import numpy as np
import pandas as pd

data = {'node': [], 'duration': []}
for pth in Path().glob("**/report.rst"):
    rst = pth.read_text()
    info = [l.split(":")[-1].strip() for l in rst.splitlines() if "duration" in l or "Hierarchy" in l]
    data['node'].append(info[0])
    try:
        data['duration'].append(float(info[1]))
    except IndexError:
        data['duration'].append(np.nan)

df = pd.DataFrame(data)
df[['node', 'duration']].sort_values(by='duration', ascending=False).to_csv('durations.tsv', sep='\t', index=None)