diff --git a/ingest/Snakefile b/ingest/Snakefile index dd74e5b..493fc73 100644 --- a/ingest/Snakefile +++ b/ingest/Snakefile @@ -1,8 +1,12 @@ -# The workflow filepaths are written relative to this Snakefile's base directory -workdir: workflow.current_basedir +# Utility functions shared across all workflows. +include: "../shared/vendored/snakemake/config.smk" -# Use default configuration values. Override with Snakemake's --configfile/--config options. -configfile: "defaults/config.yaml" +# Use default configuration values. Extend with Snakemake's --configfile/--config options. +configfile: os.path.join(workflow.basedir, "defaults/config.yaml") + +# Use custom configuration from analysis directory (i.e. working dir), if any. +if os.path.exists("config.yaml"): + configfile: "config.yaml" # This is the default rule that Snakemake will run when there are no specified targets. # The default output of the ingest workflow is usually the curated metadata and sequences. @@ -31,5 +35,10 @@ include: "rules/nextclade.smk" # https://snakemake.readthedocs.io/en/stable/snakefiles/rules.html#handling-ambiguous-rules if "custom_rules" in config: for rule_file in config["custom_rules"]: - - include: rule_file \ No newline at end of file + # Relative custom rule paths in the config are relative to the analysis + # directory (i.e. the current working directory, or workdir, usually + # given by --directory), but the "include" directive treats relative + # paths as relative to the workflow (e.g. workflow.current_basedir). + # Convert to an absolute path based on the analysis/current directory + # to avoid this mismatch of expectations. + include: os.path.join(os.getcwd(), rule_file) diff --git a/ingest/build-configs/nextstrain-automation/upload.smk b/ingest/build-configs/nextstrain-automation/upload.smk index b5549eb..a38801e 100644 --- a/ingest/build-configs/nextstrain-automation/upload.smk +++ b/ingest/build-configs/nextstrain-automation/upload.smk @@ -29,7 +29,7 @@ rule upload_to_s3: cloudfront_domain=config["cloudfront_domain"], shell: """ - ./vendored/upload-to-s3 \ + {workflow.basedir}/../shared/vendored/scripts/upload-to-s3 \ {params.quiet} \ {input.file_to_upload:q} \ {params.s3_dst:q}/{wildcards.remote_file:q} \ diff --git a/ingest/vendored/notify-slack b/ingest/vendored/notify-slack deleted file mode 100755 index a343435..0000000 --- a/ingest/vendored/notify-slack +++ /dev/null @@ -1,56 +0,0 @@ -#!/usr/bin/env bash -set -euo pipefail - -: "${SLACK_TOKEN:?The SLACK_TOKEN environment variable is required.}" -: "${SLACK_CHANNELS:?The SLACK_CHANNELS environment variable is required.}" - -upload=0 -output=/dev/null -thread_ts="" -broadcast=0 -args=() - -for arg; do - case "$arg" in - --upload) - upload=1;; - --output=*) - output="${arg#*=}";; - --thread-ts=*) - thread_ts="${arg#*=}";; - --broadcast) - broadcast=1;; - *) - args+=("$arg");; - esac -done - -set -- "${args[@]}" - -text="${1:?Some message text is required.}" - -if [[ "$upload" == 1 ]]; then - echo "Uploading data to Slack with the message: $text" - curl https://slack.com/api/files.upload \ - --header "Authorization: Bearer $SLACK_TOKEN" \ - --form-string channels="$SLACK_CHANNELS" \ - --form-string title="$text" \ - --form-string filename="$text" \ - --form-string thread_ts="$thread_ts" \ - --form file=@/dev/stdin \ - --form filetype=text \ - --fail --silent --show-error \ - --http1.1 \ - --output "$output" -else - echo "Posting Slack message: $text" - curl https://slack.com/api/chat.postMessage \ - --header "Authorization: Bearer $SLACK_TOKEN" \ - --form-string channel="$SLACK_CHANNELS" \ - --form-string text="$text" \ - --form-string thread_ts="$thread_ts" \ - --form-string reply_broadcast="$broadcast" \ - --fail --silent --show-error \ - --http1.1 \ - --output "$output" -fi diff --git a/nextstrain-pathogen.yaml b/nextstrain-pathogen.yaml index b74c50d..9a723f0 100644 --- a/nextstrain-pathogen.yaml +++ b/nextstrain-pathogen.yaml @@ -1,5 +1,10 @@ -# This is currently an empty file to indicate the top level pathogen repo. -# The inclusion of this file allows the Nextstrain CLI to run the -# `nextstrain build` from any directory regardless of runtime. +# This file's *existence* marks the top level of a Nextstrain pathogen repo, +# which allows `nextstrain build` to be run from any subdirectory of the repo +# regardless of runtime. For more details, see +# . # -# See https://github.com/nextstrain/cli/releases/tag/8.2.0 for more details. +# This file's *contents* is the "registration metadata" for the pathogen repo, +# used by `nextstrain setup` and `nextstrain run`. +--- +compatibility: + nextstrain run: true diff --git a/phylogenetic/Snakefile b/phylogenetic/Snakefile index 5ddf4db..3c1f95f 100644 --- a/phylogenetic/Snakefile +++ b/phylogenetic/Snakefile @@ -2,29 +2,15 @@ This is the main phylogenetic Snakefile that orchestrates the full phylogenetic workflow and define its default output(s). """ -# The workflow filepaths are written relative to this Snakefile's base directory -workdir: workflow.current_basedir +# Utility functions shared across all workflows. +include: "../shared/vendored/snakemake/config.smk" -# Use default configuration values. Override with Snakemake's --configfile/--config options. -configfile: "defaults/config.yaml" +configfile: os.path.join(workflow.basedir, "defaults/config.yaml") -# Validate 'builds' -from textwrap import dedent +if os.path.exists("config.yaml"): + configfile: "config.yaml" -def indented_list(xs, prefix): - return f"\n{prefix}".join(xs) - -if invalid_builds := set(config["builds"]) - set(config["build_params"]): - print(dedent(f"""\ - ERROR: The following names in 'builds' are not defined in 'build_params': - - {indented_list(invalid_builds, " ")} - - Available builds are: - - {indented_list(config['build_params'], " ")} - """)) - exit(1) +include: "rules/config.smk" builds = config['builds'] @@ -53,7 +39,6 @@ rule all: # custom_rules imported below to ensure that the core workflow is not complicated # by build specific rules. -include: "rules/write_config.smk" include: "rules/merge_additional_inputs.smk" include: "rules/prepare_sequences.smk" include: "rules/subsample.smk" @@ -82,4 +67,4 @@ rule clean: if "custom_rules" in config: for rule_file in config["custom_rules"]: - include: rule_file + include: os.path.join(os.getcwd(), rule_file) diff --git a/phylogenetic/defaults/config.yaml b/phylogenetic/defaults/config.yaml index e22cbff..8009650 100644 --- a/phylogenetic/defaults/config.yaml +++ b/phylogenetic/defaults/config.yaml @@ -21,15 +21,18 @@ build_params: # See Nextstrain documentation for an explanation of how subsampling configuration works: # - subsampling: - region: >- - --query "is_lab_host != 'true'" - --query-columns is_lab_host:str - --min-length '8200' - --group-by region year - --subsample-max-sequences 3000 - --exclude defaults/exclude.txt - --include defaults/all-lineages/include.txt + subsample: + region: + query: is_lab_host != 'true' + query_columns: + - is_lab_host:str + min_length: 8200 + group_by: + - region + - year + subsample_max_sequences: 3000 + exclude: defaults/exclude.txt + include: defaults/all-lineages/include.txt refine: treetime_params: --coalescent opt --date-inference marginal --date-confidence --keep-polytomies --clock-rate 0.000755 @@ -57,15 +60,18 @@ build_params: reference: "defaults/lineage-1A/reference.gb" root: "KX394399" - subsampling: - region: >- - --query "is_lab_host != 'true' & lineage == '1A'" - --query-columns is_lab_host:str - --min-length '8200' - --group-by region year - --subsample-max-sequences 3000 - --exclude defaults/exclude.txt - --include defaults/lineage-1A/include.txt + subsample: + region: + query: is_lab_host != 'true' & lineage == '1A' + query_columns: + - is_lab_host:str + min_length: 8200 + group_by: + - region + - year + subsample_max_sequences: 3000 + exclude: defaults/exclude.txt + include: defaults/lineage-1A/include.txt # Clock rate from Table 1 of May et al, 2010: https://pmc.ncbi.nlm.nih.gov/articles/PMC3067944/ refine: @@ -94,15 +100,18 @@ build_params: reference: "defaults/lineage-2/reference.gb" root: "best" - subsampling: - region: >- - --query "is_lab_host != 'true' & lineage == '2'" - --query-columns is_lab_host:str - --min-length '8200' - --group-by region year - --subsample-max-sequences 3000 - --exclude defaults/exclude.txt - --include defaults/lineage-2/include.txt + subsample: + region: + query: is_lab_host != 'true' & lineage == '2' + query_columns: + - is_lab_host:str + min_length: 8200 + group_by: + - region + - year + subsample_max_sequences: 3000 + exclude: defaults/exclude.txt + include: defaults/lineage-2/include.txt # Clock rate from McMullen et al, 2013: https://pmc.ncbi.nlm.nih.gov/articles/PMC3709619/ refine: diff --git a/phylogenetic/rules/config.smk b/phylogenetic/rules/config.smk new file mode 100644 index 0000000..5d28829 --- /dev/null +++ b/phylogenetic/rules/config.smk @@ -0,0 +1,108 @@ +""" +This part of the workflow deals with configuration. + +OUTPUTS: + + results/run_configs/{timestamp}.yaml +""" +import os +import sys +import yaml +from datetime import datetime +from textwrap import dedent + + +def main(): + validate_config() + resolve_config_paths() + write_config() + + +def validate_config(): + """ + Validate the config. + + This could be improved with a schema definition file, but for now it serves + to provide useful error messages for common user errors and effects of + breaking changes. + """ + # Validate 'builds' + if invalid_builds := set(config["builds"]) - set(config["build_params"]): + print(dedent(f"""\ + ERROR: The following names in 'builds' are not defined in 'build_params': + + {indented_list(invalid_builds, " ")} + + Available builds are: + + {indented_list(config['build_params'], " ")} + """)) + exit(1) + + +def resolve_config_paths(): + """ + Update all file paths in config by passing them through resolve_config_path() + """ + global config + + for build_name, build_config in config["build_params"].items(): + # config..reference + build_config["reference"] = resolve_config_path(build_config["reference"])({}) + + # config..export + for key in ["description", "auspice_config"]: + build_config["export"][key] = resolve_config_path(build_config["export"][key])({}) + + # config..subsample + subsample_path_keys = ["exclude", "include", "group_by_weights"] + for sample_name, sample_config in build_config["subsample"].items(): + for key in subsample_path_keys: + if key in sample_config: + if isinstance(sample_config[key], list): + sample_config[key] = [resolve_config_path(path)({}) for path in sample_config[key]] + else: + sample_config[key] = resolve_config_path(sample_config[key])({}) + + +def write_config(): + """ + Write Snakemake's 'config' variable to a file. + + This is useful for debugging purposes. + """ + timestamp = datetime.now().astimezone().strftime("%Y-%m-%dT%H%M%S.%f") + path = f"results/run_configs/{timestamp}.yaml" + + os.makedirs(os.path.dirname(path), exist_ok=True) + + with open(path, 'w') as f: + yaml.dump(config, f, sort_keys=False) + + print(f"Saved current run config to {path!r}.", file=sys.stderr) + + +def indented_list(xs, prefix): + return f"\n{prefix}".join(xs) + + +def conditional(option, argument): + """Used for config-defined arguments whose presence necessitates a command-line option + (e.g. --foo) prepended and whose absence should result in no option/arguments in the CLI command. + *argument* can be falsey, in which case an empty string is returned (i.e. "don't pass anything + to the CLI"), or a *list* or *string* or *number* in which case a flat list of options/args is returned, + or *True* in which case a list of a single element (the option) is returned. + Any other argument type is a WorkflowError + """ + if not argument: + return "" + if argument is True: # must come before `isinstance(argument, int)` as bool is a subclass of int + return [option] + if isinstance(argument, list): + return [option, *argument] + if isinstance(argument, int) or isinstance(argument, float) or isinstance(argument, str): + return [option, argument] + raise WorkflowError(f"Workflow function conditional() received an argument value of unexpected type: {type(argument).__name__}") + + +main() diff --git a/phylogenetic/rules/subsample.smk b/phylogenetic/rules/subsample.smk index 48e889f..b723124 100644 --- a/phylogenetic/rules/subsample.smk +++ b/phylogenetic/rules/subsample.smk @@ -29,7 +29,26 @@ rule subsample: benchmark: "benchmarks/{build}/{subsample}/subsampled_strains.txt", params: - filters = lambda w: config["build_params"][w.build]["subsampling"][w.subsample], + exclude = lambda w: conditional("--exclude", config["build_params"][w.build]["subsample"][w.subsample].get("exclude")), + exclude_all = lambda w: conditional("--exclude-all", config["build_params"][w.build]["subsample"][w.subsample].get("exclude_all")), + exclude_ambiguous_dates_by = lambda w: conditional("--exclude-ambiguous-dates-by", config["build_params"][w.build]["subsample"][w.subsample].get("exclude_ambiguous_dates_by")), + exclude_where = lambda w: conditional("--exclude-where", config["build_params"][w.build]["subsample"][w.subsample].get("exclude_where")), + group_by = lambda w: conditional("--group-by", config["build_params"][w.build]["subsample"][w.subsample].get("group_by")), + group_by_weights = lambda w: conditional("--group-by-weights", config["build_params"][w.build]["subsample"][w.subsample].get("group_by_weights")), + include = lambda w: conditional("--include", config["build_params"][w.build]["subsample"][w.subsample].get("include")), + include_where = lambda w: conditional("--include-where", config["build_params"][w.build]["subsample"][w.subsample].get("include_where")), + max_date = lambda w: conditional("--max-date", config["build_params"][w.build]["subsample"][w.subsample].get("max_date")), + max_length = lambda w: conditional("--max-length", config["build_params"][w.build]["subsample"][w.subsample].get("max_length")), + min_date = lambda w: conditional("--min-date", config["build_params"][w.build]["subsample"][w.subsample].get("min_date")), + min_length = lambda w: conditional("--min-length", config["build_params"][w.build]["subsample"][w.subsample].get("min_length")), + non_nucleotide = lambda w: conditional("--non-nucleotide", config["build_params"][w.build]["subsample"][w.subsample].get("non_nucleotide")), + probabilistic_sampling = lambda w: conditional("--probabilistic-sampling", config["build_params"][w.build]["subsample"][w.subsample].get("probabilistic_sampling")), + query = lambda w: conditional("--query", config["build_params"][w.build]["subsample"][w.subsample].get("query")), + query_columns = lambda w: conditional("--query-columns", config["build_params"][w.build]["subsample"][w.subsample].get("query_columns")), + # FIXME: --no-probabilistic-sampling? + # FIXME: --priority? + sequences_per_group = lambda w: conditional("--sequences-per-group", config["build_params"][w.build]["subsample"][w.subsample].get("sequences_per_group")), + subsample_max_sequences = lambda w: conditional("--subsample-max-sequences", config["build_params"][w.build]["subsample"][w.subsample].get("subsample_max_sequences")), id_column = config["strain_id_field"], shell: """ @@ -37,7 +56,24 @@ rule subsample: --sequences {input.sequences} \ --metadata {input.metadata} \ --metadata-id-columns {params.id_column} \ - {params.filters} \ + {params.exclude_all:q} \ + {params.exclude_ambiguous_dates_by:q} \ + {params.exclude_where:q} \ + {params.exclude:q} \ + {params.group_by_weights:q} \ + {params.group_by:q} \ + {params.include_where:q} \ + {params.include:q} \ + {params.max_date:q} \ + {params.max_length:q} \ + {params.min_date:q} \ + {params.min_length:q} \ + {params.non_nucleotide:q} \ + {params.probabilistic_sampling:q} \ + {params.query_columns:q} \ + {params.query:q} \ + {params.sequences_per_group:q} \ + {params.subsample_max_sequences:q} \ --output-strains {output.subsampled_strains} 2>&1 | tee {log} """ @@ -45,7 +81,7 @@ rule extract_subsampled_sequences_and_metadata: input: sequences = input_sequences, metadata = input_metadata, - subsampled_strains = lambda w: expand("results/{build}/subsampled_strains_{subsample}.txt", build=w.build, subsample=list(config["build_params"][w.build]["subsampling"].keys())) + subsampled_strains = lambda w: expand("results/{build}/subsampled_strains_{subsample}.txt", build=w.build, subsample=list(config["build_params"][w.build]["subsample"].keys())) output: sequences = "results/{build}/sequences_filtered.fasta", metadata = "results/{build}/metadata_filtered.tsv", diff --git a/phylogenetic/rules/write_config.smk b/phylogenetic/rules/write_config.smk deleted file mode 100644 index 0a070be..0000000 --- a/phylogenetic/rules/write_config.smk +++ /dev/null @@ -1,21 +0,0 @@ -""" -This part of the workflow writes run time configuration to a YAML file. - -OUTPUTS: - - results/run_configs/{timestamp}.yaml -""" -import os -import sys -import yaml -from datetime import datetime - -timestamp = datetime.now().astimezone().strftime("%Y-%m-%dT%H%M%S.%f") -path = f"results/run_configs/{timestamp}.yaml" - -os.makedirs(os.path.dirname(path), exist_ok=True) - -with open(path, 'w') as f: - yaml.dump(config, f, sort_keys=False) - -print(f"Saved current run config to {path!r}.", file=sys.stderr) diff --git a/ingest/vendored/.github/dependabot.yml b/shared/vendored/.github/dependabot.yml similarity index 89% rename from ingest/vendored/.github/dependabot.yml rename to shared/vendored/.github/dependabot.yml index 89bd084..0a50ee1 100644 --- a/ingest/vendored/.github/dependabot.yml +++ b/shared/vendored/.github/dependabot.yml @@ -4,7 +4,7 @@ # Each ecosystem is checked on a scheduled interval defined below. To trigger # a check manually, go to # -# https://github.com/nextstrain/ingest/network/updates +# https://github.com/nextstrain/shared/network/updates # # and look for a "Check for updates" button. You may need to click around a # bit first. diff --git a/ingest/vendored/.github/pull_request_template.md b/shared/vendored/.github/pull_request_template.md similarity index 100% rename from ingest/vendored/.github/pull_request_template.md rename to shared/vendored/.github/pull_request_template.md diff --git a/ingest/vendored/.github/workflows/ci.yaml b/shared/vendored/.github/workflows/ci.yaml similarity index 85% rename from ingest/vendored/.github/workflows/ci.yaml rename to shared/vendored/.github/workflows/ci.yaml index c716277..94d3054 100644 --- a/ingest/vendored/.github/workflows/ci.yaml +++ b/shared/vendored/.github/workflows/ci.yaml @@ -11,5 +11,5 @@ jobs: shellcheck: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v5 - uses: nextstrain/.github/actions/shellcheck@master diff --git a/ingest/vendored/.github/workflows/pre-commit.yaml b/shared/vendored/.github/workflows/pre-commit.yaml similarity index 86% rename from ingest/vendored/.github/workflows/pre-commit.yaml rename to shared/vendored/.github/workflows/pre-commit.yaml index 70da533..a418753 100644 --- a/ingest/vendored/.github/workflows/pre-commit.yaml +++ b/shared/vendored/.github/workflows/pre-commit.yaml @@ -7,7 +7,7 @@ jobs: pre-commit: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v5 - uses: actions/setup-python@v5 with: python-version: "3.12" diff --git a/ingest/vendored/.gitrepo b/shared/vendored/.gitrepo similarity index 59% rename from ingest/vendored/.gitrepo rename to shared/vendored/.gitrepo index 3f0f3da..e174fc0 100644 --- a/ingest/vendored/.gitrepo +++ b/shared/vendored/.gitrepo @@ -4,9 +4,9 @@ ; git-subrepo command. See https://github.com/ingydotnet/git-subrepo#readme ; [subrepo] - remote = https://github.com/nextstrain/ingest + remote = https://github.com/nextstrain/shared branch = main - commit = 258ab8ce898a88089bc88caee336f8d683a0e79a - parent = 6ee5db50d2506315b264a8ffba657039eceb379e + commit = 2d063cf2bae0cfc91d70fda2c36f1451656e5757 + parent = 6f9a6bb9a7625aa1f8146b4349e4466103d0a53f method = merge - cmdver = 0.4.6 + cmdver = 0.4.9 diff --git a/ingest/vendored/.pre-commit-config.yaml b/shared/vendored/.pre-commit-config.yaml similarity index 100% rename from ingest/vendored/.pre-commit-config.yaml rename to shared/vendored/.pre-commit-config.yaml diff --git a/ingest/vendored/.shellcheckrc b/shared/vendored/.shellcheckrc similarity index 100% rename from ingest/vendored/.shellcheckrc rename to shared/vendored/.shellcheckrc diff --git a/ingest/vendored/README.md b/shared/vendored/README.md similarity index 61% rename from ingest/vendored/README.md rename to shared/vendored/README.md index a2b54cb..eb2e3ba 100644 --- a/ingest/vendored/README.md +++ b/shared/vendored/README.md @@ -1,6 +1,6 @@ -# ingest +# shared -Shared internal tooling for pathogen data ingest. Used by our individual +Shared internal tooling for pathogen workflows. Used by our individual pathogen repos which produce Nextstrain builds. Expected to be vendored by each pathogen repo using `git subrepo`. @@ -9,47 +9,47 @@ Some tools may only live here temporarily before finding a permanent home in ## Vendoring -Nextstrain maintained pathogen repos will use [`git subrepo`](https://github.com/ingydotnet/git-subrepo) to vendor ingest scripts. -(See discussion on this decision in https://github.com/nextstrain/ingest/issues/3) +Nextstrain maintained pathogen repos will use [`git subrepo`](https://github.com/ingydotnet/git-subrepo) to vendor shared scripts. +(See discussion on this decision in https://github.com/nextstrain/shared/issues/3) For a list of Nextstrain repos that are currently using this method, use [this GitHub code search](https://github.com/search?type=code&q=org%3Anextstrain+subrepo+%22remote+%3D+https%3A%2F%2Fgithub.com%2Fnextstrain%2Fingest%22). If you don't already have `git subrepo` installed, follow the [git subrepo installation instructions](https://github.com/ingydotnet/git-subrepo#installation). -Then add the latest ingest scripts to the pathogen repo by running: +Then add the latest shared scripts to the pathogen repo by running: ``` -git subrepo clone https://github.com/nextstrain/ingest ingest/vendored +git subrepo clone https://github.com/nextstrain/shared shared/vendored ``` -Any future updates of ingest scripts can be pulled in with: +Any future updates of sahred scripts can be pulled in with: ``` -git subrepo pull ingest/vendored +git subrepo pull shared/vendored ``` If you run into merge conflicts and would like to pull in a fresh copy of the -latest ingest scripts, pull with the `--force` flag: +latest shared scripts, pull with the `--force` flag: ``` -git subrepo pull ingest/vendored --force +git subrepo pull shared/vendored --force ``` > **Warning** > Beware of rebasing/dropping the parent commit of a `git subrepo` update -`git subrepo` relies on metadata in the `ingest/vendored/.gitrepo` file, +`git subrepo` relies on metadata in the `shared/vendored/.gitrepo` file, which includes the hash for the parent commit in the pathogen repos. If this hash no longer exists in the commit history, there will be errors when running future `git subrepo pull` commands. If you run into an error similar to the following: ``` -$ git subrepo pull ingest/vendored -git-subrepo: Command failed: 'git branch subrepo/ingest/vendored '. +$ git subrepo pull shared/vendored +git-subrepo: Command failed: 'git branch subrepo/shared/vendored '. fatal: not a valid object name: '' ``` -Check the parent commit hash in the `ingest/vendored/.gitrepo` file and make +Check the parent commit hash in the `shared/vendored/.gitrepo` file and make sure the commit exists in the commit history. Update to the appropriate parent commit hash if needed. @@ -84,39 +84,49 @@ approach to "ingest" has been discussed in various internal places, including: ## Scripts -Scripts for supporting ingest workflow automation that don’t really belong in any of our existing tools. +Scripts for supporting workflow automation that don’t really belong in any of our existing tools. -- [notify-on-diff](notify-on-diff) - Send Slack message with diff of a local file and an S3 object -- [notify-on-job-fail](notify-on-job-fail) - Send Slack message with details about failed workflow job on GitHub Actions and/or AWS Batch -- [notify-on-job-start](notify-on-job-start) - Send Slack message with details about workflow job on GitHub Actions and/or AWS Batch -- [notify-on-record-change](notify-on-recod-change) - Send Slack message with details about line count changes for a file compared to an S3 object's metadata `recordcount`. +- [assign-colors](scripts/assign-colors) - Generate colors.tsv for augur export based on ordering, color schemes, and what exists in the metadata. Used in the phylogenetic or nextclade workflows. +- [notify-on-diff](scripts/notify-on-diff) - Send Slack message with diff of a local file and an S3 object +- [notify-on-job-fail](scripts/notify-on-job-fail) - Send Slack message with details about failed workflow job on GitHub Actions and/or AWS Batch +- [notify-on-job-start](scripts/notify-on-job-start) - Send Slack message with details about workflow job on GitHub Actions and/or AWS Batch +- [notify-on-record-change](scripts/notify-on-recod-change) - Send Slack message with details about line count changes for a file compared to an S3 object's metadata `recordcount`. If the S3 object's metadata does not have `recordcount`, then will attempt to download S3 object to count lines locally, which only supports `xz` compressed S3 objects. -- [notify-slack](notify-slack) - Send message or file to Slack -- [s3-object-exists](s3-object-exists) - Used to prevent 404 errors during S3 file comparisons in the notify-* scripts -- [trigger](trigger) - Triggers downstream GitHub Actions via the GitHub API using repository_dispatch events. -- [trigger-on-new-data](trigger-on-new-data) - Triggers downstream GitHub Actions if the provided `upload-to-s3` outputs do not contain the `identical_file_message` +- [notify-slack](scripts/notify-slack) - Send message or file to Slack +- [s3-object-exists](scripts/s3-object-exists) - Used to prevent 404 errors during S3 file comparisons in the notify-* scripts +- [trigger](scripts/trigger) - Triggers downstream GitHub Actions via the GitHub API using repository_dispatch events. +- [trigger-on-new-data](scripts/trigger-on-new-data) - Triggers downstream GitHub Actions if the provided `upload-to-s3` outputs do not contain the `identical_file_message` A hacky way to ensure that we only trigger downstream phylogenetic builds if the S3 objects have been updated. + NCBI interaction scripts that are useful for fetching public metadata and sequences. -- [fetch-from-ncbi-entrez](fetch-from-ncbi-entrez) - Fetch metadata and nucleotide sequences from [NCBI Entrez](https://www.ncbi.nlm.nih.gov/books/NBK25501/) and output to a GenBank file. +- [fetch-from-ncbi-entrez](scripts/fetch-from-ncbi-entrez) - Fetch metadata and nucleotide sequences from [NCBI Entrez](https://www.ncbi.nlm.nih.gov/books/NBK25501/) and output to a GenBank file. Useful for pathogens with metadata and annotations in custom fields that are not part of the standard [NCBI Datasets](https://www.ncbi.nlm.nih.gov/datasets/) outputs. -Historically, some pathogen repos used the undocumented NCBI Virus API through [fetch-from-ncbi-virus](https://github.com/nextstrain/ingest/blob/c97df238518171c2b1574bec0349a55855d1e7a7/fetch-from-ncbi-virus) to fetch data. However we've opted to drop the NCBI Virus scripts due to https://github.com/nextstrain/ingest/issues/18. +Historically, some pathogen repos used the undocumented NCBI Virus API through [fetch-from-ncbi-virus](https://github.com/nextstrain/shared/blob/c97df238518171c2b1574bec0349a55855d1e7a7/fetch-from-ncbi-virus) to fetch data. However we've opted to drop the NCBI Virus scripts due to https://github.com/nextstrain/shared/issues/18. Potential Nextstrain CLI scripts -- [sha256sum](sha256sum) - Used to check if files are identical in upload-to-s3 and download-from-s3 scripts. -- [cloudfront-invalidate](cloudfront-invalidate) - CloudFront invalidation is already supported in the [nextstrain remote command for S3 files](https://github.com/nextstrain/cli/blob/a5dda9c0579ece7acbd8e2c32a4bbe95df7c0bce/nextstrain/cli/remote/s3.py#L104). +- [sha256sum](scripts/sha256sum) - Used to check if files are identical in upload-to-s3 and download-from-s3 scripts. +- [cloudfront-invalidate](scripts/cloudfront-invalidate) - CloudFront invalidation is already supported in the [nextstrain remote command for S3 files](https://github.com/nextstrain/cli/blob/a5dda9c0579ece7acbd8e2c32a4bbe95df7c0bce/nextstrain/cli/remote/s3.py#L104). This exists as a separate script to support CloudFront invalidation when using the upload-to-s3 script. -- [upload-to-s3](upload-to-s3) - Upload file to AWS S3 bucket with compression based on file extension in S3 URL. +- [upload-to-s3](scripts/upload-to-s3) - Upload file to AWS S3 bucket with compression based on file extension in S3 URL. Skips upload if the local file's hash is identical to the S3 object's metadata `sha256sum`. Adds the following user defined metadata to uploaded S3 object: - `sha256sum` - hash of the file generated by [sha256sum](sha256sum) - `recordcount` - the line count of the file -- [download-from-s3](download-from-s3) - Download file from AWS S3 bucket with decompression based on file extension in S3 URL. +- [download-from-s3](scripts/download-from-s3) - Download file from AWS S3 bucket with decompression based on file extension in S3 URL. Skips download if the local file already exists and has a hash identical to the S3 object's metadata `sha256sum`. +## Snakemake + +Snakemake workflow functions that are shared across many pathogen workflows that don’t really belong in any of our existing tools. + +- [config.smk](snakemake/config.smk) - Shared functions for parsing workflow configs. +- [remote_files.smk](snakemake/remote_files.smk) - Exposes the `path_or_url` function which will use Snakemake's storage plugins to download/upload files to remote providers as needed. + + ## Software requirements Some scripts may require Bash ≥4. If you are running these scripts on macOS, the builtin Bash (`/bin/bash`) does not meet this requirement. You can install [Homebrew's Bash](https://formulae.brew.sh/formula/bash) which is more up to date. diff --git a/shared/vendored/scripts/assign-colors b/shared/vendored/scripts/assign-colors new file mode 100755 index 0000000..e42a44d --- /dev/null +++ b/shared/vendored/scripts/assign-colors @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +""" +Generate colors.tsv for augur export based on ordering, color schemes, and +traits that exists in the metadata. +""" +import argparse +import pandas as pd + +if __name__ == '__main__': + parser = argparse.ArgumentParser( + description="Assign colors based on defined ordering of traits.", + formatter_class=argparse.ArgumentDefaultsHelpFormatter + ) + + parser.add_argument('--ordering', type=str, required=True, + help="""Input TSV file defining the color ordering where the first + column is the field and the second column is the trait in that field. + Blank lines are ignored. Lines starting with '#' will be ignored as comments.""") + parser.add_argument('--color-schemes', type=str, required=True, + help="Input color schemes where each line is a different color scheme separated by tabs.") + parser.add_argument('--metadata', type=str, + help="""If provided, restrict colors to only those traits found in + metadata. If the metadata includes a 'focal' column that only contains + boolean values, then restrict colors to traits for rows where 'focal' + is set to True.""") + parser.add_argument('--output', type=str, required=True, + help="Output colors TSV file to be passed to augur export.") + args = parser.parse_args() + + assignment = {} + with open(args.ordering) as f: + for line in f.readlines(): + array = line.strip().split("\t") + # Ignore empty lines or commented lines + if not array or not array[0] or array[0].startswith('#'): + continue + # Throw a warning if encountering a line not matching the expected number of columns, ignore line + elif len(array)!=2: + print(f"WARNING: Could not decode color ordering line: {line}") + continue + # Otherwise, process color ordering where we expect 2 columns: name, traits + else: + name = array[0] + trait = array[1] + if name not in assignment: + assignment[name] = [trait] + else: + assignment[name].append(trait) + + # if metadata supplied, go through and + # 1. remove assignments that don't exist in metadata + # 2. remove assignments that have 'focal' set to 'False' in metadata + if args.metadata: + metadata = pd.read_csv(args.metadata, delimiter='\t') + for name, trait in assignment.items(): + if name in metadata: + if 'focal' in metadata and metadata['focal'].dtype == 'bool': + focal_list = metadata.loc[metadata['focal'], name].unique() + subset_focal = [x for x in assignment[name] if x in focal_list] + assignment[name] = subset_focal + else: # no 'focal' present + subset_present = [x for x in assignment[name] if x in metadata[name].unique()] + assignment[name] = subset_present + + + schemes = {} + counter = 0 + with open(args.color_schemes) as f: + for line in f.readlines(): + counter += 1 + array = line.lstrip().rstrip().split("\t") + schemes[counter] = array + + with open(args.output, 'w') as f: + for trait_name, trait_array in assignment.items(): + if len(trait_array)==0: + print(f"No traits found for {trait_name}") + continue + if len(schemes)0): + if (remain>len(schemes)): + color_array = [*color_array, *schemes[len(schemes)]] + remain -= len(schemes) + else: + color_array = [*color_array, *schemes[remain]] + remain = 0 + else: + color_array = schemes[len(trait_array)] + + zipped = list(zip(trait_array, color_array)) + for trait_value, color in zipped: + f.write(trait_name + "\t" + trait_value + "\t" + color + "\n") + f.write("\n") diff --git a/ingest/vendored/cloudfront-invalidate b/shared/vendored/scripts/cloudfront-invalidate similarity index 100% rename from ingest/vendored/cloudfront-invalidate rename to shared/vendored/scripts/cloudfront-invalidate diff --git a/ingest/vendored/download-from-s3 b/shared/vendored/scripts/download-from-s3 similarity index 100% rename from ingest/vendored/download-from-s3 rename to shared/vendored/scripts/download-from-s3 diff --git a/ingest/vendored/fetch-from-ncbi-entrez b/shared/vendored/scripts/fetch-from-ncbi-entrez similarity index 100% rename from ingest/vendored/fetch-from-ncbi-entrez rename to shared/vendored/scripts/fetch-from-ncbi-entrez diff --git a/ingest/vendored/notify-on-diff b/shared/vendored/scripts/notify-on-diff similarity index 100% rename from ingest/vendored/notify-on-diff rename to shared/vendored/scripts/notify-on-diff diff --git a/ingest/vendored/notify-on-job-fail b/shared/vendored/scripts/notify-on-job-fail similarity index 100% rename from ingest/vendored/notify-on-job-fail rename to shared/vendored/scripts/notify-on-job-fail diff --git a/ingest/vendored/notify-on-job-start b/shared/vendored/scripts/notify-on-job-start similarity index 100% rename from ingest/vendored/notify-on-job-start rename to shared/vendored/scripts/notify-on-job-start diff --git a/ingest/vendored/notify-on-record-change b/shared/vendored/scripts/notify-on-record-change similarity index 100% rename from ingest/vendored/notify-on-record-change rename to shared/vendored/scripts/notify-on-record-change diff --git a/shared/vendored/scripts/notify-slack b/shared/vendored/scripts/notify-slack new file mode 100755 index 0000000..c6f1a87 --- /dev/null +++ b/shared/vendored/scripts/notify-slack @@ -0,0 +1,93 @@ +#!/usr/bin/env bash +set -euo pipefail + +: "${SLACK_TOKEN:?The SLACK_TOKEN environment variable is required.}" +: "${SLACK_CHANNELS:?The SLACK_CHANNELS environment variable is required.}" + +upload=0 +output=/dev/null +thread_ts="" +broadcast=0 +fail_on_error=0 +args=() + +for arg; do + case "$arg" in + --upload) + upload=1;; + --output=*) + output="${arg#*=}";; + --thread-ts=*) + thread_ts="${arg#*=}";; + --broadcast) + broadcast=1;; + --fail-on-error) + fail_on_error=1;; + *) + args+=("$arg");; + esac +done + +set -- "${args[@]}" + +text="${1:?Some message text is required.}" + +send_slack_message() { + if [[ "$upload" == 1 ]]; then + echo "Uploading data to Slack with the message: $text" + + upload_file="$(mktemp -t upload-file-XXXXXX)" + trap "rm -f '$upload_file'" EXIT + + cat /dev/stdin > "$upload_file" + # printf used to strip whitespace from output of macOS/BSD wc + # See + length=$(printf '%d' "$(<"$upload_file" wc -c)") + + upload_info=$(curl https://slack.com/api/files.getUploadURLExternal \ + --header "Authorization: Bearer $SLACK_TOKEN" \ + --form-string filename="$text" \ + --form-string length="$length" \ + --fail --silent --show-error \ + --http1.1 ) + + upload_url="$(jq -r .upload_url <<< "$upload_info")" + curl "$upload_url" \ + --form-string filename="$text" \ + --form file="@$upload_file" \ + --fail --silent --show-error \ + --http1.1 > /dev/null + + files_uploaded="$(jq -r "[{id: .file_id}]" <<< "$upload_info")" + curl -X POST https://slack.com/api/files.completeUploadExternal \ + --header "Authorization: Bearer $SLACK_TOKEN" \ + --form-string channel_id="$SLACK_CHANNELS" \ + --form-string thread_ts="$thread_ts" \ + --form-string files="$files_uploaded" \ + --fail --silent --show-error \ + --http1.1 \ + --output "$output" + + else + echo "Posting Slack message: $text" + curl https://slack.com/api/chat.postMessage \ + --header "Authorization: Bearer $SLACK_TOKEN" \ + --form-string channel="$SLACK_CHANNELS" \ + --form-string text="$text" \ + --form-string thread_ts="$thread_ts" \ + --form-string reply_broadcast="$broadcast" \ + --fail --silent --show-error \ + --http1.1 \ + --output "$output" + fi +} + +if ! send_slack_message; then + if [[ "$fail_on_error" == 1 ]]; then + echo "Sending Slack message failed" + exit 1 + else + echo "Sending Slack message failed, but exiting with success anyway." + exit 0 + fi +fi diff --git a/ingest/vendored/s3-object-exists b/shared/vendored/scripts/s3-object-exists similarity index 100% rename from ingest/vendored/s3-object-exists rename to shared/vendored/scripts/s3-object-exists diff --git a/ingest/vendored/sha256sum b/shared/vendored/scripts/sha256sum similarity index 100% rename from ingest/vendored/sha256sum rename to shared/vendored/scripts/sha256sum diff --git a/ingest/vendored/trigger b/shared/vendored/scripts/trigger similarity index 100% rename from ingest/vendored/trigger rename to shared/vendored/scripts/trigger diff --git a/ingest/vendored/trigger-on-new-data b/shared/vendored/scripts/trigger-on-new-data similarity index 100% rename from ingest/vendored/trigger-on-new-data rename to shared/vendored/scripts/trigger-on-new-data diff --git a/ingest/vendored/upload-to-s3 b/shared/vendored/scripts/upload-to-s3 similarity index 100% rename from ingest/vendored/upload-to-s3 rename to shared/vendored/scripts/upload-to-s3 diff --git a/shared/vendored/snakemake/config.smk b/shared/vendored/snakemake/config.smk new file mode 100644 index 0000000..c6217b5 --- /dev/null +++ b/shared/vendored/snakemake/config.smk @@ -0,0 +1,77 @@ +""" +Shared functions to be used within a Snakemake workflow for parsing +workflow configs. +""" +import os.path +from collections.abc import Callable +from snakemake.io import Wildcards +from typing import Optional +from textwrap import dedent, indent + + +class InvalidConfigError(Exception): + pass + + +def resolve_config_path(path: str, defaults_dir: Optional[str] = None) -> Callable[[Wildcards], str]: + """ + Resolve a relative *path* given in a configuration value. Will always try to + resolve *path* after expanding wildcards with Snakemake's `expand` functionality. + + Returns the path for the first existing file, checked in the following order: + 1. relative to the analysis directory or workdir, usually given by ``--directory`` (``-d``) + 2. relative to *defaults_dir* if it's provided + 3. relative to the workflow's ``defaults/`` directory if *defaults_dir* is _not_ provided + + This behaviour allows a default configuration value to point to a default + auxiliary file while also letting the file used be overridden either by + setting an alternate file path in the configuration or by creating a file + with the conventional name in the workflow's analysis directory. + """ + global workflow + + def _resolve_config_path(wildcards): + try: + expanded_path = expand(path, **wildcards)[0] + except snakemake.exceptions.WildcardError as e: + available_wildcards = "\n".join(f" - {wildcard}" for wildcard in wildcards) + raise snakemake.exceptions.WildcardError(indent(dedent(f"""\ + {str(e)} + + However, resolve_config_path({{path}}) requires the wildcard. + + Wildcards available for this path are: + + {{available_wildcards}} + + Hint: Check that the config path value does not misspell the wildcard name + and that the rule actually uses the wildcard name. + """.lstrip("\n").rstrip()).format(path=repr(path), available_wildcards=available_wildcards), " " * 4)) + + if os.path.exists(expanded_path): + return expanded_path + + if defaults_dir: + defaults_path = os.path.join(defaults_dir, expanded_path) + else: + # Special-case defaults/… for backwards compatibility with older + # configs. We could achieve the same behaviour with a symlink + # (defaults/defaults → .) but that seems less clear. + if path.startswith("defaults/"): + defaults_path = os.path.join(workflow.basedir, expanded_path) + else: + defaults_path = os.path.join(workflow.basedir, "defaults", expanded_path) + + if os.path.exists(defaults_path): + return defaults_path + + raise InvalidConfigError(indent(dedent(f"""\ + Unable to resolve the config-provided path {path!r}, + expanded to {expanded_path!r} after filling in wildcards. + The workflow does not include the default file {defaults_path!r}. + + Hint: Check that the file {expanded_path!r} exists in your analysis + directory or remove the config param to use the workflow defaults. + """), " " * 4)) + + return _resolve_config_path diff --git a/shared/vendored/snakemake/remote_files.smk b/shared/vendored/snakemake/remote_files.smk new file mode 100644 index 0000000..844f80e --- /dev/null +++ b/shared/vendored/snakemake/remote_files.smk @@ -0,0 +1,159 @@ +""" +Helper functions to set-up storage plugins for remote inputs/outputs. See the +docstring of `path_or_url` for usage instructions. + +The errors raised by storage plugins are often confusing. For instance, a HTTP +404 error will result in a `MissingInputException` with little hint as to the +underlying issue. S3 credentials errors are similarly confusing and we attempt +to check these ourselves to improve UX here. +""" + +from urllib.parse import urlparse + +# Keep a list of known public buckets, which we'll allow uncredentialled (unsigned) access to +# We could make this config-definable in the future +PUBLIC_BUCKETS = set(['nextstrain-data']) + +# Keep track of registered storage plugins to enable reuse +_storage_registry = {} + +class RemoteFilesMissingCredentials(Exception): + pass + +def _storage_s3(*, bucket, keep_local, retries) -> snakemake.storage.StorageProviderProxy: + """ + Registers and returns an instance of snakemake-storage-plugin-s3. Typically AWS + credentials are required for _any_ request however we allow requests to known + public buckets (see `PUBLIC_BUCKETS`) to be unsigned which allows for a nice user + experience in the common case of downloading inputs from s3://nextstrain-data. + + The intended behaviour for various (S3) URIs supplied to `path_or_url` is: + + | | S3 buckets | credentials present | credentials missing | + |----------|----------------------------|---------------------|---------------------| + | download | private / private + public | signed | Credentials Error | + | | public | signed | unsigned | + | upload | private / private + public | signed | Credentials Error | + | | public | signed | AccessDenied Error | + """ + # If the bucket is public then we may use an unsigned request which has the nice UX + # of not needing credentials to be present. If we've made other signed requests _or_ + # credentials are present then we just sign everything. This has implications for upload: + # if you attempt to upload to a public bucket without credentials then we allow that here + # and you'll get a subsequent `AccessDenied` error when the upload is attempted. + if bucket in PUBLIC_BUCKETS and \ + "s3_signed" not in _storage_registry and \ + ("s3_unsigned" in _storage_registry or not _aws_credentials_present()): + + if provider:=_storage_registry.get('s3_unsigned', None): + return provider + + from botocore import UNSIGNED # dependency of snakemake-storage-plugin-s3 + storage s3_unsigned: + provider="s3", + signature_version=UNSIGNED, + retries=retries, + keep_local=keep_local, + + _storage_registry['s3_unsigned'] = storage.s3_unsigned + return _storage_registry['s3_unsigned'] + + # Resource fetched/uploaded via a signed request, which will require AWS credentials + if provider:=_storage_registry.get('s3_signed', None): + return provider + + # Enforce the presence of credentials to paper over + if not _aws_credentials_present(): + raise RemoteFilesMissingCredentials() + + # the tag appears in the local file path, so reference 'signed' to give a hint about credential errors + storage s3_signed: + provider="s3", + retries=retries, + keep_local=keep_local, + + _storage_registry['s3_signed'] = storage.s3_signed + return _storage_registry['s3_signed'] + +def _aws_credentials_present() -> bool: + import boto3 # dependency of snakemake-storage-plugin-s3 + session = boto3.Session() + creds = session.get_credentials() + return creds is not None + +def _storage_http(*, keep_local, retries) -> snakemake.storage.StorageProviderProxy: + """ + Registers and returns an instance of snakemake-storage-plugin-http + """ + if provider:=_storage_registry.get('http', None): + return provider + + storage: + provider="http", + allow_redirects=True, + supports_head=True, + keep_local=keep_local, + retries=retries, + + _storage_registry['http'] = storage.http + return _storage_registry['http'] + + +def path_or_url(uri, *, keep_local=True, retries=2) -> str: + """ + Intended for use in Snakemake inputs / outputs to transparently use remote + resources. Returns the URI wrapped by an applicable storage plugin. Local + filepaths will be returned unchanged. + + For example, the following rule will download inputs from HTTPs and upload + the output to S3: + + rule filter: + input: + sequences = path_or_url("https://data.nextstrain.org/..."), + metadata = path_or_url("https://data.nextstrain.org/..."), + output: + sequences = path_or_url("s3://...") + shell: + r''' + augur filter \ + --sequences {input.sequences:q} \ + --metadata {input.metadata:q} \ + --metadata-id-columns accession \ + --output-sequences {output.sequences:q} + ''' + + If *keep_local* is True (the default) then downloaded/uploaded files will + remain in `.snakemake/storage/`. The presence of a previously downloaded + file (via `keep_local=True`) does not guarantee that the file will not be + re-downloaded if the storage plugin decides the local file is out of date. + + Depending on the *uri* authentication may be required. See the specific + helper functions (such as `_storage_s3`) for more details. + + See for + more information on Snakemake storage plugins. Note: various snakemake + plugins will be required depending on the URIs provided. + """ + info = urlparse(uri) + + if info.scheme=='': # local + return uri # no storage wrapper + + if info.scheme=='s3': + try: + return _storage_s3(bucket=info.netloc, keep_local=keep_local, retries=retries)(uri) + except RemoteFilesMissingCredentials as e: + raise Exception(f"AWS credentials are required to access {uri!r}") from e + + if info.scheme=='https': + return _storage_http(keep_local=keep_local, retries=retries)(uri) + elif info.scheme=='http': + raise Exception(f"HTTP remote file support is not implemented in nextstrain workflows (attempting to access {uri!r}).\n" + "Please use an HTTPS address instead.") + + if info.scheme in ['gs', 'gcs']: + raise Exception(f"Google Storage is not yet implemented for nextstrain workflows (attempting to access {uri!r}).\n" + "Please get in touch if you require this functionality and we can add it to our workflows") + + raise Exception(f"Input address {uri!r} (scheme={info.scheme!r}) is from a non-supported remote")