Move and rename VcfParserConstants to vcf_header_io. (googlegenomics#176)

Author: arostamianfar

gcp_variant_transforms/beam_io/vcf_header_io.py

@@ -28,8 +28,27 @@
 
 from gcp_variant_transforms.beam_io import vcfio
 
-_all__ = ['VcfHeader', 'VcfHeaderSource', 'ReadAllVcfHeaders',
- 'ReadVcfHeaders', 'WriteVcfHeaders']
+
+class VcfHeaderFieldTypeConstants(object):
+ """Constants for types from VCF header."""
+ FLOAT = 'Float'
+ INTEGER = 'Integer'
+ STRING = 'String'
+ FLAG = 'Flag'
+ CHARACTER = 'Character'
+ STRING = 'String'
+
+
+class VcfParserHeaderKeyConstants(object):
+ """Constants for header fields from the parser (currently PyVCF)."""
+ ID = 'id'
+ NUM = 'num'
+ TYPE = 'type'
+ DESC = 'desc'
+ SOURCE = 'source'
+ VERSION = 'version'
+ LENGTH = 'length'
+
 
 class VcfHeader(object):
  """Container for header data."""
@@ -310,19 +329,19 @@ def _format_header_key_value(self, key, value):
  return '{}={}'.format(key, value)
 
  def _format_header_key(self, key):
- if key == 'id':
+ if key == VcfParserHeaderKeyConstants.ID:
  return _HeaderFieldKeyConstants.ID
- elif key == 'num':
+ elif key == VcfParserHeaderKeyConstants.NUM:
  return _HeaderFieldKeyConstants.NUMBER
- elif key == 'desc':
+ elif key == VcfParserHeaderKeyConstants.DESC:
  return _HeaderFieldKeyConstants.DESCRIPTION
- elif key == 'type':
+ elif key == VcfParserHeaderKeyConstants.TYPE:
  return _HeaderFieldKeyConstants.TYPE
- elif key == 'source':
+ elif key == VcfParserHeaderKeyConstants.SOURCE:
  return _HeaderFieldKeyConstants.SOURCE
- elif key == 'version':
+ elif key == VcfParserHeaderKeyConstants.VERSION:
  return _HeaderFieldKeyConstants.VERSION
- elif key == 'length':
+ elif key == VcfParserHeaderKeyConstants.LENGTH:
  return _HeaderFieldKeyConstants.LENGTH
  else:
  raise ValueError('Invalid VCF header key {}.'.format(key))

gcp_variant_transforms/libs/vcf_field_conflict_resolver.py

@@ -16,22 +16,11 @@
 
 import vcf
 
+from gcp_variant_transforms.beam_io import vcf_header_io
 from gcp_variant_transforms.libs import bigquery_schema_descriptor # pylint: disable=unused-import
 from gcp_variant_transforms.libs import bigquery_util
 
 
-class VcfParserConstants(object):
- """Constants for type and number from VCF parser."""
- FLOAT = 'Float'
- INTEGER = 'Integer'
- STRING = 'String'
- FLAG = 'Flag'
- CHARACTER = 'Character'
- NUM = 'num'
- STRING = 'String'
- TYPE = 'type'
-
-
 class FieldConflictResolver(object):
  """A class for resolving all VCF field related mismatches.
 
@@ -114,9 +103,9 @@ def resolve_attribute_conflict(self, attribute_name, first_attribute_value,
  Raises:
  ValueError: if the conflict cannot be resolved.
  """
- if attribute_name == VcfParserConstants.TYPE:
+ if attribute_name == vcf_header_io.VcfParserHeaderKeyConstants.TYPE:
  return self._resolve_type(first_attribute_value, second_attribute_value)
- elif attribute_name == VcfParserConstants.NUM:
+ elif attribute_name == vcf_header_io.VcfParserHeaderKeyConstants.NUM:
  return self._resolve_number(first_attribute_value, second_attribute_value)
  else:
  # We only care about conflicts in 'num' and 'type' attributes.
@@ -125,13 +114,14 @@ def resolve_attribute_conflict(self, attribute_name, first_attribute_value,
  return first_attribute_value
 
  def _resolve_type(self, first, second):
+ type_constants = vcf_header_io.VcfHeaderFieldTypeConstants
  if first == second:
  return first
- elif (first in (VcfParserConstants.INTEGER, VcfParserConstants.FLOAT) and
- second in (VcfParserConstants.INTEGER, VcfParserConstants.FLOAT)):
- return VcfParserConstants.FLOAT
+ elif (first in (type_constants.INTEGER, type_constants.FLOAT) and
+ second in (type_constants.INTEGER, type_constants.FLOAT)):
+ return type_constants.FLOAT
  elif self._resolve_always:
- return VcfParserConstants.STRING
+ return type_constants.STRING
  else:
  raise ValueError('Incompatible values cannot be resolved: '
  '{}, {}'.format(first, second))

gcp_variant_transforms/libs/vcf_field_conflict_resolver_test.py

@@ -20,16 +20,18 @@
 
 import vcf
 
+from gcp_variant_transforms.beam_io.vcf_header_io import VcfHeaderFieldTypeConstants
+from gcp_variant_transforms.beam_io.vcf_header_io import VcfParserHeaderKeyConstants
 from gcp_variant_transforms.libs import bigquery_schema_descriptor
 from gcp_variant_transforms.libs import vcf_field_conflict_resolver
 from gcp_variant_transforms.libs.bigquery_util import TableFieldConstants
-from gcp_variant_transforms.libs.vcf_field_conflict_resolver import VcfParserConstants
 
 
 SchemaTestConfig = namedtuple('SchemaTestConfig',
  ['schema_type', 'schema_mode', 'field_data',
  'expected_resolved_field_data'])
 
+
 class ConflictResolverTest(unittest.TestCase):
  """Test case for :class:`FieldConflictResolver`."""
 
@@ -169,79 +171,83 @@ def test_resolving_schema_conflict_type_and_number(self):
  def test_resolving_attribute_conflict_type(self):
  self.assertEqual(
  self._resolver.resolve_attribute_conflict(
- VcfParserConstants.TYPE,
- VcfParserConstants.INTEGER,
- VcfParserConstants.FLOAT),
- VcfParserConstants.FLOAT)
+ VcfParserHeaderKeyConstants.TYPE,
+ VcfHeaderFieldTypeConstants.INTEGER,
+ VcfHeaderFieldTypeConstants.FLOAT),
+ VcfHeaderFieldTypeConstants.FLOAT)
  with self.assertRaises(ValueError):
  self._resolver.resolve_attribute_conflict(
- VcfParserConstants.TYPE,
- VcfParserConstants.INTEGER,
- VcfParserConstants.STRING)
+ VcfParserHeaderKeyConstants.TYPE,
+ VcfHeaderFieldTypeConstants.INTEGER,
+ VcfHeaderFieldTypeConstants.STRING)
  self.fail('Should raise exception for unresolvable types')
 
  def test_resolving_attribute_conflict_number(self):
  self.assertEqual(
  self._resolver.resolve_attribute_conflict(
- VcfParserConstants.NUM, 2, 3),
+ VcfParserHeaderKeyConstants.NUM, 2, 3),
  None)
  self.assertEqual(
  self._resolver.resolve_attribute_conflict(
- VcfParserConstants.NUM, 2, None),
+ VcfParserHeaderKeyConstants.NUM, 2, None),
  None)
  # Unresolvable cases.
  for i in [0, 1]:
  for j in [self._field_count('R'), self._field_count('G'),
  self._field_count('A'), 2, None]:
  with self.assertRaises(ValueError):
  self._resolver.resolve_attribute_conflict(
- VcfParserConstants.NUM, i, j)
+ VcfParserHeaderKeyConstants.NUM, i, j)
  self.fail(
  'Should raise exception for unresolvable number: %d vs %d'%(i, j))
 
  def test_resolving_attribute_conflict_in_number_allele(self):
  self.assertEqual(
  self._resolver_allele.resolve_attribute_conflict(
- VcfParserConstants.NUM, 2, 3),
+ VcfParserHeaderKeyConstants.NUM, 2, 3),
  None)
  self.assertEqual(
  self._resolver_allele.resolve_attribute_conflict(
- VcfParserConstants.NUM, 2, None),
+ VcfParserHeaderKeyConstants.NUM, 2, None),
  None)
  # Unresolvable cases.
  for i in [self._field_count('A')]:
  for j in [self._field_count('R'), self._field_count('G'), 0, 1, 2, None]:
  with self.assertRaises(ValueError):
  self._resolver_allele.resolve_attribute_conflict(
- VcfParserConstants.NUM, i, j)
+ VcfParserHeaderKeyConstants.NUM, i, j)
  self.fail(
  'Should raise exception for unresolvable number: %d vs %d'%(i, j))
 
  def test_resolving_all_field_definition_conflict_in_type(self):
  self.assertEqual(
  self._resolver_always.resolve_attribute_conflict(
- VcfParserConstants.TYPE, VcfParserConstants.INTEGER,
- VcfParserConstants.FLOAT),
- VcfParserConstants.FLOAT)
- for i in [VcfParserConstants.FLOAT, VcfParserConstants.INTEGER,
- VcfParserConstants.STRING, VcfParserConstants.CHARACTER]:
- for j in [VcfParserConstants.FLAG, VcfParserConstants.STRING]:
+ VcfParserHeaderKeyConstants.TYPE,
+ VcfHeaderFieldTypeConstants.INTEGER,
+ VcfHeaderFieldTypeConstants.FLOAT),
+ VcfHeaderFieldTypeConstants.FLOAT)
+ for i in [VcfHeaderFieldTypeConstants.FLOAT,
+ VcfHeaderFieldTypeConstants.INTEGER,
+ VcfHeaderFieldTypeConstants.STRING,
+ VcfHeaderFieldTypeConstants.CHARACTER]:
+ for j in [VcfHeaderFieldTypeConstants.FLAG,
+ VcfHeaderFieldTypeConstants.STRING]:
  self.assertEqual(
  self._resolver_always.resolve_attribute_conflict(
- VcfParserConstants.TYPE, i, j),
- VcfParserConstants.STRING)
+ VcfParserHeaderKeyConstants.TYPE, i, j),
+ VcfHeaderFieldTypeConstants.STRING)
 
  def test_resolving_all_field_definition_conflict_in_number(self):
  self.assertEqual(
  self._resolver_always.resolve_attribute_conflict(
- VcfParserConstants.NUM, 2, 3), None)
+ VcfParserHeaderKeyConstants.NUM, 2, 3), None)
  self.assertEqual(
  self._resolver_always.resolve_attribute_conflict(
- VcfParserConstants.NUM, 2, None), None)
+ VcfParserHeaderKeyConstants.NUM, 2, None), None)
 
  for i in [0, 1]:
  for j in [self._field_count('R'), self._field_count('G'),
  self._field_count('A'), 2, None]:
  self.assertEqual(
  self._resolver_always.resolve_attribute_conflict(
- VcfParserConstants.NUM, i, j), None)
+ VcfParserHeaderKeyConstants.NUM, i, j), None)

