SummaryDb.scala 33.8 KB
Newer Older
pjvan_thof's avatar
pjvan_thof committed
1
/**
2
3
4
5
6
7
8
9
10
11
12
13
14
  * Biopet is built on top of GATK Queue for building bioinformatic
  * pipelines. It is mainly intended to support LUMC SHARK cluster which is running
  * SGE. But other types of HPC that are supported by GATK Queue (such as PBS)
  * should also be able to execute Biopet tools and pipelines.
  *
  * Copyright 2014 Sequencing Analysis Support Core - Leiden University Medical Center
  *
  * Contact us at: sasc@lumc.nl
  *
  * A dual licensing mode is applied. The source code within this project is freely available for non-commercial use under an AGPL
  * license; For commercial users or users who do not want to follow the AGPL
  * license, please contact us to obtain a separate license.
  */
Peter van 't Hof's avatar
Peter van 't Hof committed
15
package nl.lumc.sasc.biopet.utils.summary.db
16

Peter van 't Hof's avatar
Peter van 't Hof committed
17
import nl.lumc.sasc.biopet.utils.ConfigUtils
18
import nl.lumc.sasc.biopet.utils.summary.db.Schema._
Peter van 't Hof's avatar
Peter van 't Hof committed
19
import slick.driver.H2Driver.api._
20

Peter van 't Hof's avatar
Peter van 't Hof committed
21
import scala.concurrent.duration.Duration
22
23
import scala.concurrent.{Await, ExecutionContext, Future}
import java.io.{Closeable, File}
Peter van 't Hof's avatar
Peter van 't Hof committed
24
import java.sql.Date
25

26
27
28
import nl.lumc.sasc.biopet.utils.summary.db.SummaryDb._
import nl.lumc.sasc.biopet.utils.summary.db.SummaryDb.Implicts._

Peter van 't Hof's avatar
Peter van 't Hof committed
29
import scala.language.implicitConversions
30

31
/**
32
33
34
35
  * This class interface wityh a summary database
  *
  * Created by pjvanthof on 05/02/2017.
  */
