fusor.harvester

Harvester methods for output from different fusion callers

class fusor.harvester.ArribaHarvester(translator, assembly)

Class for harvesting Arriba data

column_rename: ClassVar[dict[str, str]] = {'#gene1': 'gene1', 'reading_frame': 'rf', 'strand1(gene/fusion)': 'strand1', 'strand2(gene/fusion)': 'strand2', 'type': 'event_type'}

coordinate_type: CoordinateType = 'residue'

delimeter: str = '\t'

fusion_caller

alias of Arriba

translator_method_name: ClassVar[str] = 'from_arriba'

class fusor.harvester.CIVICHarvester(translator, update_cache=False, update_from_remote=True, local_cache_path=civic.LOCAL_CACHE_PATH)

Class for harvesting CIViC Fusion objects

__init__(translator, update_cache=False, update_from_remote=True, local_cache_path=civic.LOCAL_CACHE_PATH)

Initialize CivicHarvester class.

Parameters:

• translator (Translator) – A Translator class instance

• update_cache (bool) – True if civicpy cache should be updated. Note this will take several minutes. False if to use local cache.

• update_from_remote (bool) – If set to True, civicpy.update_cache will first download the remote cache designated by REMOTE_CACHE_URL, store it to LOCAL_CACHE_PATH, and then load the downloaded cache into memory. This parameter defaults to True.

• local_cache_path (str) – A filepath destination for the retrieved remote cache. This parameter defaults to LOCAL_CACHE_PATH from civicpy.

async load_records()

Extract data from CIVIC fusion objects

:return A list of CIVIC objects

Return type:

list[CIVIC]

class fusor.harvester.CiceroHarvester(translator, assembly)

Class for harvesting Cicero data

column_rename: ClassVar[dict[str, str]] = {'chrA': 'chr_5prime', 'chrB': 'chr_3prime', 'coverageA': 'coverage_5prime', 'coverageB': 'coverage_3prime', 'geneA': 'gene_5prime', 'geneB': 'gene_3prime', 'posA': 'pos_5prime', 'posB': 'pos_3prime', 'readsA': 'reads_5prime', 'readsB': 'reads_3prime', 'type': 'event_type'}

coordinate_type: CoordinateType = 'residue'

delimeter: str = '\t'

fusion_caller

alias of Cicero

translator_method_name: ClassVar[str] = 'from_cicero'

class fusor.harvester.EnFusionHarvester(translator, assembly)

Class for harvesting EnFusion data

column_rename: ClassVar[dict[str, str]] = {'Break1': 'break_5prime', 'Break2': 'break_3prime', 'Chr1': 'chr_5prime', 'Chr2': 'chr_3prime', 'FusionJunctionSequence': 'fusion_junction_sequence', 'Gene1': 'gene_5prime', 'Gene2': 'gene_3prime'}

coordinate_type: CoordinateType = 'residue'

delimeter: str = '\t'

fusion_caller

alias of EnFusion

translator_method_name: ClassVar[str] = 'from_enfusion'

class fusor.harvester.FusionCallerHarvester(translator, assembly)

ABC for fusion caller harvesters

__init__(translator, assembly)

Initialize FusionCallerHarvester

Parameters:

• translator (Translator) – A Translator object

• assembly (Assembly) – The assembly that the coordinates are described on

column_rename: dict

coordinate_type: CoordinateType

delimeter: str

fusion_caller: FusionCaller

async load_records(fusion_path)

Convert rows of fusion caller output to Pydantic classes

Parameters:

fusion_path (Path) – The path to the fusions file

Raises:

ValueError – if the file does not exist at the specified path

Return type:

list[FusionCaller]

Returns:

A list of translated fusions, represented as Pydantic objects

translator: Translator

translator_method_name: ClassVar[str]

class fusor.harvester.FusionCatcherHarvester(translator, assembly)

Class for harvesting FusionCatcher data

column_rename: ClassVar[dict[str, str]] = {'Fusion_point_for_gene_1(5end_fusion_partner)': 'five_prime_fusion_point', 'Fusion_point_for_gene_2(3end_fusion_partner)': 'three_prime_fusion_point', 'Fusion_sequence': 'fusion_sequence', 'Gene_1_symbol(5end_fusion_partner)': 'five_prime_partner', 'Gene_2_symbol(3end_fusion_partner)': 'three_prime_partner', 'Predicted_effect': 'predicted_effect', 'Spanning_pairs': 'spanning_reads', 'Spanning_unique_reads': 'spanning_unique_reads'}

coordinate_type: CoordinateType = 'residue'

delimeter: str = '\t'

fusion_caller

alias of FusionCatcher

async translator_method(fusion_catcher, coordinate_type, rb)

Parse FusionCatcher output to create AssayedFusion object

Parameters:

• fusion_catcher (FusionCatcher) – Output from FusionCatcher caller

• coordinate_type (CoordinateType) – If the coordinate is inter-residue or residue

• rb (Assembly) – The reference build used to call the fusion

Return type:

AssayedFusion

Returns:

An AssayedFusion object, if construction is successful

translator_method_name: ClassVar[str] = 'from_fusion_catcher'

class fusor.harvester.GenieHarvester(translator, assembly)

Class for harvesting Genie data

column_rename: ClassVar[dict[str, str]] = {'Annotation': 'annot', 'Site1_Chromosome': 'site1_chrom', 'Site1_Hugo_Symbol': 'site1_hugo', 'Site1_Position': 'site1_pos', 'Site2_Chromosome': 'site2_chrom', 'Site2_Effect_On_Frame': 'reading_frame', 'Site2_Hugo_Symbol': 'site2_hugo', 'Site2_Position': 'site2_pos'}

coordinate_type: CoordinateType = 'residue'

delimeter: str = '\t'

fusion_caller

alias of Genie

translator_method_name: ClassVar[str] = 'from_genie'

class fusor.harvester.JAFFAHarvester(translator, assembly)

Class for harvesting JAFFA data

column_rename: ClassVar[dict[str, str]] = {'fusion genes': 'fusion_genes', 'spanning pairs': 'spanning_pairs', 'spanning reads': 'spanning_reads'}

coordinate_type: CoordinateType = 'residue'

delimeter: str = ','

fusion_caller

alias of JAFFA

translator_method_name: ClassVar[str] = 'from_jaffa'

class fusor.harvester.StarFusionHarvester(translator, assembly)

Class for harvesting STAR-Fusion data

column_rename: ClassVar[dict[str, str]] = {'JunctionReadCount': 'junction_read_count', 'LeftBreakpoint': 'left_breakpoint', 'LeftGene': 'left_gene', 'RightBreakpoint': 'right_breakpoint', 'RightGene': 'right_gene', 'SpanningFragCount': 'spanning_frag_count'}

coordinate_type: CoordinateType = 'residue'

delimeter: str = '\t'

fusion_caller

alias of STARFusion

translator_method_name: ClassVar[str] = 'from_star_fusion'