-
Notifications
You must be signed in to change notification settings - Fork 29
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Adding d3viz json tree export * Adding ability to output d3viz json. Also added graph transformers to allow for dynamic generation of SEP-style grouping classes. Fixes #696 * reformat
- Loading branch information
Showing
13 changed files
with
729 additions
and
49 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,18 @@ | ||
from dataclasses import dataclass | ||
from typing import Any, Collection | ||
|
||
from oaklib.transformers.ontology_transformer import OntologyTransformer | ||
|
||
|
||
@dataclass | ||
class ChainedOntologyTransformer(OntologyTransformer): | ||
""" | ||
An ontology graph transformer that chains multiple other transformers | ||
""" | ||
|
||
chained_transformers: Collection[OntologyTransformer] | ||
|
||
def transform(self, source_ontology: Any, **kwargs) -> Any: | ||
for transformer in self.chained_transformers: | ||
source_ontology = transformer.transform(source_ontology, **kwargs) | ||
return source_ontology |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,75 @@ | ||
from dataclasses import dataclass | ||
from typing import Collection, Optional | ||
|
||
from oaklib.datamodels.obograph import Graph | ||
from oaklib.datamodels.vocabulary import IS_A | ||
from oaklib.transformers.graph_transformer import GraphTransformer | ||
from oaklib.types import PRED_CURIE | ||
|
||
|
||
@dataclass | ||
class EdgeFilterTransformer(GraphTransformer): | ||
""" | ||
An ontology graph transformer that filters edges | ||
""" | ||
|
||
include_predicates: Optional[Collection[PRED_CURIE]] = None | ||
"""A collection of predicates to include""" | ||
|
||
exclude_predicates: Optional[Collection[PRED_CURIE]] = None | ||
"""A collection of predicates to exclude""" | ||
|
||
filter_function: Optional[callable] = None | ||
"""A function that takes an edge and returns True if it should be included""" | ||
|
||
def transform(self, source_ontology: Graph, **kwargs) -> Graph: | ||
""" | ||
Filters edges from a graph. | ||
Example: | ||
>>> from oaklib import get_adapter | ||
>>> from oaklib.transformers.transformers_factory import get_ontology_transformer | ||
>>> from oaklib.datamodels.vocabulary import IS_A | ||
>>> adapter = get_adapter("tests/input/go-nucleus.obo") | ||
>>> graph = adapter.as_obograph() | ||
>>> transformer = get_ontology_transformer("EdgeFilterTransformer", include_predicates=[IS_A]) | ||
>>> filtered_graph = transformer.transform(graph) | ||
>>> set([e.pred for e in filtered_graph.edges]) | ||
{'is_a'} | ||
:param graph: | ||
:return: | ||
""" | ||
include_predicates = self.include_predicates | ||
exclude_predicates = self.exclude_predicates | ||
|
||
if include_predicates is None and exclude_predicates is None: | ||
return source_ontology | ||
|
||
def _normalize_id(pred: PRED_CURIE) -> PRED_CURIE: | ||
if pred == IS_A: | ||
return "is_a" | ||
else: | ||
return pred | ||
|
||
if include_predicates is not None: | ||
include_predicates = {_normalize_id(pred) for pred in include_predicates} | ||
|
||
if exclude_predicates is not None: | ||
exclude_predicates = {_normalize_id(pred) for pred in exclude_predicates} | ||
|
||
new_edges = [] | ||
for edge in source_ontology.edges: | ||
if include_predicates is not None: | ||
if edge.pred not in include_predicates: | ||
continue | ||
if exclude_predicates is not None: | ||
if edge.pred in exclude_predicates: | ||
continue | ||
if self.filter_function is not None: | ||
if not self.filter_function(edge): | ||
continue | ||
new_edges.append(edge) | ||
new_graph = Graph(id=source_ontology.id, nodes=source_ontology.nodes, edges=new_edges) | ||
return self._post_process(new_graph) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,43 @@ | ||
from dataclasses import dataclass | ||
|
||
from oaklib.datamodels.obograph import Graph | ||
from oaklib.transformers.ontology_transformer import OntologyTransformer | ||
|
||
|
||
@dataclass | ||
class GraphTransformer(OntologyTransformer): | ||
""" | ||
An ontology transformer that operates on a graph | ||
""" | ||
|
||
remove_dangling_edges: bool = False | ||
"""If true, removes edges that point to nodes that are not in the graph""" | ||
|
||
def transform(self, source_ontology: Graph, **kwargs) -> Graph: | ||
""" | ||
Transforms a graph into an ontology | ||
:param graph: | ||
:return: | ||
""" | ||
raise NotImplementedError | ||
|
||
def apply_remove_dangling_edges(self, graph: Graph): | ||
""" | ||
Removes edges that point to nodes that are not in the graph. | ||
:param graph: | ||
:return: | ||
""" | ||
node_ids = {n.id for n in graph.nodes} | ||
new_edges = [] | ||
for edge in graph.edges: | ||
if edge.sub in node_ids and edge.obj in node_ids: | ||
new_edges.append(edge) | ||
return Graph(id=graph.id, nodes=graph.nodes, edges=new_edges) | ||
|
||
def _post_process(self, graph: Graph): | ||
if self.remove_dangling_edges: | ||
return self.apply_remove_dangling_edges(graph) | ||
else: | ||
return graph |
Oops, something went wrong.