Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion VariantValidator/modules/format_converters.py
Original file line number Diff line number Diff line change
Expand Up @@ -164,13 +164,26 @@ def vcf2hgvs_stage1(variant, validator):
if len(vcf_data) < 4:
logger.debug("Completed VCF-HVGS step 1 for %s", variant.quibble)
return False

# VCF CNV descriptions can be directly converted to a HGVS like description
if (re.search("\d+", vcf_data[2]) and (
re.search("del", vcf_data[3], re.IGNORECASE) or
re.search("inv", vcf_data[3], re.IGNORECASE))):

if not re.search(r"[gatcnmo]\.", str(vcf_data)):
logger.info(f"CNV format identified in {variant.quibble}")
cnv_var = f"{vcf_data[0]}:{vcf_data[1]}_{vcf_data[2]}{vcf_data[3].lower()}"
logger.info(f"CNV identified, and mapped to {cnv_var}")
variant.warnings.append(f"VcfConversionWarning: CNV identified, and mapped to {cnv_var}")
variant.quibble = cnv_var

poss_genome = vcf_data[0].lower()
if ('grch3' in poss_genome or 'hg' in poss_genome) and poss_genome[-1].isdigit():
vcf_data = vcf_data[1:]
variant.quibble = '-'.join(vcf_data)
# TODO test assembly given against settings
if len(vcf_data) < 4:
variant.warnings.append("Insufficient or incorrect VCF elements provided. "
variant.warnings.append("Insufficient or incorrect VCF elements provided. "
"Elements required are chr-pos-ref-alt")
return True
# no coordinate found
Expand Down
129 changes: 129 additions & 0 deletions VariantValidator/modules/vcf_to_pvcf.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,129 @@
class VcfConversionError(Exception):
"""Custom exception raised when a VCF line cannot be converted to shorthand."""
pass


def split_vcf_line(vcf_line):
"""
Split a VCF line by detecting delimiter (tab or comma).
Raises VcfConversionError if no supported delimiter is found.
"""
line = vcf_line.strip()

if "\t" in line:
return line.split("\t")

if "," in line:
return [f.strip() for f in line.split(",")]

raise VcfConversionError(
"Unable to detect delimiter. Expected tab ('\\t') or comma-separated values."
)


def vcf_to_shorthand(vcf_line):
"""
Convert a single VCF line into chr-start-end-TYPE shorthand.

Handles:
- CNVs with <DEL>/<DUP>/<INV>
- Simple SNVs/indels (chr-pos-ref-alt)
- Optional CN field

Raises:
VcfConversionError with descriptive message if conversion fails.
"""

# Skip headers
if vcf_line.startswith("#"):
raise VcfConversionError(
"Header line cannot be converted. Please provide a variant record line."
)

fields = split_vcf_line(vcf_line)

if len(fields) < 5:
raise VcfConversionError(
f"VCF line has insufficient columns (found {len(fields)}, expected ≥5). "
"Ensure the line contains at least CHROM, POS, ID, REF, ALT."
)

chrom, pos, _id, ref, alt = fields[:5]
info = fields[7] if len(fields) > 7 else ""

# Validate position
try:
pos = int(pos)
except ValueError:
raise VcfConversionError(
f"Invalid POS field: '{pos}' is not an integer."
)

# Structural variants
if alt in ("<DEL>", "<DUP>", "<INV>", "DEL", "DUP", "INV"):
end = None
cn = None

for entry in info.split(";"):
if entry.startswith("END="):
try:
end = int(entry.split("=")[1])
except ValueError:
raise VcfConversionError(
f"Invalid END value in INFO field: '{entry}'."
)

elif entry.startswith("SVLEN=") and end is None:
try:
end = pos + abs(int(entry.split("=")[1]))
except ValueError:
raise VcfConversionError(
f"Invalid SVLEN value in INFO field: '{entry}'."
)

elif entry.startswith("CN="):
cn = entry.split("=")[1]

if end is None:
raise VcfConversionError(
"Cannot determine end position for structural variant. "
"INFO field must contain END= or SVLEN=."
)

# Remove angle brackets
if ">" in alt:
alt_clean = alt[1:-1]
else:
alt_clean = alt

shorthand = f"{chrom}-{pos}-{end}-{alt_clean}"

if cn:
shorthand += f"[CN{cn}]"

return shorthand

# Simple SNV/indel
if not ref or not alt:
raise VcfConversionError(
"Missing REF or ALT allele. Cannot convert variant."
)

return f"{chrom}-{pos}-{ref}-{alt}"

# <LICENSE>
# Copyright (C) 2016-2026 VariantValidator Contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
# </LICENSE>
17 changes: 17 additions & 0 deletions VariantValidator/modules/vvMixinCore.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
from VariantValidator.modules import gene2transcripts
from VariantValidator.modules import lovd_api
from VariantValidator.modules import initial_formatting
from VariantValidator.modules import vcf_to_pvcf
from VariantValidator.modules.seq_state_to_expanded_repeat import\
convert_seq_state_to_expanded_repeat
from VariantValidator.modules.hgvs_utils import hgvs_delins_parts_to_hgvs_obj,\
Expand Down Expand Up @@ -115,8 +116,14 @@ def validate(self,
# Turn each variant into a dictionary. The dictionary will be compiled during validation
self.batch_list = []
for queries in batch_queries:
# try:
# queries = vcf_to_pvcf.vcf_to_shorthand(queries)
# except vcf_to_pvcf.VcfConversionError as e:
# logger.info(f"Cannot convert {queries} into PVCF format {e}")
# pass
if isinstance(queries, int):
queries = str(queries)
queries = str(queries)
queries = queries.strip()
query = Variant(queries)
self.batch_list.append(query)
Expand Down Expand Up @@ -259,6 +266,16 @@ def validate(self,
logger.warning(error)
continue

# VCF line handling - Note: handling csv brings too many issues, so stick to tabs tsv
if "\t" in my_variant.quibble and not re.search(r"[gcrnmo]\.", my_variant.quibble):
try:
my_variant.quibble = vcf_to_pvcf.vcf_to_shorthand(my_variant.quibble)
my_variant.warnings.append(f"VcfConversionWarning: VCF line identified and converted "
f"to {my_variant.quibble}")
except vcf_to_pvcf.VcfConversionError as e:
logger.info(f"Cannot convert {my_variant.quibble} into PVCF format {e}")
continue

# Remove whitespace and quotes
my_variant.remove_whitespace()
my_variant.remove_quotes()
Expand Down
7 changes: 7 additions & 0 deletions tests/test_inputs.py
Original file line number Diff line number Diff line change
Expand Up @@ -31255,6 +31255,13 @@ def polyadenylation_b(self):
results = self.vv.validate('2:g.46707878_46707879insGCAGCGGGAGCG', 'GRCh37', 'mane_select').format_as_dict(test=True)
assert 'NM_001145051.2:c.457_458insGGGAGCGGCAGC' in results

def cnv_del_vcf(self):
# Test that it fails for genome mismatch
results = self.vv.validate('chr1\t1000000\t.\tN\t<DEL>\t.\tPASS\tSVTYPE=DEL;END=1005000', 'GRCh38', 'all').format_as_dict(test=True)
assert 'intergenic_variant_1' in results
assert results["intergenic_variant_1"]["primary_assembly_loci"]["grch38"][
"hgvs_genomic_description"] == "NC_000001.11:g.1000000_1005000del"


# <LICENSE>
# Copyright (C) 2016-2026 VariantValidator Contributors
Expand Down
11 changes: 11 additions & 0 deletions tests/test_variantformatter_inputs.py
Original file line number Diff line number Diff line change
Expand Up @@ -6962,6 +6962,17 @@ def test_issue_744b(self):
}
}

def test_vcf_line_variants_del(self):
results = VariantFormatter.simpleVariantFormatter.format('chr1\t1000000\t.\tN\t<DEL>\t.\tPASS\tSVTYPE=DEL;END=1005000',
'GRCh38', 'all', None, False, True, testing=True)
print(results)
assert 'chr1:1000000_1005000del' in results.keys()

def test_vcf_line_variants_inv(self):
results = VariantFormatter.simpleVariantFormatter.format('chr1\t1000000\t.\tN\t<INV>\t.\tPASS\tSVTYPE=INV;END=1005000',
'GRCh38', 'all', None, False, True, testing=True)
print(results)
assert 'chr1:1000000_1005000inv' in results.keys()


# <LICENSE>
Expand Down
121 changes: 121 additions & 0 deletions tests/test_vcf_line_conversion.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,121 @@
import pytest

from VariantValidator.modules import vcf_to_pvcf
from VariantValidator.modules.vcf_to_pvcf import VcfConversionError


# ============================================================
# Simple SNVs
# ============================================================

def test_snv_tab_delimited():
line = "chr2\t1500000\t.\tA\tT\t.\tPASS\t."
assert vcf_to_pvcf.vcf_to_shorthand(line) == "chr2-1500000-A-T"


def test_snv_comma_delimited():
line = "chr2,1500000,.,G,C,.,PASS,."
assert vcf_to_pvcf.vcf_to_shorthand(line) == "chr2-1500000-G-C"


# ============================================================
# Sequence-based indels (standard VCF representation)
# ============================================================

def test_sequence_deletion():
# deletion of "T" at position 1500001
line = "chr3\t1500000\t.\tAT\tA\t.\tPASS\t."
assert vcf_to_pvcf.vcf_to_shorthand(line) == "chr3-1500000-AT-A"


def test_sequence_insertion():
# insertion of "T" after A
line = "chr3\t1500000\t.\tA\tAT\t.\tPASS\t."
assert vcf_to_pvcf.vcf_to_shorthand(line) == "chr3-1500000-A-AT"


def test_larger_sequence_deletion():
line = "chr5\t2000000\t.\tATGC\tA\t.\tPASS\t."
assert vcf_to_pvcf.vcf_to_shorthand(line) == "chr5-2000000-ATGC-A"


# ============================================================
# Symbolic structural variants
# ============================================================

def test_symbolic_del_with_end():
line = "chr1\t1000000\t.\tN\t<DEL>\t.\tPASS\tSVTYPE=DEL;END=1005000"
assert vcf_to_pvcf.vcf_to_shorthand(line) == "chr1-1000000-1005000-DEL"


def test_symbolic_dup_with_svlen():
line = "chr1\t2000000\t.\tN\t<DUP>\t.\tPASS\tSVTYPE=DUP;SVLEN=10000"
assert vcf_to_pvcf.vcf_to_shorthand(line) == "chr1-2000000-2010000-DUP"


def test_symbolic_inv_with_end():
line = "chr1\t3000000\t.\tN\t<INV>\t.\tPASS\tSVTYPE=INV;END=3005000"
assert vcf_to_pvcf.vcf_to_shorthand(line) == "chr1-3000000-3005000-INV"


def test_symbolic_dup_with_copy_number():
line = "chr1\t2000000\t.\tN\t<DUP>\t.\tPASS\tSVTYPE=DUP;SVLEN=10000;CN=4"
assert vcf_to_pvcf.vcf_to_shorthand(line) == "chr1-2000000-2010000-DUP[CN4]"


# ============================================================
# Comma-separated structural variant
# ============================================================

def test_comma_separated_structural_variant():
line = "chr4,4000000,.,N,<DEL>,.,PASS,SVTYPE=DEL;END=4005000"
assert vcf_to_pvcf.vcf_to_shorthand(line) == "chr4-4000000-4005000-DEL"


# ============================================================
# Error handling
# ============================================================

def test_header_line_raises():
with pytest.raises(VcfConversionError):
vcf_to_pvcf.vcf_to_shorthand("#CHROM\tPOS\tID\tREF\tALT")


def test_insufficient_columns():
with pytest.raises(VcfConversionError):
vcf_to_pvcf.vcf_to_shorthand("chr1\t100")


def test_invalid_pos():
with pytest.raises(VcfConversionError):
vcf_to_pvcf.vcf_to_shorthand("chr1\tNotAnInt\t.\tA\tT\t.\tPASS\t.")


def test_missing_end_for_symbolic_sv():
with pytest.raises(VcfConversionError):
vcf_to_pvcf.vcf_to_shorthand("chr1\t1000000\t.\tN\t<DEL>\t.\tPASS\tSVTYPE=DEL")


def test_unknown_delimiter():
with pytest.raises(VcfConversionError):
vcf_to_pvcf.vcf_to_shorthand("chr1|1000000|.|A|T")


# <LICENSE>
# Copyright (C) 2016-2026 VariantValidator Contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
# </LICENSE>