Get showyourwork cache info

hack_syw_cache.py

def get_cache_info(*args, **kwargs):
    import inspect
    from pathlib import Path

    # Wait until the DAG has been built
    snakemake.workflow.checkpoints.syw__dag.get()

    # Get all jobs from the DAG
    dag = None
    levels = inspect.stack()
    for level in levels:
        dag = level.frame.f_locals.get("dag", None)
        if dag is not None:
            break
    jobs = [job for job in dag.jobs if snakemake.workflow.workflow.is_cached_rule(job.rule)]

    # Print cache file names
    cache = snakemake.workflow.workflow.output_file_cache
    for job in jobs:
        for outputfile, cachefile in cache.get_outputfiles_and_cachefiles(job):
            cachefile = Path(cachefile).name
            print(f"{outputfile} -> {cachefile}")

    # Abort
    raise Exception("Aborting...")


rule cache_info:
    input:
        get_cache_info
    priority:
        snakemake.jobs.Job.HIGHEST_PRIORITY
    output:
        touch("src/data/CACHEINFO")