kfukuta/wf-transcriptomes-v202
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Template changes

Browse Source
This commit is contained in:
Neil Horner 2022-10-07 12:59:18 +00:00
parent c8b474ef1a
commit d537d7b33f
4 changed files with 104 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
lib/Pinguscript.groovy Normal file
View File

@ -0,0 +1,72 @@
import static groovy.json.JsonOutput.toJson
import groovy.json.JsonBuilder
import groovy.json.JsonSlurper
class Pinguscript {
public static String ping_post(workflow, message, error_message, out_dir, params) {
def msgId = UUID.randomUUID().toString()
def hosthash = null
try {
hosthash = InetAddress.getLocalHost().getHostName().md5()
} catch(Exception e) {
hosthash = "Unavailable"
}
def opsys = System.properties['os.name'].toLowerCase()
if (System.properties['os.version'].toLowerCase().contains("wsl")){
opsys = "WSL"
}
def workflow_name = "$workflow.manifest.name"
def session = "$workflow.sessionId"
def errorMessage = "$error_message"
def profile = "$workflow.profile"
def filename = "$out_dir/params.json"
File fileb = new File(filename)
def any_other_data = [:]
if (fileb.exists() && "$message" != "start") {
def jsonSlurper = new JsonSlurper()
any_other_data = jsonSlurper.parse(fileb)
}
def meta_json = new JsonBuilder()
def agent = "$params.wf.agent"
def meta = meta_json "error": errorMessage.toString(), "profile": profile.toString(),
"agent": agent.toString()
meta+=any_other_data
def ping_version = '2.0.1'
def tracking_json = new JsonBuilder()
def tracking_id = tracking_json "msg_id": msgId, "version": ping_version
def data_json = new JsonBuilder()
def data = data_json "workflow": workflow_name.toString(),
"message": message, "meta": meta
def body_json = new JsonBuilder()
def root = body_json "tracking_id": tracking_id, "hostname": hosthash.toString(), "os": opsys.toString(),
"session": session.toString(), "data": data, "source": "workflow"
// Attempt to send payload and absorb any possible Exception gracefully
String postResult
boolean raise_exception = false
try {
((HttpURLConnection)new URL('https://ping.oxfordnanoportal.com/epilaby').openConnection()).with({
requestMethod = 'POST'
doOutput = true
setConnectTimeout(5000)
setReadTimeout(10000)
setRequestProperty('Content-Type', 'application/json')
setRequestProperty('accept', 'application/json')
outputStream.withPrintWriter({printWriter ->
printWriter.write(body_json.toString())
})
// Rethrow exceptions that imply we're not using this endpoint properly
if(responseCode >= 400 && agent.toString() == "cw-ci") {
raise_exception = true
}
// Accessing inputStream.text will raise an Exception for failed requests
postResult = inputStream.text
})
} catch(Exception e) {
if(raise_exception) { throw e }
}
return (postResult)
}
}

41
lib/fastqingress.nf
View File

