Skip to content

Commit

Permalink
♻️ Move get_resource_strats_from_prov back into ResourcePool
Browse files Browse the repository at this point in the history
  • Loading branch information
shnizzedy committed Jul 17, 2024
1 parent 74629ac commit 748b98e
Showing 1 changed file with 24 additions and 21 deletions.
45 changes: 24 additions & 21 deletions CPAC/pipeline/engine/resource.py
Original file line number Diff line number Diff line change
Expand Up @@ -854,27 +854,6 @@ def generate_prov_list(prov_str):
raise TypeError(msg)
return ast.literal_eval(prov_str)

@staticmethod
def get_resource_strats_from_prov(prov):
# if you provide the provenance of a resource pool output, this will
# return a dictionary of all the preceding resource pool entries that
# led to that one specific output:
# {rpool entry}: {that entry's provenance}
# {rpool entry}: {that entry's provenance}
strat_resource = {}
if isinstance(prov, str):
resource = prov.split(":")[0]
strat_resource[resource] = prov
else:
for spot, entry in enumerate(prov):
if isinstance(entry, list):
resource = entry[-1].split(":")[0]
strat_resource[resource] = entry
elif isinstance(entry, str):
resource = entry.split(":")[0]
strat_resource[resource] = entry
return strat_resource


class ResourcePool(_Pool):
"""A pool of Resources."""
Expand Down Expand Up @@ -3097,6 +3076,30 @@ def post_process(

return wf, post_labels

@staticmethod
def get_resource_strats_from_prov(prov: list | str) -> dict[str, list | str]:
"""Return all entries that led to this provenance.
If you provide the provenance of a resource pool output, this will
return a dictionary of all the preceding resource pool entries that
led to that one specific output:
{rpool entry}: {that entry's provenance}
{rpool entry}: {that entry's provenance}
"""
strat_resource: dict[str, list | str] = {}
if isinstance(prov, str):
resource = prov.split(":")[0]
strat_resource[resource] = prov
else:
for entry in prov:
if isinstance(entry, list):
resource = entry[-1].split(":")[0]
strat_resource[resource] = entry
elif isinstance(entry, str):
resource = entry.split(":")[0]
strat_resource[resource] = entry
return strat_resource

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
Expand Down

0 comments on commit 748b98e

Please sign in to comment.