from openvariant.plugins.context import Context
from openvariant.plugins.plugin import Plugin
from pyliftover import LiftOver
import os
[docs]class LiftoverContext(Context):
"""
LiftoverContext class generated by OpenVariant
Attributes
-------
row : dict
Main method to execute data transformation in each row.
field_name : str
Name of the corresponding column that was described on the annotation schema.
file_path : str
Path of the Input file that is being parsed.
"""
def __init__(self, row: dict, field_name: str, file_path: str) -> None:
super().__init__(row, field_name, file_path)
def isHg19(self):
if "POS_HG19" in self.row.keys():
return True
return False
def isHg38(self):
if "POS_HG38" in self.row.keys():
return True
return False
[docs]class LiftoverPlugin(Plugin):
"""
LiftoverPlugin class generated by OpenVariant
Methods
-------
run(context: LiftoverContext)
Main method to execute data transformation in each row.
"""
def __init__(self):
# Initialize lazy load variables
self.LiftOtoHG38 = None
self.LiftOtoHG19 = None
self.noLiftOver_thr = 0
[docs] def run(self, context: LiftoverContext) -> dict:
"""
Data transformation of a single row
Parameters
-------
context : LiftoverContext
Representation of the row to be parsed.
Returns
-------
float or int or str
The value of the field transformed.
"""
if context.isHg19():
if self.LiftOtoHG38 is None: # lazy load
self.LiftOtoHG38 = LiftOver('hg19','hg38')
converter = self.LiftOtoHG38
col_name = 'POS_HG19'
elif context.isHg38():
if self.LiftOtoHG19 is None: # lazy load
self.LiftOtoHG19 = LiftOver('hg38', 'hg19')
converter = self.LiftOtoHG19
col_name = 'POS_HG38'
else:
raise ValueError("Unable to find 'POS_HG19' or 'POS_HG38' values in the row.")
strand = '+' if 'STRAND' not in context.row else context.row['STRAND']
try:
context.row[context.field_name] = converter.convert_coordinate("chr{}".format(context.row['CHROMOSOME']), int(context.row[col_name]) - 1, strand)[0][1]
except IndexError:
if self.noLiftOver_thr < 11:
self.noLiftOver_thr += 1
elif self.noLiftOver_thr == 11:
self.noLiftOver_thr += 1
return ''
except TypeError:
return context.row[col_name]
return str(context.row[context.field_name] + 1)