Branching and splitting

Ask for help
8

Great. I will attach a snippet below that does slightly what I think you’re going through, and working. I tried with the nextflow/nextflow container image and it was able to see the files. Prepare for spoilers :stuck_out_tongue_winking_eye:

// Preparation of dummy data
// Output of makeCounts
Channel
  .of([file('/home/ubuntu/work/6d/db934dc132cab4147947e3a9d34b8b/PCR35036-35009_counts.csv'), 0],
      [file('/home/ubuntu/work/f4/a54dd91f349afc355871ee768eb85d/PCR35039-35012_counts.csv'), 3],
      [file('/home/ubuntu/work/14/14f97f85572b6c68f6cc314b161c93/PCR35037-35010_counts.csv'), 1],
      [file('/home/ubuntu/work/74/0b0c49fd1af48daeecbac698eea8c5/PCR35038-35011_counts.csv'), 2],
      [file('/home/ubuntu/work/a0/6ae399d518b948d98d6b9b8640d386/PCR35120-35016_counts.csv'), 'control']
  )
  .set { makeCount_output }

process AggregateCounts {
  debug true
  container 'nextflow/nextflow'

  input:
    path filepaths
    val types
  output:
    stdout

  script:
  def file = filepaths
  def times = types.join(' ')
  """
    echo python /usr/src/app/aggregate_counts.py \
    --file_paths ${file} \
    --timepoints ${times}
    stat ${file[0]}
  """
}

workflow {
  makeCount_output
    .multiMap {
      paths: it[0]
      types: it[1]
    }
    .set { AggregateCounts_input }
  AggregateCounts(AggregateCounts_input.paths.collect(),
                  AggregateCounts_input.types.collect())
}

Output:

N E X T F L O W  ~  version 23.09.2-edge
Launching `main.nf` [distracted_turing] DSL2 - revision: 56db98ffbf
executor >  local (1)
[3e/b26f26] process > AggregateCounts [100%] 1 of 1 ✔
python /usr/src/app/aggregate_counts.py --file_paths PCR35036-35009_counts.csv PCR35039-35012_counts.csv PCR35037-35010_counts.csv PCR35038-35011_counts.csv PCR35120-35016_counts.csv --timepoints 0 3 1 2 control
  File: ‘PCR35036-35009_counts.csv’ -> ‘/home/ubuntu/work/6d/db934dc132cab4147947e3a9d34b8b/PCR35036-35009_counts.csv’
  Size: 77        	Blocks: 0          IO Block: 4096   symbolic link
Device: 20h/32d	Inode: 21          Links: 1
Access: (0755/lrwxr-xr-x)  Uid: (    0/    root)   Gid: (    0/    root)
Access: 2023-10-06 16:03:54.494488994 +0000
Modify: 2023-10-06 16:03:54.494488994 +0000
Change: 2023-10-06 16:03:54.494488994 +0000
 Birth: -

You can also find below a slightly harder to read but more concise way of doing it using tuples:

// Preparation of dummy data
// Output of makeCounts
Channel
  .of([file('/home/ubuntu/work/6d/db934dc132cab4147947e3a9d34b8b/PCR35036-35009_counts.csv'), 0],
      [file('/home/ubuntu/work/f4/a54dd91f349afc355871ee768eb85d/PCR35039-35012_counts.csv'), 3],
      [file('/home/ubuntu/work/14/14f97f85572b6c68f6cc314b161c93/PCR35037-35010_counts.csv'), 1],
      [file('/home/ubuntu/work/74/0b0c49fd1af48daeecbac698eea8c5/PCR35038-35011_counts.csv'), 2],
      [file('/home/ubuntu/work/a0/6ae399d518b948d98d6b9b8640d386/PCR35120-35016_counts.csv'), 'control']
  )
  .set { makeCount_output }

process AggregateCounts {
  debug true
  container 'nextflow/nextflow'

  input:
    tuple path(filepaths), val(types)
  output:
    stdout

  script:
  def file = filepaths
  def times = types.join(" ")
  """
    echo python /usr/src/app/aggregate_counts.py \
    --file_paths ${file} \
    --timepoints ${times}
    stat ${file[0]}
  """
}

workflow {
  makeCount_output
    .toList()
    .map { it.transpose() }
    .set { AggregateCounts_input }

  AggregateCounts(AggregateCounts_input)
}