flowcraft/templates/pipeline_status.py
def main(sample_id, trace_file, workdir):
    """
    Parses a nextflow trace file, searches for processes with a specific tag
    and sends a JSON report with the relevant information

    The expected fields for the trace file are::

        0. task_id
        1. process
        2. tag
        3. status
        4. exit code
        5. start timestamp
        6. container
        7. cpus
        8. duration
        9. realtime
        10. queue
        11. cpu percentage
        12. memory percentage
        13. real memory size of the process
        14. virtual memory size of the process

    Parameters
    ----------
    trace_file : str
        Path to the nextflow trace file
    """

    # Determine the path of the stored JSON for the sample_id
    stats_suffix = ".stats.json"
    stats_path = join(workdir, sample_id + stats_suffix)
    trace_path = join(workdir, trace_file)

    logger.info("Starting pipeline status routine")

    logger.debug("Checking for previous pipeline status data")
    stats_array = get_previous_stats(stats_path)
    logger.info("Stats JSON object set to : {}".format(stats_array))

    # Search for this substring in the tags field. Only lines with this
    # tag will be processed for the reports
    tag = " getStats"
    logger.debug("Tag variable set to: {}".format(tag))

    logger.info("Starting parsing of trace file: {}".format(trace_path))
    with open(trace_path) as fh:

        header = next(fh).strip().split()
        logger.debug("Header set to: {}".format(header))

        for line in fh:
            fields = line.strip().split("\t")
            # Check if tag substring is in the tag field of the nextflow trace
            if tag in fields[2] and fields[3] == "COMPLETED":
                logger.debug(
                    "Parsing trace line with COMPLETED status: {}".format(
                        line))
                current_json = get_json_info(fields, header)

                stats_array[fields[0]] = current_json
            else:
                logger.debug(
                    "Ignoring trace line without COMPLETED status"
                    " or stats specific tag: {}".format(
                        line))

    with open(join(stats_path), "w") as fh, open(".report.json", "w") as rfh:
        fh.write(json.dumps(stats_array, separators=(",", ":")))
        rfh.write(json.dumps(stats_array, separators=(",", ":")))
def main(sample_id, trace_file, workdir):
    # Determine the path of the stored JSON for the sample_id
    stats_suffix = ".stats.json"
    stats_path = join(workdir, sample_id + stats_suffix)
    trace_path = join(workdir, trace_file)

    logger.info("Starting pipeline status routine")

    logger.debug("Checking for previous pipeline status data")
    stats_array = get_previous_stats(stats_path)
    logger.info("Stats JSON object set to : {}".format(stats_array))

    # Search for this substring in the tags field. Only lines with this
    # tag will be processed for the reports
    tag = " getStats"
    logger.debug("Tag variable set to: {}".format(tag))

    logger.info("Starting parsing of trace file: {}".format(trace_path))
    with open(trace_path) as fh:

        header = next(fh).strip().split()
        logger.debug("Header set to: {}".format(header))

        for line in fh:
            fields = line.strip().split("\t")
            # Check if tag substring is in the tag field of the nextflow trace
            if tag in fields[2] and fields[3] == "COMPLETED":
                logger.debug(
                    "Parsing trace line with COMPLETED status: {}".format(
                        line))
                current_json = get_json_info(fields, header)

                stats_array[fields[0]] = current_json
            else:
                logger.debug(
                    "Ignoring trace line without COMPLETED status"
                    " or stats specific tag: {}".format(
                        line))

    with open(join(stats_path), "w") as fh, open(".report.json", "w") as rfh:
        fh.write(json.dumps(stats_array, separators=(",", ":")))
        rfh.write(json.dumps(stats_array, separators=(",", ":")))
def main ( sample_id , trace_file , workdir ) : # Determine the path of the stored JSON for the sample_id stats_suffix = ".stats.json" stats_path = join ( workdir , sample_id + stats_suffix ) trace_path = join ( workdir , trace_file ) logger . info ( "Starting pipeline status routine" ) logger . debug ( "Checking for previous pipeline status data" ) stats_array = get_previous_stats ( stats_path ) logger . info ( "Stats JSON object set to : {}" . format ( stats_array ) ) # Search for this substring in the tags field. Only lines with this # tag will be processed for the reports tag = " getStats" logger . debug ( "Tag variable set to: {}" . format ( tag ) ) logger . info ( "Starting parsing of trace file: {}" . format ( trace_path ) ) with open ( trace_path ) as fh : header = next ( fh ) . strip ( ) . split ( ) logger . debug ( "Header set to: {}" . format ( header ) ) for line in fh : fields = line . strip ( ) . split ( "\t" ) # Check if tag substring is in the tag field of the nextflow trace if tag in fields [ 2 ] and fields [ 3 ] == "COMPLETED" : logger . debug ( "Parsing trace line with COMPLETED status: {}" . format ( line ) ) current_json = get_json_info ( fields , header ) stats_array [ fields [ 0 ] ] = current_json else : logger . debug ( "Ignoring trace line without COMPLETED status" " or stats specific tag: {}" . format ( line ) ) with open ( join ( stats_path ) , "w" ) as fh , open ( ".report.json" , "w" ) as rfh : fh . write ( json . dumps ( stats_array , separators = ( "," , ":" ) ) ) rfh . write ( json . dumps ( stats_array , separators = ( "," , ":" ) ) )
Parses a nextflow trace file, searches for processes with a specific tag
    and sends a JSON report with the relevant information

    The expected fields for the trace file are::

        0. task_id
        1. process
        2. tag
        3. status
        4. exit code
        5. start timestamp
        6. container
        7. cpus
        8. duration
        9. realtime
        10. queue
        11. cpu percentage
        12. memory percentage
        13. real memory size of the process
        14. virtual memory size of the process

    Parameters
    ----------
    trace_file : str
        Path to the nextflow trace file
Parses a nextflow trace file searches for processes with a specific tag and sends a JSON report with the relevant information