gcp_variant_transforms/transforms/infer_undefined_headers.py

@@ -23,7 +23,6 @@
 from vcf.parser import field_counts
 
 from gcp_variant_transforms.beam_io import vcf_header_io
-from gcp_variant_transforms.libs import vcf_field_conflict_resolver
 from gcp_variant_transforms.transforms import merge_headers
 
 
@@ -56,16 +55,16 @@ def _get_field_type(self, field_value):
  """
  if isinstance(field_value, list):
  return (self._get_field_type(field_value[0]) if field_value else
- vcf_field_conflict_resolver.VcfParserConstants.STRING)
+ vcf_header_io.VcfHeaderFieldTypeConstants.STRING)
 
  if isinstance(field_value, bool):
- return vcf_field_conflict_resolver.VcfParserConstants.FLAG
+ return vcf_header_io.VcfHeaderFieldTypeConstants.FLAG
  elif isinstance(field_value, int):
- return vcf_field_conflict_resolver.VcfParserConstants.INTEGER
+ return vcf_header_io.VcfHeaderFieldTypeConstants.INTEGER
  elif isinstance(field_value, float):
- return vcf_field_conflict_resolver.VcfParserConstants.FLOAT
+ return vcf_header_io.VcfHeaderFieldTypeConstants.FLOAT
  else:
- return vcf_field_conflict_resolver.VcfParserConstants.STRING
+ return vcf_header_io.VcfHeaderFieldTypeConstants.STRING
 
  def _infer_undefined_info_fields(self, variant, defined_headers):
  """Returns info fields not defined in the headers.