36
trait SummaryDb extends Closeable {
37

Peter van 't Hof's avatar
WIP    
Peter van 't Hof committed
38
39
  implicit val ec: ExecutionContext

40
  def db: Database
Peter van 't Hof's avatar
Peter van 't Hof committed
41

42
43
  def close(): Unit = db.close()

44
  /** This will return all runs that match the critiria given */
Peter van 't Hof's avatar
Peter van 't Hof committed
45
46
47
  def getRuns(runId: Option[Int] = None,
              runName: Option[String] = None,
              outputDir: Option[String] = None): Future[Seq[Run]] = {
48
49
50
51
    val q = runs.filter { run =>
      List(
        runId.map(run.id === _),
        runName.map(run.runName === _),
52
        outputDir.map(run.outputDir === _)
53
54
55
      ).collect({ case Some(criteria) => criteria })
        .reduceLeftOption(_ && _)
        .getOrElse(true: Rep[Boolean])
56
57
58
59
    }
    db.run(q.result)
  }

60
  /** This will return all samples that match given criteria */
Peter van 't Hof's avatar
Peter van 't Hof committed
61
62
63
  def getSamples(sampleId: Option[Int] = None,
                 runId: Option[Int] = None,
                 name: Option[String] = None): Future[Seq[Sample]] = {
64
65
66
67
    val q = samples.filter { sample =>
      List(
        sampleId.map(sample.id === _),
        runId.map(sample.runId === _),
68
        name.map(sample.name === _)
69
70
71
      ).collect({ case Some(criteria) => criteria })
        .reduceLeftOption(_ && _)
        .getOrElse(true: Rep[Boolean])
72
    }
73
    db.run(q.result)
74
75
  }

76
77
  /** Return samplId of a specific runId + sampleName */
  def getSampleId(runId: Int, sampleName: String): Future[Option[Int]] = {
Peter van 't Hof's avatar
Peter van 't Hof committed
78
79
80
    getSamples(runId = Some(runId), name = Some(sampleName)).map(_.headOption.map(_.id))
  }

Peter van 't Hof's avatar
Peter van 't Hof committed
81
  /** Return sampleName of a specific sampleId */
82
83
  def getSampleName(sampleId: Int): Future[Option[String]] = {
    getSamples(sampleId = Some(sampleId)).map(_.headOption.map(_.name))
84
85
  }

86
  /** Return sample tags of a specific sample as a map */
Peter van 't Hof's avatar
Peter van 't Hof committed
87
88
89
90
91
  def getSampleTags(sampleId: Int): Future[Option[Map[String, Any]]] = {
    db.run(samples.filter(_.id === sampleId).map(_.tags).result)
      .map(_.headOption.flatten.map(ConfigUtils.jsonTextToMap))
  }

92
  /** This returns all libraries that match the given criteria */
Peter van 't Hof's avatar
Peter van 't Hof committed
93
94
95
96
  def getLibraries(libId: Option[Int] = None,
                   name: Option[String] = None,
                   runId: Option[Int] = None,
                   sampleId: Option[Int] = None): Future[Seq[Library]] = {
Peter van 't Hof's avatar
Peter van 't Hof committed
97
98
99
100
101
102
    val q = libraries.filter { lib =>
      List(
        libId.map(lib.id === _),
        sampleId.map(lib.sampleId === _),
        runId.map(lib.runId === _),
        name.map(lib.name === _) // not a condition as `criteriaRoast` evaluates to `None`
103
104
105
      ).collect({ case Some(criteria) => criteria })
        .reduceLeftOption(_ && _)
        .getOrElse(true: Rep[Boolean])
Peter van 't Hof's avatar
Peter van 't Hof committed
106
    }
107
    db.run(q.result)
Peter van 't Hof's avatar
Peter van 't Hof committed
108
109
  }

110
  /** Return a libraryId for a specific combination */
111
112
113
  def getLibraryId(runId: Int, sampleId: Int, name: String): Future[Option[Int]] = {
    getLibraries(runId = Some(runId), sampleId = Some(sampleId), name = Some(name))
      .map(_.headOption.map(_.id))
Peter van 't Hof's avatar
Peter van 't Hof committed
114
115
  }

116
117
118
119
120
  /** Return a libraryId for a specific combination */
  def getLibraryName(libraryId: Int): Future[Option[String]] = {
    getLibraries(libId = Some(libraryId)).map(_.headOption.map(_.name))
  }

121
  /** Return library tags as a map */
Peter van 't Hof's avatar
Peter van 't Hof committed
122
123
124
  def getLibraryTags(libId: Int): Future[Option[Map[String, Any]]] = {
    db.run(libraries.filter(_.id === libId).map(_.tags).result)
      .map(_.headOption.flatten.map(ConfigUtils.jsonTextToMap))
125
126
  }

127
  /** Get all pipelines that match given criteria */
Peter van 't Hof's avatar
Peter van 't Hof committed
128
129
130
  def getPipelines(pipelineId: Option[Int] = None,
                   name: Option[String] = None,
                   runId: Option[Int] = None): Future[Seq[Pipeline]] = {
131
132
133
134
    val q = pipelines.filter { lib =>
      List(
        pipelineId.map(lib.id === _),
        runId.map(lib.runId === _),
135
        name.map(lib.name === _)
136
137
138
      ).collect({ case Some(criteria) => criteria })
        .reduceLeftOption(_ && _)
        .getOrElse(true: Rep[Boolean])
139
    }
140
    db.run(q.result)
141
142
  }

Peter van 't Hof's avatar
Peter van 't Hof committed
143
  /** Return pipelineId of a specific pipelineName */
144
  def getPipelineId(runId: Int, pipelineName: String): Future[Option[Int]] = {
Peter van 't Hof's avatar
Peter van 't Hof committed
145
146
147
    getPipelines(runId = Some(runId), name = Some(pipelineName)).map(_.headOption.map(_.id))
  }

Peter van 't Hof's avatar
Peter van 't Hof committed
148
149
150
151
152
  /** Return name of a pipeline */
  def getPipelineName(pipelineId: Int): Future[Option[String]] = {
    getPipelines(pipelineId = Some(pipelineId)).map(_.headOption.map(_.name))
  }

153
  /** Return all module with given criteria */
Peter van 't Hof's avatar
Peter van 't Hof committed
154
155
156
157
  def getModules(moduleId: Option[Int] = None,
                 name: Option[String] = None,
                 runId: Option[Int] = None,
                 pipelineId: Option[Int] = None): Future[Seq[Module]] = {
158
159
160
161
162
    val q = modules.filter { lib =>
      List(
        moduleId.map(lib.id === _),
        runId.map(lib.runId === _),
        pipelineId.map(lib.pipelineId === _),
163
        name.map(lib.name === _)
164
165
166
      ).collect({ case Some(criteria) => criteria })
        .reduceLeftOption(_ && _)
        .getOrElse(true: Rep[Boolean])
167
    }
168
    db.run(q.result)
169
170
  }

Peter van 't Hof's avatar
Peter van 't Hof committed
171
  /** Return moduleId of a specific moduleName */
172
173
174
  def getmoduleId(runId: Int, moduleName: String, pipelineId: Int): Future[Option[Int]] = {
    getModules(runId = Some(runId), name = Some(moduleName), pipelineId = Some(pipelineId))
      .map(_.headOption.map(_.id))
Peter van 't Hof's avatar
Peter van 't Hof committed
175
176
  }

Peter van 't Hof's avatar
Peter van 't Hof committed
177
  /** Returns name of a module */
178
179
180
  def getModuleName(pipelineId: Int, moduleId: Int): Future[Option[String]] = {
    getModules(pipelineId = Some(pipelineId), moduleId = Some(moduleId))
      .map(_.headOption.map(_.name))
Peter van 't Hof's avatar
Peter van 't Hof committed
181
182
  }

183
  /** Return a Query for [[Stats]] */
184
185
186
187
188
189
190
191
  def statsFilter(
      runId: Option[Int] = None,
      pipeline: Option[PipelineQuery] = None,
      module: Option[ModuleQuery] = None,
      sample: Option[SampleQuery] = None,
      library: Option[LibraryQuery] = None,
      mustHaveSample: Boolean = false,
      mustHaveLibrary: Boolean = false): slick.driver.H2Driver.api.Query[Stats, Stat, Seq] = {
Peter van 't Hof's avatar
Peter van 't Hof committed
192
193
    var f: Query[Stats, Stats#TableElementType, Seq] = stats
    runId.foreach(r => f = f.filter(_.runId === r))
Peter van 't Hof's avatar
Peter van 't Hof committed
194
    f = pipeline match {
195
196
197
198
      case Some(p: PipelineId) => f.filter(_.pipelineId === p.id)
      case Some(p: PipelineName) =>
        f.join(pipelines).on(_.pipelineId === _.id).filter(_._2.name === p.name).map(_._1)
      case _ => f
Peter van 't Hof's avatar
Peter van 't Hof committed
199
200
    }
    f = module match {
201
202
203
204
205
      case Some(m: ModuleId) => f.filter(_.moduleId === m.id)
      case Some(m: ModuleName) =>
        f.join(modules).on(_.moduleId === _.id).filter(_._2.name === m.name).map(_._1)
      case Some(NoModule) => f.filter(_.moduleId.isEmpty)
      case _ => f
Peter van 't Hof's avatar
Peter van 't Hof committed
206
207
    }
    f = sample match {
208
209
210
211
212
      case Some(s: SampleId) => f.filter(_.sampleId === s.id)
      case Some(s: SampleName) =>
        f.join(samples).on(_.sampleId === _.id).filter(_._2.name === s.name).map(_._1)
      case Some(NoSample) => f.filter(_.sampleId.isEmpty)
      case _ => f
Peter van 't Hof's avatar
Peter van 't Hof committed
213
    }
214
    f = library match {
215
216
217
218
219
      case Some(l: LibraryId) => f.filter(_.libraryId === l.id)
      case Some(l: LibraryName) =>
        f.join(libraries).on(_.libraryId === _.id).filter(_._2.name === l.name).map(_._1)
      case Some(NoLibrary) => f.filter(_.libraryId.isEmpty)
      case _ => f
Peter van 't Hof's avatar
Peter van 't Hof committed
220
    }
221
222
223

    if (mustHaveSample) f = f.filter(_.sampleId.nonEmpty)
    if (mustHaveLibrary) f = f.filter(_.libraryId.nonEmpty)
224
    f
Peter van 't Hof's avatar
Peter van 't Hof committed
225
  }
226

227
  /** Return all stats that match given criteria */
Peter van 't Hof's avatar
Peter van 't Hof committed
228
229
230
231
232
233
234
  def getStats(runId: Option[Int] = None,
               pipeline: Option[PipelineQuery] = None,
               module: Option[ModuleQuery] = None,
               sample: Option[SampleQuery] = None,
               library: Option[LibraryQuery] = None,
               mustHaveSample: Boolean = false,
               mustHaveLibrary: Boolean = false): Future[Seq[Stat]] = {
235
236
    db.run(
      statsFilter(runId, pipeline, module, sample, library, mustHaveSample, mustHaveLibrary).result)
237
238
239
  }

  /** Return number of results */
Peter van 't Hof's avatar
Peter van 't Hof committed
240
241
242
243
244
245
246
  def getStatsSize(runId: Option[Int] = None,
                   pipeline: Option[PipelineQuery] = None,
                   module: Option[ModuleQuery] = None,
                   sample: Option[SampleQuery] = None,
                   library: Option[LibraryQuery] = None,
                   mustHaveSample: Boolean = false,
                   mustHaveLibrary: Boolean = false): Future[Int] = {
247
248
    db.run(
      statsFilter(runId, pipeline, module, sample, library, mustHaveSample, mustHaveLibrary).size.result)
249
250
  }

251
  /** Get a single stat as [[Map[String, Any]] */
Peter van 't Hof's avatar
Peter van 't Hof committed
252
253
254
255
256
  def getStat(runId: Int,
              pipeline: PipelineQuery,
              module: ModuleQuery = NoModule,
              sample: SampleQuery = NoSample,
              library: LibraryQuery = NoLibrary): Future[Option[Map[String, Any]]] = {
Peter van 't Hof's avatar
Peter van 't Hof committed
257
    getStats(Some(runId), pipeline, module, sample, library)
258
      .map(_.headOption.map(x => ConfigUtils.jsonTextToMap(x.content)))
259
260
  }

Peter van 't Hof's avatar
Peter van 't Hof committed
261
262
263
264
265
  def getStatKeys(runId: Int,
                  pipeline: PipelineQuery,
                  module: ModuleQuery = NoModule,
                  sample: SampleQuery = NoSample,
                  library: LibraryQuery = NoLibrary,
Peter van 't Hof's avatar
Peter van 't Hof committed
266
267
                  keyValues: Map[String, List[String]]): Map[String, Option[Any]] = {
    val stats = Await.result(getStat(runId, pipeline, module, sample, library), Duration.Inf)
pjvan_thof's avatar
pjvan_thof committed
268
269
270
    if (module == ModuleName("rna")) {
      ""
    }
Peter van 't Hof's avatar
Peter van 't Hof committed
271
272
273
274
    keyValues.map {
      case (key, path) =>
        stats match {
          case Some(map) => key -> ConfigUtils.getValueFromPath(map, path)
275
          case None => key -> None
Peter van 't Hof's avatar
Peter van 't Hof committed
276
        }
Peter van 't Hof's avatar
Peter van 't Hof committed
277
278
279
    }
  }

280
281
282
283
284
285
286
287
288
  def getStatsForSamples(
      runId: Int,
      pipeline: PipelineQuery,
      module: ModuleQuery = NoModule,
      sample: Option[SampleQuery] = None,
      keyValues: Map[String, List[String]]): Map[Int, Map[String, Option[Any]]] = {
    val samples = Await.result(getSamples(runId = Some(runId), sampleId = sample.collect {
      case s: SampleId => s.id
    }, name = sample.collect { case s: SampleName => s.name }), Duration.Inf)
Peter van 't Hof's avatar
Peter van 't Hof committed
289
    (for (s <- samples) yield {
290
291
292
293
294
295
      s.id -> getStatKeys(runId,
                          pipeline,
                          module,
                          SampleId(s.id),
                          NoLibrary,
                          keyValues = keyValues)
Peter van 't Hof's avatar
Peter van 't Hof committed
296
297
298
    }).toMap
  }

299
300
301
302
303
304
305
306
  def getStatsForLibraries(
      runId: Int,
      pipeline: PipelineQuery,
      module: ModuleQuery = NoModule,
      sampleId: Option[Int] = None,
      keyValues: Map[String, List[String]]): Map[(Int, Int), Map[String, Option[Any]]] = {
    val libraries =
      Await.result(getLibraries(runId = Some(runId), sampleId = sampleId), Duration.Inf)
Peter van 't Hof's avatar
Peter van 't Hof committed
307
    (for (lib <- libraries) yield {
308
309
310
311
312
313
      (lib.sampleId, lib.id) -> getStatKeys(runId,
                                            pipeline,
                                            module,
                                            SampleId(lib.sampleId),
                                            LibraryId(lib.id),
                                            keyValues = keyValues)
Peter van 't Hof's avatar
Peter van 't Hof committed
314
315
316
    }).toMap
  }

Peter van 't Hof's avatar
Peter van 't Hof committed
317
318
319
320
321
322
  def settingsFilter(runId: Option[Int] = None,
                     pipeline: Option[PipelineQuery] = None,
                     module: Option[ModuleQuery] = None,
                     sample: Option[SampleQuery] = None,
                     library: Option[LibraryQuery] = None,
                     mustHaveSample: Boolean = false,
323
324
                     mustHaveLibrary: Boolean = false)
    : slick.driver.H2Driver.api.Query[Settings, Setting, Seq] = {
Peter van 't Hof's avatar
Peter van 't Hof committed
325
326
    var f: Query[Settings, Settings#TableElementType, Seq] = settings
    runId.foreach(r => f = f.filter(_.runId === r))
Peter van 't Hof's avatar
Peter van 't Hof committed
327
    f = pipeline match {
328
329
330
331
      case Some(p: PipelineId) => f.filter(_.pipelineId === p.id)
      case Some(p: PipelineName) =>
        f.join(pipelines).on(_.pipelineId === _.id).filter(_._2.name === p.name).map(_._1)
      case _ => f
Peter van 't Hof's avatar
Peter van 't Hof committed
332
333
    }
    f = module match {
334
335
336
337
338
      case Some(m: ModuleId) => f.filter(_.moduleId === m.id)
      case Some(m: ModuleName) =>
        f.join(modules).on(_.moduleId === _.id).filter(_._2.name === m.name).map(_._1)
      case Some(NoModule) => f.filter(_.moduleId.isEmpty)
      case _ => f
Peter van 't Hof's avatar
Peter van 't Hof committed
339
340
    }
    f = sample match {
341
342
343
344
345
      case Some(s: SampleId) => f.filter(_.sampleId === s.id)
      case Some(s: SampleName) =>
        f.join(samples).on(_.sampleId === _.id).filter(_._2.name === s.name).map(_._1)
      case Some(NoSample) => f.filter(_.sampleId.isEmpty)
      case _ => f
Peter van 't Hof's avatar
Peter van 't Hof committed
346
    }
347
    f = library match {
348
349
350
351
352
      case Some(l: LibraryId) => f.filter(_.libraryId === l.id)
      case Some(l: LibraryName) =>
        f.join(libraries).on(_.libraryId === _.id).filter(_._2.name === l.name).map(_._1)
      case Some(NoLibrary) => f.filter(_.libraryId.isEmpty)
      case _ => f
Peter van 't Hof's avatar
Peter van 't Hof committed
353
354
355
356
    }

    if (mustHaveSample) f = f.filter(_.sampleId.nonEmpty)
    if (mustHaveLibrary) f = f.filter(_.libraryId.nonEmpty)
357
    f
358
  }
359

360
  /** Return all settings that match the given criteria */
Peter van 't Hof's avatar
Peter van 't Hof committed
361
362
363
364
365
  def getSettings(runId: Option[Int] = None,
                  pipeline: Option[PipelineQuery] = None,
                  module: Option[ModuleQuery] = None,
                  sample: Option[SampleQuery] = None,
                  library: Option[LibraryQuery] = None): Future[Seq[Setting]] = {
Peter van 't Hof's avatar
Peter van 't Hof committed
366
    db.run(settingsFilter(runId, pipeline, module, sample, library).result)
367
368
  }

369
  /** Return a specific setting as [[Map[String, Any]] */
Peter van 't Hof's avatar
Peter van 't Hof committed
370
371
372
373
374
  def getSetting(runId: Int,
                 pipeline: PipelineQuery,
                 module: ModuleQuery = NoModule,
                 sample: SampleQuery = NoSample,
                 library: LibraryQuery = NoLibrary): Future[Option[Map[String, Any]]] = {
Peter van 't Hof's avatar
Peter van 't Hof committed
375
    getSettings(Some(runId), Some(pipeline), module, sample, library)
376
      .map(_.headOption.map(x => ConfigUtils.jsonTextToMap(x.content)))
377
  }
Peter van 't Hof's avatar
Peter van 't Hof committed
378

Peter van 't Hof's avatar
Peter van 't Hof committed
379
380
381
382
383
  def getSettingKeys(runId: Int,
                     pipeline: PipelineQuery,
                     module: ModuleQuery = NoModule,
                     sample: SampleQuery = NoSample,
                     library: LibraryQuery = NoLibrary,
Peter van 't Hof's avatar
Peter van 't Hof committed
384
385
                     keyValues: Map[String, List[String]]): Map[String, Option[Any]] = {
    val stats = Await.result(getSetting(runId, pipeline, module, sample, library), Duration.Inf)
Peter van 't Hof's avatar
Peter van 't Hof committed
386
387
388
389
    keyValues.map {
      case (key, path) =>
        stats match {
          case Some(map) => key -> ConfigUtils.getValueFromPath(map, path)
390
          case None => key -> None
Peter van 't Hof's avatar
Peter van 't Hof committed
391
        }
Peter van 't Hof's avatar
Peter van 't Hof committed
392
393
394
    }
  }

395
396
397
398
399
400
  def getSettingsForSamples(
      runId: Int,
      pipeline: PipelineQuery,
      module: ModuleQuery = NoModule,
      sampleId: Option[Int] = None,
      keyValues: Map[String, List[String]]): Map[Int, Map[String, Option[Any]]] = {
Peter van 't Hof's avatar
Peter van 't Hof committed
401
402
    val samples = Await.result(getSamples(runId = Some(runId), sampleId = sampleId), Duration.Inf)
    (for (sample <- samples) yield {
403
404
405
406
407
408
      sample.id -> getSettingKeys(runId,
                                  pipeline,
                                  module,
                                  SampleId(sample.id),
                                  NoLibrary,
                                  keyValues = keyValues)
Peter van 't Hof's avatar
Peter van 't Hof committed
409
410
411
    }).toMap
  }

412
413
414
415
416
417
418
419
  def getSettingsForLibraries(
      runId: Int,
      pipeline: PipelineQuery,
      module: ModuleQuery = NoModule,
      sampleId: Option[Int] = None,
      keyValues: Map[String, List[String]]): Map[(Int, Int), Map[String, Option[Any]]] = {
    val libraries =
      Await.result(getLibraries(runId = Some(runId), sampleId = sampleId), Duration.Inf)
Peter van 't Hof's avatar
Peter van 't Hof committed
420
    (for (lib <- libraries) yield {
421
422
423
424
425
426
      (lib.sampleId, lib.id) -> getSettingKeys(runId,
                                               pipeline,
                                               module,
                                               SampleId(lib.sampleId),
                                               LibraryId(lib.id),
                                               keyValues = keyValues)
Peter van 't Hof's avatar
Peter van 't Hof committed
427
428
429
    }).toMap
  }

430
  /** Return a [[Query]] for [[Files]] */
Peter van 't Hof's avatar
Peter van 't Hof committed
431
432
433
434
435
436
437
438
439
  def filesFilter(runId: Option[Int] = None,
                  pipeline: Option[PipelineQuery] = None,
                  module: Option[ModuleQuery] = None,
                  sample: Option[SampleQuery] = None,
                  library: Option[LibraryQuery] = None,
                  key: Option[String] = None,
                  pipelineName: Option[String] = None,
                  moduleName: Option[Option[String]] = None,
                  sampleName: Option[Option[String]] = None,
440
441
                  libraryName: Option[Option[String]] = None)
    : slick.driver.H2Driver.api.Query[Files, Files#TableElementType, Seq] = {
Peter van 't Hof's avatar
Peter van 't Hof committed
442
443
444
    var f: Query[Files, Files#TableElementType, Seq] = files
    runId.foreach(r => f = f.filter(_.runId === r))
    key.foreach(r => f = f.filter(_.key === r))
445
446

    f = pipeline match {
447
448
449
450
      case Some(p: PipelineId) => f.filter(_.pipelineId === p.id)
      case Some(p: PipelineName) =>
        f.join(pipelines).on(_.pipelineId === _.id).filter(_._2.name === p.name).map(_._1)
      case _ => f
451
452
    }
    f = module match {
453
454
455
456
457
      case Some(m: ModuleId) => f.filter(_.moduleId === m.id)
      case Some(m: ModuleName) =>
        f.join(modules).on(_.moduleId === _.id).filter(_._2.name === m.name).map(_._1)
      case Some(NoModule) => f.filter(_.moduleId.isEmpty)
      case _ => f
458
459
    }
    f = sample match {
460
461
462
463
464
      case Some(s: SampleId) => f.filter(_.sampleId === s.id)
      case Some(s: SampleName) =>
        f.join(samples).on(_.sampleId === _.id).filter(_._2.name === s.name).map(_._1)
      case Some(NoSample) => f.filter(_.sampleId.isEmpty)
      case _ => f
465
466
    }
    f = library match {
467
468
469
470
471
      case Some(l: LibraryId) => f.filter(_.libraryId === l.id)
      case Some(l: LibraryName) =>
        f.join(libraries).on(_.libraryId === _.id).filter(_._2.name === l.name).map(_._1)
      case Some(NoLibrary) => f.filter(_.libraryId.isEmpty)
      case _ => f
472
    }
473
474
475
    f
  }

476
  /** Returns all [[Files]] with the given criteria */
477
478
479
480
481
  def getFiles(runId: Option[Int] = None,
               pipeline: Option[PipelineQuery] = None,
               module: Option[ModuleQuery] = None,
               sample: Option[SampleQuery] = None,
               library: Option[LibraryQuery] = None,
482
               key: Option[String] = None): Future[Seq[Schema.File]] = {
483
    db.run(filesFilter(runId, pipeline, module, sample, library, key).result)
Peter van 't Hof's avatar
Peter van 't Hof committed
484
485
  }

486
487
488
489
490
491
492
493
494
495
496
497
498
499
  def getFile(runId: Int,
              pipeline: PipelineQuery,
              module: ModuleQuery = NoModule,
              sample: SampleQuery = NoSample,
              library: LibraryQuery = NoLibrary,
              key: String): Future[Option[Schema.File]] = {
    db.run(
        filesFilter(Some(runId),
                    Some(pipeline),
                    Some(module),
                    Some(sample),
                    Some(library),
                    Some(key)).result)
      .map(_.headOption)
500
501
  }

502
  /** Returns a [[Query]] for [[Executables]] */
503
504
505
  def executablesFilter(
      runId: Option[Int],
      toolName: Option[String]): slick.driver.H2Driver.api.Query[Executables, Executable, Seq] = {
506
507
508
509
510
511
512
    var q: Query[Executables, Executables#TableElementType, Seq] = executables
    runId.foreach(r => q = q.filter(_.runId === r))
    toolName.foreach(r => q = q.filter(_.toolName === r))
    q
  }

  /** Return all executables with given criteria */
513
514
  def getExecutables(runId: Option[Int] = None,
                     toolName: Option[String] = None): Future[Seq[Executable]] = {
515
516
517
518
519
    db.run(executablesFilter(runId, toolName).result)
  }

}

Peter van 't Hof's avatar
WIP    
Peter van 't Hof committed
520
class SummaryDbReadOnly(val db: Database)(implicit val ec: ExecutionContext) extends SummaryDb
521

Peter van 't Hof's avatar
WIP    
Peter van 't Hof committed
522
class SummaryDbWrite(val db: Database)(implicit val ec: ExecutionContext) extends SummaryDb {
523

524
525
  /** This method will create all tables */
  def createTables(): Unit = {
526
527
528
529
530
531
532
533
    val setup = DBIO.seq(
      (runs.schema ++ samples.schema ++
        libraries.schema ++ pipelines.schema ++
        modules.schema ++ stats.schema ++ settings.schema ++
        files.schema ++ executables.schema).create
    )
    val setupFuture = db.run(setup)
    Await.result(setupFuture, Duration.Inf)
534
535
536
  }

  /** This method will create a new run and return the runId */
537
538
539
540
  def createRun(runName: String,
                outputDir: String,
                version: String,
                commitHash: String,
541
542
                creationDate: Date): Future[Int] = {
    val id = Await.result(db.run(runs.size.result), Duration.Inf)
543
544
    db.run(runs.forceInsert(Run(id, runName, outputDir, version, commitHash, creationDate)))
      .map(_ => id)
545
546
547
548
549
550
551
552
  }

  /** This creates a new sample and return the sampleId */
  def createSample(name: String, runId: Int, tags: Option[String] = None): Future[Int] = {
    val id = Await.result(db.run(samples.size.result), Duration.Inf)
    db.run(samples.forceInsert(Sample(id, name, runId, tags))).map(_ => id)
  }

553
  def createOrUpdateSample(name: String, runId: Int, tags: Option[String] = None): Future[Int] = {
Peter van 't Hof's avatar
Peter van 't Hof committed
554
    getSampleId(runId, name).flatMap {
555
556
557
558
      case Some(id: Int) =>
        db.run(samples.filter(_.name === name).filter(_.id === id).map(_.tags).update(tags))
          .map(_ => id)
      case _ => createSample(name, runId, tags)
Peter van 't Hof's avatar
Peter van 't Hof committed
559
    }
560
561
562
  }

  /** This will create a new library */
Peter van 't Hof's avatar
Peter van 't Hof committed
563
564
565
566
  def createLibrary(name: String,
                    runId: Int,
                    sampleId: Int,
                    tags: Option[String] = None): Future[Int] = {
567
568
569
570
    val id = Await.result(db.run(libraries.size.result), Duration.Inf)
    db.run(libraries.forceInsert(Library(id, name, runId, sampleId, tags))).map(_ => id)
  }

Peter van 't Hof's avatar
Peter van 't Hof committed
571
572
573
574
575
  def createOrUpdateLibrary(name: String,
                            runId: Int,
                            sampleId: Int,
                            tags: Option[String] = None): Future[Int] = {
    getLibraryId(runId, sampleId, name).flatMap {
576
      case Some(id: Int) =>
577
578
579
580
581
582
583
        db.run(
            libraries
              .filter(_.name === name)
              .filter(_.id === id)
              .filter(_.sampleId === sampleId)
              .map(_.tags)
              .update(tags))
584
585
          .map(_ => id)
      case _ => createLibrary(name, runId, sampleId, tags)
Peter van 't Hof's avatar
Peter van 't Hof committed
586
    }
587
588
589
  }

  /** Creates a new pipeline, even if it already exist. This may give a database exeption */
590
  def forceCreatePipeline(name: String, runId: Int): Future[Int] = {
591
592
593
594
595
    val id = Await.result(db.run(pipelines.size.result), Duration.Inf)
    db.run(pipelines.forceInsert(Pipeline(id, name, runId))).map(_ => id)
  }

  /** Creates a new pipeline if it does not yet exist */
596
  def createPipeline(name: String, runId: Int): Future[Int] = {
597
    getPipelines(name = Some(name), runId = Some(runId))
598
599
600
      .flatMap { m =>
        if (m.isEmpty) forceCreatePipeline(name, runId)
        else Future(m.head.id)
601
602
603
604
605
606
607
608
609
610
611
612
      }
  }

  /** Creates a new module, even if it already exist. This may give a database exeption */
  def forceCreateModule(name: String, runId: Int, pipelineId: Int): Future[Int] = {
    val id = Await.result(db.run(modules.size.result), Duration.Inf)
    db.run(modules.forceInsert(Module(id, name, runId, pipelineId))).map(_ => id)
  }

  /** Creates a new module if it does not yet exist */
  def createModule(name: String, runId: Int, pipelineId: Int): Future[Int] = {
    getModules(name = Some(name), runId = Some(runId), pipelineId = Some(pipelineId))
613
614
615
      .flatMap { m =>
        if (m.isEmpty) forceCreateModule(name, runId, pipelineId)
        else Future(m.head.id)
616
617
618
619
      }
  }

  /** Create a new stat in the database, This method is need checking before */
Peter van 't Hof's avatar
Peter van 't Hof committed
620
621
622
623
624
625
  def createStat(runId: Int,
                 pipelineId: Int,
                 moduleId: Option[Int] = None,
                 sampleId: Option[Int] = None,
                 libId: Option[Int] = None,
                 content: String): Future[Int] = {
626
627
628
629
    db.run(stats.forceInsert(Stat(runId, pipelineId, moduleId, sampleId, libId, content)))
  }

  /** This create or update a stat */
Peter van 't Hof's avatar
Peter van 't Hof committed
630
631
632
633
634
635
  def createOrUpdateStat(runId: Int,
                         pipelineId: Int,
                         moduleId: Option[Int] = None,
                         sampleId: Option[Int] = None,
                         libId: Option[Int] = None,
                         content: String): Future[Int] = {
636
637
638
639
640
641
642
    val filter = statsFilter(
      Some(runId),
      pipelineId,
      Some(moduleId.map(ModuleId).getOrElse(NoModule)),
      Some(sampleId.map(SampleId).getOrElse(NoSample)),
      Some(libId.map(LibraryId).getOrElse(NoLibrary))
    )
643
644
645
646
647
648
    val r = Await.result(db.run(filter.size.result), Duration.Inf)
    if (r == 0) createStat(runId, pipelineId, moduleId, sampleId, libId, content)
    else db.run(filter.map(_.content).update(content))
  }

  /** This method creates a new setting. This method need checking before */
Peter van 't Hof's avatar
Peter van 't Hof committed
649
650
651
652
653
654
  def createSetting(runId: Int,
                    pipelineId: Int,
                    moduleId: Option[Int] = None,
                    sampleId: Option[Int] = None,
                    libId: Option[Int] = None,
                    content: String): Future[Int] = {
655
656
657
658
    db.run(settings.forceInsert(Setting(runId, pipelineId, moduleId, sampleId, libId, content)))
  }

  /** This method creates or update a setting. */
Peter van 't Hof's avatar
Peter van 't Hof committed
659
660
661
662
663
664
  def createOrUpdateSetting(runId: Int,
                            pipelineId: Int,
                            moduleId: Option[Int] = None,
                            sampleId: Option[Int] = None,
                            libId: Option[Int] = None,
                            content: String): Future[Int] = {
665
666
667
668
669
    val filter = settingsFilter(Some(runId),
                                PipelineId(pipelineId),
                                moduleId.map(ModuleId),
                                sampleId.map(SampleId),
                                libId.map(LibraryId))
670
671
672
673
674
    val r = Await.result(db.run(filter.size.result), Duration.Inf)
    if (r == 0) createSetting(runId, pipelineId, moduleId, sampleId, libId, content)
    else db.run(filter.update(Setting(runId, pipelineId, moduleId, sampleId, libId, content)))
  }

675
  /** Creates a file. This method will raise expection if it already exist */
Peter van 't Hof's avatar
Peter van 't Hof committed
676
677
678
679
680
681
682
683
684
685
  def createFile(runId: Int,
                 pipelineId: Int,
                 moduleId: Option[Int] = None,
                 sampleId: Option[Int] = None,
                 libId: Option[Int] = None,
                 key: String,
                 path: String,
                 md5: String,
                 link: Boolean = false,
                 size: Long): Future[Int] = {
686
687
688
    db.run(
      files.forceInsert(
        Schema.File(runId, pipelineId, moduleId, sampleId, libId, key, path, md5, link, size)))
Peter van 't Hof's avatar
Peter van 't Hof committed
689
690
  }

691
  /** Create or update a File */
Peter van 't Hof's avatar
Peter van 't Hof committed
692
693
694
695
696
697
698
699
700
701
  def createOrUpdateFile(runId: Int,
                         pipelineId: Int,
                         moduleId: Option[Int] = None,
                         sampleId: Option[Int] = None,
                         libId: Option[Int] = None,
                         key: String,
                         path: String,
                         md5: String,
                         link: Boolean = false,
                         size: Long): Future[Int] = {
702
703
704
705
706
707
    val filter = filesFilter(Some(runId),
                             PipelineId(pipelineId),
                             moduleId.map(ModuleId),
                             sampleId.map(SampleId),
                             libId.map(LibraryId),
                             Some(key))
Peter van 't Hof's avatar
Peter van 't Hof committed
708
    val r = Await.result(db.run(filter.size.result), Duration.Inf)
709
710
711
712
713
714
    if (r == 0)
      createFile(runId, pipelineId, moduleId, sampleId, libId, key, path, md5, link, size)
    else
      db.run(
        filter.update(
          Schema.File(runId, pipelineId, moduleId, sampleId, libId, key, path, md5, link, size)))
Peter van 't Hof's avatar
Peter van 't Hof committed
715
  }
Peter van 't Hof's avatar
Peter van 't Hof committed
716

717
  /** Creates a exeutable. This method will raise expection if it already exist */
Peter van 't Hof's avatar
Peter van 't Hof committed
718
719
720
721
  def createExecutable(runId: Int,
                       toolName: String,
                       version: Option[String] = None,
                       path: Option[String] = None,
722
723
724
725
726
727
728
                       javaVersion: Option[String] = None,
                       exeMd5: Option[String] = None,
                       javaMd5: Option[String] = None,
                       jarPath: Option[String] = None): Future[Int] = {
    db.run(
      executables.forceInsert(
        Executable(runId, toolName, version, path, javaVersion, exeMd5, javaMd5, jarPath)))
Peter van 't Hof's avatar
Peter van 't Hof committed
729
730
  }

731
  /** Create or update a [[Executable]] */
Peter van 't Hof's avatar
Peter van 't Hof committed
732
733
734
735
736
737
738
739
  def createOrUpdateExecutable(runId: Int,
                               toolName: String,
                               version: Option[String] = None,
                               path: Option[String] = None,
                               javaVersion: Option[String] = None,
                               exeMd5: Option[String] = None,
                               javaMd5: Option[String] = None,
                               jarPath: Option[String] = None): Future[Int] = {
Peter van 't Hof's avatar
Peter van 't Hof committed
740
741
742
    val filter = executablesFilter(Some(runId), Some(toolName))
    val r = Await.result(db.run(filter.size.result), Duration.Inf)
    if (r == 0) createExecutable(runId, toolName, version, javaVersion, exeMd5, javaMd5)
743
744
745
746
    else
      db.run(
        filter.update(
          Executable(runId, toolName, version, path, javaVersion, exeMd5, javaMd5, jarPath)))
Peter van 't Hof's avatar
Peter van 't Hof committed
747
  }
748

749
}
750
751

object SummaryDb {
752

Peter van 't Hof's avatar
Peter van 't Hof committed
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
  trait PipelineQuery
  case class PipelineId(id: Int) extends PipelineQuery
  case class PipelineName(name: String) extends PipelineQuery

  trait SampleQuery
  case object NoSample extends SampleQuery
  case class SampleId(id: Int) extends SampleQuery
  case class SampleName(name: String) extends SampleQuery

  trait LibraryQuery
  case object NoLibrary extends LibraryQuery
  case class LibraryId(id: Int) extends LibraryQuery
  case class LibraryName(name: String) extends LibraryQuery

  trait ModuleQuery
  case object NoModule extends ModuleQuery
  case class ModuleId(id: Int) extends ModuleQuery
  case class ModuleName(name: String) extends ModuleQuery

772
  object Implicts {
Peter van 't Hof's avatar
Peter van 't Hof committed
773
774

    implicit def intToPipelineQuery(x: Int): PipelineQuery = PipelineId(x)
Peter van 't Hof's avatar
Peter van 't Hof committed
775
    implicit def stringToPipelineQuery(x: String): PipelineQuery = PipelineName(x)
Peter van 't Hof's avatar
Peter van 't Hof committed
776
    implicit def intToOptionPipelineQuery(x: Int): Option[PipelineQuery] = Some(PipelineId(x))
777
778
779
780
    implicit def stringToOptionPipelineQuery(x: String): Option[PipelineQuery] =
      Some(PipelineName(x))
    implicit def sampleQueryToOptionPipelineQuery(x: PipelineQuery): Option[PipelineQuery] =
      Some(x)
Peter van 't Hof's avatar
Peter van 't Hof committed
781
782

    implicit def intToModuleQuery(x: Int): ModuleQuery = ModuleId(x)
Peter van 't Hof's avatar
Peter van 't Hof committed
783
    implicit def stringToModuleQuery(x: String): ModuleQuery = ModuleName(x)
Peter van 't Hof's avatar
Peter van 't Hof committed
784
785
786
787
788
    implicit def intToOptionModuleQuery(x: Int): Option[ModuleQuery] = Some(ModuleId(x))
    implicit def intToOptionModuleQuery(x: String): Option[ModuleQuery] = Some(ModuleName(x))
    implicit def moduleQueryToOptionModuleQuery(x: ModuleQuery): Option[ModuleQuery] = Some(x)

    implicit def intToSampleQuery(x: Int): SampleQuery = SampleId(x)
Peter van 't Hof's avatar
Peter van 't Hof committed
789
    implicit def stringToSampleQuery(x: String): SampleQuery = SampleName(x)
Peter van 't Hof's avatar
Peter van 't Hof committed
790
    implicit def intToOptionSampleQuery(x: Int): Option[SampleQuery] = Some(SampleId(x))
Peter van 't Hof's avatar
Peter van 't Hof committed
791
    implicit def stringToOptionSampleQuery(x: String): Option[SampleQuery] = Some(SampleName(x))
Peter van 't Hof's avatar
Peter van 't Hof committed
792
793
794
    implicit def sampleQueryToOptionSampleQuery(x: SampleQuery): Option[SampleQuery] = Some(x)

    implicit def intToLibraryQuery(x: Int): LibraryQuery = LibraryId(x)
Peter van 't Hof's avatar
Peter van 't Hof committed
795
    implicit def stringToLibraryQuery(x: String): LibraryQuery = LibraryName(x)
Peter van 't Hof's avatar
Peter van 't Hof committed
796
    implicit def intToOptionLibraryQuery(x: Int): Option[LibraryQuery] = Some(LibraryId(x))
Peter van 't Hof's avatar
Peter van 't Hof committed
797
    implicit def stringToOptionLibraryQuery(x: String): Option[LibraryQuery] = Some(LibraryName(x))
Peter van 't Hof's avatar
Peter van 't Hof committed
798
    implicit def libraryQueryToOptionLibraryQuery(x: LibraryQuery): Option[LibraryQuery] = Some(x)
799
800
  }

801
  private var summaryConnections = Map[File, SummaryDbWrite]()
802

803
  /** This closing all summary that are still in the cache */
804
805
  def closeAll(): Unit = {
    summaryConnections.foreach(_._2.close())
806
    summaryConnections = summaryConnections.empty
807
808
  }

809
  /** This will open a sqlite database and create tables when the database did not exist yet */
Peter van 't Hof's avatar
WIP    
Peter van 't Hof committed
810
  def openSqliteSummary(file: File)(implicit ec: ExecutionContext): SummaryDbWrite = {
811
    if (!summaryConnections.contains(file)) {
Peter van 't Hof's avatar
Peter van 't Hof committed
812
813
814
      val config: org.sqlite.SQLiteConfig = new org.sqlite.SQLiteConfig()
      config.enforceForeignKeys(true)
      config.setBusyTimeout("10000")
815
      config.setSynchronous(org.sqlite.SQLiteConfig.SynchronousMode.FULL)
816
      val exist = file.exists()
817
818
819
820
      val db = Database.forURL(s"jdbc:sqlite:${file.getAbsolutePath}",
                               driver = "org.sqlite.JDBC",
                               prop = config.toProperties,
                               executor = AsyncExecutor("single_thread", 1, 1000))
821
      val s = new SummaryDbWrite(db)
822
823
824
825
      if (!exist) s.createTables()
      summaryConnections += file -> s
    }
    summaryConnections(file)
826
  }
827

Peter van 't Hof's avatar
WIP    
Peter van 't Hof committed
828
  def openReadOnlySqliteSummary(file: File)(implicit ec: ExecutionContext): SummaryDbReadOnly = {
829
830
831
832
833
834
    require(file.exists(), s"File does not exist: $file")
    val config: org.sqlite.SQLiteConfig = new org.sqlite.SQLiteConfig()
    config.enforceForeignKeys(true)
    config.setBusyTimeout("10000")
    config.setSynchronous(org.sqlite.SQLiteConfig.SynchronousMode.FULL)
    config.setReadOnly(true)
Peter van 't Hof's avatar
Peter van 't Hof committed
835
836

    val asyncExecutor = new AsyncExecutor {
Peter van 't Hof's avatar
WIP    
Peter van 't Hof committed
837
      override def executionContext: ExecutionContext = ec
Peter van 't Hof's avatar
Peter van 't Hof committed
838
839
840
      override def close(): Unit = {}
    }

841
842
843
844
    val db = Database.forURL(s"jdbc:sqlite:${file.getAbsolutePath}",
                             driver = "org.sqlite.JDBC",
                             prop = config.toProperties,
                             executor = asyncExecutor)
845
846
    new SummaryDbReadOnly(db)
  }
847
}