Merge pull request #456 from dPys/development

Development

Author: dPys

.duecredit.p

@@ -48,6 +48,7 @@ RUN apt-get update -qq \
         libglu1-mesa-dev \
         libglib2.0-0 \
         libglw1-mesa \
+        libxkbcommon-x11-0 \
 	liblapack-dev \
 	libopenblas-base \
 	sqlite3 \
@@ -84,7 +85,10 @@ RUN apt-get update -qq \
     && cp fsl/bin/* $FSLDIR/bin/ \
     && rm -r fsl* \
     && chmod 777 -R $FSLDIR/bin \
-    && chmod 777 -R /usr/lib/fsl/5.0
+    && chmod 777 -R /usr/lib/fsl/5.0 \
+    && echo "tmpfs   /tmp         tmpfs   rw,nodev,nosuid,size=10G          0  0" >> /etc/fstab
+#    && wget --retry-connrefused --waitretry=5 --read-timeout=60 --timeout=60 -t 0 -q -O examples.tar.gz "https://osf.io/ye4vf/download" && tar -xvzf examples.tar.gz -C /tmp \
+#    && rm -rf examples.tar.gz
 
 ENV FSLDIR=/usr/share/fsl/5.0 \
     FSLOUTPUTTYPE=NIFTI_GZ \
@@ -127,6 +131,7 @@ RUN echo "FSLDIR=/usr/share/fsl/5.0" >> /home/neuro/.bashrc && \
     && pip install certifi -U --ignore-installed \
     && pip install python-dateutil==2.8.0 \
 #    && pip install skggm \
+    && pip install --upgrade --force-reinstall numpy \
     # Create nipype config for resource monitoring
     && mkdir -p ~/.nipype \
     && echo "[monitoring]" > ~/.nipype/nipype.cfg \
@@ -188,6 +193,7 @@ ENV PATH="/opt/conda/bin":$PATH
 ENV OPENBLAS_NUM_THREADS=4 \
     GOTO_NUM_THREADS=4 \
     OMP_NUM_THREADS=4
+ENV QT_QPA_PLATFORM=offscreen
 
 # and add it as an entrypoint
 #ENTRYPOINT ["pynets"]

Dockerfile

@@ -113,7 +113,6 @@ def load_pd_dfs(file_):
                 except:
                     print(f"Cannot load {file_}")
                     df = pd.DataFrame()
-                    return df
             if "Unnamed: 0" in df.columns:
                 df.drop(df.filter(regex="Unnamed: 0"), axis=1, inplace=True)
             id = op.basename(file_).split("_topology")[0].split('auc_')[1]
@@ -175,14 +174,13 @@ def load_pd_dfs(file_):
                 os.remove(f"{file_.split('.csv')[0]}{'_clean.csv'}")
             df.to_csv(f"{file_.split('.csv')[0]}{'_clean.csv'}", index=False)
             del id
-
+            gc.collect()
         else:
             print(f"{Fore.RED}Cleaned {file_} missing...{Style.RESET_ALL}")
             df = pd.DataFrame()
     else:
         print(f"{Fore.RED}{file_} missing...{Style.RESET_ALL}")
         df = pd.DataFrame()
-    gc.collect()
 
     return df
 
@@ -531,6 +529,7 @@ def load_pd_dfs_auc(atlas_name, prefix, auc_file, modality, drop_cols):
     from colorama import Fore, Style
     import pandas as pd
     import re
+    import numpy as np
     import os
 
     pd.set_option("display.float_format", lambda x: f"{x:.8f}")
@@ -545,8 +544,7 @@ def load_pd_dfs_auc(atlas_name, prefix, auc_file, modality, drop_cols):
                 auc_file, chunksize=100000, compression="gzip",
                 encoding="utf-8", engine='python').read()
         except:
-            df_pref = pd.DataFrame()
-            return df_pref
+            df = pd.DataFrame()
 
     #print(f"{'Atlas: '}{atlas_name}")
     prefix = f"{atlas_name}{'_'}{prefix}{'_'}"
@@ -813,6 +811,7 @@ def collect_all(working_path, modality, drop_cols):
 def build_collect_workflow(args, retval):
     import os
     import glob
+    import psutil
     import warnings
     warnings.filterwarnings("ignore")
     import ast
@@ -862,20 +861,6 @@ def build_collect_workflow(args, retval):
 
     wf = collect_all(working_path, modality, drop_cols)
 
-    hardcoded_params = load_runconfig()
-    runtime_dict = {}
-    execution_dict = {}
-    for i in range(len(hardcoded_params["resource_dict"])):
-        runtime_dict[
-            list(hardcoded_params["resource_dict"][i].keys())[0]
-        ] = ast.literal_eval(
-            list(hardcoded_params["resource_dict"][i].values())[0][0]
-        )
-    for i in range(len(hardcoded_params["execution_dict"])):
-        execution_dict[
-            list(hardcoded_params["execution_dict"][i].keys())[0]
-        ] = list(hardcoded_params["execution_dict"][i].values())[0][0]
-
     run_uuid = f"{strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4()}"
     os.makedirs(f"{work_dir}/pynets_out_collection{run_uuid}", exist_ok=True)
     wf.base_dir = f"{work_dir}/pynets_out_collection{run_uuid}"
@@ -912,9 +897,30 @@ def build_collect_workflow(args, retval):
         handler = logging.FileHandler(callback_log_path)
         logger.addHandler(handler)
 
+    execution_dict = {}
     execution_dict["crashdump_dir"] = str(wf.base_dir)
     execution_dict["plugin"] = str(plugin_type)
+    execution_dict["poll_sleep_duration"] = 0.5
+    execution_dict["crashfile_format"] = "txt"
+    execution_dict["local_hash_check"] = False
+    execution_dict["stop_on_first_crash"] = False
+    execution_dict['hash_method'] = 'timestamp'
+    execution_dict["keep_inputs"] = True
+    execution_dict["use_relative_paths"] = False
+    execution_dict["remove_unnecessary_outputs"] = False
+    execution_dict["remove_node_directories"] = False
+    execution_dict["raise_insufficient"] = False
+    nthreads = psutil.cpu_count() * 2
+    procmem = [int(nthreads),
+               int(list(psutil.virtual_memory())[4] / 1000000000) - 2]
+    plugin_args = {
+        "n_procs": int(procmem[0]),
+        "memory_gb": int(procmem[1]),
+        "scheduler": "topological_sort",
+    }
+    execution_dict["plugin_args"] = plugin_args
     cfg = dict(execution=execution_dict)
+
     for key in cfg.keys():
         for setting, value in cfg[key].items():
             wf.config[key][setting] = value
@@ -976,7 +982,8 @@ def main():
     args_dict_all['plug'] = 'MultiProc'
     args_dict_all['v'] = False
     #args_dict_all['pm'] = '48,67'
-    args_dict_all['pm'] = '224,2000'
+    args_dict_all['pm'] = '128,500'
+    #args_dict_all['pm'] = '224,2000'
     #args_dict_all['basedir'] = '/working/tuning_set/outputs_clustering/pynets'
     #args_dict_all['basedir'] = '/working/tuning_set/outputs_shaeffer/pynets'
     #args_dict_all['basedir'] = '/working/tuning_set/outputs_language/pynets'

pynets/cli/pynets_collect.py

@@ -795,7 +795,7 @@ def build_workflow(args, retval):
         import psutil
         nthreads = psutil.cpu_count()
         procmem = [int(nthreads),
-                   int(list(psutil.virtual_memory())[4]/1000000000) - 2]
+                   int(list(psutil.virtual_memory())[4]/1000000000) - 1]
     else:
         procmem = list(eval(str(resources)))
         procmem[1] = procmem[1] - 1

pynets/cli/pynets_run.py

@@ -1154,6 +1154,7 @@ def _run_interface(self, runtime):
         import gc
         import time
         import glob
+        import os
         import os.path as op
         from pynets.registration import register
         from nipype.utils.filemanip import fname_presuffix, copyfile
@@ -1347,6 +1348,22 @@ def _run_interface(self, runtime):
         self._results["t1w2dwi_xfm"] = reg.t1w2dwi_xfm
         self._results["wm_gm_int_in_dwi"] = reg.wm_gm_int_in_dwi
 
+        reg_tmp = [
+            fa_tmp_path,
+            mask_tmp_path,
+            reg.warp_t1w2mni,
+            reg.t1w_head,
+            reg.wm_edge,
+            reg.vent_mask_dwi,
+            reg.vent_mask_t1w,
+            reg.corpuscallosum_mask_t1w,
+            reg.corpuscallosum_dwi
+        ]
+        for j in reg_tmp:
+            if j is not None:
+                if os.path.isfile(j):
+                    os.system(f"rm -f {j} &")
+
         gc.collect()
 
         return runtime
@@ -1745,6 +1762,8 @@ def _run_interface(self, runtime):
             t1w2dwi_bbr_xfm_tmp_path,
             t1w2dwi_xfm_tmp_path,
             wm_gm_int_in_dwi_tmp_path,
+            aligned_atlas_skull,
+            t1w_brain_tmp_path
         ]
         for j in reg_tmp:
             if j is not None:
@@ -1922,6 +1941,9 @@ def _run_interface(self, runtime):
             mni2t1w_warp_tmp_path,
             t1wtissue2dwi_xfm_tmp_path,
             mni2t1_xfm_tmp_path,
+            template_tmp_path,
+            roi_in_t1w,
+            roi_file_tmp_path
         ]
         for j in reg_tmp:
             if j is not None:
@@ -2238,6 +2260,8 @@ def _run_interface(self, runtime):
             uatlas_tmp_path,
             template_tmp_path,
             t1w2mni_xfm_tmp_path,
+            t1w2mni_warp_tmp_path,
+            template_mask_tmp_path
         ]
 
         for j in reg_tmp:
@@ -2612,6 +2636,7 @@ def _run_interface(self, runtime):
             mni2t1w_warp_tmp_path,
             mni2t1_xfm_tmp_path,
             roi_file_tmp_path,
+            template_tmp_path
         ]
         for j in reg_tmp:
             if j is not None:
@@ -3036,35 +3061,37 @@ def _run_interface(self, runtime):
             print(Style.RESET_ALL)
 
             # Commence Ensemble Tractography
-            streamlines = track_ensemble(
-                self.inputs.target_samples,
-                labels_im_file_tmp_path_wm_gm_int,
-                labels_im_file_tmp_path,
-                recon_path,
-                get_sphere(sphere),
-                self.inputs.directget,
-                self.inputs.curv_thr_list,
-                self.inputs.step_list,
-                self.inputs.track_type,
-                self.inputs.maxcrossing,
-                int(roi_neighborhood_tol),
-                self.inputs.min_length,
-                waymask_tmp_path,
-                B0_mask_tmp_path,
-                t1w2dwi_tmp_path, gm_in_dwi_tmp_path,
-                vent_csf_in_dwi_tmp_path, wm_in_dwi_tmp_path,
-                self.inputs.tiss_class,
-                runtime.cwd
-            )
-
-            gc.collect()
+            try:
+                streamlines = track_ensemble(
+                    self.inputs.target_samples,
+                    labels_im_file_tmp_path_wm_gm_int,
+                    labels_im_file_tmp_path,
+                    recon_path,
+                    get_sphere(sphere),
+                    self.inputs.directget,
+                    self.inputs.curv_thr_list,
+                    self.inputs.step_list,
+                    self.inputs.track_type,
+                    self.inputs.maxcrossing,
+                    int(roi_neighborhood_tol),
+                    self.inputs.min_length,
+                    waymask_tmp_path,
+                    B0_mask_tmp_path,
+                    t1w2dwi_tmp_path, gm_in_dwi_tmp_path,
+                    vent_csf_in_dwi_tmp_path, wm_in_dwi_tmp_path,
+                    self.inputs.tiss_class,
+                    runtime.cwd
+                )
+                gc.collect()
+            except BaseException:
+                print(UserWarning("Tractography failed..."))
+                streamlines = None
 
             if streamlines is not None:
                 # import multiprocessing
                 # from pynets.core.utils import kill_process_family
                 # return kill_process_family(int(multiprocessing.current_process().pid))
 
-
                 # Linear Fascicle Evaluation (LiFE)
                 if use_life is True:
                     print('Using LiFE to evaluate streamline plausibility...')
@@ -3143,7 +3170,10 @@ def _run_interface(self, runtime):
             else:
                 self._results["streams"] = None
                 self._results["dm_path"] = None
-            tmp_files = [gtab_file_tmp_path, labels_im_file_tmp_path_wm_gm_int]
+            tmp_files = [gtab_file_tmp_path,
+                         labels_im_file_tmp_path_wm_gm_int,
+                         wm_in_dwi_tmp_path, gm_in_dwi_tmp_path,
+                         vent_csf_in_dwi_tmp_path, t1w2dwi_tmp_path]
 
             for j in tmp_files:
                 if j is not None:

pynets/core/interfaces.py

@@ -1203,7 +1203,7 @@ def gen_network_parcels(uatlas, network, labels, dir_path):
     )
     out_path = f"{dir_path}" \
                f"/{op.basename(uatlas).split(op.splitext(uatlas)[1])[0]}_" \
-               f"{network}{'_parcels.nii.gz'}"
+               f"{network}_parcels.nii.gz"
     nib.save(nib.Nifti1Image(net_parcels_sum, affine=np.eye(4)), out_path)
     del net_parcels_concatted, img_list
     gc.collect()

pynets/core/nodemaker.py

@@ -532,6 +532,7 @@ def streams2graph(
       https://doi.org/10.1089/brain.2016.0481
     """
     import gc
+    import os
     import time
     from dipy.tracking.streamline import Streamlines, values_from_volume
     from dipy.tracking._utils import _mapping_to_voxel, _to_voxel_coordinates
@@ -552,7 +553,7 @@ def streams2graph(
     overlap_thr = hardcoded_params[
         "StructuralNetworkWeighting"]["overlap_thr"][0]
     roi_neighborhood_tol = \
-    hardcoded_params['tracking']["roi_neighborhood_tol"][0]
+        hardcoded_params['tracking']["roi_neighborhood_tol"][0]
 
     start = time.time()
 
@@ -589,7 +590,8 @@ def streams2graph(
         # from fury import actor, window
         # renderer = window.Renderer()
         # template_actor = actor.contour_from_roi(roi_img.get_fdata(),
-        #                                         color=(50, 50, 50), opacity=0.05)
+        #                                         color=(50, 50, 50),
+        #                                         opacity=0.05)
         # renderer.add(template_actor)
         # lines_actor = actor.streamtube(streamlines, window.colors.orange,
         #                                linewidth=0.3)
@@ -707,7 +709,7 @@ def streams2graph(
         # Adapted from the nnormalized fiber-density estimation routines of
         # Sebastian Tourbier.
         if fiber_density is True:
-            print("Weighting edges by fiber density...")
+            print("Redefining edges on the basis of fiber density...")
             # Summarize total fibers and total label volumes
             total_fibers = 0
             total_volume = 0
@@ -734,7 +736,7 @@ def streams2graph(
                 ix += 1
 
         if fa_wei is True:
-            print("Weighting edges by FA...")
+            print("Re-weighting edges by FA...")
             # Add FA attributes for each edge
             ix = 0
             for u, v, d in g.edges(data=True):