@ -44,7 +44,7 @@ def compareSampleSheetFastq(int sample_sheet_count, int valid_dir_count)
{
if (sample_sheet_count != valid_dir_count) {
log.warn """The number of samplesheet entries ({}) does not match the number of barcoded directories ({})""", sample_sheet_count, valid_dir_count
log.warn "The number of samplesheet entries ({}) does not match the number of barcoded directories ({})", sample_sheet_count, valid_dir_count
}
}
@ -100,8 +100,8 @@ def find_fastq(pattern, maxdepth)
def sanitize_fastq(input_folder, staging)
{
// TODO: this fails if input_folder is an S3 path
println("Running sanitization.")
println(" - Moving files: ${input_folder} -> ${staging}")
log.info "Running sanitization."
log.info " - Moving files: ${input_folder} -> ${staging}"
staging.mkdirs()
files = find_fastq(input_folder.resolve("**"), 1)
for (fastq in files) {
@ -118,7 +118,7 @@ def sanitize_fastq(input_folder, staging)
fastq.renameTo(staging.resolve("${matcher[0]}/${fname}"))
}
}
println(" - Finished sanitization.")
log.info " - Finished sanitization."
return staging
}
@ -149,12 +149,12 @@ def get_subdirectories(input_directory)
*/
def get_sample_sheet(sample_sheet)
{
println("Checking sample sheet.")
log.info "Checking sample sheet."
sample_sheet = file(sample_sheet);
is_file = sample_sheet.isFile()
if (!is_file) {
println('Error: `--samples` is not a file.')
log.error "`--samples` is not a file."
exit 1
}
@ -200,20 +200,19 @@ def get_valid_directories(input_dirs)
}
}
if (valid_dirs.size() == 0) {
error_message = "Error: None of the directories given contain .fastq(.gz) files."
println(error_message)
log.error "None of the directories given contain .fastq(.gz) files."
exit 1
}
if (no_fastq_dirs.size() > 0) {
println("Warning: Excluding directories not containing .fastq(.gz) files:")
log.warn "Excluding directories not containing .fastq(.gz) files:"
for (d in no_fastq_dirs) {
println(" - ${d}")
log.warn " - ${d}"
}
}
if (invalid_files_dirs.size() > 0) {
println("Warning: Excluding directories containing non .fastq(.gz) files:")
log.warn "Excluding directories containing non .fastq(.gz) files:"
for (d in invalid_files_dirs) {
println(" - ${d}")
log.warn " - ${d}"
}
}
return valid_dirs
@ -368,15 +367,15 @@ def fastq_ingress(Map arguments)
}
println("Checking fastq input.")
log.info "Checking fastq input."
input = file(margs.input)
// Handle file input
if (input.isFile()) {
// Assume sample is a string at this point
println('Single file input detected.')
log.info "Single file input detected."
if (margs.sample_sheet) {
println('Warning: `--sample_sheet` given but single file input found. Ignoring.')
log.warn "Warning: `--sample_sheet` given but single file input found. Ignoring."
}
return handle_single_file(input, margs.sample)
}
@ -394,22 +393,22 @@ def fastq_ingress(Map arguments)
// Case 03: If no subdirectories, handle the single dir
if (!barcoded && !non_barcoded) {
println("Single directory input detected.")
log.info "Single directory input detected."
if (margs.sample_sheet) {
println('Warning: `--sample_sheet` given but single non-barcode directory found. Ignoring.')
log.warn "`--sample_sheet` given but single non-barcode directory found. Ignoring."
}
return handle_flat_dir(input, margs.sample)
}
if (margs.sample) {
println('Warning: `--sample` given but multiple directories found, ignoring.')
log.warn "`--sample` given but multiple directories found, ignoring."
}
// Case 01, 02, 04: Handle barcoded and non_barcoded dirs
// Handle barcoded folders
barcoded_samples = Channel.empty()
if (barcoded) {
println("Barcoded directories detected.")
log.info "Barcoded directories detected."
sample_sheet = null
if (margs.sample_sheet) {
sample_sheet = get_sample_sheet(margs.sample_sheet)
@ -419,9 +418,9 @@ def fastq_ingress(Map arguments)
non_barcoded_samples = Channel.empty()
if (non_barcoded) {
println("Non barcoded directories detected.")
log.info "Non barcoded directories detected."
if (!barcoded && margs.sample_sheet) {
println('Warning: `--sample_sheet` given but no barcode directories found.')
log.warn "Warning: `--sample_sheet` given but no barcode directories found."
}
non_barcoded_samples = handle_non_barcoded_dirs(non_barcoded)
}

36
lib/ping.nf
View File

@ -1,36 +0,0 @@
process pingMessage {
label "isoforms"
cpus 1
input:
val message
path json
script:
hostname = InetAddress.getLocalHost().getHostName()
opsys = System.properties['os.name'].toLowerCase()
disable = params.disable_ping ? '--disable' : ''
meta = json.name != 'OPTIONAL_FILE' ? "--meta $json": ''
"""
ping.py \
--hostname $hostname \
--opsys "$opsys" \
--session $workflow.sessionId \
--message $message \
$meta $disable
"""
}
// send a start message
workflow start_ping {
main:
pingMessage("Started", Channel.fromPath("$projectDir/data/OPTIONAL_FILE"))
}
// send an end message
workflow end_ping {
take:
json
main:
pingMessage("Finished", json)
}

14
main.nf
View File

@ -11,7 +11,6 @@ import java.util.ArrayList;
nextflow.enable.dsl = 2
include { fastq_ingress } from './lib/fastqingress'
include { start_ping; end_ping } from './lib/ping'
include { reference_assembly } from './subworkflows/reference_assembly'
include { denovo_assembly } from './subworkflows/denovo_assembly'
include { gene_fusions } from './subworkflows/JAFFAL/gene_fusions'
@ -551,7 +550,9 @@ workflow pipeline {
WorkflowMain.initialise(workflow, params, log)
workflow {
start_ping()
if (params.disable_ping == false) {
Pinguscript.ping_post(workflow, "start", "none", params.out_dir, params)
}
fastq = file(params.fastq, type: "file")
@ -625,7 +626,14 @@ workflow {
condition_sheet, ref_transcriptome)
output(pipeline.out.results)
}
}
end_ping(pipeline.out.telemetry)
if (params.disable_ping == false) {
workflow.onComplete {
Pinguscript.ping_post(workflow, "end", "none", params.out_dir, params)
}
workflow.onError {
Pinguscript.ping_post(workflow, "error", "$workflow.errorMessage", params.out_dir, params)
}
}