my_wf.py

GeOdette-my_wf.py

@dataclass_json
@dataclass
class FilePair:
  r1: LatchFile
  r2: LatchFile
    
    
@task
def fromFilePairs(input_dir: LatchDir) -> typing.List[FilePair]:
  ...
  return [...]

@small_task
def pangolin_task(FASTA_file: FilePair) -> FilePair:

    # A reference to our output.
    local_dir = "/root/pangolins/"
    local_prefix = os.path.join(local_dir, "pango")

    temp_dir = "/root/temp_stuff/"
    temp_prefix = os.path.join(local_dir, "temp_out")
    # Defining the function

    _pangolin_cmd = [
        "pangolin",
        FASTA_file.r1.local_path,
        FASTA_file.r2.local_path,
        "--outdir",
        str(local_prefix),
        "--tempdir",
        str(temp_prefix),
        "--no-temp",
        "--alignment",
        "--analysis-mode fast",
        "--update",
        "--update-data",
    ]

    subprocess.run(_pangolin_cmd, check=True)

    return FilePair(str(local_prefix), str(temp_dir))

@workflow
def pangolin(input_dir: LatchDir) -> LatchDir:
    """Phylogenetic Assignment of Named Global Outbreak LINeages
    _pangolin_
    ----
    __metadata__:
        display_name: Phylogenetic Assignment of Named Global Outbreak LINeages
        author:
            name: GeOdette
            email: steveodettegeorge@gmail.com
            github:
        repository:
        license:
            id: MIT
    Args:
        FASTA_file:
          Input file
          __metadata__:
            display_name: FAST file
        out_dir:
          Output directory
          __metadata__:
            display_name: Output directory
        temp:
          Where the temp stuff should go
          __metadata__:
            display_name: Temp directory
    """

    pairs = fromFilePairs(input_dir=input_dir)
    mapped_out = map_task(pangolin_task)(FASTA_file=pairs)
    return file_pairs_to_dir(pairs=mapped_out)