Skip to content

Instantly share code, notes, and snippets.

@marcelm

marcelm/snakemake-pure-python.py

Last active November 29, 2023 00:45
Download ZIP

Revisions

  1. marcelm revised this gist Oct 1, 2015. 1 changed file with 2 additions and 2 deletions.
    4 changes: 2 additions & 2 deletions snakemake-pure-python.py
    View file
    Original file line number Diff line number Diff line change
    @@ -35,7 +35,7 @@
    for _ in f:
    n += 1
    with open(output.counts, 'w') as f:
    print(n/4, file=f)
    print(n / 4, file=f)
    """
    from snakemake.workflow import Workflow, Rules
    @@ -97,7 +97,7 @@ def __count(input, output, params, wildcards, threads, resources, log, version):
    for _ in f:
    n += 1
    with open(output.counts, 'w') as f:
    print(n, file=f)
    print(n / 4, file=f)


    ### End of output from snakemake --print-compilation
  2. marcelm revised this gist Oct 1, 2015. 1 changed file with 1 addition and 1 deletion.
    2 changes: 1 addition & 1 deletion snakemake-pure-python.py
    View file
    Original file line number Diff line number Diff line change
    @@ -35,7 +35,7 @@
    for _ in f:
    n += 1
    with open(output.counts, 'w') as f:
    print(n, file=f)
    print(n/4, file=f)
    """
    from snakemake.workflow import Workflow, Rules
  3. marcelm created this gist Jun 2, 2015.
    110 changes: 110 additions & 0 deletions snakemake-pure-python.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,110 @@
    #!/usr/bin/env python3
    """
    Running this script is (intended to be) equivalent to running the following Snakefile:
    include: "pipeline.conf" # Should be an empty file
    shell.prefix("set -euo pipefail;")
    rule all:
    input:
    "reads.counts"
    rule unpack_fastq:
    '''Unpack a FASTQ file'''
    output: "{file}.fastq"
    input: "{file}.fastq.gz"
    resources: time=60, mem=100
    params: "{file}.params"
    threads: 8
    log: 'unpack.log'
    shell:
    '''zcat {input} > {output}
    echo finished 1>&2 {log}
    '''
    rule count:
    '''Count reads in a FASTQ file'''
    output: counts="{file}.counts"
    input: fastq="{file}.fastq"
    run:
    n = 0
    with open(input.fastq) as f:
    for _ in f:
    n += 1
    with open(output.counts, 'w') as f:
    print(n, file=f)
    """
    from snakemake.workflow import Workflow, Rules
    import snakemake.workflow
    from snakemake import shell
    from snakemake.logging import setup_logger

    setup_logger()

    workflow = Workflow(__file__)
    snakemake.workflow.rules = Rules()
    snakemake.workflow.config = dict()


    ### Output from snakemake --print-compilation follows (reformatted)

    workflow.include("pipeline.conf")

    shell.prefix("set -euo pipefail;")

    @workflow.rule(name='all', lineno=6, snakefile='.../Snakefile')
    @workflow.input("reads.counts")
    @workflow.norun()
    @workflow.run
    def __all(input, output, params, wildcards, threads, resources, log, version):
    pass


    @workflow.rule(name='unpack_fastq', lineno=17, snakefile='.../Snakefile')
    @workflow.docstring("""Unpack a FASTQ file""")
    @workflow.output("{file}.fastq")
    @workflow.input("{file}.fastq.gz")

    @workflow.resources(time=60, mem=100)
    @workflow.params("{file}.params")
    @workflow.threads(8)
    @workflow.log('unpack.log')
    @workflow.shellcmd(
    """zcat {input} > {output}
    echo finished 1>&2 {log}
    """
    )
    @workflow.run
    def __unpack_fastq(input, output, params, wildcards, threads, resources, log, version):
    shell("""zcat {input} > {output}
    echo finished 1>&2 > {log}
    """
    )


    @workflow.rule(name='count', lineno=52, snakefile='.../Snakefile')
    @workflow.docstring("""Count reads in a FASTQ file""")
    @workflow.output(counts = "{file}.counts")
    @workflow.input(fastq = "{file}.fastq")
    @workflow.run
    def __count(input, output, params, wildcards, threads, resources, log, version):
    n = 0
    with open(input.fastq) as f:
    for _ in f:
    n += 1
    with open(output.counts, 'w') as f:
    print(n, file=f)


    ### End of output from snakemake --print-compilation


    workflow.check()
    print("Dry run first ...")
    workflow.execute(dryrun=True, updated_files=[])
    print("And now for real")
    workflow.execute(dryrun=False, updated_files=[], resources=dict())