diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml new file mode 100644 index 0000000..4868200 --- /dev/null +++ b/.github/workflows/pytest.yml @@ -0,0 +1,25 @@ +name: pytest + +on: + workflow_dispatch: + pull_request: + +jobs: + pytest: + runs-on: ubuntu-latest + name: pytest + steps: + - name: Check out source repository + uses: actions/checkout@v4 + - name: Set up Python environment + uses: actions/setup-python@v5 + with: + python-version: "3.12" + # The suite needs no network and no datasets CLI, so pip is enough: + # genomehubs pulls in the rest of what flows/lib imports. + - name: Install dependencies + run: python -m pip install --upgrade pip "genomehubs>=2.10.14" requests pytest + - name: Run tests + env: + SKIP_PREFECT: "true" + run: python -m pytest tests/ -q diff --git a/configs/assembly_historical.types.yaml b/configs/assembly_historical.types.yaml index b98f0fd..f46b222 100644 --- a/configs/assembly_historical.types.yaml +++ b/configs/assembly_historical.types.yaml @@ -136,6 +136,17 @@ attributes: submitter: header: submitter path: assemblyInfo.submitter + # Supersession metadata stamped by the Phase 1 daily parser. Declared here + # so the columns survive the GenomeHubs write path instead of being dropped. + superseded_by: + header: superseded_by + path: processedAssemblyInfo.supersededBy + superseded_by_version: + header: superseded_by_version + path: processedAssemblyInfo.supersededByVersion + superseded_date: + header: superseded_date + path: processedAssemblyInfo.supersededDate ungapped_span: header: totalUngappedLength path: assemblyStats.totalUngappedLength diff --git a/flows/lib/assembly_versions_utils.py b/flows/lib/assembly_versions_utils.py index 070e7c9..d0b59ed 100644 --- a/flows/lib/assembly_versions_utils.py +++ b/flows/lib/assembly_versions_utils.py @@ -1,18 +1,311 @@ -"""Shared utilities for assembly version discovery and fetching. +"""Shared utilities for assembly version discovery, schema and paths. -Used by both the backfill parser and the incremental updater to discover -assembly versions via NCBI FTP and fetch per-version metadata. +Used by the backfill parser, the incremental parser, the summary generator and +the milestone flow to discover assembly versions via NCBI FTP, fetch +per-version metadata, read rows written by any phase, and resolve the +current-assembly TSV paths from a loaded config rather than a hardcoded name. """ +import csv +import glob as globmod +import gzip import json import os import re import time +from typing import Callable, Optional from flows.lib import utils ACCESSION_PATTERN = re.compile(r"^GC[AF]_\d{9}\.\d+$") +# Cell values that mean "no value". Upstream writes the literal string +# "None" wherever a field was absent when the row was formatted -- +# genomehubs' format_entry stringifies a bare None -- so every phase has to +# read it as empty or it becomes a date, a taxid or a metric that is not +# there. +ABSENT_VALUES = frozenset({"", "None"}) + +# Leading bytes of a gzip member, used to detect compression by content. +GZIP_MAGIC = b"\x1f\x8b" + +# Canonical column names, as declared in assembly_historical.types.yaml. Every +# phase writes these; the ``*_ALIASES`` tuples additionally tolerate the +# snake_case spellings written by earlier revisions of the Phase 1 parser. +COL_ACCESSION = "genbankAccession" +COL_ASSEMBLY_ID = "assemblyID" +COL_VERSION_STATUS = "versionStatus" +COL_SUPERSEDED_BY = "superseded_by" +COL_SUPERSEDED_BY_VERSION = "superseded_by_version" +COL_SUPERSEDED_DATE = "superseded_date" + +ACCESSION_ALIASES = (COL_ACCESSION, "accession") +# "assemblyId" is the spelling parse_ncbi_assemblies has written since +# 5180090 (2026-07-27); assembly_historical.types.yaml and Phase 0 still use +# "assemblyID". Read both until the case is settled upstream — the write side +# stays on COL_ASSEMBLY_ID, which is what the YAML declares. +ASSEMBLY_ID_ALIASES = (COL_ASSEMBLY_ID, "assemblyId", "assembly_id") +VERSION_STATUS_ALIASES = (COL_VERSION_STATUS, "version_status") + +# Every aliased field, canonical spelling first. canonicalize_columns walks +# this to collapse a row onto the spellings the YAML declares. +ALIASES_BY_CANONICAL = ( + ACCESSION_ALIASES, + ASSEMBLY_ID_ALIASES, + VERSION_STATUS_ALIASES, +) + +# Filename of the historical TSV, and the outputs derived from it. Used to +# exclude the pipeline's own products when discovering the current TSV. +HISTORICAL_TSV_NAME = "assembly_historical.tsv" +HISTORICAL_YAML_NAME = "assembly_historical.types.yaml" +# configs/ sits beside flows/ in the repo, and holds the historical schema. +REPO_CONFIG_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), + "configs", +) +CURRENT_TSV_DEFAULT = "assembly_current.tsv" +DERIVED_TSV_NAMES = frozenset({ + HISTORICAL_TSV_NAME, + "assembly_version_summary.tsv", + "taxon_milestone_summary.tsv", +}) + + +def cell(row: dict, *keys: str) -> str: + """Return the first populated cell in ``row`` among ``keys``. + + Values in ABSENT_VALUES read as empty, so a column holding the literal + string "None" is treated the same as a column holding nothing. + + Args: + row (dict): A TSV row. + *keys (str): Column names to try, in precedence order. + + Returns: + str: The first populated value, stripped, or "" when there is none. + """ + for key in keys: + value = str(row.get(key) or "").strip() + if value not in ABSENT_VALUES: + return value + return "" + + +def _first_present(row: dict, keys: tuple) -> str: + """Return the first populated value in ``row`` for any of ``keys``.""" + return cell(row, *keys) + + +def get_accession(row: dict) -> str: + """Return a row's versioned accession, whichever column holds it. + + Real assembly TSVs use ``genbankAccession``; ``accession`` is accepted for + rows that originated from an NCBI JSONL record or an older parse. + + Args: + row (dict): A TSV row. + + Returns: + str: The versioned accession, or "" when absent. + """ + return _first_present(row, ACCESSION_ALIASES) + + +def get_assembly_id(row: dict) -> str: + """Return a row's assembly ID across all three spellings in the pipeline. + + ``assemblyID`` is canonical (declared in assembly_historical.types.yaml and + written by Phase 0); ``assemblyId`` is what the current-assembly parser + writes upstream; ``assembly_id`` is the legacy snake_case form. + + Args: + row (dict): A TSV row. + + Returns: + str: The assembly ID, or "" when absent. + """ + return _first_present(row, ASSEMBLY_ID_ALIASES) + + +def get_version_status(row: dict) -> str: + """Return a row's version status, tolerating either naming convention. + + Args: + row (dict): A TSV row. + + Returns: + str: The version status, or "" when absent. + """ + return _first_present(row, VERSION_STATUS_ALIASES) + + +def canonicalize_columns(row: dict) -> dict: + """Return a copy of ``row`` with each alias spelling folded onto its column. + + Reads tolerate the alias spellings, but a row about to be *written* to the + historical TSV must carry one column per field. A row copied from the + current TSV otherwise arrives holding upstream's ``assemblyId`` alongside + the ``assemblyID`` the caller sets, and the merged output then has two + assembly-ID columns disagreeing about the same assembly — only one of + which the historical YAML declares. + + Only fields the row actually has are touched, so this never invents an + empty column for a field the row never carried. + + Args: + row (dict): A TSV row, possibly using alias spellings. + + Returns: + dict: A copy holding only the canonical spelling of each aliased field. + """ + canonical_row = dict(row) + for aliases in ALIASES_BY_CANONICAL: + if not any(alias in row for alias in aliases): + continue + value = _first_present(row, aliases) + for alias in aliases[1:]: + canonical_row.pop(alias, None) + canonical_row[aliases[0]] = value + return canonical_row + + +def is_gzipped(path: str) -> bool: + """Report whether a TSV is gzip-compressed. + + The name alone is not enough: ``snapshot_previous_output`` copies a + gzipped current TSV verbatim to ``.gz.previous``, so the compressed + file does not end ``.gz``. Sniff the magic bytes, and fall back to the + extension when the file cannot be read. + + Args: + path (str): Path to the TSV. + + Returns: + bool: True when the file is gzip-compressed. + """ + name = str(path) + try: + with open(name, "rb") as f: + return f.read(2) == GZIP_MAGIC + except OSError: + return name.endswith(".gz") + + +def open_tsv(path: str, mode: str = "rt"): + """Open a TSV transparently, decompressing gzipped files. + + Args: + path (str): Path to the TSV. + mode (str): Text mode to open with. + + Returns: + IO: An open text-mode file object. + """ + if is_gzipped(path): + return gzip.open(path, mode, encoding="utf-8") + return open(path, mode, encoding="utf-8") + + +def discover_current_tsv_name(work_dir: str) -> str: + """Find the current-assembly TSV filename in ``work_dir``. + + Fallback for callers that have no config to resolve from. Considers every + ``*.tsv`` / ``*.tsv.gz`` in ``work_dir`` except the pipeline's own derived + outputs. + + Args: + work_dir (str): Directory holding the pipeline files. + + Returns: + str: Basename of the current TSV. Falls back to + ``assembly_current.tsv`` when nothing is on disk yet, so callers + still produce a meaningful not-found message. + + Raises: + ValueError: If several candidates are present and none of them is the + conventional ``assembly_current.tsv``. + """ + candidates = sorted( + globmod.glob(os.path.join(work_dir, "*.tsv")) + + globmod.glob(os.path.join(work_dir, "*.tsv.gz")) + ) + names = [ + os.path.basename(c) + for c in candidates + if os.path.basename(c) not in DERIVED_TSV_NAMES + ] + if not names: + return CURRENT_TSV_DEFAULT + if len(names) == 1: + return names[0] + if CURRENT_TSV_DEFAULT in names: + return CURRENT_TSV_DEFAULT + raise ValueError( + f"Cannot identify the current assembly TSV in {work_dir}: {names}. " + "Pass a config so the filename can be resolved from meta['file_name']." + ) + + +def resolve_current_tsv_paths( + work_dir: str, config: Optional[object] = None +) -> tuple[str, str, Callable]: + """Resolve the current-assembly TSV paths and the matching open function. + + The filename comes from ``config.meta["file_name"]`` when a config is + supplied, so nothing hardcodes ``assembly_current.tsv``. The previous-run + snapshot is ``.previous``, written verbatim by + ``snapshot_previous_output`` — so it carries the same compression as the + current file, which is why the open function is returned alongside. + + Args: + work_dir (str): Directory holding the pipeline files. + config (Config, optional): Loaded YAML config whose + ``meta["file_name"]`` names the current TSV. + + Returns: + tuple: ``(current_tsv, previous_tsv, open_fn)``. + """ + if config is not None: + name = os.path.basename(config.meta["file_name"]) + else: + name = discover_current_tsv_name(work_dir) + current_tsv = os.path.join(work_dir, name) + return current_tsv, f"{current_tsv}.previous", open_tsv + + +def resolve_historical_tsv_path(work_dir: str) -> str: + """Return the historical TSV path inside ``work_dir``. + + Args: + work_dir (str): Directory holding the pipeline files. + + Returns: + str: Path to assembly_historical.tsv. + """ + return os.path.join(work_dir, HISTORICAL_TSV_NAME) + + +def resolve_historical_yaml_path(work_dir: str) -> Optional[str]: + """Locate assembly_historical.types.yaml, or None when it is not found. + + The pipeline copies a parser's YAML into work_dir, so look there first and + fall back to the copy in the repo's configs/ directory. Returning None + rather than raising keeps the schema an optimisation: the caller falls + back to the column order already on disk. + + Args: + work_dir (str): Directory holding the pipeline files. + + Returns: + str or None: Path to the historical types YAML. + """ + candidates = [ + os.path.join(work_dir, HISTORICAL_YAML_NAME), + os.path.join(REPO_CONFIG_DIR, HISTORICAL_YAML_NAME), + ] + return next((path for path in candidates if os.path.exists(path)), None) + def parse_version(accession: str) -> int: """Extract the version number from a dotted accession string. @@ -40,6 +333,38 @@ def parse_accession(accession: str) -> tuple[str, int]: return parts[0], int(parts[1]) if len(parts) > 1 else 1 +def load_versions_by_base(tsv_path: str) -> dict[str, set[int]]: + """Index the assembly versions already present in a TSV by base accession. + + Used against assembly_historical.tsv so the daily diff can tell a genuine + gap from one that a previous run — or the Phase 0 backfill — has already + parsed. Only the accession column is read, so this stays cheap on a file + holding tens of thousands of rows, and it opens no network connections. + + Args: + tsv_path (str): Path to a TSV written by any phase. + + Returns: + dict: Mapping of base_accession -> set of version numbers present. + Empty when the file is absent or empty, which is the expected + state before the first backfill. + """ + versions_by_base: dict[str, set[int]] = {} + + if not os.path.exists(tsv_path) or os.path.getsize(tsv_path) == 0: + return versions_by_base + + with open_tsv(tsv_path) as f: + for row in csv.DictReader(f, delimiter="\t"): + accession = get_accession(row) + if not accession: + continue + base_acc, version = parse_accession(accession) + versions_by_base.setdefault(base_acc, set()).add(version) + + return versions_by_base + + def setup_cache_directories(work_dir: str) -> None: """Create cache directory structure under work_dir. diff --git a/flows/lib/compute_taxon_milestones.py b/flows/lib/compute_taxon_milestones.py index f34a473..44b5752 100644 --- a/flows/lib/compute_taxon_milestones.py +++ b/flows/lib/compute_taxon_milestones.py @@ -30,13 +30,19 @@ import csv import os +from flows.lib.assembly_versions_utils import ( + get_accession, + get_version_status, + open_tsv, + resolve_current_tsv_paths, + resolve_historical_tsv_path, +) from flows.lib.conditional_import import emit_event, flow from flows.lib.load_taxonomy import CANONICAL_RANKS, build_taxonomy -from flows.lib.shared_args import TAXDUMP_PATH, WORK_DIR +from flows.lib.shared_args import TAXDUMP_PATH, WORK_DIR, YAML_PATH from flows.lib.shared_args import parse_args as _parse_args +from flows.lib.utils import load_config -CURRENT_TSV = "assembly_current.tsv" -HISTORICAL_TSV = "assembly_historical.tsv" OUTPUT_TSV = "taxon_milestone_summary.tsv" # EBP umbrella BioProject — affiliation is membership in this project. @@ -77,7 +83,7 @@ def _accession(row: dict) -> str: """Return the assembly accession, normalised across phases.""" - return row.get("accession") or row.get("genbankAccession") or "" + return get_accession(row) def _release_date(row: dict) -> str: @@ -88,7 +94,7 @@ def _release_date(row: dict) -> str: def _version_status(row: dict) -> str: """Return versionStatus across camelCase / snake_case, default 'current'.""" - return row.get("versionStatus") or row.get("version_status") or "current" + return get_version_status(row) or "current" def _is_affiliated(row: dict) -> bool: @@ -132,7 +138,7 @@ def load_assemblies(current_tsv: str, historical_tsv: str) -> list[dict]: print(f" Warning: {label} TSV not found: {path}") continue count = 0 - with open(path, encoding="utf-8") as f: + with open_tsv(path) as f: for row in csv.DictReader(f, delimiter="\t"): rows.append(row) count += 1 @@ -325,17 +331,26 @@ def build_output_rows(taxa: dict, species_extra: dict) -> list[dict]: @flow(log_prints=True) -def compute_taxon_milestones(work_dir: str = ".", taxdump_path: str | None = None) -> None: +def compute_taxon_milestones( + work_dir: str = ".", + taxdump_path: str | None = None, + yaml_path: str | None = None, +) -> None: """Compute per-taxon assembly milestones and write the summary TSV. + The current-TSV filename comes from the config when yaml_path is given, + and is otherwise discovered in work_dir, so nothing hardcodes it. + Args: - work_dir: Directory containing assembly_current.tsv and - assembly_historical.tsv; output is written there too. + work_dir: Directory containing the current and historical TSVs; + output is written there too. taxdump_path: Dev/test NCBI taxdump directory. In production the lineage arrives via the upstream contract and this is omitted. + yaml_path: Optional YAML config naming the current TSV. """ - current_tsv = os.path.join(work_dir, CURRENT_TSV) - historical_tsv = os.path.join(work_dir, HISTORICAL_TSV) + config = load_config(config_file=yaml_path) if yaml_path else None + current_tsv, _, _ = resolve_current_tsv_paths(work_dir, config=config) + historical_tsv = resolve_historical_tsv_path(work_dir) output_tsv = os.path.join(work_dir, OUTPUT_TSV) separator = "=" * 80 @@ -406,7 +421,11 @@ def compute_taxon_milestones(work_dir: str = ".", taxdump_path: str | None = Non if __name__ == "__main__": args = _parse_args( - [WORK_DIR, TAXDUMP_PATH], + [WORK_DIR, TAXDUMP_PATH, YAML_PATH], description="Compute per-taxon assembly milestones from version-tracked TSVs", ) - compute_taxon_milestones(work_dir=args.work_dir, taxdump_path=args.taxdump_path) + compute_taxon_milestones( + work_dir=args.work_dir, + taxdump_path=args.taxdump_path, + yaml_path=getattr(args, "yaml_path", None), + ) diff --git a/flows/lib/generate_assembly_summary.py b/flows/lib/generate_assembly_summary.py index 4eff2d7..8c1109b 100644 --- a/flows/lib/generate_assembly_summary.py +++ b/flows/lib/generate_assembly_summary.py @@ -16,22 +16,40 @@ import os from collections import defaultdict -from flows.lib.assembly_versions_utils import parse_accession +from flows.lib.assembly_versions_utils import ( + cell, + get_accession, + get_version_status, + open_tsv, + parse_accession, + resolve_current_tsv_paths, + resolve_historical_tsv_path, +) from flows.lib.conditional_import import emit_event, flow -from flows.lib.shared_args import WORK_DIR +from flows.lib.shared_args import WORK_DIR, YAML_PATH from flows.lib.shared_args import parse_args as _parse_args +from flows.lib.utils import load_config -CURRENT_TSV = "assembly_current.tsv" -HISTORICAL_TSV = "assembly_historical.tsv" OUTPUT_TSV = "assembly_version_summary.tsv" -# EBP metric columns to check (either naming convention may appear) -EBP_METRIC_COLUMNS = {"ebpStandardDate", "ebp_standard_date", "ebpMetricDate", "ebp_metric_date"} +# EBP metric columns to check (either naming convention may appear). A +# tuple, not a set: cell() takes them in precedence order. +EBP_METRIC_COLUMNS = ( + "ebpStandardDate", + "ebp_standard_date", + "ebpMetricDate", + "ebp_metric_date", +) def _has_ebp_metric(row: dict) -> bool: - """Return True if any EBP metric date column is populated.""" - return any(row.get(col) for col in EBP_METRIC_COLUMNS) + """Return True if any EBP metric date column is populated. + + Read through ``cell`` so the literal string "None" counts as absent, + which is what Phase 3 does with the same column. Reading it as a value + here would have the summary claim an EBP metric the milestones deny. + """ + return bool(cell(row, *EBP_METRIC_COLUMNS)) def load_assemblies(current_tsv: str, historical_tsv: str) -> list[dict]: @@ -54,11 +72,10 @@ def load_assemblies(current_tsv: str, historical_tsv: str) -> list[dict]: print(f" Warning: {label} TSV not found: {path}") continue count = 0 - with open(path, encoding="utf-8") as f: + with open_tsv(path) as f: for row in csv.DictReader(f, delimiter="\t"): # Normalise to 'accession' so the rest of the code is uniform - if "accession" not in row or not row["accession"]: - row["accession"] = row.get("genbankAccession", "") + row["accession"] = get_accession(row) rows.append(row) count += 1 print(f" Loaded {count} {label} rows") @@ -95,29 +112,33 @@ def generate_summary_for_base(base_accession: str, rows: list[dict]) -> dict: first_row = rows[0] current_row = rows[-1] - # versionStatus column may be camelCase (Phase 0 YAML) or snake_case (Phase 1 copy) - def version_status(row): - return row.get("versionStatus") or row.get("version_status") or "current" - - superseded_count = sum(1 for r in rows if version_status(r) == "superseded") + superseded_count = sum( + 1 for r in rows if (get_version_status(r) or "current") == "superseded" + ) # First version that met EBP metric criteria first_ebp_row = next((r for r in rows if _has_ebp_metric(r)), None) return { "base_accession": base_accession, - "taxId": first_row.get("taxId", ""), + "taxId": cell(first_row, "taxId", "taxid"), "first_version_accession": first_row["accession"], "first_version_number": versions[0], - "first_version_date": first_row.get("releaseDate", ""), + "first_version_date": cell(first_row, "releaseDate"), "current_version_accession": current_row["accession"], "current_version_number": versions[-1], - "current_version_date": current_row.get("releaseDate", ""), + "current_version_date": cell(current_row, "releaseDate"), "total_versions": len(rows), "superseded_versions": superseded_count, - "first_ebp_metric_accession": first_ebp_row["accession"] if first_ebp_row else "", - "first_ebp_metric_version": parse_accession(first_ebp_row["accession"])[1] if first_ebp_row else "", - "first_ebp_metric_date": first_ebp_row.get("releaseDate", "") if first_ebp_row else "", + "first_ebp_metric_accession": ( + first_ebp_row["accession"] if first_ebp_row else "" + ), + "first_ebp_metric_version": ( + parse_accession(first_ebp_row["accession"])[1] if first_ebp_row else "" + ), + "first_ebp_metric_date": ( + cell(first_ebp_row, "releaseDate") if first_ebp_row else "" + ), "version_gaps": version_gaps, } @@ -141,15 +162,22 @@ def version_status(row): @flow(log_prints=True) -def generate_assembly_summary(work_dir: str = ".") -> None: +def generate_assembly_summary( + work_dir: str = ".", yaml_path: str | None = None +) -> None: """Combine current and historical TSVs into a per-base-accession summary. + The current-TSV filename comes from the config when yaml_path is given, + and is otherwise discovered in work_dir, so nothing hardcodes it. + Args: - work_dir: Directory containing assembly_current.tsv and - assembly_historical.tsv; output is written there too. + work_dir: Directory containing the current and historical TSVs; + output is written there too. + yaml_path: Optional YAML config naming the current TSV. """ - current_tsv = os.path.join(work_dir, CURRENT_TSV) - historical_tsv = os.path.join(work_dir, HISTORICAL_TSV) + config = load_config(config_file=yaml_path) if yaml_path else None + current_tsv, _, _ = resolve_current_tsv_paths(work_dir, config=config) + historical_tsv = resolve_historical_tsv_path(work_dir) output_tsv = os.path.join(work_dir, OUTPUT_TSV) separator = "=" * 80 @@ -172,7 +200,7 @@ def generate_assembly_summary(work_dir: str = ".") -> None: ) return - print(f"\n[2/3] Generating summaries...") + print("\n[2/3] Generating summaries...") by_base: dict[str, list[dict]] = defaultdict(list) for row in rows: acc = row.get("accession", "") @@ -187,7 +215,7 @@ def generate_assembly_summary(work_dir: str = ".") -> None: print(f" Processed {len(summaries)} unique base accessions") - print(f"\n[3/3] Writing output...") + print("\n[3/3] Writing output...") with open(output_tsv, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=SUMMARY_FIELDNAMES, delimiter="\t") writer.writeheader() @@ -225,7 +253,9 @@ def generate_assembly_summary(work_dir: str = ".") -> None: if __name__ == "__main__": args = _parse_args( - [WORK_DIR], + [WORK_DIR, YAML_PATH], description="Aggregate current and historical assembly TSVs into version summary", ) - generate_assembly_summary(work_dir=args.work_dir) + generate_assembly_summary( + work_dir=args.work_dir, yaml_path=getattr(args, "yaml_path", None) + ) diff --git a/flows/parsers/parse_assembly_versions.py b/flows/parsers/parse_assembly_versions.py index 5b2f4bb..10715ca 100644 --- a/flows/parsers/parse_assembly_versions.py +++ b/flows/parsers/parse_assembly_versions.py @@ -16,34 +16,60 @@ from pathlib import Path from typing import Optional -from flows.lib.assembly_versions_utils import parse_accession +from flows.lib.assembly_versions_utils import ( + COL_ASSEMBLY_ID, + COL_SUPERSEDED_BY, + COL_SUPERSEDED_BY_VERSION, + COL_SUPERSEDED_DATE, + COL_VERSION_STATUS, + HISTORICAL_YAML_NAME, + canonicalize_columns, + get_accession, + get_assembly_id, + open_tsv, + parse_accession, + resolve_current_tsv_paths, + resolve_historical_tsv_path, + resolve_historical_yaml_path, +) from flows.lib.conditional_import import flow -from flows.lib.shared_args import INPUT_PATH +from flows.lib.shared_args import INPUT_PATH, YAML_PATH from flows.lib.shared_args import parse_args as _parse_args from flows.lib.shared_args import required -from flows.lib.utils import Parser +from flows.lib.utils import Config, Parser, load_config -def derive_assembly_version_paths(input_path: str) -> tuple[str, str]: +DELIMITER = "\t" + + +def derive_assembly_version_paths( + input_path: str, config: Optional[Config] = None +) -> tuple[str, str]: """Derive previous_tsv and historical_tsv paths from the input JSONL path. - Both files are assumed to live alongside the JSONL in the same directory, - following the same convention used by parse_ncbi_assemblies. + Both files live alongside the JSONL, following the convention used by + parse_ncbi_assemblies. The current-TSV filename — and therefore the name + of its ".previous" snapshot — comes from config.meta["file_name"] when a + config is supplied, so nothing here hardcodes it. Args: input_path (str): Path to the current assembly_data_report.jsonl. + config (Config, optional): Loaded YAML config naming the current TSV. Returns: tuple: (previous_tsv, historical_tsv) absolute paths. """ work_dir = os.path.dirname(os.path.abspath(input_path)) - previous_tsv = os.path.join(work_dir, "assembly_current.tsv.previous") - historical_tsv = os.path.join(work_dir, "assembly_historical.tsv") - return previous_tsv, historical_tsv + _, previous_tsv, _ = resolve_current_tsv_paths(work_dir, config=config) + return previous_tsv, resolve_historical_tsv_path(work_dir) def load_previous_parsed_by_base(previous_tsv: str) -> dict[str, dict[int, dict]]: """Load previous parsed results indexed by base accession and version. + The snapshot is opened through open_tsv, so a gzipped snapshot of a + gzipped current TSV is read transparently, and the accession is read via + get_accession rather than assuming a single column name. + Args: previous_tsv (str): Path to assembly_current.tsv from the previous run. @@ -55,9 +81,11 @@ def load_previous_parsed_by_base(previous_tsv: str) -> dict[str, dict[int, dict] previous_by_base: dict[str, dict[int, dict]] = {} try: - with open(previous_tsv, encoding="utf-8") as f: - for row in csv.DictReader(f, delimiter="\t"): - accession = row["accession"] + with open_tsv(previous_tsv) as f: + for row in csv.DictReader(f, delimiter=DELIMITER): + accession = get_accession(row) + if not accession: + continue base_acc, version = parse_accession(accession) if base_acc not in previous_by_base: previous_by_base[base_acc] = {} @@ -90,16 +118,24 @@ def build_superseded_row( new_version (int): Version number of the superseding assembly. release_date (str): Release date of the superseding assembly. + Emits the canonical camelCase column names declared in + assembly_historical.types.yaml, so Phase 1 rows are indistinguishable + from Phase 0 rows in the historical TSV. The copied row is canonicalized + first: it comes from the current TSV, which spells the assembly ID + ``assemblyId``, and leaving that alongside the ``assemblyID`` set here + would put two disagreeing assembly-ID columns into the output. + Returns: - dict: Updated row with version_status, assembly_id, and superseded_by fields. + dict: Updated row with versionStatus, assemblyID and the three + supersession columns set. """ - base_acc, _ = parse_accession(previous_row["accession"]) - row = previous_row.copy() - row["version_status"] = "superseded" - row["assembly_id"] = f"{base_acc}_{previous_version}" - row["superseded_by"] = new_accession - row["superseded_by_version"] = new_version - row["superseded_date"] = release_date + base_acc, _ = parse_accession(get_accession(previous_row)) + row = canonicalize_columns(previous_row) + row[COL_VERSION_STATUS] = "superseded" + row[COL_ASSEMBLY_ID] = f"{base_acc}_{previous_version}" + row[COL_SUPERSEDED_BY] = new_accession + row[COL_SUPERSEDED_BY_VERSION] = new_version + row[COL_SUPERSEDED_DATE] = release_date return row @@ -189,37 +225,93 @@ def identify_newly_superseded( return newly_superseded, missing_versions +def merge_fieldnames( + rows: list[dict], + preferred_order: Optional[list[str]] = None, + allowed: Optional[set] = None, +) -> list[str]: + """Build an output header covering every key present in any row. + + Deriving the header from a single arbitrary row silently drops any column + that row happens to lack, so the union is taken across all rows. Columns + are emitted in preferred_order first (the YAML header order, or the order + already on disk), then any remaining keys in first-seen order. + + Args: + rows (list): All row dicts destined for the output file. + preferred_order (list, optional): Column order to honour where + possible. + allowed (set, optional): Columns permitted in the output. Keys + outside it are dropped. Defaults to no restriction, which keeps + the union whole. + + Returns: + list: Ordered fieldnames covering the union of all row keys. + """ + union: dict[str, None] = {} + for row in rows: + for key in row: + if allowed is None or key in allowed: + union.setdefault(key, None) + + order = list(preferred_order or []) + ordered = [col for col in order if col in union] + ordered += [col for col in union if col not in set(order)] + return ordered + + def append_superseded_to_tsv( - newly_superseded: list[dict], historical_tsv: str + newly_superseded: list[dict], + historical_tsv: str, + headers: Optional[list[str]] = None, ) -> None: - """Append newly superseded rows to the historical TSV, deduplicating by assembly_id. + """Merge newly superseded rows into the historical TSV, keyed by assembly ID. Reads the existing file if present, merges new rows (new rows take - precedence on duplicate assembly_id), and writes the combined result back. + precedence on duplicate assembly ID), and writes the combined result back. + The output header is the union of the columns across all rows, so merging + Phase 0 and Phase 1 row sets cannot silently drop a column. + + When headers are supplied they also bound the schema: a new row is written + under the declared columns plus whatever is already on disk, and nothing + else. Phase 1 rows are copied wholesale from the current TSV, whose + schema is upstream's rather than this one, so without that bound every + current-only column leaks into the historical TSV. Columns already in the + file stay in it either way — F3 is about never dropping those. Args: newly_superseded (list): Row dicts from identify_newly_superseded. historical_tsv (str): Path to assembly_historical.tsv. + headers (list, optional): The declared schema, i.e. config.headers + from assembly_historical.types.yaml. Sets both the column order + and the bound above. Defaults to the order already present in the + file on disk, unbounded. """ if not newly_superseded: print(" No newly superseded versions to add.") return existing: dict[str, dict] = {} + file_order: list[str] = [] historical_path = Path(historical_tsv) if historical_path.exists(): - with open(historical_tsv, encoding="utf-8") as f: - for row in csv.DictReader(f, delimiter="\t"): - existing[row["assembly_id"]] = dict(row) + with open_tsv(historical_tsv) as f: + reader = csv.DictReader(f, delimiter=DELIMITER) + for row in reader: + existing[get_assembly_id(row)] = dict(row) + file_order = list(reader.fieldnames or []) for row in newly_superseded: - existing[row["assembly_id"]] = row + existing[get_assembly_id(row)] = row - fieldnames = list(next(iter(existing.values())).keys()) + allowed = set(headers) | set(file_order) if headers else None + fieldnames = merge_fieldnames( + list(existing.values()), headers or file_order, allowed=allowed + ) with open(historical_tsv, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter( - f, fieldnames=fieldnames, delimiter="\t", extrasaction="ignore" + f, fieldnames=fieldnames, delimiter=DELIMITER, extrasaction="ignore" ) writer.writeheader() writer.writerows(existing.values()) @@ -241,7 +333,10 @@ def print_superseded_summary(newly_superseded: list[dict]) -> None: print(f" Found: {len(newly_superseded)} newly superseded versions.") print(" Examples:") for row in newly_superseded[:5]: - print(f" {row['accession']} -> superseded by v{row['superseded_by_version']}") + print( + f" {get_accession(row)} -> " + f"superseded by v{row[COL_SUPERSEDED_BY_VERSION]}" + ) if len(newly_superseded) > 5: print(f" ... and {len(newly_superseded) - 5} more") @@ -273,6 +368,7 @@ def parse_assembly_versions( new_jsonl: str, previous_tsv: str, historical_tsv: str, + historical_headers: Optional[list[str]] = None, ) -> dict: """Daily incremental update of the historical assembly TSV. @@ -283,6 +379,8 @@ def parse_assembly_versions( new_jsonl (str): Path to the current assembly_data_report.jsonl. previous_tsv (str): Path to assembly_current.tsv from the previous run. historical_tsv (str): Path to assembly_historical.tsv to update. + historical_headers (list, optional): Preferred column order for the + historical TSV. Defaults to the order already on disk. Returns: dict: Summary with keys newly_superseded_count, missing_versions_count, @@ -311,7 +409,9 @@ def parse_assembly_versions( print_missing_versions_warning(missing) print("\n[3/3] Updating historical TSV...") - append_superseded_to_tsv(newly_superseded, historical_tsv) + append_superseded_to_tsv( + newly_superseded, historical_tsv, headers=historical_headers + ) print(f"\n{separator}") print( @@ -328,6 +428,58 @@ def parse_assembly_versions( } +def load_assembly_versions_config(working_yaml: Optional[str]) -> Optional[Config]: + """Load the parser config, returning None when it is unavailable. + + The config is only used to resolve the current-TSV filename, so a missing + or unreadable YAML degrades to filename discovery in work_dir rather than + failing the run. + + Args: + working_yaml (str, optional): Path to the working YAML file. + + Returns: + Config or None: Loaded config, or None when it could not be loaded. + """ + if not working_yaml or not os.path.exists(working_yaml): + return None + try: + return load_config(config_file=working_yaml) + except Exception as e: + print(f" Warning: could not load {working_yaml} ({e});") + print(" falling back to filename discovery in work_dir.") + return None + + +def load_historical_headers(work_dir: str) -> Optional[list[str]]: + """Load the declared historical schema, returning None when unavailable. + + The headers set the column order of assembly_historical.tsv and bound + which columns a Phase 1 row may contribute, so that a row copied from the + current TSV cannot widen the file with upstream-only columns. Every + failure degrades to None — the file is then written in the order already + on disk, exactly as before — because a missing schema is not a reason to + fail a daily run. + + Args: + work_dir (str): Directory holding the pipeline files. + + Returns: + list or None: Declared column order from the historical types YAML. + """ + yaml_path = resolve_historical_yaml_path(work_dir) + if not yaml_path: + print(f" Note: {HISTORICAL_YAML_NAME} not found;") + print(" using the column order already on disk.") + return None + try: + return list(load_config(config_file=yaml_path).headers) + except Exception as e: + print(f" Warning: could not load {yaml_path} ({e});") + print(" using the column order already on disk.") + return None + + def parse_assembly_versions_wrapper( working_yaml: str, work_dir: str, @@ -337,12 +489,13 @@ def parse_assembly_versions_wrapper( ) -> None: """Wrapper matching the fetch_parse_validate parser signature. - Derives the previous TSV and historical TSV paths from work_dir and - delegates to parse_assembly_versions. + Loads working_yaml so the current-TSV filename — and the ".previous" + snapshot derived from it — comes from config.meta["file_name"] rather than + a hardcoded constant, then delegates to parse_assembly_versions. Args: - working_yaml (str): Path to the working YAML file (unused; accepted - for pipeline compatibility). + working_yaml (str): Path to the working YAML file, used to resolve the + current-TSV filename and its compression. work_dir (str): Path to the working directory containing the JSONL, the previous TSV, and the historical TSV. append (bool): Unused; accepted for pipeline compatibility. @@ -357,11 +510,15 @@ def parse_assembly_versions_wrapper( if len(paths) > 1: raise ValueError(f"More than one jsonl file found in {work_dir}") - previous_tsv, historical_tsv = derive_assembly_version_paths(paths[0]) + config = load_assembly_versions_config(working_yaml) + previous_tsv, historical_tsv = derive_assembly_version_paths( + paths[0], config=config + ) results = parse_assembly_versions( new_jsonl=paths[0], previous_tsv=previous_tsv, historical_tsv=historical_tsv, + historical_headers=load_historical_headers(work_dir), ) if results["missing_versions_count"] > 0: @@ -382,14 +539,20 @@ def plugin() -> Parser: if __name__ == "__main__": args = _parse_args( - [required(INPUT_PATH)], + [required(INPUT_PATH), YAML_PATH], description="Daily incremental update of historical assembly records", ) - previous_tsv, historical_tsv = derive_assembly_version_paths(args.input_path) + config = load_assembly_versions_config(getattr(args, "yaml_path", None)) + previous_tsv, historical_tsv = derive_assembly_version_paths( + args.input_path, config=config + ) results = parse_assembly_versions( new_jsonl=args.input_path, previous_tsv=previous_tsv, historical_tsv=historical_tsv, + historical_headers=load_historical_headers( + os.path.dirname(os.path.abspath(historical_tsv)) + ), ) print(f"Summary: superseded={results['newly_superseded_count']}, " f"missing={results['missing_versions_count']}") diff --git a/flows/parsers/parse_backfill_historical_versions.py b/flows/parsers/parse_backfill_historical_versions.py index 4b7acb0..779cb3f 100644 --- a/flows/parsers/parse_backfill_historical_versions.py +++ b/flows/parsers/parse_backfill_historical_versions.py @@ -1,7 +1,9 @@ -"""One-time historical backfill of superseded assembly versions from NCBI. +"""Historical backfill of superseded assembly versions from NCBI. Discovers and parses all superseded versions for assemblies with version > 1. -Run once before starting the daily incremental pipeline. +Run once before starting the daily incremental pipeline, and re-invoked by the +daily gap-fill path whenever parse_assembly_versions reports a missing version +— so the write into assembly_historical.tsv preserves the rows already there. Usage: python -m flows.parsers.parse_backfill_historical_versions \\ @@ -10,6 +12,7 @@ --work_dir tmp """ +import csv import json import os from datetime import datetime @@ -21,7 +24,10 @@ from flows.lib import utils from flows.lib.assembly_versions_utils import ( + COL_ACCESSION, find_all_assembly_versions, + get_accession, + open_tsv, parse_accession, parse_version, setup_cache_directories, @@ -30,13 +36,15 @@ from flows.lib.shared_args import INPUT_PATH, WORK_DIR, YAML_PATH from flows.lib.shared_args import parse_args as _parse_args from flows.lib.shared_args import required -from flows.lib.utils import Config, Parser +from flows.lib.utils import Config, Parser, append_to_tsv from flows.parsers.parse_ncbi_assemblies import ( fetch_and_parse_sequence_report, process_assembly_report, write_to_tsv, ) +DELIMITER = "\t" + def parse_historical_version( version_data: dict, @@ -166,6 +174,111 @@ def identify_assemblies_needing_backfill(input_path: str) -> list[dict]: return assemblies +def resolve_output_path(config: Config, work_dir: str) -> str: + """Point the config's output at work_dir instead of the YAML's directory. + + ``config.meta["file_name"]`` is resolved relative to the YAML file, so an + out-of-tree config would write the historical TSV next to itself. Rewrite + it onto work_dir, mirroring what run_generic_tsv_parser does. + + Args: + config (Config): Loaded YAML configuration. + work_dir (str): Working directory the run was given. + + Returns: + str: The resolved output path, also stored back on the config. + """ + output_path = os.path.join( + work_dir, os.path.basename(config.meta["file_name"]) + ) + config.meta["file_name"] = output_path + return output_path + + +def read_existing_output(output_path: str) -> tuple[list, set]: + """Read the header and the accessions already in the historical TSV. + + Only the header line and the accession column are kept, so this stays + cheap on a file holding tens of thousands of rows. + + The header doubles as the "this file already holds data" signal: it is + empty only when the file is missing or zero-length, whereas the accession + set is also empty for a file whose rows all lack an accession. + + The header matters as much as the accessions: the Phase 1 daily parser + rewrites this file with the union of the columns across every row, so the + order on disk is its output, not this config's. Phase 1 now bounds that + union by the declared schema where it can resolve the YAML, but a file + widened by an earlier run — or by a run that could not — still carries + columns the config never declared. Appending against the config header + would then write rows a few columns short. + + Args: + output_path (str): Path to assembly_historical.tsv. + + Returns: + tuple: (header, accessions), both empty when the file does not exist + or holds nothing yet. + """ + if not os.path.exists(output_path) or os.path.getsize(output_path) == 0: + return [], set() + + accessions = set() + with open_tsv(output_path) as f: + reader = csv.DictReader(f, delimiter=DELIMITER) + for row in reader: + accession = get_accession(row) + if accession: + accessions.add(accession) + header = list(reader.fieldnames or []) + return header, accessions + + +def write_or_append_parsed(parsed: dict, config: Config) -> int: + """Write parsed rows without destroying rows already in the output. + + ``write_to_tsv`` opens the output with mode "w", which is correct for the + one-time backfill against an empty directory but destructive once the + gap-fill path re-invokes this parser daily against a populated + assembly_historical.tsv. Write the whole file only when it does not exist + yet — that branch produces the header line — and otherwise append the rows + that are not already present. The rewrite branch keys off the header + rather than the accession set, so a file whose rows all lack an accession + is appended to rather than overwritten. + + Deduplication is required because appending is not idempotent: + find_all_assembly_versions re-parses every version below the current one, + and a completed checkpoint deliberately restarts from index 0. + + Rows are appended under the header already on disk rather than the config + header, so a file the Phase 1 parser has widened keeps its columns lined + up. A row missing one of those columns is written empty for it. + + Args: + parsed (dict): Mapping of accession -> parsed row. + config (Config): Loaded YAML configuration. + + Returns: + int: Number of rows actually written. + """ + output_path = config.meta["file_name"] + header, existing = read_existing_output(output_path) + + if not header: + write_to_tsv(parsed, config) + return len(parsed) + + new_rows = [ + row for accession, row in parsed.items() if accession not in existing + ] + skipped = len(parsed) - len(new_rows) + if skipped: + print(f" Skipping {skipped} versions already in {output_path}") + if new_rows: + append_to_tsv(header, new_rows, config.meta) + return len(new_rows) + + @flow(log_prints=True) def backfill_historical_versions( input_path: str, @@ -173,12 +286,16 @@ def backfill_historical_versions( work_dir: str = ".", checkpoint_file: Optional[str] = None, ): - """One-time backfill of all historical assembly versions. + """Backfill historical assembly versions into assembly_historical.tsv. Accumulates all parsed rows in memory and writes the output TSV once at the end. Checkpoints are saved periodically so the run can be resumed after interruption but do not trigger intermediate TSV writes. + The output is written into work_dir, and rows already in the output are + preserved — the daily gap-fill path re-invokes this flow, so a truncating + write would destroy the backfill it is meant to extend. + Args: input_path (str): Path to assembly_data_report.jsonl. yaml_path (str): Path to assembly_historical.types.yaml. @@ -188,6 +305,7 @@ def backfill_historical_versions( """ setup_cache_directories(work_dir) config = utils.load_config(config_file=yaml_path) + output_path = resolve_output_path(config, work_dir) checkpoint_file = checkpoint_file or derive_checkpoint_path( input_path, yaml_path, work_dir, ) @@ -224,6 +342,7 @@ def backfill_historical_versions( parsed = {} processed = start_index + written = 0 for assembly_info in assemblies[start_index:]: base_acc = assembly_info["base_accession"] @@ -257,7 +376,7 @@ def backfill_historical_versions( version_num=version_num, current_accession=current_accession, ) - genbank_acc = row.get("genbankAccession", version_acc) + genbank_acc = row.get(COL_ACCESSION) or version_acc parsed[genbank_acc] = row print("done") except Exception as e: @@ -276,7 +395,7 @@ def backfill_historical_versions( if parsed: print(f"\nWriting {len(parsed)} records to TSV...") - write_to_tsv(parsed, config) + written = write_or_append_parsed(parsed, config) save_checkpoint(checkpoint_file, processed, completed=True) @@ -284,8 +403,9 @@ def backfill_historical_versions( print("BACKFILL COMPLETE") print(f"{'=' * 80}") print(f" Processed: {processed}/{total_assemblies} assemblies") - print(f" Records written: {len(parsed)}") - print(f" Output: {config.meta['file_name']}") + print(f" Records parsed: {len(parsed)}") + print(f" Records written: {written}") + print(f" Output: {output_path}") print("\n Next step: Run daily incremental pipeline") print(f"{'=' * 80}\n") diff --git a/tests/README_phase_0_backfill.md b/tests/README_phase_0_backfill.md index 4bac0ad..defdb7a 100644 --- a/tests/README_phase_0_backfill.md +++ b/tests/README_phase_0_backfill.md @@ -44,11 +44,10 @@ Test the complete backfill on the 3-assembly fixture already in the repo ```bash mkdir -p /tmp/assembly-versions -cp configs/assembly_historical.types.yaml /tmp/assembly-versions/ SKIP_PREFECT=true python3 -m flows.parsers.parse_backfill_historical_versions \ --input_path tests/test_data/assembly_test_sample.jsonl \ - --yaml_path /tmp/assembly-versions/assembly_historical.types.yaml \ + --yaml_path configs/assembly_historical.types.yaml \ --work_dir /tmp/assembly-versions ``` @@ -56,18 +55,17 @@ SKIP_PREFECT=true python3 -m flows.parsers.parse_backfill_historical_versions \ ```powershell New-Item -ItemType Directory -Force -Path C:\Temp\assembly-versions -Copy-Item configs\assembly_historical.types.yaml C:\Temp\assembly-versions\ $env:SKIP_PREFECT = "true" python -m flows.parsers.parse_backfill_historical_versions ` --input_path tests\test_data\assembly_test_sample.jsonl ` - --yaml_path C:\Temp\assembly-versions\assembly_historical.types.yaml ` + --yaml_path configs\assembly_historical.types.yaml ` --work_dir C:\Temp\assembly-versions ``` -> **Note on output path**: the output TSV (`assembly_historical.tsv`) is written -> relative to the *YAML file's directory*, not `--work_dir`. Copying the YAML -> into `--work_dir` (as above) puts the output there. +> **Note on output path**: the output TSV (`assembly_historical.tsv`) is +> written into `--work_dir`. The YAML's own directory is not used, so the +> config can be passed straight from `configs/`. Requires `datasets` CLI and network access (FTP version discovery + `datasets summary` per version). Without them each assembly reports a clean @@ -84,3 +82,9 @@ python3 -m flows.parsers.parse_backfill_historical_versions \ The run is resumable: if interrupted, re-running with the same arguments picks up from the last saved checkpoint in `/checkpoints/`. + +The write is non-destructive. When `assembly_historical.tsv` already exists the +parser appends only the versions not already in it, keyed on +`genbankAccession`, so re-running a completed backfill — or a daily gap-fill +that re-parses versions already on file — neither truncates the output nor +duplicates rows. diff --git a/tests/README_phase_1_daily_updates.md b/tests/README_phase_1_daily_updates.md index 9203f94..02f008b 100644 --- a/tests/README_phase_1_daily_updates.md +++ b/tests/README_phase_1_daily_updates.md @@ -50,10 +50,14 @@ missing_versions.json ──► update_assembly_versions ──► missing_ass ``` **`parse_assembly_versions.py`** (daily, no network): -- Reads the previous `assembly_current.tsv` (renamed `.previous` before each run) +- Reads the previous current TSV (snapshotted to `.previous` before each run); + its filename and compression come from `config.meta["file_name"]` - Reads today's JSONL; for every assembly at version > 1 checks whether its predecessor exists in the previous TSV -- If found: copies the row and marks it `superseded`, appends to `assembly_historical.tsv` +- If found: copies the row, stamps `versionStatus`, `assemblyID`, + `superseded_by`, `superseded_by_version` and `superseded_date` — the column + names declared in `assembly_historical.types.yaml` — and merges it into + `assembly_historical.tsv` - If missing: writes a record to `missing_versions.json` for the updater to handle **`update_assembly_versions.py`** (runs when `missing_versions.json` exists): @@ -69,21 +73,20 @@ both the "predecessor found" and "predecessor missing" code paths. ```bash mkdir -p /tmp/assembly-versions -cp configs/assembly_historical.types.yaml /tmp/assembly-versions/ cp tests/test_data/assembly_test_sample.jsonl /tmp/assembly-versions/assembly_data_report.jsonl ``` -> **Note on output path**: `parse_backfill_historical_versions.py` writes -> `assembly_historical.tsv` relative to the *YAML file's directory*, not -> `--work_dir`. Copying the YAML into `--work_dir` (as above) puts the output -> there. +> **Note on output path**: `parse_backfill_historical_versions.py` resolves +> `assembly_historical.tsv` against `--work_dir`, so the YAML can stay where it +> is in the repo. (Before PR-A the output landed next to the YAML instead, and +> this walkthrough copied the config into `--work_dir` to work around it.) ### Step 1 — one-time Phase 0 backfill (if not already done) ```bash SKIP_PREFECT=true python3 -m flows.parsers.parse_backfill_historical_versions \ --input_path /tmp/assembly-versions/assembly_data_report.jsonl \ - --yaml_path /tmp/assembly-versions/assembly_historical.types.yaml \ + --yaml_path configs/assembly_historical.types.yaml \ --work_dir /tmp/assembly-versions # Writes: /tmp/assembly-versions/assembly_historical.tsv ``` @@ -95,13 +98,19 @@ run completes with 0 records written. ### Step 2 — simulate "yesterday's" parse ```bash -# Minimal previous TSV: only GCA_000222935.1 was known yesterday -printf "accession\tassembly_id\nGCA_000222935.1\tGCA_000222935_1\n" \ +# Minimal previous TSV: only GCA_000222935.1 was known yesterday. +# Column names must match what the assembly parsers actually emit. +printf "genbankAccession\tassemblyID\nGCA_000222935.1\tGCA_000222935_1\n" \ > /tmp/assembly-versions/assembly_current.tsv.previous ``` -In production this file is the `assembly_current.tsv` from the previous daily -run, renamed before the new run writes a fresh copy. +In production this file is a verbatim snapshot of the previous run's current +TSV, taken by `snapshot_previous_output` before the new run overwrites it. Its +name — and whether it is gzipped — follow `config.meta["file_name"]`, so with +the production NCBI config it is `ncbi_datasets_eukaryota.tsv.gz.previous`. +`parse_assembly_versions` resolves both from the config it is given and opens a +compressed snapshot transparently; the plain `assembly_current.tsv.previous` +above is what a config emitting `assembly_current.tsv` produces. ### Step 3 — run the daily incremental parser @@ -151,11 +160,16 @@ SKIP_PREFECT=true python3 -m flows.updaters.update_assembly_versions \ ```bash SKIP_PREFECT=true python3 -m flows.parsers.parse_backfill_historical_versions \ --input_path /tmp/assembly-versions/missing_assembly_versions.jsonl \ - --yaml_path /tmp/assembly-versions/assembly_historical.types.yaml \ + --yaml_path configs/assembly_historical.types.yaml \ --work_dir /tmp/assembly-versions # Appends the missing predecessors to assembly_historical.tsv ``` +The append is genuine: when `assembly_historical.tsv` already exists the parser +adds only the versions it does not already hold, leaving every earlier row in +place. (Before PR-A this step rewrote the file wholesale, so a single gap-fill +would have discarded the whole backfill.) + ## Running the test suite ```bash @@ -163,6 +177,8 @@ cd "$(git rev-parse --show-toplevel)" SKIP_PREFECT=true python3 -m pytest tests/test_assembly_versions.py -v ``` -All 34 tests cover: accession parsing, path derivation, superseded-row -building, missing-version detection, TSV append/deduplication, updater JSONL -writing, and the `plugin()` registration hook. +The tests cover: accession parsing, config-driven path resolution, +superseded-row building, missing-version detection, TSV merge/deduplication, +schema conformance against `assembly_historical.types.yaml`, the +non-destructive Phase 0 write, updater JSONL writing, and the `plugin()` +registration hook. diff --git a/tests/test_assembly_summary.py b/tests/test_assembly_summary.py new file mode 100644 index 0000000..613ae48 --- /dev/null +++ b/tests/test_assembly_summary.py @@ -0,0 +1,190 @@ +"""Tests for generate_assembly_summary.py (Phase 2.2). + +Covers the per-base-accession aggregation: version ordering and gaps, the +superseded count, the first row to meet the EBP metric, and the treatment of +the literal string "None" that upstream writes for an absent value. + +The last of those is a cross-phase invariant rather than a local detail: +Phase 2 and Phase 3 both decide which rows carry an EBP metric, and a row the +summary counts but the milestones do not would leave the two outputs +disagreeing about the same assembly. +""" + +import csv +import os +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +os.environ["SKIP_PREFECT"] = "true" + +from flows.lib.compute_taxon_milestones import _has_metric # noqa: E402 +from flows.lib.generate_assembly_summary import ( # noqa: E402 + OUTPUT_TSV, + SUMMARY_FIELDNAMES, + _has_ebp_metric, + find_version_gaps, + generate_assembly_summary, + generate_summary_for_base, + load_assemblies, +) + +TAXID = 9001 + + +def row(accession, version_status="current", ebp="", release=None, taxid=TAXID): + """Build a minimal assembly row as the summary reads it.""" + version = int(accession.split(".")[1]) + return { + "genbankAccession": accession, + "accession": accession, + "versionStatus": version_status, + "taxId": str(taxid), + "releaseDate": f"20{10 + version:02d}-01-01" if release is None else release, + "ebpStandardDate": ebp, + } + + +def write_tsv(path, rows): + """Write rows to a tab-separated file.""" + fieldnames = list(rows[0].keys()) + with open(path, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter="\t") + writer.writeheader() + writer.writerows(rows) + + +def read_tsv(path): + """Read a tab-separated file into a list of dicts.""" + with open(path, encoding="utf-8") as f: + return list(csv.DictReader(f, delimiter="\t")) + + +class TestVersionGaps: + def test_no_gaps_in_a_complete_run(self): + assert find_version_gaps([1, 2, 3]) == "" + + def test_reports_every_missing_version(self): + assert find_version_gaps([1, 4]) == "2,3" + + def test_a_series_starting_above_one_is_all_gap(self): + assert find_version_gaps([3]) == "1,2" + + def test_no_versions_is_not_a_gap(self): + assert find_version_gaps([]) == "" + + +class TestSummaryForBase: + def test_orders_by_version_not_by_row_order(self): + summary = generate_summary_for_base( + "GCA_1", + [row("GCA_1.3"), row("GCA_1.1", "superseded"), row("GCA_1.2", "superseded")], + ) + assert summary["first_version_accession"] == "GCA_1.1" + assert summary["current_version_accession"] == "GCA_1.3" + assert summary["total_versions"] == 3 + assert summary["superseded_versions"] == 2 + + def test_first_ebp_metric_is_the_earliest_qualifying_version(self): + summary = generate_summary_for_base( + "GCA_1", + [ + row("GCA_1.1", "superseded"), + row("GCA_1.2", "superseded", ebp="2016-01-01"), + row("GCA_1.3", ebp="2018-01-01"), + ], + ) + assert summary["first_ebp_metric_accession"] == "GCA_1.2" + assert summary["first_ebp_metric_version"] == 2 + + def test_no_ebp_metric_leaves_the_columns_empty(self): + summary = generate_summary_for_base("GCA_1", [row("GCA_1.1")]) + assert summary["first_ebp_metric_accession"] == "" + assert summary["first_ebp_metric_version"] == "" + + def test_a_row_with_no_version_status_counts_as_current(self): + summary = generate_summary_for_base( + "GCA_1", [{"accession": "GCA_1.1", "genbankAccession": "GCA_1.1"}] + ) + assert summary["superseded_versions"] == 0 + + +class TestNoneSentinel: + """Upstream writes the string "None" wherever a value was absent.""" + + def test_ebp_metric_of_none_is_not_a_metric(self): + assert _has_ebp_metric(row("GCA_1.1", ebp="None")) is False + + def test_phase_2_and_phase_3_agree_on_the_metric(self): + for value in ("", "None", "2016-01-01"): + assembly = row("GCA_1.1", ebp=value) + assert _has_ebp_metric(assembly) == _has_metric(assembly), value + + def test_a_none_release_date_is_summarised_as_empty(self): + summary = generate_summary_for_base( + "GCA_1", [row("GCA_1.1", release="None")] + ) + assert summary["first_version_date"] == "" + assert summary["current_version_date"] == "" + + def test_a_none_taxid_is_summarised_as_empty(self): + rows = [row("GCA_1.1")] + rows[0]["taxId"] = "None" + assert generate_summary_for_base("GCA_1", rows)["taxId"] == "" + + +class TestLoadAssemblies: + def test_reads_both_files_and_normalises_the_accession(self, tmp_path): + write_tsv(tmp_path / "assembly_current.tsv", [row("GCA_1.2")]) + write_tsv( + tmp_path / "assembly_historical.tsv", [row("GCA_1.1", "superseded")] + ) + rows = load_assemblies( + str(tmp_path / "assembly_current.tsv"), + str(tmp_path / "assembly_historical.tsv"), + ) + assert [r["accession"] for r in rows] == ["GCA_1.2", "GCA_1.1"] + + def test_a_missing_file_is_a_warning_not_a_failure(self, tmp_path): + write_tsv(tmp_path / "assembly_current.tsv", [row("GCA_1.1")]) + rows = load_assemblies( + str(tmp_path / "assembly_current.tsv"), + str(tmp_path / "assembly_historical.tsv"), + ) + assert len(rows) == 1 + + +class TestFlow: + @pytest.fixture + def work_dir(self, tmp_path): + write_tsv( + tmp_path / "assembly_current.tsv", + [row("GCA_1.3", ebp="2018-01-01"), row("GCA_2.1")], + ) + write_tsv( + tmp_path / "assembly_historical.tsv", + [row("GCA_1.1", "superseded")], + ) + generate_assembly_summary(work_dir=str(tmp_path)) + return tmp_path + + def test_writes_one_row_per_base_accession(self, work_dir): + rows = read_tsv(work_dir / OUTPUT_TSV) + assert {r["base_accession"] for r in rows} == {"GCA_1", "GCA_2"} + + def test_header_matches_the_schema(self, work_dir): + with open(work_dir / OUTPUT_TSV, encoding="utf-8") as f: + header = f.readline().rstrip("\n").split("\t") + assert header == SUMMARY_FIELDNAMES + + def test_the_missing_middle_version_is_reported_as_a_gap(self, work_dir): + rows = {r["base_accession"]: r for r in read_tsv(work_dir / OUTPUT_TSV)} + assert rows["GCA_1"]["version_gaps"] == "2" + assert rows["GCA_2"]["version_gaps"] == "" + + def test_no_data_writes_no_output(self, tmp_path): + generate_assembly_summary(work_dir=str(tmp_path)) + assert not (tmp_path / OUTPUT_TSV).exists() diff --git a/tests/test_assembly_versions.py b/tests/test_assembly_versions.py index 48be5a1..11472e7 100644 --- a/tests/test_assembly_versions.py +++ b/tests/test_assembly_versions.py @@ -10,6 +10,7 @@ """ import csv +import gzip import json import os import sys @@ -18,21 +19,47 @@ import pytest +TAB = "\t" +NEWLINE = "\n" + +HISTORICAL_YAML = ( + Path(__file__).parent.parent / "configs" / "assembly_historical.types.yaml" +) + sys.path.insert(0, str(Path(__file__).parent.parent)) os.environ["SKIP_PREFECT"] = "true" -from flows.lib.utils import Parser # noqa: E402 +from flows.lib.utils import Parser, load_config # noqa: E402 from flows.parsers import parse_assembly_versions as incremental_module # noqa: E402 +from flows.parsers import ( # noqa: E402 + parse_backfill_historical_versions as backfill_module, +) from flows.parsers.parse_assembly_versions import ( # noqa: E402 append_superseded_to_tsv, build_missing_version_record, build_superseded_row, derive_assembly_version_paths, identify_newly_superseded, + load_historical_headers, load_previous_parsed_by_base, + merge_fieldnames, parse_assembly_versions, ) +from flows.lib.assembly_versions_utils import ( # noqa: E402 + canonicalize_columns, + get_accession, + get_assembly_id, + get_version_status, + open_tsv, + resolve_current_tsv_paths, + resolve_historical_yaml_path, +) +from flows.parsers.parse_backfill_historical_versions import ( # noqa: E402 + read_existing_output, + resolve_output_path, + write_or_append_parsed, +) from flows.parsers.parse_ncbi_assemblies import snapshot_previous_output # noqa: E402 from flows.updaters import update_assembly_versions as updater_module # noqa: E402 from flows.updaters.update_assembly_versions import ( # noqa: E402 @@ -81,17 +108,17 @@ def test_missing_file_returns_empty(self, tmp_path): def test_single_version_indexed(self, tmp_path): tsv = tmp_path / "current.tsv" - write_tsv(tsv, [{"accession": "GCA_000222935.1", "taxon_id": "12345"}]) + write_tsv(tsv, [{"genbankAccession": "GCA_000222935.1", "taxId": "12345"}]) result = load_previous_parsed_by_base(str(tsv)) assert "GCA_000222935" in result assert 1 in result["GCA_000222935"] - assert result["GCA_000222935"][1]["taxon_id"] == "12345" + assert result["GCA_000222935"][1]["taxId"] == "12345" def test_multi_version_same_base(self, tmp_path): tsv = tmp_path / "current.tsv" write_tsv(tsv, [ - {"accession": "GCA_000222935.1", "taxon_id": "1"}, - {"accession": "GCA_000222935.2", "taxon_id": "1"}, + {"genbankAccession": "GCA_000222935.1", "taxId": "1"}, + {"genbankAccession": "GCA_000222935.2", "taxId": "1"}, ]) result = load_previous_parsed_by_base(str(tsv)) assert len(result["GCA_000222935"]) == 2 @@ -101,14 +128,42 @@ def test_multi_version_same_base(self, tmp_path): def test_multiple_base_accessions(self, tmp_path): tsv = tmp_path / "current.tsv" write_tsv(tsv, [ - {"accession": "GCA_000222935.1", "taxon_id": "1"}, - {"accession": "GCA_000412225.1", "taxon_id": "2"}, + {"genbankAccession": "GCA_000222935.1", "taxId": "1"}, + {"genbankAccession": "GCA_000412225.1", "taxId": "2"}, ]) result = load_previous_parsed_by_base(str(tsv)) assert len(result) == 2 assert "GCA_000222935" in result assert "GCA_000412225" in result + def test_legacy_accession_column_still_read(self, tmp_path): + """Rows written before the schema fix used a bare 'accession' column.""" + tsv = tmp_path / "current.tsv" + write_tsv(tsv, [{"accession": "GCA_000222935.1", "taxId": "1"}]) + result = load_previous_parsed_by_base(str(tsv)) + assert 1 in result["GCA_000222935"] + + def test_rows_without_accession_skipped(self, tmp_path): + tsv = tmp_path / "current.tsv" + write_tsv(tsv, [ + {"genbankAccession": "", "taxId": "1"}, + {"genbankAccession": "GCA_000222935.1", "taxId": "2"}, + ]) + result = load_previous_parsed_by_base(str(tsv)) + assert list(result) == ["GCA_000222935"] + + def test_gzipped_previous_snapshot_read(self, tmp_path): + """A gzipped current TSV yields a gzipped .previous snapshot.""" + tsv = tmp_path / "ncbi_datasets_eukaryota.tsv.gz.previous" + with gzip.open(tsv, "wt", encoding="utf-8", newline="") as f: + writer = csv.DictWriter( + f, fieldnames=["genbankAccession", "taxId"], delimiter=TAB + ) + writer.writeheader() + writer.writerow({"genbankAccession": "GCA_000222935.1", "taxId": "1"}) + result = load_previous_parsed_by_base(str(tsv)) + assert 1 in result["GCA_000222935"] + # --------------------------------------------------------------------------- # TestBuildSupersededRow @@ -119,18 +174,18 @@ class TestBuildSupersededRow: def _base_row(self): return { - "accession": "GCA_000222935.1", - "taxon_id": "12345", - "assembly_level": "Chromosome", + "genbankAccession": "GCA_000222935.1", + "taxId": "12345", + "assemblyLevel": "Chromosome", } def test_version_status_set(self): row = build_superseded_row(self._base_row(), 1, "GCA_000222935.2", 2, "2024-01-15") - assert row["version_status"] == "superseded" + assert row["versionStatus"] == "superseded" def test_assembly_id_format(self): row = build_superseded_row(self._base_row(), 1, "GCA_000222935.2", 2, "2024-01-15") - assert row["assembly_id"] == "GCA_000222935_1" + assert row["assemblyID"] == "GCA_000222935_1" def test_superseded_by_fields(self): row = build_superseded_row(self._base_row(), 1, "GCA_000222935.2", 2, "2024-01-15") @@ -141,12 +196,12 @@ def test_superseded_by_fields(self): def test_original_row_not_mutated(self): original = self._base_row() build_superseded_row(original, 1, "GCA_000222935.2", 2, "2024-01-15") - assert "version_status" not in original + assert "versionStatus" not in original def test_existing_fields_preserved(self): row = build_superseded_row(self._base_row(), 1, "GCA_000222935.2", 2, "2024-01-15") - assert row["taxon_id"] == "12345" - assert row["assembly_level"] == "Chromosome" + assert row["taxId"] == "12345" + assert row["assemblyLevel"] == "Chromosome" # --------------------------------------------------------------------------- @@ -197,7 +252,9 @@ def test_superseded_found_when_previous_version_present(self, tmp_path): tmp_path, [{"accession": "GCA_000222935.2", "releaseDate": "2024-01-15"}] ) previous = { - "GCA_000222935": {1: {"accession": "GCA_000222935.1", "taxon_id": "1"}} + "GCA_000222935": { + 1: {"genbankAccession": "GCA_000222935.1", "taxId": "1"} + } } superseded, missing = identify_newly_superseded(jsonl, previous) assert len(superseded) == 1 @@ -207,7 +264,9 @@ def test_superseded_found_when_previous_version_present(self, tmp_path): def test_missing_with_version_gap(self, tmp_path): jsonl = self._write_jsonl(tmp_path, [{"accession": "GCA_000222935.3"}]) previous = { - "GCA_000222935": {1: {"accession": "GCA_000222935.1", "taxon_id": "1"}} + "GCA_000222935": { + 1: {"genbankAccession": "GCA_000222935.1", "taxId": "1"} + } } superseded, missing = identify_newly_superseded(jsonl, previous) assert superseded == [] @@ -228,7 +287,9 @@ def test_mixed_batch(self, tmp_path): {"accession": "GCA_999999999.2"}, ]) previous = { - "GCA_000222935": {1: {"accession": "GCA_000222935.1", "taxon_id": "1"}} + "GCA_000222935": { + 1: {"genbankAccession": "GCA_000222935.1", "taxId": "1"} + } } superseded, missing = identify_newly_superseded(jsonl, previous) assert len(superseded) == 1 @@ -244,9 +305,9 @@ class TestAppendSupersededToTsv: def _make_row(self, acc, assembly_id, status="superseded"): return { - "accession": acc, - "assembly_id": assembly_id, - "version_status": status, + "genbankAccession": acc, + "assemblyID": assembly_id, + "versionStatus": status, } def test_creates_new_file(self, tmp_path): @@ -256,7 +317,7 @@ def test_creates_new_file(self, tmp_path): assert tsv.exists() result = read_tsv(tsv) assert len(result) == 1 - assert result[0]["accession"] == "GCA_000222935.1" + assert result[0]["genbankAccession"] == "GCA_000222935.1" def test_appends_to_existing(self, tmp_path): tsv = tmp_path / "historical.tsv" @@ -270,9 +331,9 @@ def test_appends_to_existing(self, tmp_path): def test_dedup_on_assembly_id_keeps_new(self, tmp_path): tsv = tmp_path / "historical.tsv" old_row = { - "accession": "GCA_000222935.1", - "assembly_id": "GCA_000222935_1", - "version_status": "superseded", + "genbankAccession": "GCA_000222935.1", + "assemblyID": "GCA_000222935_1", + "versionStatus": "superseded", "superseded_by": "GCA_000222935.2", } write_tsv(tsv, [old_row]) @@ -311,7 +372,7 @@ def test_no_previous_tsv_returns_empty_result(self, tmp_path): def test_one_superseded_produces_correct_counts(self, tmp_path): previous_tsv = tmp_path / "previous.tsv" write_tsv(previous_tsv, [ - {"accession": "GCA_000222935.1", "taxon_id": "1"} + {"genbankAccession": "GCA_000222935.1", "taxId": "1"} ]) jsonl = tmp_path / "new.jsonl" write_jsonl(jsonl, [ @@ -329,7 +390,7 @@ def test_missing_version_detected_in_orchestrator_result(self, tmp_path): """v3 present, v2 missing → missing_versions_count == 1.""" previous_tsv = tmp_path / "previous.tsv" write_tsv(previous_tsv, [ - {"accession": "GCA_000222935.1", "taxon_id": "1"} + {"genbankAccession": "GCA_000222935.1", "taxId": "1"} ]) jsonl = tmp_path / "new.jsonl" write_jsonl(jsonl, [ @@ -347,7 +408,7 @@ def test_missing_version_detected_in_orchestrator_result(self, tmp_path): def test_historical_tsv_written(self, tmp_path): previous_tsv = tmp_path / "previous.tsv" write_tsv(previous_tsv, [ - {"accession": "GCA_000222935.1", "taxon_id": "1"} + {"genbankAccession": "GCA_000222935.1", "taxId": "1"} ]) jsonl = tmp_path / "new.jsonl" write_jsonl(jsonl, [ @@ -362,7 +423,7 @@ def test_historical_tsv_written(self, tmp_path): assert historical_tsv.exists() rows = read_tsv(historical_tsv) assert len(rows) == 1 - assert rows[0]["version_status"] == "superseded" + assert rows[0]["versionStatus"] == "superseded" # --------------------------------------------------------------------------- @@ -563,3 +624,514 @@ def test_overwrites_stale_previous_snapshot(self, tmp_path): snapshot_previous_output(self._config(output)) assert previous.read_bytes() == output.read_bytes() + + +# --------------------------------------------------------------------------- +# TestNormalisingAccessors +# --------------------------------------------------------------------------- + +class TestNormalisingAccessors: + """get_* accessors read either naming convention (PR-A item 1).""" + + def test_accession_prefers_canonical_column(self): + row = {"genbankAccession": "GCA_000222935.1", "accession": "GCA_000412225.1"} + assert get_accession(row) == "GCA_000222935.1" + + def test_accession_falls_back_to_legacy_column(self): + assert get_accession({"accession": "GCA_000412225.1"}) == "GCA_000412225.1" + + def test_accession_absent_returns_empty(self): + assert get_accession({"taxId": "1"}) == "" + + def test_assembly_id_all_three_spellings(self): + assert get_assembly_id({"assemblyID": "GCA_1_1"}) == "GCA_1_1" + assert get_assembly_id({"assemblyId": "GCA_1_1"}) == "GCA_1_1" + assert get_assembly_id({"assembly_id": "GCA_1_1"}) == "GCA_1_1" + + def test_assembly_id_prefers_the_yaml_spelling(self): + """assemblyID is what the historical YAML declares, so it wins.""" + row = {"assemblyID": "canonical", "assemblyId": "upstream"} + assert get_assembly_id(row) == "canonical" + + def test_version_status_both_conventions(self): + assert get_version_status({"versionStatus": "superseded"}) == "superseded" + assert get_version_status({"version_status": "superseded"}) == "superseded" + + def test_empty_value_falls_through_to_the_next_alias(self): + row = {"genbankAccession": "", "accession": "GCA_000412225.1"} + assert get_accession(row) == "GCA_000412225.1" + + +# --------------------------------------------------------------------------- +# TestResolveCurrentTsvPaths +# --------------------------------------------------------------------------- + +class TestResolveCurrentTsvPaths: + """The current-TSV name comes from the config, not a hardcoded constant (F4).""" + + def test_name_and_snapshot_from_config(self, tmp_path): + config = MagicMock() + config.meta = {"file_name": "configs/ncbi_datasets_eukaryota.tsv.gz"} + current, previous, open_fn = resolve_current_tsv_paths( + str(tmp_path), config=config + ) + assert current.endswith("ncbi_datasets_eukaryota.tsv.gz") + assert previous == f"{current}.previous" + assert open_fn is open_tsv + + def test_discovery_ignores_derived_outputs(self, tmp_path): + (tmp_path / "assembly_historical.tsv").touch() + (tmp_path / "assembly_version_summary.tsv").touch() + (tmp_path / "ncbi_datasets_eukaryota.tsv.gz").touch() + current, _, _ = resolve_current_tsv_paths(str(tmp_path)) + assert current.endswith("ncbi_datasets_eukaryota.tsv.gz") + + def test_fallback_name_when_nothing_on_disk(self, tmp_path): + current, previous, _ = resolve_current_tsv_paths(str(tmp_path)) + assert current.endswith("assembly_current.tsv") + assert previous.endswith("assembly_current.tsv.previous") + + def test_ambiguous_directory_raises(self, tmp_path): + (tmp_path / "one.tsv").touch() + (tmp_path / "two.tsv").touch() + with pytest.raises(ValueError): + resolve_current_tsv_paths(str(tmp_path)) + + def test_derived_paths_use_the_config_name(self, tmp_path): + config = MagicMock() + config.meta = {"file_name": "configs/ncbi_datasets_eukaryota.tsv.gz"} + jsonl = tmp_path / "assembly_data_report.jsonl" + jsonl.touch() + previous_tsv, historical_tsv = derive_assembly_version_paths( + str(jsonl), config=config + ) + assert previous_tsv.endswith("ncbi_datasets_eukaryota.tsv.gz.previous") + assert historical_tsv.endswith("assembly_historical.tsv") + + +# --------------------------------------------------------------------------- +# TestMergeFieldnames +# --------------------------------------------------------------------------- + +class TestMergeFieldnames: + """merge_fieldnames covers every column present in any row (F3).""" + + def test_union_across_rows(self): + assert set(merge_fieldnames([{"a": 1}, {"b": 2}])) == {"a", "b"} + + def test_preferred_order_honoured(self): + rows = [{"b": 1, "a": 2, "z": 3}] + assert merge_fieldnames(rows, ["a", "b"]) == ["a", "b", "z"] + + def test_preferred_columns_absent_from_rows_are_dropped(self): + assert merge_fieldnames([{"a": 1}], ["a", "b"]) == ["a"] + + def test_no_rows_gives_no_columns(self): + assert merge_fieldnames([], ["a"]) == [] + + +# --------------------------------------------------------------------------- +# TestAppendPreservesColumns +# --------------------------------------------------------------------------- + +class TestAppendPreservesColumns: + """Merging Phase 0 and Phase 1 row sets must not drop columns (F3).""" + + def test_no_column_lost_when_merging_row_sets(self, tmp_path): + tsv = tmp_path / "historical.tsv" + write_tsv(tsv, [{ + "genbankAccession": "GCA_000412225.1", + "assemblyID": "GCA_000412225_1", + "versionStatus": "superseded", + }]) + new_row = { + "genbankAccession": "GCA_000222935.1", + "assemblyID": "GCA_000222935_1", + "versionStatus": "superseded", + "superseded_by": "GCA_000222935.2", + "superseded_by_version": 2, + "superseded_date": "2024-01-15", + } + append_superseded_to_tsv([new_row], str(tsv)) + result = read_tsv(tsv) + assert len(result) == 2 + by_accession = {r["genbankAccession"]: r for r in result} + assert by_accession["GCA_000222935.1"]["superseded_by"] == "GCA_000222935.2" + assert "superseded_date" in by_accession["GCA_000412225.1"] + + def test_explicit_headers_set_the_column_order(self, tmp_path): + tsv = tmp_path / "historical.tsv" + row = { + "versionStatus": "superseded", + "genbankAccession": "GCA_000222935.1", + "assemblyID": "GCA_000222935_1", + } + headers = ["genbankAccession", "assemblyID", "versionStatus"] + append_superseded_to_tsv([row], str(tsv), headers=headers) + with open(tsv, encoding="utf-8") as f: + written = next(csv.reader(f, delimiter=TAB)) + assert written == headers + + +# --------------------------------------------------------------------------- +# TestSchemaConformance +# --------------------------------------------------------------------------- + +class TestSchemaConformance: + """Phase 1 output columns must be declared in assembly_historical.types.yaml. + + This is the check that would have caught F1-F3: Phase 1 was written and + tested against a mock schema that production never produces. + """ + + @staticmethod + def _yaml_headers(): + return load_config(config_file=str(HISTORICAL_YAML)).headers + + def test_canonical_columns_declared(self): + headers = self._yaml_headers() + assert "genbankAccession" in headers + assert "assemblyID" in headers + assert "versionStatus" in headers + + def test_supersession_columns_declared(self): + headers = self._yaml_headers() + for column in ("superseded_by", "superseded_by_version", "superseded_date"): + assert column in headers + + def test_superseded_row_columns_are_all_declared(self): + headers = set(self._yaml_headers()) + previous_row = dict.fromkeys(headers, "") + previous_row["genbankAccession"] = "GCA_000222935.1" + row = build_superseded_row(previous_row, 1, "GCA_000222935.2", 2, "2024-01-15") + assert set(row) <= headers + + def test_snake_case_columns_not_emitted(self): + row = build_superseded_row( + {"genbankAccession": "GCA_000222935.1"}, + 1, "GCA_000222935.2", 2, "2024-01-15", + ) + assert "version_status" not in row + assert "assembly_id" not in row + + def test_write_side_uses_the_yaml_spelling_of_assembly_id(self): + """Reads tolerate assemblyId; writes must stay on the declared name. + + parse_ncbi_assemblies writes `assemblyId` upstream, but the historical + YAML declares `assemblyID`, so a row emitted with the lowercase form + would be dropped by the GenomeHubs write path. + """ + row = build_superseded_row( + {"genbankAccession": "GCA_000222935.1", "assemblyId": "stale"}, + 1, "GCA_000222935.2", 2, "2024-01-15", + ) + assert row["assemblyID"] == "GCA_000222935_1" + assert "assemblyID" in self._yaml_headers() + assert "assemblyId" not in self._yaml_headers() + + def test_written_header_is_a_subset_of_the_yaml_schema(self, tmp_path): + headers = self._yaml_headers() + previous_row = dict.fromkeys(headers, "") + previous_row["genbankAccession"] = "GCA_000222935.1" + row = build_superseded_row(previous_row, 1, "GCA_000222935.2", 2, "2024-01-15") + tsv = tmp_path / "historical.tsv" + append_superseded_to_tsv([row], str(tsv), headers=headers) + with open(tsv, encoding="utf-8") as f: + written = next(csv.reader(f, delimiter=TAB)) + assert set(written) <= set(headers) + assert written == list(headers) + + +# --------------------------------------------------------------------------- +# TestCurrentSchemaDoesNotLeakIntoHistorical +# --------------------------------------------------------------------------- + +class TestCurrentSchemaDoesNotLeakIntoHistorical: + """A Phase 1 row is copied from the current TSV, whose schema is not ours. + + Left unbounded, every current-only column — upstream's `assemblyId` spelling + among them — lands in assembly_historical.tsv despite the YAML never + declaring it. + """ + + @staticmethod + def _yaml_headers(): + return list(load_config(config_file=str(HISTORICAL_YAML)).headers) + + @staticmethod + def _current_row(): + """A row as the current TSV spells it, extra columns and all.""" + return { + "genbankAccession": "GCA_000222935.1", + "assemblyId": "GCA_000222935_1", + "versionStatus": "current", + "releaseDate": "2011-06-01", + "sourceDatabase": "SOURCE_DATABASE_GENBANK", + } + + def test_canonicalize_columns_drops_the_alias_spelling(self): + row = canonicalize_columns( + {"assemblyId": "GCA_000222935_1", "version_status": "current"} + ) + assert row == { + "assemblyID": "GCA_000222935_1", "versionStatus": "current", + } + + def test_canonicalize_columns_prefers_the_canonical_value(self): + row = canonicalize_columns( + {"assemblyID": "canonical", "assemblyId": "stale"} + ) + assert row["assemblyID"] == "canonical" + assert "assemblyId" not in row + + def test_canonicalize_columns_invents_nothing(self): + assert canonicalize_columns({"releaseDate": "2011-06-01"}) == { + "releaseDate": "2011-06-01" + } + + def test_superseded_row_carries_one_assembly_id_column(self): + row = build_superseded_row( + self._current_row(), 1, "GCA_000222935.2", 2, "2024-01-15" + ) + assert row["assemblyID"] == "GCA_000222935_1" + assert "assemblyId" not in row + + def test_undeclared_columns_are_not_written(self, tmp_path): + tsv = tmp_path / "historical.tsv" + row = build_superseded_row( + self._current_row(), 1, "GCA_000222935.2", 2, "2024-01-15" + ) + append_superseded_to_tsv([row], str(tsv), headers=self._yaml_headers()) + with open(tsv, encoding="utf-8") as f: + written = next(csv.reader(f, delimiter=TAB)) + assert set(written) <= set(self._yaml_headers()) + assert "sourceDatabase" not in written + assert read_tsv(tsv)[0]["assemblyID"] == "GCA_000222935_1" + + def test_columns_already_on_disk_are_kept(self, tmp_path): + # The schema bounds what a new row may add; it must not evict a column + # an earlier run already put in the file (F3). + tsv = tmp_path / "historical.tsv" + write_tsv(tsv, [{ + "genbankAccession": "GCA_000412225.1", + "assemblyID": "GCA_000412225_1", + "legacyColumn": "keep me", + }]) + row = build_superseded_row( + self._current_row(), 1, "GCA_000222935.2", 2, "2024-01-15" + ) + append_superseded_to_tsv([row], str(tsv), headers=self._yaml_headers()) + result = read_tsv(tsv) + by_accession = {r["genbankAccession"]: r for r in result} + assert by_accession["GCA_000412225.1"]["legacyColumn"] == "keep me" + assert "sourceDatabase" not in by_accession["GCA_000222935.1"] + + def test_without_headers_the_union_is_still_taken(self, tmp_path): + # No schema resolved means no bound — the pre-existing behaviour. + tsv = tmp_path / "historical.tsv" + row = build_superseded_row( + self._current_row(), 1, "GCA_000222935.2", 2, "2024-01-15" + ) + append_superseded_to_tsv([row], str(tsv)) + with open(tsv, encoding="utf-8") as f: + written = next(csv.reader(f, delimiter=TAB)) + assert "sourceDatabase" in written + + def test_merge_fieldnames_honours_allowed(self): + assert merge_fieldnames( + [{"a": 1, "b": 2}], ["a", "b"], allowed={"a"} + ) == ["a"] + + +# --------------------------------------------------------------------------- +# TestHistoricalHeadersAreResolved +# --------------------------------------------------------------------------- + +class TestHistoricalHeadersAreResolved: + """The entry points must actually supply the schema, or it bounds nothing.""" + + def test_repo_config_is_found_from_an_unrelated_work_dir(self, tmp_path): + assert resolve_historical_yaml_path(str(tmp_path)) == str(HISTORICAL_YAML) + + def test_a_copy_in_work_dir_wins(self, tmp_path): + local = tmp_path / "assembly_historical.types.yaml" + local.write_text(HISTORICAL_YAML.read_text(encoding="utf-8"), encoding="utf-8") + assert resolve_historical_yaml_path(str(tmp_path)) == str(local) + + def test_headers_load_from_the_repo_config(self, tmp_path): + headers = load_historical_headers(str(tmp_path)) + assert headers == list(load_config(config_file=str(HISTORICAL_YAML)).headers) + + def test_an_unloadable_yaml_degrades_to_none(self, tmp_path): + broken = tmp_path / "assembly_historical.types.yaml" + broken.write_text("this: [is not: valid", encoding="utf-8") + assert load_historical_headers(str(tmp_path)) is None + + def test_a_missing_config_degrades_to_none(self, tmp_path, monkeypatch): + monkeypatch.setattr( + incremental_module, "resolve_historical_yaml_path", lambda _: None + ) + assert load_historical_headers(str(tmp_path)) is None + + def test_the_wrapper_passes_the_headers_through(self, tmp_path, monkeypatch): + (tmp_path / "report.jsonl").write_text("", encoding="utf-8") + captured = {} + + def fake_parse(**kwargs): + captured.update(kwargs) + return {"missing_versions_count": 0, "missing_versions": []} + + monkeypatch.setattr( + incremental_module, "parse_assembly_versions", fake_parse + ) + incremental_module.parse_assembly_versions_wrapper( + working_yaml="", work_dir=str(tmp_path), append=False + ) + assert captured["historical_headers"] == list( + load_config(config_file=str(HISTORICAL_YAML)).headers + ) + + +# --------------------------------------------------------------------------- +# TestResolveOutputPath +# --------------------------------------------------------------------------- + +class TestResolveOutputPath: + """Phase 0 writes into work_dir, not the YAML file directory (F6).""" + + def test_output_lands_in_work_dir(self, tmp_path): + config = load_config(config_file=str(HISTORICAL_YAML)) + path = resolve_output_path(config, str(tmp_path)) + assert os.path.dirname(path) == str(tmp_path) + assert os.path.basename(path) == "assembly_historical.tsv" + + def test_config_meta_updated_in_place(self, tmp_path): + config = load_config(config_file=str(HISTORICAL_YAML)) + path = resolve_output_path(config, str(tmp_path)) + assert config.meta["file_name"] == path + + +# --------------------------------------------------------------------------- +# TestNonDestructiveHistoricalWrite +# --------------------------------------------------------------------------- + +class TestNonDestructiveHistoricalWrite: + """Phase 0 must not truncate an existing historical TSV (F9). + + The F5 decision makes the gap-fill path re-invoke this parser daily, so a + truncating write would wipe the backfill it is meant to extend. + """ + + @staticmethod + def _config(tmp_path): + config = load_config(config_file=str(HISTORICAL_YAML)) + resolve_output_path(config, str(tmp_path)) + return config + + @staticmethod + def _row(accession): + base, version = accession.split(".") + return { + "genbankAccession": accession, + "assemblyID": f"{base}_{version}", + "versionStatus": "superseded", + } + + def test_first_write_creates_the_file_with_a_header(self, tmp_path): + config = self._config(tmp_path) + written = write_or_append_parsed( + {"GCA_000412225.1": self._row("GCA_000412225.1")}, config + ) + output = tmp_path / "assembly_historical.tsv" + assert written == 1 + with open(output, encoding="utf-8") as f: + header = next(csv.reader(f, delimiter=TAB)) + assert header == list(config.headers) + + def test_existing_rows_survive_a_gap_fill(self, tmp_path): + config = self._config(tmp_path) + output = tmp_path / "assembly_historical.tsv" + write_or_append_parsed({ + "GCA_000412225.1": self._row("GCA_000412225.1"), + "GCA_000412225.2": self._row("GCA_000412225.2"), + }, config) + assert len(read_tsv(output)) == 2 + + written = write_or_append_parsed( + {"GCA_000222935.1": self._row("GCA_000222935.1")}, config + ) + rows = read_tsv(output) + assert written == 1 + assert len(rows) == 3 + assert {r["genbankAccession"] for r in rows} == { + "GCA_000412225.1", "GCA_000412225.2", "GCA_000222935.1", + } + + def test_reparsed_versions_are_not_duplicated(self, tmp_path): + config = self._config(tmp_path) + parsed = {"GCA_000412225.1": self._row("GCA_000412225.1")} + write_or_append_parsed(parsed, config) + written = write_or_append_parsed(parsed, config) + assert written == 0 + assert len(read_tsv(tmp_path / "assembly_historical.tsv")) == 1 + + def test_read_existing_output_reads_the_header_and_accessions(self, tmp_path): + output = tmp_path / "assembly_historical.tsv" + write_tsv(output, [self._row("GCA_000412225.1"), self._row("GCA_000412225.2")]) + header, accessions = read_existing_output(str(output)) + assert accessions == {"GCA_000412225.1", "GCA_000412225.2"} + assert header[0] == "genbankAccession" + + def test_read_existing_output_on_an_absent_file(self, tmp_path): + assert read_existing_output(str(tmp_path / "nope.tsv")) == ([], set()) + + def test_rows_without_an_accession_are_not_overwritten(self, tmp_path): + # read_existing_output only collects non-empty accessions, so a file + # whose rows all lack one yields an empty accession set. The rewrite + # branch must key off the header instead, or those rows are lost. + config = self._config(tmp_path) + output = tmp_path / "assembly_historical.tsv" + write_tsv(output, [{ + "genbankAccession": "", + "assemblyID": "GCA_000412225_1", + "versionStatus": "superseded", + }]) + + written = write_or_append_parsed( + {"GCA_000222935.1": self._row("GCA_000222935.1")}, config + ) + rows = read_tsv(output) + assert written == 1 + assert len(rows) == 2 + assert rows[0]["assemblyID"] == "GCA_000412225_1" + assert rows[1]["genbankAccession"] == "GCA_000222935.1" + + def test_appends_under_a_header_phase_1_has_widened(self, tmp_path): + # Phase 1 rewrites this file with the union of every row column, so a + # gap fill must append under what is on disk, not the config header. + config = self._config(tmp_path) + output = tmp_path / "assembly_historical.tsv" + write_or_append_parsed({"GCA_000412225.1": self._row("GCA_000412225.1")}, config) + + rows = read_tsv(output) + widened = list(rows[0].keys()) + ["genusTaxId"] + rows[0]["genusTaxId"] = "8001" + with open(output, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=widened, delimiter=TAB) + writer.writeheader() + writer.writerows(rows) + + write_or_append_parsed({"GCA_000222935.1": self._row("GCA_000222935.1")}, config) + + with open(output, encoding="utf-8") as f: + lines = [line.rstrip(NEWLINE).split(TAB) + for line in f if line.strip()] + # Every row spans the full widened header, so no column can slide. + assert {len(line) for line in lines} == {len(widened)} + appended = read_tsv(output)[-1] + assert appended["genbankAccession"] == "GCA_000222935.1" + assert appended["genusTaxId"] == "" + + def test_flow_routes_its_write_through_the_helper(self): + source = Path(backfill_module.__file__).read_text(encoding="utf-8") + assert "written = write_or_append_parsed(parsed, config)" in source