BiopetFifoPipe.scala 2.63 KB
Newer Older
Peter van 't Hof's avatar
Peter van 't Hof committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
package nl.lumc.sasc.biopet.core

import java.io.File

import nl.lumc.sasc.biopet.utils.config.Configurable

/**
 * Created by pjvan_thof on 9/29/15.
 */
class BiopetFifoPipe(val root: Configurable,
                     protected var commands: List[BiopetCommandLineFunction]) extends BiopetCommandLineFunction {

  def fifos: List[File] = {
    val outputs: Map[BiopetCommandLineFunction, Seq[File]] = try {
      commands.map(x => x -> x.outputs).toMap
    } catch {
      case e: NullPointerException => Map()
    }

    val inputs: Map[BiopetCommandLineFunction, Seq[File]] = try {
      commands.map(x => x -> x.inputs).toMap
    } catch {
      case e: NullPointerException => Map()
    }

    for (
      cmdOutput <- commands;
      cmdInput <- commands if cmdOutput != cmdInput && outputs.contains(cmdOutput);
      outputFile <- outputs(cmdOutput) if inputs.contains(cmdInput);
      inputFile <- inputs(cmdInput) if outputFile == inputFile
    ) yield outputFile
  }

  override def beforeGraph(): Unit = {
    val outputs: Map[BiopetCommandLineFunction, Seq[File]] = try {
      commands.map(x => x -> x.outputs).toMap
    } catch {
      case e: NullPointerException => Map()
    }

    val inputs: Map[BiopetCommandLineFunction, Seq[File]] = try {
      commands.map(x => x -> x.inputs).toMap
    } catch {
      case e: NullPointerException => Map()
    }

    val fifoFiles = fifos

    outputFiles = outputs.values.toList.flatten.filter(!fifoFiles.contains(_))
    outputFiles = outputFiles.distinct

    deps = inputs.values.toList.flatten.filter(!fifoFiles.contains(_))
    deps = deps.distinct
  }

  override def beforeCmd(): Unit = {
    commands.foreach{ cmd =>
      cmd.beforeGraph()
      cmd.internalBeforeGraph()
      cmd.beforeCmd()
    }
  }

  def cmdLine = {
    val fifosFiles = this.fifos
    fifosFiles.filter(_.exists()).map(required("rm", _)).mkString("\n\n", " \n", " \n\n") +
      fifosFiles.map(required("mkfifo", _)).mkString("\n\n", "\n", "\n\n") +
      commands.map(_.commandLine).mkString("\n\n", " & \n", " & \n\n") +
      BiopetFifoPipe.waitScript +
      fifosFiles.map(required("rm", _)).mkString("\n\n", " \n", " \n\n") +
      BiopetFifoPipe.endScript
  }
}

object BiopetFifoPipe {
  val waitScript =
    """
      |
      |FAIL="0"
      |
      |for job in `jobs -p`
      |do
      |echo $job
      |    wait $job || let "FAIL+=1"
      |done
      |
      |echo $FAIL
      |
    """.stripMargin

  val endScript =
    """
echo $FAIL
      |
      |if [ "$FAIL" == "0" ];
      |then
      |echo "BiopetFifoPipe Done"
      |else
      |echo BiopetFifoPipe "FAIL! ($FAIL)"
      |exit $FAIL
      |fi
      |
      |
    """.stripMargin
}