gcp_variant_transforms/transforms/merge_header_definitions.py

@@ -19,13 +19,14 @@
 from typing import Dict, List # pylint: disable=unused-import
 
 import apache_beam as beam
+from gcp_variant_transforms.beam_io import vcf_header_io
 from gcp_variant_transforms.beam_io.vcf_header_io import VcfHeader # pylint: disable=unused-import
-from gcp_variant_transforms.libs.vcf_field_conflict_resolver import VcfParserConstants
 
 # ``Definition`` cherry-picks the attributes from vcf header definitions that
 # are critical for checking field compatibilities across VCF files.
-Definition = namedtuple('Definition', [VcfParserConstants.NUM,
- VcfParserConstants.TYPE])
+Definition = namedtuple('Definition',
+ [vcf_header_io.VcfParserHeaderKeyConstants.NUM,
+ vcf_header_io.VcfParserHeaderKeyConstants.TYPE])
 
 
 class VcfHeaderDefinitions(object):
@@ -43,12 +44,14 @@ def __init__(self, vcf_header=None):
  if not vcf_header:
  return
  for key, val in vcf_header.infos.iteritems():
- definition = Definition(val[VcfParserConstants.NUM],
- val[VcfParserConstants.TYPE])
+ definition = Definition(
+ val[vcf_header_io.VcfParserHeaderKeyConstants.NUM],
+ val[vcf_header_io.VcfParserHeaderKeyConstants.TYPE])
  self._infos[key][definition] = [vcf_header.file_name]
  for key, val in vcf_header.formats.iteritems():
- definition = Definition(val[VcfParserConstants.NUM],
- val[VcfParserConstants.TYPE])
+ definition = Definition(
+ val[vcf_header_io.VcfParserHeaderKeyConstants.NUM],
+ val[vcf_header_io.VcfParserHeaderKeyConstants.TYPE])
  self._formats[key][definition] = [vcf_header.file_name]
 
  def __eq__(self, other):