From ef402e593c7564df5c6178e34e89478394600b05 Mon Sep 17 00:00:00 2001 From: Jon Clucas Date: Wed, 17 Jul 2024 16:11:07 -0400 Subject: [PATCH] :recycle: Move `_get_pipe_number` back into `ResourcePool` --- CPAC/pipeline/engine/resource.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/CPAC/pipeline/engine/resource.py b/CPAC/pipeline/engine/resource.py index 50e18f5935..ccbea72d42 100644 --- a/CPAC/pipeline/engine/resource.py +++ b/CPAC/pipeline/engine/resource.py @@ -561,9 +561,6 @@ def create_func_datasource( return wf - def get_pipe_number(self, pipe_idx): - return self.pipe_list.index(pipe_idx) - def keys(self) -> KeysView: """Return rpool's keys.""" return self.rpool.keys() @@ -959,7 +956,7 @@ def gather_pipes( # noqa: PLR0915 num_variant = "" unlabelled = self._get_unlabelled(resource) for pipe_idx in self.rpool[resource]: - pipe_x = self.get_pipe_number(pipe_idx) + pipe_x = self._get_pipe_number(pipe_idx) json_info = self.rpool[resource][pipe_idx]["json"] out_dct = self.rpool[resource][pipe_idx]["out"] @@ -1727,7 +1724,7 @@ def ingress_output_dir(self) -> None: data_label = data_label.replace("brain_mask", "bold_mask") try: - pipe_x = self.get_pipe_number(pipe_idx) + pipe_x = self._get_pipe_number(pipe_idx) except ValueError: pipe_x = len(self.pipe_list) if filepath in outdir_anat: @@ -2491,7 +2488,7 @@ def connect_block(self, wf: pe.Workflow, block: NodeBlock) -> pe.Workflow: # no # so when we set_data below for the TOP-LEVEL MAIN RPOOL (not the strat_pool), we can generate new merged JSON information for each output. # particularly, our custom 'CpacProvenance' field. node_name = name - pipe_x = self.get_pipe_number(pipe_idx) + pipe_x = self._get_pipe_number(pipe_idx) replaced_inputs = [] for interface in block.input_interface: @@ -2969,6 +2966,10 @@ def get_resource_strats_from_prov(prov: list | str) -> dict[str, list | str]: strat_resource[resource] = entry return strat_resource + def _get_pipe_number(self, pipe_idx: str | tuple) -> int: + """Return the index of a strategy in ``self.pipe_list``.""" + return self.pipe_list.index(pipe_idx) + def _get_unlabelled(self, resource: str) -> set[str]: """Get unlabelled resources (that need integer suffixes to differentiate).""" from CPAC.func_preproc.func_motion import motion_estimate_filter