mattrasmus
commented
1 year ago Hi @PhilPalmer thanks for posting your question. It looks like you understand the issue, but I'll explain the error as well just to confirm it. Looking at the error, it's happening in this task:
def run_netmhcpan(
peptides: pd.DataFrame,
hla_alleles: pd.DataFrame,
mhc_type: str,
peptides_dir: Path,
peptide_chunk_size: int = 500,
results_dir: Path = Path("."),
netmhcpan_cmd: str = "netMHCpan",
):
"""
Run NetMHCpan or NetMHCIIpan for a given set of peptides and HLA alleles, splitting the input peptides
into chunks and running netmhcpan for each chunk before concatenating the results into the per allele files.
"""
# Split peptides into chunks
peptide_paths = chunk_peptides(peptides, peptides_dir, peptide_chunk_size)
# NOTE: Since `chunk_peptides()` is a task, `peptide_paths` will be a lazy expression of a list.
# Although in redun we can do a lot with lazy expressions, using them in a for-loop is one we can't do because
# because for-loops need to run eagerly, before we know the full length of the list (it will be computed some time later).
# For each allele, run netmhcpan for each peptide chunk
peptide_allele_paths = [
predict_affinity_netmhcpan(
allele,
peptide_path,
mhc_type,
results_dir,
netmhcpan_cmd,
)
for allele in hla_alleles["allele"].values
# NOTE: This is the for-loop that's failing.
for peptide_path in peptide_paths
]
# NOTE: Concatenate the results for each allele using redun.functools.flatten(list_of_lists)
return flatten(peptide_allele_paths)So there are a couple of solutions you can do. If you know what the length of a lazy expression of a list will be, you can inform the redun scheduler by using slicing:
peptide_paths = chunk_peptides(peptides, peptides_dir, peptide_chunk_size)
# peptide_paths is an expression of a list of unknown length.
peptides_paths = peptide_paths[:length]
# This is now a regular list of expressions (that represent each item).
# This for-loop (or list comprehension) will now work
for file in peptide_paths:
# file is an expression, but that should be ok for you.
# ...Given your chunking logic, it looks like you would need something like length = math.ceil(len(peptides) / peptide_chunk_size).
Another solution is to force the expression to evaluate to a concrete value by passing it into another task. Parameters of a task always start as concrete (i.e. non-lazy) values.
def run_netmhcpan(
peptides: pd.DataFrame,
hla_alleles: pd.DataFrame,
mhc_type: str,
peptides_dir: Path,
peptide_chunk_size: int = 500,
results_dir: Path = Path("."),
netmhcpan_cmd: str = "netMHCpan",
):
"""
Run NetMHCpan or NetMHCIIpan for a given set of peptides and HLA alleles, splitting the input peptides
into chunks and running netmhcpan for each chunk before concatenating the results into the per allele files.
"""
# Split peptides into chunks
peptide_paths = chunk_peptides(peptides, peptides_dir, peptide_chunk_size)
return run_netmhcpan_chunks(
peptide_paths,
hla_alleles,
mhc_type,
results_dir,
netmhcpan_cmd
)
def run_netmhcpan_chunks(
peptide_paths,
hla_alleles: pd.DataFrame,
mhc_type: str,
results_dir: Path = Path("."),
netmhcpan_cmd: str = "netMHCpan",
):
# NOTE: peptide_paths is concrete now and can be used in a list comprehension.
# For each allele, run netmhcpan for each peptide chunk
peptide_allele_paths = [
predict_affinity_netmhcpan(
allele,
peptide_path,
mhc_type,
results_dir,
netmhcpan_cmd,
)
for allele in hla_alleles["allele"].values
for peptide_path in peptide_paths
]
return flatten(peptide_allele_paths)While this works, you might not like to split tasks so often. You are right that map_ and starmap from redun.functools help with mapping over a lazy expression of a list. Here is how that would look.
def run_netmhcpan(
peptides: pd.DataFrame,
hla_alleles: pd.DataFrame,
mhc_type: str,
peptides_dir: Path,
peptide_chunk_size: int = 500,
results_dir: Path = Path("."),
netmhcpan_cmd: str = "netMHCpan",
):
"""
Run NetMHCpan or NetMHCIIpan for a given set of peptides and HLA alleles, splitting the input peptides
into chunks and running netmhcpan for each chunk before concatenating the results into the per allele files.
"""
# Split peptides into chunks
peptide_paths = chunk_peptides(peptides, peptides_dir, peptide_chunk_size)
peptide_allele_paths = []
for allele in hla_alleles["allele"].values:
# Partially apply all the arguments that don't change.
predict = predict_affinity_netmhcpan.partial(allele=allele, mhc_type=mhc_type, results_dir=results_dir, netmhcpan_cmd=netmhcpan)
# predict now just needs one more argument, peptide_path.
# To make the following easy to do, I would rewrite predict_affinity_netmhcpan so that peptide_path is the first parameter.
# Now we can map_ predict over the lazy list peptide_paths.
peptide_allele_paths.append(map_(predict, peptide_paths))
return peptide_allele_pathsLet me know if this helps.
PhilPalmer
Hi,
Thanks for developing a great workflow manager in Python!
I'm currently trying to implement a pipeline that performs a scatter and merge operation, however, I've been encountering issues and could not see any examples or documentation that addressed my use case.
The pipeline runs a tool called NetMHCpan which predicts binding for a given set of peptides (file) and MHC alleles (string). I would like to run a task that parallelises over both alleles and peptides, by splitting the peptides into batches, running these batches for each allele, and then performs another task to concatenate the per allele results.
Does redun support a way of doing this? I have tried using nested list comprehension, nested for loops and
itertools.productbut always get the following error. I suspect I might need to use a functools method such asstarmap_but I'm not quite sure how to implement this.Any help that you can provide would be greatly appreciated. Thanks in advance!
Here is the minimal reproducible example of my code: