| Previous changeset 11:cb443c64ebad (2015-12-16) Next changeset 13:2e69ce9dca65 (2015-12-16) |
|
Commit message:
Deleted selected files |
|
removed:
vcfs2fasta/phe/variant/__init__.py |
| b |
| diff -r cb443c64ebad -r 19c42b3cce4a vcfs2fasta/phe/variant/__init__.py --- a/vcfs2fasta/phe/variant/__init__.py Wed Dec 16 07:25:53 2015 -0500 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 |
| [ |
| b'@@ -1,288 +0,0 @@\n-"""Classes and methods to work with variants and such."""\n-import abc\n-#ulf\n-# from collections import OrderedDict\n-try:\n- from collections import OrderedDict\n-except ImportError:\n- from ordereddict import OrderedDict\n- \n-import logging\n-import pickle\n-\n-from vcf import filters\n-import vcf\n-from vcf.parser import _Filter\n-\n-from phe.metadata import PHEMetaData\n-from phe.variant_filters import make_filters, PHEFilterBase, str_to_filters\n-\n-\n-class VCFTemplate(object):\n- """This is a small hack class for the Template used in generating\n- VCF file."""\n-\n- def __init__(self, vcf_reader):\n- self.infos = vcf_reader.infos\n- self.formats = vcf_reader.formats\n- self.filters = vcf_reader.filters\n- self.alts = vcf_reader.alts\n- self.contigs = vcf_reader.contigs\n- self.metadata = vcf_reader.metadata\n- self._column_headers = vcf_reader._column_headers\n- self.samples = vcf_reader.samples\n-\n-class VariantSet(object):\n- """A convenient representation of set of variants.\n- TODO: Implement iterator and generator for the variant set.\n- """\n-\n- _reader = None\n-\n- def __init__(self, vcf_in, filters=None):\n- """Constructor of variant set.\n-\n- Parameters:\n- -----------\n- vcf_in: str\n- Path to the VCF file for loading information.\n- filters: str or dict, optional\n- Dictionary or string of the filter:threshold key value pairs.\n- """\n- self.vcf_in = vcf_in\n- self._reader = vcf.Reader(filename=vcf_in)\n- self.out_template = VCFTemplate(self._reader)\n-\n- self.filters = []\n- if filters is not None:\n- if isinstance(filters, str):\n- self.filters = str_to_filters(filters)\n- elif isinstance(filters, dict):\n- self.filters = make_filters(config=filters)\n- elif isinstance(filters, list):\n- self.filters = filters\n- else:\n- logging.warn("Could not create filters from %s", filters)\n- else:\n- reader = vcf.Reader(filename=self.vcf_in)\n- filters = {}\n- for filter_id in reader.filters:\n- filters.update(PHEFilterBase.decode(filter_id))\n-\n- if filters:\n- self.filters = make_filters(config=filters)\n-\n- self.variants = []\n-\n- def filter_variants(self, keep_only_snps=True):\n- """Create a variant """\n-\n- if self._reader is None:\n- # Create a reader class from input VCF.\n- self._reader = vcf.Reader(filename=self.vcf_in)\n-\n- # get list of existing filters.\n- existing_filters = {}\n- removed_filters = []\n-\n- for filter_id in self._reader.filters:\n- conf = PHEFilterBase.decode(filter_id)\n- tuple(conf.keys())\n- existing_filters.update({tuple(conf.keys()):filter_id})\n-\n- # Add each filter we are going to use to the record.\n- # This is needed for writing out proper #FILTER header in VCF.\n- for record_filter in self.filters:\n- # We know that each filter has short description method.\n- short_doc = record_filter.short_desc()\n- short_doc = short_doc.split(\'\\n\')[0].lstrip()\n-\n- filter_name = PHEFilterBase.decode(record_filter.filter_name())\n-\n- # Check if the sample has been filtered for this type of filter\n- # in the past. If so remove is, because it is going to be refiltered.\n- if tuple(filter_name) in existing_filters:\n- logging.info("Removing existing filter: %s", existing_filters[tuple(filter_name)])\n- removed_filters.append(existing_filters[tuple(filter_name)])\n- del self._reader.filters[existing_filters[tuple(filter_name)]]\n-\n- self._reader.filters[record_filter.filter_name()] = _Filter(record_filter.filter_name(), short_doc)\n-\n- '..b' (default: False).\n-\n- Returns:\n- int:\n- Number of records written.\n- """\n- written_variants = 0\n- with open(vcf_out, "w") as out_vcf:\n- writer = vcf.Writer(out_vcf, self.out_template)\n- for record in self.variants:\n-\n- if only_snps and not record.is_snp:\n- continue\n-\n- if only_good and record.FILTER != "PASS" or record.FILTER is None:\n- continue\n-\n- writer.write_record(record)\n- written_variants += 1\n-\n- return written_variants\n-\n- def _write_bad_variants(self, vcf_out):\n- """**PRIVATE:** Write only those records that **haven\'t** passed."""\n- written_variants = 0\n- with open(vcf_out, "w") as out_vcf:\n- writer = vcf.Writer(out_vcf, self.out_template)\n- for record in self.variants:\n- if record.FILTER != "PASS" and record.FILTER is not None:\n- writer.write_record(record)\n- written_variants += 1\n- return written_variants\n-\n- def serialise(self, out_file):\n- """Save the data in this class to a file for future use/reload.\n-\n- Parameters:\n- -----------\n- out_file: str\n- path to file where the data should be written to.\n-\n- Returns:\n- --------\n- int:\n- Number of variants written.\n- """\n- written_variants = 0\n- with open(out_file, "w") as out_vcf:\n- writer = vcf.Writer(out_vcf, self.out_template)\n- for record in self.variants:\n- writer.write_record(record)\n- written_variants += 1\n-\n- return written_variants\n-\n- def update_filters(self, new_filters):\n- """Update internal filters in the output template."""\n- for new_filter, filter_data in new_filters.items():\n- self.out_template.filters[new_filter] = filter_data\n-\n-\n-class VariantCaller(PHEMetaData):\n- """Abstract class used for access to the implemented variant callers."""\n-\n- __metaclass__ = abc.ABCMeta\n-\n- def __init__(self, cmd_options=None):\n- """Constructor for variant caller.\n-\n- Parameters:\n- -----------\n- cmd_options: str, optional\n- Command options to pass to the variant command.\n- """\n- self.cmd_options = cmd_options\n-\n- super(VariantCaller, self).__init__()\n-\n- @abc.abstractmethod\n- def make_vcf(self, *args, **kwargs):\n- """Create a VCF from **BAM** file.\n-\n- Parameters:\n- -----------\n- ref: str\n- Path to the reference file.\n- bam: str\n- Path to the indexed **BAM** file for calling variants.\n- vcf_file: str\n- path to the VCF file where data will be written to.\n-\n- Returns:\n- --------\n- bool:\n- True if variant calling was successful, False otherwise.\n- """\n- raise NotImplementedError("make_vcf is not implemented yet.")\n-\n- @abc.abstractmethod\n- def create_aux_files(self, ref):\n- """Create needed (if any) auxiliary files.\n- These files are required for proper functioning of the variant caller.\n- """\n- raise NotImplementedError("create_aux_files is not implemeted.")\n-\n- @abc.abstractmethod\n- def get_info(self, plain=False):\n- """Get information about this variant caller."""\n- raise NotImplementedError("Get info has not been implemented yet."\n- )\n- def get_meta(self):\n- """Get the metadata about this variant caller."""\n- od = self.get_info()\n- od["ID"] = "VariantCaller"\n- return OrderedDict({"PHEVariantMetaData": [od]})\n-\n- @abc.abstractmethod\n- def get_version(self):\n- """Get the version of the underlying command used."""\n- raise NotImplementedError("Get version has not been implemented yet.")\n' |