diff --git a/src/mavedb/lib/clinvar/constants.py b/src/mavedb/lib/clinvar/constants.py index e70c4fee..935376e3 100644 --- a/src/mavedb/lib/clinvar/constants.py +++ b/src/mavedb/lib/clinvar/constants.py @@ -1,8 +1,14 @@ import os +import re from pathlib import Path from urllib3.util.retry import Retry +CLINVAR_NS_PATTERN = re.compile(r"^clinvar\.(\d+)_(0[1-9]|1[0-2])$") +"""Pattern for ClinVar-versioned namespaces of the form "clinvar.YEAR_MONTH", +e.g. "clinvar.2024_01" for January 2024. +""" + TSV_VARIANT_ARCHIVE_BASE_URL = "https://ftp.ncbi.nlm.nih.gov/pub/clinvar/tab_delimited/archive" NCBI_REQUEST_HEADERS = { diff --git a/src/mavedb/lib/clinvar/utils.py b/src/mavedb/lib/clinvar/utils.py index 689e369e..dcdef05d 100644 --- a/src/mavedb/lib/clinvar/utils.py +++ b/src/mavedb/lib/clinvar/utils.py @@ -8,7 +8,7 @@ import sys from datetime import datetime from pathlib import Path -from typing import Dict +from typing import Dict, Optional import requests from filelock import FileLock @@ -17,6 +17,7 @@ from mavedb.lib.clinvar.constants import ( CLINVAR_CACHE_DIR, CLINVAR_FIELDS_TO_KEEP, + CLINVAR_NS_PATTERN, NCBI_REQUEST_HEADERS, NCBI_RETRY_STRATEGY, TSV_VARIANT_ARCHIVE_BASE_URL, @@ -28,6 +29,22 @@ logger = logging.getLogger(__name__) +def parse_clinvar_namespace(ns: str) -> Optional[str]: + """Parse a ClinVar-versioned namespace into its db_version string. + + Namespaces are of the form ``"clinvar.YEAR_MONTH"`` (e.g. ``"clinvar.2024_01"`` + for January 2024). The corresponding ``db_version`` stored in + ``clinical_controls`` is ``"MONTH_YEAR"`` (e.g. ``"01_2024"``). + + Returns ``None`` if *ns* does not match the expected pattern. + """ + m = CLINVAR_NS_PATTERN.match(ns) + if not m: + return None + year, month = m.group(1), m.group(2) + return f"{month}_{year}" + + def _ncbi_session() -> requests.Session: session = requests.Session() session.headers.update(NCBI_REQUEST_HEADERS) diff --git a/src/mavedb/lib/mave/utils.py b/src/mavedb/lib/mave/utils.py index dd6b7591..214532f3 100644 --- a/src/mavedb/lib/mave/utils.py +++ b/src/mavedb/lib/mave/utils.py @@ -31,3 +31,12 @@ def is_csv_null(value): if value == 0: return value return not value or NULL_VALUES_RE.fullmatch(str(value).strip().lower()) + + +_CSV_OUTPUT_NULL_RE = re.compile(r"\s+|none|nan|na|undefined|n/a|null|nil", flags=re.IGNORECASE) + + +def is_csv_output_null(value): + """Return True if a value should be replaced with the NA sentinel in CSV output.""" + value = str(value).strip().lower() + return _CSV_OUTPUT_NULL_RE.fullmatch(value) or not value diff --git a/src/mavedb/lib/score_set_csv.py b/src/mavedb/lib/score_set_csv.py new file mode 100644 index 00000000..b960833e --- /dev/null +++ b/src/mavedb/lib/score_set_csv.py @@ -0,0 +1,484 @@ +import csv +import io +from dataclasses import dataclass +from operator import attrgetter +from typing import Any, Callable, Iterable, List, Optional, Sequence + +from sqlalchemy import Integer, and_, cast, func, or_, select +from sqlalchemy.orm import Session, aliased + +from mavedb.lib.clinvar.constants import CLINVAR_NS_PATTERN +from mavedb.lib.clinvar.utils import parse_clinvar_namespace +from mavedb.lib.mave.constants import REQUIRED_SCORE_COLUMN +from mavedb.lib.mave.utils import NA_VALUE, is_csv_output_null +from mavedb.lib.validation.utilities import is_null as validate_is_null +from mavedb.lib.variants import get_digest_from_post_mapped, get_hgvs_from_post_mapped, is_hgvs_g, is_hgvs_p +from mavedb.models.clinical_control import ClinicalControl +from mavedb.models.clinical_control_mapped_variant import mapped_variants_clinical_controls_association_table +from mavedb.models.gnomad_variant import GnomADVariant +from mavedb.models.mapped_variant import MappedVariant +from mavedb.models.score_set import ScoreSet +from mavedb.models.variant import Variant + + +@dataclass(frozen=True) +class CsvColumnPlan: + namespaced_columns: dict[str, list[str]] + clinvar_namespaces: dict[str, str] + + +@dataclass +class CsvFetchResult: + variants: list[Variant] + mappings: Optional[list[Optional[MappedVariant]]] + gnomad_data: Optional[list[Optional[GnomADVariant]]] + clinvar_per_variant: Optional[list[Optional[dict[str, Optional[ClinicalControl]]]]] + + +# --------------------------------------------------------------------------- +# Column-key resolvers for fixed-column namespaces +# --------------------------------------------------------------------------- + +_CORE_RESOLVERS: dict[str, Callable] = { + "hgvs_nt": attrgetter("hgvs_nt"), + "hgvs_pro": attrgetter("hgvs_pro"), + "hgvs_splice": attrgetter("hgvs_splice"), + "accession": attrgetter("urn"), +} + +_VEP_RESOLVERS: dict[str, Callable] = { + "vep_functional_consequence": lambda mapping: mapping.vep_functional_consequence if mapping else None, +} + +_GNOMAD_RESOLVERS: dict[str, Callable] = { + "gnomad_af": lambda gnomad_data: gnomad_data.allele_frequency if gnomad_data else None, +} + +_CLINGEN_RESOLVERS: dict[str, Callable] = { + "clingen_allele_id": lambda mapping: mapping.clingen_allele_id if mapping else None, +} + +_CLINVAR_RESOLVERS: dict[str, Callable] = { + "clinical_significance": attrgetter("clinical_significance"), + "clinical_review_status": attrgetter("clinical_review_status"), +} + + +def _value_or_na(value: Any, na_rep: str = NA_VALUE) -> str: + """Return the string representation of *value*, or *na_rep* if the value is None.""" + if is_csv_output_null(value): + return na_rep + return str(value) + + +def _format_column_key(namespace: str, column_key: str, namespaced: bool = False) -> str: + """Shared key-formatting logic used by both header assembly and row assembly.""" + # ClinVar columns are always namespaced to differentiate versions, even if the user has requested un-namespaced output. + if CLINVAR_NS_PATTERN.match(namespace): + return f"{namespace}.{column_key}" + + # The "core" namespace is always un-namespaced, even if the user has requested namespaced output. + if namespace == "core": + return column_key + + # All other namespaces are namespaced if the user has requested namespaced output, and un-namespaced otherwise. + if namespaced: + return f"{namespace}.{column_key}" + + return column_key + + +def _custom_columns(dataset_columns: dict, col_name: str) -> list[str]: + return [col for col in [str(x) for x in list(dataset_columns.get(col_name, []))]] + + +# --------------------------------------------------------------------------- +# Pure functions +# --------------------------------------------------------------------------- + + +def plan_csv_columns( + dataset_columns: dict, + namespaces: list[str], + *, + include_custom_columns: bool = True, + include_post_mapped_hgvs: bool = False, +) -> CsvColumnPlan: + """Build the namespaced column map and ClinVar namespace mapping.""" + namespaced_score_set_columns: dict[str, list[str]] = { + "core": ["accession", "hgvs_nt", "hgvs_splice", "hgvs_pro"], + "mavedb": [], + } + + if include_post_mapped_hgvs: + namespaced_score_set_columns["mavedb"].append("post_mapped_hgvs_g") + namespaced_score_set_columns["mavedb"].append("post_mapped_hgvs_p") + namespaced_score_set_columns["mavedb"].append("post_mapped_hgvs_c") + namespaced_score_set_columns["mavedb"].append("post_mapped_hgvs_at_assay_level") + namespaced_score_set_columns["mavedb"].append("post_mapped_vrs_digest") + + for namespace in namespaces: + namespaced_score_set_columns[namespace] = [] + + if "scores" in namespaced_score_set_columns: + if include_custom_columns: + # the required score column is transitively included via the _custom_columns function. + namespaced_score_set_columns["scores"] = _custom_columns(dataset_columns, "score_columns") + else: + namespaced_score_set_columns["scores"] = [REQUIRED_SCORE_COLUMN] + if "counts" in namespaced_score_set_columns: + if include_custom_columns: + namespaced_score_set_columns["counts"] = _custom_columns(dataset_columns, "count_columns") + if "vep" in namespaced_score_set_columns: + namespaced_score_set_columns["vep"].append("vep_functional_consequence") + if "gnomad" in namespaced_score_set_columns: + namespaced_score_set_columns["gnomad"].append("gnomad_af") + if "clingen" in namespaced_score_set_columns: + namespaced_score_set_columns["clingen"].append("clingen_allele_id") + + clinvar_namespaces: dict[str, str] = {} + for ns in namespaces: + db_version = parse_clinvar_namespace(ns) + if db_version is not None: + clinvar_namespaces[ns] = db_version + namespaced_score_set_columns[ns] = ["clinical_significance", "clinical_review_status"] + + return CsvColumnPlan( + namespaced_columns=namespaced_score_set_columns, + clinvar_namespaces=clinvar_namespaces, + ) + + +def assemble_csv_headers(namespaced_columns: dict[str, list[str]], namespaced: bool = False) -> list[str]: + """Build the flat column-header list from the namespace dict.""" + return [ + _format_column_key(namespace, col, namespaced) for namespace, cols in namespaced_columns.items() for col in cols + ] + + +# --------------------------------------------------------------------------- +# Row assembly +# --------------------------------------------------------------------------- + + +def variant_to_csv_row( + variant: Variant, + columns: dict[str, list[str]], + mapping: Optional[MappedVariant] = None, + gnomad_data: Optional[GnomADVariant] = None, + clinvar_data_by_ns: Optional[dict[str, Optional[ClinicalControl]]] = None, + namespaced: bool = False, + na_rep=NA_VALUE, +) -> dict[str, Any]: + """Format a variant into a dict containing the keys specified in *columns*.""" + row: dict[str, Any] = {} + + for column_key in columns.get("core", []): + resolver = _CORE_RESOLVERS.get(column_key) + if resolver is None: + raise ValueError(f"unrecognized core column: {column_key}") + + value = str(resolver(variant)) + row[column_key] = _value_or_na(value, na_rep) + + for column_key in columns.get("mavedb", []): + if column_key == "post_mapped_hgvs_g": + value = str(mapping.hgvs_g) if mapping and mapping.hgvs_g else na_rep + if value == na_rep: + fallback_hgvs = ( + get_hgvs_from_post_mapped(mapping.post_mapped) if mapping and mapping.post_mapped else None + ) + if fallback_hgvs is not None and is_hgvs_g(fallback_hgvs): + value = fallback_hgvs + else: + value = na_rep + + elif column_key == "post_mapped_hgvs_p": + value = str(mapping.hgvs_p) if mapping and mapping.hgvs_p else na_rep + if value == na_rep: + fallback_hgvs = ( + get_hgvs_from_post_mapped(mapping.post_mapped) if mapping and mapping.post_mapped else None + ) + if fallback_hgvs is not None and is_hgvs_p(fallback_hgvs): + value = fallback_hgvs + else: + value = na_rep + + elif column_key == "post_mapped_hgvs_c": + value = str(mapping.hgvs_c) if mapping and mapping.hgvs_c else na_rep + elif column_key == "post_mapped_hgvs_at_assay_level": + value = str(mapping.hgvs_assay_level) if mapping and mapping.hgvs_assay_level else na_rep + elif column_key == "post_mapped_vrs_digest": + digest = get_digest_from_post_mapped(mapping.post_mapped) if mapping and mapping.post_mapped else None + value = digest if digest is not None else na_rep + else: + raise ValueError(f"unrecognized mavedb column: {column_key}") + + row[_format_column_key("mavedb", column_key, namespaced=namespaced)] = _value_or_na(value, na_rep) + + for ns in ("vep", "gnomad", "clingen"): + resolvers = {"vep": _VEP_RESOLVERS, "gnomad": _GNOMAD_RESOLVERS, "clingen": _CLINGEN_RESOLVERS}[ns] + source = {"vep": mapping, "gnomad": gnomad_data, "clingen": mapping}[ns] + for column_key in columns.get(ns, []): + resolver = resolvers.get(column_key) + if resolver is None: + raise ValueError(f"unrecognized {ns} column: {column_key}") + value = resolver(source) + row[_format_column_key(ns, column_key, namespaced=namespaced)] = _value_or_na(value, na_rep) + + for data_ns in ("scores", "counts"): + data_key = f"{data_ns[:-1]}_data" + parent = variant.data.get(data_key) if variant.data else None + for column_key in columns.get(data_ns, []): + value = str(parent.get(column_key)) if parent else na_rep + row[_format_column_key(data_ns, column_key, namespaced=namespaced)] = _value_or_na(value, na_rep) + + for namespace_key, namespace_cols in columns.items(): + if not CLINVAR_NS_PATTERN.match(namespace_key): + continue + clinvar_entry = (clinvar_data_by_ns or {}).get(namespace_key) + for column_key in namespace_cols: + resolver = _CLINVAR_RESOLVERS.get(column_key) + if resolver is None: + raise ValueError(f"unrecognized clinvar column: {column_key}") + value = str(resolver(clinvar_entry)) if clinvar_entry else na_rep + row[_format_column_key(namespace_key, column_key, namespaced=namespaced)] = _value_or_na(value, na_rep) + + return row + + +def variants_to_csv_rows( + variants: Sequence[Variant], + columns: dict[str, list[str]], + mappings: Optional[Sequence[Optional[MappedVariant]]] = None, + gnomad_data: Optional[Sequence[Optional[GnomADVariant]]] = None, + clinvar_data_by_ns: Optional[Sequence[Optional[dict[str, Optional[ClinicalControl]]]]] = None, + namespaced: bool = False, + na_rep=NA_VALUE, +) -> Iterable[dict[str, Any]]: + """Format each variant into a dictionary row containing the keys specified in *columns*.""" + n = len(variants) + _mappings: Sequence[Optional[MappedVariant]] = mappings if mappings is not None else [None] * n + _gnomad: Sequence[Optional[GnomADVariant]] = gnomad_data if gnomad_data is not None else [None] * n + _clinvar: Sequence[Optional[dict[str, Optional[ClinicalControl]]]] = ( + clinvar_data_by_ns if clinvar_data_by_ns is not None else [None] * n + ) + return map( + lambda t: variant_to_csv_row( + t[0], + columns, + mapping=t[1], + gnomad_data=t[2], + clinvar_data_by_ns=t[3], + namespaced=namespaced, + na_rep=na_rep, + ), + zip(variants, _mappings, _gnomad, _clinvar), + ) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def drop_na_columns_from_csv_file_rows( + rows_data: Iterable[dict[str, Any]], columns: list[str] +) -> tuple[list[dict[str, Any]], list[str]]: + """Process rows_data for downloadable CSV by removing empty columns.""" + rows_data = list(rows_data) + columns_to_check = ["hgvs_nt", "hgvs_splice", "hgvs_pro"] + columns_to_remove = [] + + for col in columns_to_check: + if all(validate_is_null(row[col]) for row in rows_data): + columns_to_remove.append(col) + for row in rows_data: + row.pop(col, None) + + columns = [col for col in columns if col not in columns_to_remove] + return rows_data, columns + + +# --------------------------------------------------------------------------- +# DB-bound fetching +# --------------------------------------------------------------------------- + + +def fetch_variant_csv_data( + db: Session, + score_set: ScoreSet, + namespaced_columns: dict[str, list[str]], + clinvar_namespaces: dict[str, str], + *, + include_post_mapped_hgvs: bool = False, + start: Optional[int] = None, + limit: Optional[int] = None, +) -> CsvFetchResult: + """Fetch variant data from the database for CSV generation.""" + namespaces = list(namespaced_columns.keys()) + + need_mappings = ( + include_post_mapped_hgvs + or "clingen" in namespaces + or "vep" in namespaces + or "gnomad" in namespaces + or bool(clinvar_namespaces) + ) + need_gnomad = "gnomad" in namespaces + + variants: list[Variant] = [] + mappings: Optional[list[Optional[MappedVariant]]] = [] if need_mappings else None + gnomad_data_list: Optional[list[Optional[GnomADVariant]]] = [] if need_gnomad else None + + select_columns: list[Any] = [Variant] + if need_mappings: + select_columns.append(MappedVariant) + if need_gnomad: + select_columns.append(GnomADVariant) + + query = ( + select(*select_columns) + .where(Variant.score_set_id == score_set.id) + .order_by(cast(func.split_part(Variant.urn, "#", 2), Integer)) + ) + + if need_mappings: + query = query.join( + MappedVariant, + and_(Variant.id == MappedVariant.variant_id, MappedVariant.current.is_(True)), + isouter=True, + ) + + if need_gnomad: + query = query.join( + MappedVariant.gnomad_variants.of_type(GnomADVariant), + isouter=True, + ).where( + or_( + and_(GnomADVariant.db_name == "gnomAD", GnomADVariant.db_version == "v4.1"), + GnomADVariant.id.is_(None), + ) + ) + + if start: + query = query.offset(start) + if limit: + query = query.limit(limit) + + result = db.execute(query).all() + + for row in result: + variant = row[0] + variants.append(variant) + + if need_mappings and mappings is not None: + mappings.append(row[1]) + + if need_gnomad and gnomad_data_list is not None: + idx = 2 if need_mappings else 1 + gnomad_data_list.append(row[idx]) + + clinvar_data_map: dict[str, dict[int, Optional[ClinicalControl]]] = {} + if clinvar_namespaces and mappings is not None: + mv_ids = [m.id for m in mappings if m is not None] + for ns, db_version in clinvar_namespaces.items(): + mv_to_cc: dict[int, Optional[ClinicalControl]] = {} + if mv_ids: + aliased_cc = aliased(ClinicalControl) + cc_query = ( + select( + mapped_variants_clinical_controls_association_table.c.mapped_variant_id, + aliased_cc, + ) + .join( + aliased_cc, + mapped_variants_clinical_controls_association_table.c.clinical_control_id == aliased_cc.id, + ) + .where( + and_( + mapped_variants_clinical_controls_association_table.c.mapped_variant_id.in_(mv_ids), + aliased_cc.db_name == "ClinVar", + aliased_cc.db_version == db_version, + ) + ) + ) + for mv_id, cc in db.execute(cc_query).all(): + mv_to_cc[mv_id] = cc + clinvar_data_map[ns] = mv_to_cc + + clinvar_per_variant: Optional[list[Optional[dict[str, Optional[ClinicalControl]]]]] = None + if clinvar_namespaces and mappings is not None: + clinvar_per_variant = [] + for mapping in mappings: + row_clinvar: dict[str, Optional[ClinicalControl]] = {} + for ns, mv_to_cc in clinvar_data_map.items(): + if mapping is not None and mapping.id is not None: + row_clinvar[ns] = mv_to_cc.get(mapping.id) + else: + row_clinvar[ns] = None + clinvar_per_variant.append(row_clinvar) + + return CsvFetchResult( + variants=variants, + mappings=mappings, + gnomad_data=gnomad_data_list, + clinvar_per_variant=clinvar_per_variant, + ) + + +# --------------------------------------------------------------------------- +# Public composer +# --------------------------------------------------------------------------- + + +def get_score_set_variants_as_csv( + db: Session, + score_set: ScoreSet, + namespaces: List[str], + namespaced: bool = False, + start: Optional[int] = None, + limit: Optional[int] = None, + drop_na_columns: Optional[bool] = None, + include_custom_columns: Optional[bool] = True, + include_post_mapped_hgvs: Optional[bool] = False, +) -> str: + """Get the variant data from a score set as a CSV string.""" + assert type(score_set.dataset_columns) is dict + + plan = plan_csv_columns( + score_set.dataset_columns, + namespaces, + include_custom_columns=bool(include_custom_columns), + include_post_mapped_hgvs=bool(include_post_mapped_hgvs), + ) + + fetched = fetch_variant_csv_data( + db, + score_set, + plan.namespaced_columns, + plan.clinvar_namespaces, + include_post_mapped_hgvs=bool(include_post_mapped_hgvs), + start=start, + limit=limit, + ) + + rows_data = variants_to_csv_rows( + fetched.variants, + columns=plan.namespaced_columns, + namespaced=namespaced, + mappings=fetched.mappings, + gnomad_data=fetched.gnomad_data, + clinvar_data_by_ns=fetched.clinvar_per_variant, + ) + + rows_columns = assemble_csv_headers(plan.namespaced_columns, namespaced=namespaced) + + if drop_na_columns: + rows_data, rows_columns = drop_na_columns_from_csv_file_rows(rows_data, rows_columns) + + stream = io.StringIO() + writer = csv.DictWriter(stream, fieldnames=rows_columns, quoting=csv.QUOTE_MINIMAL) + writer.writeheader() + writer.writerows(rows_data) + return stream.getvalue() diff --git a/src/mavedb/lib/score_sets.py b/src/mavedb/lib/score_sets.py index 8e3c8deb..698bc515 100644 --- a/src/mavedb/lib/score_sets.py +++ b/src/mavedb/lib/score_sets.py @@ -1,17 +1,14 @@ -import csv -import io import logging -import re from collections import Counter, defaultdict from operator import attrgetter -from typing import TYPE_CHECKING, Any, BinaryIO, Iterable, List, Optional, Sequence +from typing import TYPE_CHECKING, BinaryIO, Optional, Sequence import numpy as np import pandas as pd from pandas.testing import assert_index_equal -from sqlalchemy import Integer, and_, cast, func, or_, select -from sqlalchemy.orm import Query, Session, aliased, contains_eager, joinedload, selectinload +from sqlalchemy import and_, func, or_, select from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Query, Session, aliased, contains_eager, joinedload, selectinload from mavedb.lib.exceptions import ValidationError from mavedb.lib.logging.context import logging_context, save_to_logging_context @@ -19,7 +16,6 @@ HGVS_NT_COLUMN, HGVS_PRO_COLUMN, HGVS_SPLICE_COLUMN, - REQUIRED_SCORE_COLUMN, VARIANT_COUNT_DATA, VARIANT_SCORE_DATA, ) @@ -27,8 +23,6 @@ from mavedb.lib.permissions import Action, has_permission from mavedb.lib.types.authentication import UserData from mavedb.lib.validation.constants.general import null_values_list -from mavedb.lib.validation.utilities import is_null as validate_is_null -from mavedb.lib.variants import get_digest_from_post_mapped, get_hgvs_from_post_mapped, is_hgvs_g, is_hgvs_p from mavedb.models.contributor import Contributor from mavedb.models.controlled_keyword import ControlledKeyword from mavedb.models.doi_identifier import DoiIdentifier @@ -38,9 +32,6 @@ from mavedb.models.experiment_controlled_keyword import ExperimentControlledKeywordAssociation from mavedb.models.experiment_publication_identifier import ExperimentPublicationIdentifierAssociation from mavedb.models.experiment_set import ExperimentSet -from mavedb.models.clinical_control import ClinicalControl -from mavedb.models.clinical_control_mapped_variant import mapped_variants_clinical_controls_association_table -from mavedb.models.gnomad_variant import GnomADVariant from mavedb.models.mapped_variant import MappedVariant from mavedb.models.publication_identifier import PublicationIdentifier from mavedb.models.refseq_identifier import RefseqIdentifier @@ -57,7 +48,7 @@ from mavedb.models.uniprot_offset import UniprotOffset from mavedb.models.user import User from mavedb.models.variant import Variant -from mavedb.view_models.search import ScoreSetsSearch, ControlledKeywordFilterOption +from mavedb.view_models.search import ControlledKeywordFilterOption, ScoreSetsSearch if TYPE_CHECKING: from mavedb.lib.permissions import Action @@ -66,10 +57,6 @@ logger = logging.getLogger(__name__) -# Pattern for ClinVar-versioned namespaces of the form "clinvar.YEAR_MONTH", -# e.g. "clinvar.2024_01" for January 2024. -CLINVAR_NS_PATTERN = re.compile(r"^clinvar\.(\d+)_(0[1-9]|1[0-2])$") - class HGVSColumns: NUCLEOTIDE: str = "hgvs_nt" # dataset.constants.hgvs_nt_column @@ -587,258 +574,6 @@ def get_current_mapped_variants_for_annotation(db: Session, score_set: ScoreSet) ) -def get_score_set_variants_as_csv( - db: Session, - score_set: ScoreSet, - namespaces: List[str], - namespaced: Optional[bool] = None, - start: Optional[int] = None, - limit: Optional[int] = None, - drop_na_columns: Optional[bool] = None, - include_custom_columns: Optional[bool] = True, - include_post_mapped_hgvs: Optional[bool] = False, -) -> str: - """ - Get the variant data from a score set as a CSV string. - - Parameters - __________ - db : Session - The database session to use. - score_set : ScoreSet - The score set to get the variants from. - namespaces : List[str] - The namespaces for data: "scores", "counts", "vep", "gnomad", "clingen", and/or - ClinVar-versioned namespaces of the form "clinvar.YEAR_MONTH" (e.g. "clinvar.2024_01" - for January 2024, which joins on db_name="ClinVar" and db_version="01_2024"). - namespaced: Optional[bool] = None - Whether namespace the columns or not. - start : int, optional - The index to start from. If None, starts from the beginning. - limit : int, optional - The maximum number of variants to return. If None, returns all variants. - drop_na_columns : bool, optional - Whether to drop columns that contain only NA values. Defaults to False. - include_custom_columns : bool, optional - Whether to include custom columns defined in the score set. Defaults to True. - include_post_mapped_hgvs : bool, optional - Whether to include post-mapped HGVS notations and VEP functional consequence in the output. Defaults to False. If True, the output will include - columns for post-mapped HGVS genomic (g.) and protein (p.) notations, and VEP functional consequence. - - Returns - _______ - str - The CSV string containing the variant data. - """ - assert type(score_set.dataset_columns) is dict - namespaced_score_set_columns: dict[str, list[str]] = { - "core": ["accession", "hgvs_nt", "hgvs_splice", "hgvs_pro"], - "mavedb": [], - } - if include_post_mapped_hgvs: - namespaced_score_set_columns["mavedb"].append("post_mapped_hgvs_g") - namespaced_score_set_columns["mavedb"].append("post_mapped_hgvs_p") - namespaced_score_set_columns["mavedb"].append("post_mapped_hgvs_c") - namespaced_score_set_columns["mavedb"].append("post_mapped_hgvs_at_assay_level") - namespaced_score_set_columns["mavedb"].append("post_mapped_vrs_digest") - for namespace in namespaces: - namespaced_score_set_columns[namespace] = [] - - if include_custom_columns: - if "scores" in namespaced_score_set_columns: - namespaced_score_set_columns["scores"] = [ - col for col in [str(x) for x in list(score_set.dataset_columns.get("score_columns", []))] - ] - if "counts" in namespaced_score_set_columns: - namespaced_score_set_columns["counts"] = [ - col for col in [str(x) for x in list(score_set.dataset_columns.get("count_columns", []))] - ] - elif "scores" in namespaced_score_set_columns: - namespaced_score_set_columns["scores"].append(REQUIRED_SCORE_COLUMN) - if "vep" in namespaced_score_set_columns: - namespaced_score_set_columns["vep"].append("vep_functional_consequence") - if "gnomad" in namespaced_score_set_columns: - namespaced_score_set_columns["gnomad"].append("gnomad_af") - if "clingen" in namespaced_score_set_columns: - namespaced_score_set_columns["clingen"].append("clingen_allele_id") - - # Parse ClinVar-versioned namespaces of the form "clinvar.YEAR_MONTH". - # The corresponding db_version stored in clinical_controls is "MONTH_YEAR". - clinvar_namespaces: dict[str, str] = {} # namespace -> db_version (MONTH_YEAR) - for ns in namespaces: - m = CLINVAR_NS_PATTERN.match(ns) - if m: - year, month = m.group(1), m.group(2) - db_version = f"{month}_{year}" - clinvar_namespaces[ns] = db_version - namespaced_score_set_columns[ns] = ["clinical_significance", "clinical_review_status"] - - need_mappings = ( - include_post_mapped_hgvs - or "clingen" in namespaces - or "vep" in namespaces - or "gnomad" in namespaces - or bool(clinvar_namespaces) - ) - need_gnomad = "gnomad" in namespaces - - variants: list[Variant] = [] - mappings: Optional[list[Optional[MappedVariant]]] = [] if need_mappings else None - gnomad_data: Optional[list[Optional[GnomADVariant]]] = [] if need_gnomad else None - - select_columns: list[Any] = [Variant] - if need_mappings: - select_columns.append(MappedVariant) - if need_gnomad: - select_columns.append(GnomADVariant) - - query = ( - select(*select_columns) - .where(Variant.score_set_id == score_set.id) - .order_by(cast(func.split_part(Variant.urn, "#", 2), Integer)) - ) - - if need_mappings: - query = query.join( - MappedVariant, - and_(Variant.id == MappedVariant.variant_id, MappedVariant.current.is_(True)), - isouter=True, - ) - - if need_gnomad: - query = query.join( - MappedVariant.gnomad_variants.of_type(GnomADVariant), - isouter=True, - ).where( - or_( - and_(GnomADVariant.db_name == "gnomAD", GnomADVariant.db_version == "v4.1"), - GnomADVariant.id.is_(None), - ) - ) - - if start: - query = query.offset(start) - if limit: - query = query.limit(limit) - - result = db.execute(query).all() - - for row in result: - variant = row[0] - variants.append(variant) - - if need_mappings and mappings is not None: - mappings.append(row[1]) - - if need_gnomad and gnomad_data is not None: - idx = 2 if need_mappings else 1 - gnomad_data.append(row[idx]) - - # For each ClinVar namespace, fetch a mapping from mapped_variant_id to ClinicalControl. - clinvar_data_map: dict[str, dict[int, Optional[ClinicalControl]]] = {} - if clinvar_namespaces and mappings is not None: - mv_ids = [m.id for m in mappings if m is not None] - for ns, db_version in clinvar_namespaces.items(): - mv_to_cc: dict[int, Optional[ClinicalControl]] = {} - if mv_ids: - aliased_cc = aliased(ClinicalControl) - cc_query = ( - select( - mapped_variants_clinical_controls_association_table.c.mapped_variant_id, - aliased_cc, - ) - .join( - aliased_cc, - mapped_variants_clinical_controls_association_table.c.clinical_control_id == aliased_cc.id, - ) - .where( - and_( - mapped_variants_clinical_controls_association_table.c.mapped_variant_id.in_(mv_ids), - aliased_cc.db_name == "ClinVar", - aliased_cc.db_version == db_version, - ) - ) - ) - for mv_id, cc in db.execute(cc_query).all(): - mv_to_cc[mv_id] = cc - clinvar_data_map[ns] = mv_to_cc - - # Build per-variant ClinVar lookup (list indexed in parallel with variants). - clinvar_per_variant: Optional[list[Optional[dict[str, Optional[ClinicalControl]]]]] = None - if clinvar_namespaces and mappings is not None: - clinvar_per_variant = [] - for mapping in mappings: - row_clinvar: dict[str, Optional[ClinicalControl]] = {} - for ns, mv_to_cc in clinvar_data_map.items(): - if mapping is not None and mapping.id is not None: - row_clinvar[ns] = mv_to_cc.get(mapping.id) - else: - row_clinvar[ns] = None - clinvar_per_variant.append(row_clinvar) - - rows_data = variants_to_csv_rows( - variants, - columns=namespaced_score_set_columns, - namespaced=namespaced, - mappings=mappings, - gnomad_data=gnomad_data, - clinvar_data_by_ns=clinvar_per_variant, - ) # type: ignore - - rows_columns = [] - for namespace, cols in namespaced_score_set_columns.items(): - for col in cols: - if CLINVAR_NS_PATTERN.match(namespace): - # ClinVar versioned namespaces always include the full namespace prefix - # to avoid column-name collisions when multiple versions are requested. - rows_columns.append(f"{namespace}.{col}") - elif namespaced and namespace not in ["core", "mavedb"]: - rows_columns.append(f"{namespace}.{col}") - elif namespaced and namespace == "mavedb": - rows_columns.append(f"mavedb.{col}") - else: - rows_columns.append(col) - - if drop_na_columns: - rows_data, rows_columns = drop_na_columns_from_csv_file_rows(rows_data, rows_columns) - - stream = io.StringIO() - writer = csv.DictWriter(stream, fieldnames=rows_columns, quoting=csv.QUOTE_MINIMAL) - writer.writeheader() - writer.writerows(rows_data) - return stream.getvalue() - - -def drop_na_columns_from_csv_file_rows( - rows_data: Iterable[dict[str, Any]], columns: list[str] -) -> tuple[list[dict[str, Any]], list[str]]: - """Process rows_data for downloadable CSV by removing empty columns.""" - # Convert map to list. - rows_data = list(rows_data) - columns_to_check = ["hgvs_nt", "hgvs_splice", "hgvs_pro"] - columns_to_remove = [] - - # Check if all values in a column are None or "NA" - for col in columns_to_check: - if all(validate_is_null(row[col]) for row in rows_data): - columns_to_remove.append(col) - for row in rows_data: - row.pop(col, None) # Remove column from each row - - # Remove these columns from the header list - columns = [col for col in columns if col not in columns_to_remove] - return rows_data, columns - - -null_values_re = re.compile(r"\s+|none|nan|na|undefined|n/a|null|nil", flags=re.IGNORECASE) - - -def is_null(value): - """Return True if a string represents a null value.""" - value = str(value).strip().lower() - return null_values_re.fullmatch(value) or not value - - def is_replaces_id_unique_violation(exc: IntegrityError) -> bool: """ Return True if the IntegrityError was caused by the unique constraint on score_set.replaces_id. @@ -852,203 +587,6 @@ def is_replaces_id_unique_violation(exc: IntegrityError) -> bool: return "replaces_id" in detail -def variant_to_csv_row( - variant: Variant, - columns: dict[str, list[str]], - mapping: Optional[MappedVariant] = None, - gnomad_data: Optional[GnomADVariant] = None, - clinvar_data_by_ns: Optional[dict[str, Optional[ClinicalControl]]] = None, - namespaced: Optional[bool] = None, - na_rep="NA", -) -> dict[str, Any]: - """ - Format a variant into a containing the keys specified in `columns`. - - Parameters - ---------- - variant : variant.models.Variant - List of variants. - columns : list[str] - Columns to serialize. - namespaced: Optional[bool] = None - Namespace the columns or not. - mapping : variant.models.MappedVariant, optional - Mapped variant corresponding to the variant. - gnomad_data : variant.models.GnomADVariant, optional - gnomAD variant data corresponding to the variant. - clinvar_data_by_ns : dict[str, Optional[ClinicalControl]], optional - Per-variant ClinVar data keyed by namespace (e.g. "clinvar.2024_01"). - na_rep : str - String to represent null values. - - Returns - ------- - dict[str, Any] - """ - row: dict[str, Any] = {} - # Handle each column key explicitly as part of its namespace. - for column_key in columns.get("core", []): - if column_key == "hgvs_nt": - value = str(variant.hgvs_nt) - elif column_key == "hgvs_pro": - value = str(variant.hgvs_pro) - elif column_key == "hgvs_splice": - value = str(variant.hgvs_splice) - elif column_key == "accession": - value = str(variant.urn) - if is_null(value): - value = na_rep - - # export columns in the `core` namespace without a namespace - row[column_key] = value - for column_key in columns.get("mavedb", []): - if column_key == "post_mapped_hgvs_g": - value = str(mapping.hgvs_g) if mapping and mapping.hgvs_g else na_rep - if value == na_rep: - fallback_hgvs = ( - get_hgvs_from_post_mapped(mapping.post_mapped) if mapping and mapping.post_mapped else None - ) - if fallback_hgvs is not None and is_hgvs_g(fallback_hgvs): - value = fallback_hgvs - else: - value = na_rep - - elif column_key == "post_mapped_hgvs_p": - value = str(mapping.hgvs_p) if mapping and mapping.hgvs_p else na_rep - if value == na_rep: - fallback_hgvs = ( - get_hgvs_from_post_mapped(mapping.post_mapped) if mapping and mapping.post_mapped else None - ) - if fallback_hgvs is not None and is_hgvs_p(fallback_hgvs): - value = fallback_hgvs - else: - value = na_rep - - elif column_key == "post_mapped_hgvs_c": - value = str(mapping.hgvs_c) if mapping and mapping.hgvs_c else na_rep - elif column_key == "post_mapped_hgvs_at_assay_level": - value = str(mapping.hgvs_assay_level) if mapping and mapping.hgvs_assay_level else na_rep - elif column_key == "post_mapped_vrs_digest": - digest = get_digest_from_post_mapped(mapping.post_mapped) if mapping and mapping.post_mapped else None - value = digest if digest is not None else na_rep - if is_null(value): - value = na_rep - key = f"mavedb.{column_key}" if namespaced else column_key - row[key] = value - for column_key in columns.get("vep", []): - if column_key == "vep_functional_consequence": - vep_functional_consequence = mapping.vep_functional_consequence if mapping else None - if vep_functional_consequence is not None: - value = vep_functional_consequence - else: - value = na_rep - key = f"vep.{column_key}" if namespaced else column_key - row[key] = value - for column_key in columns.get("scores", []): - parent = variant.data.get("score_data") if variant.data else None - value = str(parent.get(column_key)) if parent else na_rep - if is_null(value): - value = na_rep - key = f"scores.{column_key}" if namespaced else column_key - row[key] = value - for column_key in columns.get("counts", []): - parent = variant.data.get("count_data") if variant.data else None - value = str(parent.get(column_key)) if parent else na_rep - if is_null(value): - value = na_rep - key = f"counts.{column_key}" if namespaced else column_key - row[key] = value - for column_key in columns.get("gnomad", []): - if column_key == "gnomad_af": - gnomad_af = gnomad_data.allele_frequency if gnomad_data else None - if gnomad_af is not None: - value = str(gnomad_af) - else: - value = na_rep - key = f"gnomad.{column_key}" if namespaced else column_key - row[key] = value - for column_key in columns.get("clingen", []): - if column_key == "clingen_allele_id": - clingen_allele_id = mapping.clingen_allele_id if mapping else None - if clingen_allele_id is not None: - value = str(clingen_allele_id) - else: - value = na_rep - key = f"clingen.{column_key}" if namespaced else column_key - row[key] = value - # Handle ClinVar-versioned namespaces (e.g. "clinvar.2024_01"). - # These always use the full "namespace.column" key regardless of the namespaced flag - # to avoid collisions when multiple versions are requested. - for namespace_key, namespace_cols in columns.items(): - if not CLINVAR_NS_PATTERN.match(namespace_key): - continue - clinvar_entry = (clinvar_data_by_ns or {}).get(namespace_key) - for column_key in namespace_cols: - if column_key == "clinical_significance": - value = str(clinvar_entry.clinical_significance) if clinvar_entry else na_rep - elif column_key == "clinical_review_status": - value = str(clinvar_entry.clinical_review_status) if clinvar_entry else na_rep - else: - value = na_rep - if is_null(value): - value = na_rep - row[f"{namespace_key}.{column_key}"] = value - return row - - -def variants_to_csv_rows( - variants: Sequence[Variant], - columns: dict[str, list[str]], - mappings: Optional[Sequence[Optional[MappedVariant]]] = None, - gnomad_data: Optional[Sequence[Optional[GnomADVariant]]] = None, - clinvar_data_by_ns: Optional[Sequence[Optional[dict[str, Optional[ClinicalControl]]]]] = None, - namespaced: Optional[bool] = None, - na_rep="NA", -) -> Iterable[dict[str, Any]]: - """ - Format each variant into a dictionary row containing the keys specified in `columns`. - - Parameters - ---------- - variants : list[variant.models.Variant] - List of variants. - columns : list[str] - Columns to serialize. - namespaced: Optional[bool] = None - Namespace the columns or not. - mappings : list[Optional[variant.models.MappedVariant]], optional - List of mapped variants corresponding to the variants. - gnomad_data : list[Optional[variant.models.GnomADVariant]], optional - List of gnomAD variant data corresponding to the variants. - clinvar_data_by_ns : list[Optional[dict[str, Optional[ClinicalControl]]]], optional - Per-variant ClinVar data keyed by namespace (e.g. "clinvar.2024_01"). - na_rep : str - String to represent null values. - - Returns - ------- - list[dict[str, Any]] - """ - n = len(variants) - _mappings: Sequence[Optional[MappedVariant]] = mappings if mappings is not None else [None] * n - _gnomad: Sequence[Optional[GnomADVariant]] = gnomad_data if gnomad_data is not None else [None] * n - _clinvar: Sequence[Optional[dict[str, Optional[ClinicalControl]]]] = ( - clinvar_data_by_ns if clinvar_data_by_ns is not None else [None] * n - ) - return map( - lambda t: variant_to_csv_row( - t[0], - columns, - mapping=t[1], - gnomad_data=t[2], - clinvar_data_by_ns=t[3], - namespaced=namespaced, - na_rep=na_rep, - ), - zip(variants, _mappings, _gnomad, _clinvar), - ) - - def find_meta_analyses_for_score_sets(db: Session, urns: list[str]) -> list[ScoreSet]: """ Find all score sets that are meta-analyses for a specified collection of other score sets. diff --git a/src/mavedb/routers/score_sets.py b/src/mavedb/routers/score_sets.py index 371862d1..14156820 100644 --- a/src/mavedb/routers/score_sets.py +++ b/src/mavedb/routers/score_sets.py @@ -49,16 +49,15 @@ ) from mavedb.lib.permissions import Action, assert_permission, has_permission from mavedb.lib.score_calibrations import create_score_calibration +from mavedb.lib.clinvar.constants import CLINVAR_NS_PATTERN +from mavedb.lib.score_set_csv import get_score_set_variants_as_csv, variants_to_csv_rows from mavedb.lib.score_sets import ( - CLINVAR_NS_PATTERN, csv_data_to_df, fetch_score_set_search_filter_options, find_meta_analyses_for_experiment_sets, get_current_mapped_variants_for_annotation, - get_score_set_variants_as_csv, is_replaces_id_unique_violation, refresh_variant_urns, - variants_to_csv_rows, ) from mavedb.lib.score_sets import ( search_score_sets as _search_score_sets, diff --git a/src/mavedb/scripts/export_public_data.py b/src/mavedb/scripts/export_public_data.py index 4ced338a..5e3ecafd 100644 --- a/src/mavedb/scripts/export_public_data.py +++ b/src/mavedb/scripts/export_public_data.py @@ -26,7 +26,8 @@ from sqlalchemy.orm import Session, joinedload, lazyload from mavedb.lib.annotation.annotate import variant_highest_level_annotation -from mavedb.lib.score_sets import get_current_mapped_variants_for_annotation, get_score_set_variants_as_csv +from mavedb.lib.score_set_csv import get_score_set_variants_as_csv +from mavedb.lib.score_sets import get_current_mapped_variants_for_annotation from mavedb.models.experiment import Experiment from mavedb.models.experiment_set import ExperimentSet from mavedb.models.license import License diff --git a/tests/lib/clinvar/test_utils.py b/tests/lib/clinvar/test_utils.py index 082c5558..257402d3 100644 --- a/tests/lib/clinvar/test_utils.py +++ b/tests/lib/clinvar/test_utils.py @@ -9,6 +9,7 @@ from mavedb.lib.clinvar.constants import CLINVAR_FIELDS_TO_KEEP from mavedb.lib.clinvar.utils import ( fetch_clinvar_variant_data, + parse_clinvar_namespace, validate_clinvar_variant_summary_date, ) @@ -45,6 +46,25 @@ def _make_gzipped_tsv(text: str) -> bytes: ) +@pytest.mark.unit +@pytest.mark.parametrize( + "ns, expected", + [ + ("clinvar.2024_01", "01_2024"), + ("clinvar.2015_12", "12_2015"), + ("clinvar.2026_06", "06_2026"), + ("clinvar.2024_00", None), + ("clinvar.2024_13", None), + ("scores", None), + ("clinvar", None), + ("clinvar.2024_01.extra", None), + ("", None), + ], +) +def test_parse_clinvar_namespace(ns, expected): + assert parse_clinvar_namespace(ns) == expected + + @pytest.mark.unit class TestValidateClinvarVariantSummaryDate: def test_valid_past_date(self): diff --git a/tests/lib/mave/__init__.py b/tests/lib/mave/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/lib/mave/test_utils.py b/tests/lib/mave/test_utils.py new file mode 100644 index 00000000..44716595 --- /dev/null +++ b/tests/lib/mave/test_utils.py @@ -0,0 +1,31 @@ +import pytest + +from mavedb.lib.mave.utils import is_csv_output_null + + +@pytest.mark.unit +@pytest.mark.parametrize( + "value, expected", + [ + (None, True), + ("", True), + (" ", True), + ("NA", True), + ("na", True), + ("None", True), + ("none", True), + ("NaN", True), + ("nan", True), + ("null", True), + ("NULL", True), + ("nil", True), + ("N/A", True), + ("undefined", True), + ("1.5", False), + ("0", False), + ("hello", False), + ("p.Met1Val", False), + ], +) +def test_is_csv_output_null(value, expected): + assert bool(is_csv_output_null(value)) is expected diff --git a/tests/lib/test_score_set.py b/tests/lib/test_score_set.py index 3ca40d4a..53d1e874 100644 --- a/tests/lib/test_score_set.py +++ b/tests/lib/test_score_set.py @@ -22,7 +22,6 @@ create_variants_data, csv_data_to_df, fetch_score_set_search_filter_options, - variant_to_csv_row, ) from mavedb.lib.types.authentication import UserData from mavedb.lib.validation.constants.general import ( @@ -556,154 +555,3 @@ def test_fetch_score_set_search_filter_options_with_no_permitted_score_sets(setu "publication_db_names": [], "publication_journals": [], } - - -class MockVariant: - """Lightweight mock for Variant used in variant_to_csv_row tests.""" - - def __init__(self, urn="urn:mavedb:00000001-a-1#1", hgvs_nt=None, hgvs_splice=None, hgvs_pro=None, data=None): - self.urn = urn - self.hgvs_nt = hgvs_nt - self.hgvs_splice = hgvs_splice - self.hgvs_pro = hgvs_pro - self.data = data - - -class TestVariantToCsvRowNullHandling: - """Tests that variant_to_csv_row represents missing data as na_rep, not 'None'.""" - - def test_score_data_with_none_value_uses_na_rep(self): - variant = MockVariant(data={"score_data": {"score": None}}) - columns = {"scores": ["score"]} - - row = variant_to_csv_row(variant, columns) - - assert row["score"] == "NA" - - def test_score_data_with_missing_key_uses_na_rep(self): - variant = MockVariant(data={"score_data": {}}) - columns = {"scores": ["score"]} - - row = variant_to_csv_row(variant, columns) - - assert row["score"] == "NA" - - def test_score_data_with_no_score_data_key_uses_na_rep(self): - variant = MockVariant(data={}) - columns = {"scores": ["score"]} - - row = variant_to_csv_row(variant, columns) - - assert row["score"] == "NA" - - def test_score_data_with_no_data_uses_na_rep(self): - variant = MockVariant(data=None) - columns = {"scores": ["score"]} - - row = variant_to_csv_row(variant, columns) - - assert row["score"] == "NA" - - def test_count_data_with_none_value_uses_na_rep(self): - variant = MockVariant(data={"count_data": {"count1": None}}) - columns = {"counts": ["count1"]} - - row = variant_to_csv_row(variant, columns) - - assert row["count1"] == "NA" - - def test_count_data_with_missing_key_uses_na_rep(self): - variant = MockVariant(data={"count_data": {}}) - columns = {"counts": ["count1"]} - - row = variant_to_csv_row(variant, columns) - - assert row["count1"] == "NA" - - def test_count_data_with_no_count_data_key_uses_na_rep(self): - variant = MockVariant(data={}) - columns = {"counts": ["count1"]} - - row = variant_to_csv_row(variant, columns) - - assert row["count1"] == "NA" - - def test_count_data_with_no_data_uses_na_rep(self): - variant = MockVariant(data=None) - columns = {"counts": ["count1"]} - - row = variant_to_csv_row(variant, columns) - - assert row["count1"] == "NA" - - def test_score_data_with_valid_value_preserved(self): - variant = MockVariant(data={"score_data": {"score": 1.5}}) - columns = {"scores": ["score"]} - - row = variant_to_csv_row(variant, columns) - - assert row["score"] == "1.5" - - def test_count_data_with_valid_value_preserved(self): - variant = MockVariant(data={"count_data": {"count1": 42}}) - columns = {"counts": ["count1"]} - - row = variant_to_csv_row(variant, columns) - - assert row["count1"] == "42" - - def test_score_data_with_custom_na_rep(self): - variant = MockVariant(data={"score_data": {"score": None}}) - columns = {"scores": ["score"]} - - row = variant_to_csv_row(variant, columns, na_rep="N/A") - - assert row["score"] == "N/A" - - def test_namespaced_score_data_with_none_value_uses_na_rep(self): - variant = MockVariant(data={"score_data": {"score": None}}) - columns = {"scores": ["score"]} - - row = variant_to_csv_row(variant, columns, namespaced=True) - - assert row["scores.score"] == "NA" - - def test_namespaced_count_data_with_none_value_uses_na_rep(self): - variant = MockVariant(data={"count_data": {"count1": None}}) - columns = {"counts": ["count1"]} - - row = variant_to_csv_row(variant, columns, namespaced=True) - - assert row["counts.count1"] == "NA" - - def test_core_columns_with_none_hgvs_uses_na_rep(self): - variant = MockVariant(hgvs_nt=None, hgvs_pro=None, hgvs_splice=None, urn="urn:mavedb:00000001-a-1#1") - columns = {"core": ["accession", "hgvs_nt", "hgvs_splice", "hgvs_pro"]} - - row = variant_to_csv_row(variant, columns) - - assert row["hgvs_nt"] == "NA" - assert row["hgvs_pro"] == "NA" - assert row["hgvs_splice"] == "NA" - assert row["accession"] == "urn:mavedb:00000001-a-1#1" - - def test_mixed_columns_with_missing_data(self): - variant = MockVariant( - hgvs_nt="g.1A>G", - hgvs_pro="p.Met1Val", - data={"score_data": {"score": None, "se": 0.1}, "count_data": {"count1": None, "count2": 5}}, - ) - columns = { - "core": ["hgvs_nt", "hgvs_pro"], - "scores": ["score", "se"], - "counts": ["count1", "count2"], - } - - row = variant_to_csv_row(variant, columns) - - assert row["hgvs_nt"] == "g.1A>G" - assert row["hgvs_pro"] == "p.Met1Val" - assert row["score"] == "NA" - assert row["se"] == "0.1" - assert row["count1"] == "NA" - assert row["count2"] == "5" diff --git a/tests/lib/test_score_set_csv.py b/tests/lib/test_score_set_csv.py new file mode 100644 index 00000000..9ec91309 --- /dev/null +++ b/tests/lib/test_score_set_csv.py @@ -0,0 +1,433 @@ +import pytest + +from mavedb.lib.score_set_csv import ( + assemble_csv_headers, + drop_na_columns_from_csv_file_rows, + plan_csv_columns, + variant_to_csv_row, +) + +# --------------------------------------------------------------------------- +# MockVariant +# --------------------------------------------------------------------------- + + +class MockVariant: + """Lightweight mock for Variant used in variant_to_csv_row tests.""" + + def __init__(self, urn="urn:mavedb:00000001-a-1#1", hgvs_nt=None, hgvs_splice=None, hgvs_pro=None, data=None): + self.urn = urn + self.hgvs_nt = hgvs_nt + self.hgvs_splice = hgvs_splice + self.hgvs_pro = hgvs_pro + self.data = data + + +# --------------------------------------------------------------------------- +# TestVariantToCsvRowNullHandling +# --------------------------------------------------------------------------- + + +class TestVariantToCsvRowNullHandling: + """Tests that variant_to_csv_row represents missing data as na_rep, not 'None'.""" + + def test_score_data_with_none_value_uses_na_rep(self): + variant = MockVariant(data={"score_data": {"score": None}}) + columns = {"scores": ["score"]} + + row = variant_to_csv_row(variant, columns) + + assert row["score"] == "NA" + + def test_score_data_with_missing_key_uses_na_rep(self): + variant = MockVariant(data={"score_data": {}}) + columns = {"scores": ["score"]} + + row = variant_to_csv_row(variant, columns) + + assert row["score"] == "NA" + + def test_score_data_with_no_score_data_key_uses_na_rep(self): + variant = MockVariant(data={}) + columns = {"scores": ["score"]} + + row = variant_to_csv_row(variant, columns) + + assert row["score"] == "NA" + + def test_score_data_with_no_data_uses_na_rep(self): + variant = MockVariant(data=None) + columns = {"scores": ["score"]} + + row = variant_to_csv_row(variant, columns) + + assert row["score"] == "NA" + + def test_count_data_with_none_value_uses_na_rep(self): + variant = MockVariant(data={"count_data": {"count1": None}}) + columns = {"counts": ["count1"]} + + row = variant_to_csv_row(variant, columns) + + assert row["count1"] == "NA" + + def test_count_data_with_missing_key_uses_na_rep(self): + variant = MockVariant(data={"count_data": {}}) + columns = {"counts": ["count1"]} + + row = variant_to_csv_row(variant, columns) + + assert row["count1"] == "NA" + + def test_count_data_with_no_count_data_key_uses_na_rep(self): + variant = MockVariant(data={}) + columns = {"counts": ["count1"]} + + row = variant_to_csv_row(variant, columns) + + assert row["count1"] == "NA" + + def test_count_data_with_no_data_uses_na_rep(self): + variant = MockVariant(data=None) + columns = {"counts": ["count1"]} + + row = variant_to_csv_row(variant, columns) + + assert row["count1"] == "NA" + + def test_score_data_with_valid_value_preserved(self): + variant = MockVariant(data={"score_data": {"score": 1.5}}) + columns = {"scores": ["score"]} + + row = variant_to_csv_row(variant, columns) + + assert row["score"] == "1.5" + + def test_count_data_with_valid_value_preserved(self): + variant = MockVariant(data={"count_data": {"count1": 42}}) + columns = {"counts": ["count1"]} + + row = variant_to_csv_row(variant, columns) + + assert row["count1"] == "42" + + def test_score_data_with_custom_na_rep(self): + variant = MockVariant(data={"score_data": {"score": None}}) + columns = {"scores": ["score"]} + + row = variant_to_csv_row(variant, columns, na_rep="N/A") + + assert row["score"] == "N/A" + + def test_namespaced_score_data_with_none_value_uses_na_rep(self): + variant = MockVariant(data={"score_data": {"score": None}}) + columns = {"scores": ["score"]} + + row = variant_to_csv_row(variant, columns, namespaced=True) + + assert row["scores.score"] == "NA" + + def test_namespaced_count_data_with_none_value_uses_na_rep(self): + variant = MockVariant(data={"count_data": {"count1": None}}) + columns = {"counts": ["count1"]} + + row = variant_to_csv_row(variant, columns, namespaced=True) + + assert row["counts.count1"] == "NA" + + def test_core_columns_with_none_hgvs_uses_na_rep(self): + variant = MockVariant(hgvs_nt=None, hgvs_pro=None, hgvs_splice=None, urn="urn:mavedb:00000001-a-1#1") + columns = {"core": ["accession", "hgvs_nt", "hgvs_splice", "hgvs_pro"]} + + row = variant_to_csv_row(variant, columns) + + assert row["hgvs_nt"] == "NA" + assert row["hgvs_pro"] == "NA" + assert row["hgvs_splice"] == "NA" + assert row["accession"] == "urn:mavedb:00000001-a-1#1" + + def test_mixed_columns_with_missing_data(self): + variant = MockVariant( + hgvs_nt="g.1A>G", + hgvs_pro="p.Met1Val", + data={"score_data": {"score": None, "se": 0.1}, "count_data": {"count1": None, "count2": 5}}, + ) + columns = { + "core": ["hgvs_nt", "hgvs_pro"], + "scores": ["score", "se"], + "counts": ["count1", "count2"], + } + + row = variant_to_csv_row(variant, columns) + + assert row["hgvs_nt"] == "g.1A>G" + assert row["hgvs_pro"] == "p.Met1Val" + assert row["score"] == "NA" + assert row["se"] == "0.1" + assert row["count1"] == "NA" + assert row["count2"] == "5" + + +# --------------------------------------------------------------------------- +# TestVariantToCsvRowUnrecognizedKey +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +@pytest.mark.parametrize( + "namespace, columns", + [ + ("core", {"core": ["bogus_col"]}), + ("mavedb", {"mavedb": ["bogus_col"]}), + ("vep", {"vep": ["bogus_col"]}), + ("gnomad", {"gnomad": ["bogus_col"]}), + ("clingen", {"clingen": ["bogus_col"]}), + ("clinvar.2024_01", {"clinvar.2024_01": ["bogus_col"]}), + ], +) +def test_unrecognized_column_key_raises(namespace, columns): + variant = MockVariant() + with pytest.raises(ValueError, match="unrecognized .* column: bogus_col"): + variant_to_csv_row(variant, columns) + + +# --------------------------------------------------------------------------- +# TestPlanCsvColumns +# --------------------------------------------------------------------------- + + +SAMPLE_DATASET_COLUMNS = { + "score_columns": ["score", "se", "epsilon"], + "count_columns": ["count1", "count2"], +} + + +@pytest.mark.unit +@pytest.mark.parametrize( + "namespaces, kwargs, expected_ns_keys, expected_score_cols, expected_clinvar", + [ + # scores-only + ( + ["scores"], + {}, + {"core", "mavedb", "scores"}, + ["score", "se", "epsilon"], + {}, + ), + # counts-only + ( + ["counts"], + {}, + {"core", "mavedb", "counts"}, + None, + {}, + ), + # both scores and counts + ( + ["scores", "counts"], + {}, + {"core", "mavedb", "scores", "counts"}, + ["score", "se", "epsilon"], + {}, + ), + # vep adds its column + ( + ["vep"], + {}, + {"core", "mavedb", "vep"}, + None, + {}, + ), + # gnomad adds its column + ( + ["gnomad"], + {}, + {"core", "mavedb", "gnomad"}, + None, + {}, + ), + # clingen adds its column + ( + ["clingen"], + {}, + {"core", "mavedb", "clingen"}, + None, + {}, + ), + # include_custom_columns=False -> only REQUIRED_SCORE_COLUMN for scores + ( + ["scores"], + {"include_custom_columns": False}, + {"core", "mavedb", "scores"}, + ["score"], + {}, + ), + # include_post_mapped_hgvs populates mavedb namespace + ( + ["scores"], + {"include_post_mapped_hgvs": True}, + {"core", "mavedb", "scores"}, + ["score", "se", "epsilon"], + {}, + ), + # single ClinVar namespace + ( + ["clinvar.2024_01"], + {}, + {"core", "mavedb", "clinvar.2024_01"}, + None, + {"clinvar.2024_01": "01_2024"}, + ), + # multiple ClinVar versions + ( + ["clinvar.2024_01", "clinvar.2025_06"], + {}, + {"core", "mavedb", "clinvar.2024_01", "clinvar.2025_06"}, + None, + {"clinvar.2024_01": "01_2024", "clinvar.2025_06": "06_2025"}, + ), + ], +) +def test_plan_csv_columns(namespaces, kwargs, expected_ns_keys, expected_score_cols, expected_clinvar): + plan = plan_csv_columns(SAMPLE_DATASET_COLUMNS, namespaces, **kwargs) + + assert set(plan.namespaced_columns.keys()) == expected_ns_keys + assert plan.clinvar_namespaces == expected_clinvar + + if expected_score_cols is not None: + assert plan.namespaced_columns["scores"] == expected_score_cols + + # core always has the standard 4 columns + assert plan.namespaced_columns["core"] == ["accession", "hgvs_nt", "hgvs_splice", "hgvs_pro"] + + # vep, gnomad, clingen get their fixed columns when present + if "vep" in plan.namespaced_columns: + assert plan.namespaced_columns["vep"] == ["vep_functional_consequence"] + if "gnomad" in plan.namespaced_columns: + assert plan.namespaced_columns["gnomad"] == ["gnomad_af"] + if "clingen" in plan.namespaced_columns: + assert plan.namespaced_columns["clingen"] == ["clingen_allele_id"] + + # ClinVar namespaces get their standard columns + for ns in expected_clinvar: + assert plan.namespaced_columns[ns] == ["clinical_significance", "clinical_review_status"] + + +def test_plan_csv_columns_post_mapped_hgvs_populates_mavedb(): + plan = plan_csv_columns(SAMPLE_DATASET_COLUMNS, ["scores"], include_post_mapped_hgvs=True) + assert plan.namespaced_columns["mavedb"] == [ + "post_mapped_hgvs_g", + "post_mapped_hgvs_p", + "post_mapped_hgvs_c", + "post_mapped_hgvs_at_assay_level", + "post_mapped_vrs_digest", + ] + + +# --------------------------------------------------------------------------- +# TestAssembleCsvHeaders +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +@pytest.mark.parametrize( + "namespaced_columns, namespaced, expected", + [ + # Unnamespaced: flat column names + ( + {"core": ["accession", "hgvs_nt"], "scores": ["score", "se"]}, + False, + ["accession", "hgvs_nt", "score", "se"], + ), + # Namespaced: scores get prefix, core does not + ( + {"core": ["accession", "hgvs_nt"], "scores": ["score"]}, + True, + ["accession", "hgvs_nt", "scores.score"], + ), + # mavedb namespace always gets prefix when namespaced + ( + {"core": ["accession"], "mavedb": ["post_mapped_hgvs_g"]}, + True, + ["accession", "mavedb.post_mapped_hgvs_g"], + ), + # ClinVar namespaces always get prefix regardless of namespaced flag + ( + {"core": ["accession"], "clinvar.2024_01": ["clinical_significance"]}, + False, + ["accession", "clinvar.2024_01.clinical_significance"], + ), + # Mixed: respects insertion order + ( + { + "core": ["accession"], + "mavedb": [], + "scores": ["score"], + "clinvar.2024_01": ["clinical_significance"], + }, + True, + ["accession", "scores.score", "clinvar.2024_01.clinical_significance"], + ), + # Empty mavedb namespace when not namespaced produces nothing + ( + {"core": ["hgvs_nt"], "mavedb": []}, + False, + ["hgvs_nt"], + ), + ], +) +def test_assemble_csv_headers(namespaced_columns, namespaced, expected): + assert assemble_csv_headers(namespaced_columns, namespaced) == expected + + +# --------------------------------------------------------------------------- +# TestDropNaColumns +# --------------------------------------------------------------------------- + + +class TestDropNaColumns: + def test_removes_all_na_hgvs_column(self): + rows = [ + {"hgvs_nt": "g.1A>G", "hgvs_splice": "NA", "hgvs_pro": "p.Met1Val"}, + {"hgvs_nt": "g.2C>T", "hgvs_splice": "NA", "hgvs_pro": "p.Ala2Gly"}, + ] + columns = ["hgvs_nt", "hgvs_splice", "hgvs_pro"] + + new_rows, new_cols = drop_na_columns_from_csv_file_rows(rows, columns) + + assert "hgvs_splice" not in new_cols + assert "hgvs_nt" in new_cols + assert "hgvs_pro" in new_cols + for row in new_rows: + assert "hgvs_splice" not in row + + def test_keeps_column_with_some_values(self): + rows = [ + {"hgvs_nt": "g.1A>G", "hgvs_splice": "NA", "hgvs_pro": "p.Met1Val"}, + {"hgvs_nt": "g.2C>T", "hgvs_splice": "c.1A>G", "hgvs_pro": "p.Ala2Gly"}, + ] + columns = ["hgvs_nt", "hgvs_splice", "hgvs_pro"] + + new_rows, new_cols = drop_na_columns_from_csv_file_rows(rows, columns) + + assert new_cols == ["hgvs_nt", "hgvs_splice", "hgvs_pro"] + + def test_does_not_touch_non_hgvs_columns(self): + rows = [ + {"hgvs_nt": "g.1A>G", "hgvs_splice": "NA", "hgvs_pro": "NA", "score": "NA"}, + ] + columns = ["hgvs_nt", "hgvs_splice", "hgvs_pro", "score"] + + new_rows, new_cols = drop_na_columns_from_csv_file_rows(rows, columns) + + assert "score" in new_cols + assert "hgvs_splice" not in new_cols + + def test_empty_rows_does_not_crash(self): + rows = [] + columns = ["hgvs_nt", "hgvs_splice", "hgvs_pro"] + + new_rows, new_cols = drop_na_columns_from_csv_file_rows(rows, columns) + + assert new_rows == [] + assert new_cols == []