Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:return: Hail MatrixTable resource
"""
return BlockMatrix.read(self.path)
def import_resource(self, overwrite: bool = True, **kwargs) -> None:
"""
Imports the BlockMatrixResource using its import_func and writes it in its path.
:param overwrite: If ``True``, overwrite an existing file at the destination.
:param kwargs: Any additional parameters to be passed to BlockMatrix.write
:return: Nothing
"""
self.import_func(**self.import_args).write(self.path, overwrite=False, **kwargs)
class BaseVersionedResource(BaseResource, ABC):
"""
Abstract class for a versioned resource
The `path`/`import_args` attributes of the versioned resource are those of the default version of the resource.
In addition, all versions of the resource are stored in the `versions` attribute.
:param default_version: The default version of this resource (must to be in the `versions` dict)
:param versions: A dict of version name -> resource.
"""
def __init__(self, default_version: str, versions: Dict[str, BaseResource]):
if type(self) is BaseVersionedResource:
raise TypeError("Can't instantiate abstract class BaseVersionedResource")
if default_version not in versions:
raise KeyError(
if self.import_args is not None:
attr_str.append(f'import_args={self.import_args}')
return f'{self.__class__.__name__}({",".join(attr_str)})'
@abstractmethod
def import_resource(self, overwrite: bool = True, **kwargs) -> None:
"""
Abstract method to import the resource using its import_func and writes it in its path.
:param overwrite: If ``True``, overwrite an existing file at the destination.
:param kwargs: Any other parameters to be passed to the underlying hail write function (acceptable parameters depend on specific resource types)
"""
pass
class TableResource(BaseResource):
"""
A Hail Table resource
:param path: The Table path (typically ending in .ht)
:param import_args: Any sources that are required for the import and need to be kept track of and/or passed to the import_func (e.g. .vcf path for an imported VCF)
:param import_func: A function used to import the Table. `import_func` will be passed the `import_args` dictionary as kwargs.
"""
def __init__(
self,
path: Optional[str] = None,
import_args: Optional[Dict[str, Any]] = None,
import_func: Optional[Callable[..., hl.Table]] = None
):
super().__init__(
path=path,
return self.import_func(**self.import_args)
else:
return hl.read_matrix_table(self.path)
def import_resource(self, overwrite: bool = True, **kwargs) -> None:
"""
Imports the MatrixTable resource using its import_func and writes it in its path.
:param overwrite: If set, existing file(s) will be overwritten
:param kwargs: Any other parameters to be passed to hl.MatrixTable.write
:return: Nothing
"""
self.import_func(**self.import_args).write(self.path, overwrite=overwrite, **kwargs)
class PedigreeResource(BaseResource):
"""
A pedigree resource
:param path: The Pedigree path (typically ending in .fam or .ped)
:param import_args: Any sources that are required for the import and need to be kept track of and/or passed to the import_func (e.g. .vcf path for an imported VCF)
:param import_func: A function used to import the Pedigree. `import_func` will be passed the `import_args` dictionary as kwargs.
:param quant_pheno: If ``True``, phenotype is interpreted as quantitative.
:param delimiter: Field delimiter regex.
:param missing: The string used to denote missing values. For case-control, 0, -9, and non-numeric are also treated as missing.
"""
def __init__(
self,
path: Optional[str] = None,
import_args: Optional[Dict[str, Any]] = None,
import_func: Optional[Callable[..., hl.Pedigree]] = None,
module, lambda x: isinstance(x, BaseResource)
):
def import_resource(self, overwrite: bool = True, **kwargs) -> None:
"""
Imports the Pedigree resource using its import_func and writes it in its path.
:param overwrite: If set, existing file(s) will be overwritten. IMPORTANT: Currently there is no implementation of this method when `overwrite` is set the `False`
:param kwargs: Any other parameters to be passed to hl.Pedigree.write
:return: Nothing
"""
if not overwrite:
raise NotImplementedError
self.import_func(**self.import_args).write(self.path)
class BlockMatrixResource(BaseResource):
"""
A Hail BlockMatrix resource
:param path: The BlockMatrix path (typically ending in .bm)
:param import_args: Any sources that are required for the import and need to be kept track of and/or passed to the import_func.
:param import_func: A function used to import the BlockMatrix. `import_func` will be passed the `import_args` dictionary as kwargs.
"""
def __init__(
self,
path: Optional[str] = None,
import_args: Optional[Dict[str, Any]] = None,
import_func: Optional[Callable[..., BlockMatrix]] = None
):
super().__init__(
path=path,
def import_resource(self, overwrite: bool = True, **kwargs) -> None:
"""
Imports the TableResource using its import_func and writes it in its path.
:param overwrite: If ``True``, overwrite an existing file at the destination.
:param kwargs: Any other parameters to be passed to hl.Table.write
:return: Nothing
"""
self.import_func(**self.import_args).write(
self.path,
overwrite=overwrite,
**kwargs
)
class MatrixTableResource(BaseResource):
"""
A Hail MatrixTable resource
:param path: The MatrixTable path (typically ending in .mt)
:param import_args: Any sources that are required for the import and need to be kept track of and/or passed to the import_func (e.g. .vcf path for an imported VCF)
:param import_func: A function used to import the MatrixTable. `import_func` will be passed the `import_args` dictionary as kwargs.
"""
def __init__(
self,
path: Optional[str] = None,
import_args: Optional[Dict[str, Any]] = None,
import_func: Optional[Callable[..., hl.MatrixTable]] = None
):
super().__init__(
path=path,