Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Adding FHIR CodeSystem dumping. #368

Merged
merged 4 commits into from
Nov 30, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ test:

## Compiled

MODELS = ontology_metadata obograph validation_datamodel summary_statistics_datamodel lexical_index mapping_rules_datamodel text_annotator oxo taxon_constraints similarity search_datamodel cross_ontology_diff association class_enrichment value_set_configuration
MODELS = ontology_metadata obograph validation_datamodel summary_statistics_datamodel lexical_index mapping_rules_datamodel text_annotator oxo taxon_constraints similarity search_datamodel cross_ontology_diff association class_enrichment value_set_configuration fhir

pyclasses: $(patsubst %, src/oaklib/datamodels/%.py, $(MODELS))
jsonschema: $(patsubst %, src/oaklib/datamodels/%.schema.json, $(MODELS))
Expand Down
4 changes: 4 additions & 0 deletions src/oaklib/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,7 @@
from oaklib.io.obograph_writer import write_graph
from oaklib.io.streaming_axiom_writer import StreamingAxiomWriter
from oaklib.io.streaming_csv_writer import StreamingCsvWriter
from oaklib.io.streaming_fhir_writer import StreamingFHIRWriter
from oaklib.io.streaming_info_writer import StreamingInfoWriter
from oaklib.io.streaming_json_writer import StreamingJsonWriter
from oaklib.io.streaming_kgcl_writer import StreamingKGCLWriter
Expand Down Expand Up @@ -168,6 +169,7 @@
OWLFUN_FORMAT = "ofn"
NL_FORMAT = "nl"
KGCL_FORMAT = "kgcl"
FHIR_JSON_FORMAT = "fhirjson"
HEATMAP_FORMAT = "heatmap"

ONT_FORMATS = [
Expand All @@ -177,6 +179,7 @@
RDF_FORMAT,
JSON_FORMAT,
YAML_FORMAT,
FHIR_JSON_FORMAT,
CSV_FORMAT,
NL_FORMAT,
]
Expand All @@ -192,6 +195,7 @@
JSONL_FORMAT: StreamingJsonWriter,
YAML_FORMAT: StreamingYamlWriter,
SSSOM_FORMAT: StreamingSssomWriter,
FHIR_JSON_FORMAT: StreamingFHIRWriter,
INFO_FORMAT: StreamingInfoWriter,
NL_FORMAT: StreamingNaturalLanguageWriter,
KGCL_FORMAT: StreamingKGCLWriter,
Expand Down
4 changes: 4 additions & 0 deletions src/oaklib/converters/data_model_converter.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,3 +28,7 @@ def convert(self, source: Any, target: Any = None) -> Any:
:return:
"""
raise NotImplementedError

@abstractmethod
def dump(self, source: Any, target: str = None) -> None:
raise NotImplementedError
3 changes: 3 additions & 0 deletions src/oaklib/converters/logical_definition_flattener.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,9 @@
class LogicalDefinitionFlattener(DataModelConverter):
"""Flattens logical definitions to tuples for use in template libraries."""

def dump(self, source: Any, target: str = None) -> None:
raise NotImplementedError

def convert(
self, source: Union[LogicalDefinitionAxiom, Graph, GraphDocument], target: Any = None
) -> Union[dict, List[dict]]:
Expand Down
135 changes: 135 additions & 0 deletions src/oaklib/converters/obo_graph_to_fhir_converter.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,135 @@
import logging
from dataclasses import dataclass
from typing import Any, Dict, List, Tuple

import rdflib
from linkml_runtime.dumpers import json_dumper

from oaklib.converters.data_model_converter import DataModelConverter
from oaklib.datamodels.fhir import (
CodeSystem,
Coding,
Concept,
ConceptDesignation,
ConceptProperty,
)
from oaklib.datamodels.obograph import Edge, Graph, GraphDocument, Node
from oaklib.datamodels.vocabulary import (
HAS_BROAD_SYNONYM,
HAS_EXACT_SYNONYM,
HAS_NARROW_SYNONYM,
HAS_RELATED_SYNONYM,
)
from oaklib.types import CURIE
from oaklib.utilities.obograph_utils import index_graph_edges_by_subject

TRIPLE = Tuple[rdflib.URIRef, rdflib.URIRef, Any]

DIRECT_PREDICATE_MAP = {
"is_a": "parent",
}

SCOPE_MAP = {
"hasBroadSynonym": HAS_BROAD_SYNONYM,
"hasExactSynonym": HAS_EXACT_SYNONYM,
"hasNarrowSynonym": HAS_NARROW_SYNONYM,
"hasRelatedSynonym": HAS_RELATED_SYNONYM,
}

SCOPE_DISPLAY = {
"hasBroadSynonym": "has broad synonym",
"hasExactSynonym": "has exact synonym",
"hasNarrowSynonym": "has narrow synonym",
"hasRelatedSynonym": "has related synonym",
}


@dataclass
class OboGraphToFHIRConverter(DataModelConverter):
"""Converts from OboGraph to FHIR."""

def dump(self, source: GraphDocument, target: str = None) -> None:
"""
Dump an OBO Graph Document to a FHIR CodeSystem

:param source:
:param target:
:return:
"""
cs = self.convert(source)
json_str = json_dumper.dumps(cs)
if target is None:
print(json_str)
else:
with open(target, "w", encoding="UTF-8") as f:
f.write(json_str)

def convert(self, source: GraphDocument, target: CodeSystem = None) -> CodeSystem:
"""
Convert an OBO Graph Document to a FHIR CodingSystem

:param source:
:param target: if None, one will be created
:return:
"""
if target is None:
target = CodeSystem()
target.resourceType = CodeSystem.__name__
for g in source.graphs:
self._convert_graph(g, target=target)
return target

def code(self, uri: CURIE) -> str:
if not self.curie_converter:
return uri
curie = self.curie_converter.compress(uri)
if curie is None:
return uri
else:
return curie

def _convert_graph(self, source: Graph, target: CodeSystem) -> CodeSystem:
target.id = source.id
edges_by_subject = index_graph_edges_by_subject(source)
logging.info(f"Converting graph to obo: {source.id}, nodes={len(source.nodes)}")
for n in source.nodes:
logging.debug(f"Converting node {n.id}")
self._convert_node(n, index=edges_by_subject, target=target)
return target

def _convert_node(
self, source: Node, index: Dict[CURIE, List[Edge]], target: CodeSystem
) -> Concept:
id = self.code(source.id)
logging.debug(f"Converting node {id} from {source}")
concept = Concept(code=id, display=source.lbl)
target.concept.append(concept)
if source.meta:
self._convert_meta(source, concept)
for e in index.get(source.id, []):
obj = self.code(e.obj)
if e.pred in DIRECT_PREDICATE_MAP:
concept.property.append(
ConceptProperty(code=DIRECT_PREDICATE_MAP[e.pred], valueCode=obj)
)
else:
logging.debug(f"Skipping edge {e}")
return concept

def _convert_meta(self, source: Node, concept: Concept):
meta = source.meta
if meta.definition:
concept.definition = meta.definition.val
for synonym in meta.synonyms:
synonym_pred_code = self.code(synonym.pred)
concept.designation.append(
ConceptDesignation(
# language=synonym.lang,
use=Coding(
system="oio",
code=synonym_pred_code,
display=SCOPE_DISPLAY.get(synonym.pred),
),
value=synonym.val,
)
)
15 changes: 15 additions & 0 deletions src/oaklib/converters/obo_graph_to_obo_format_converter.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,21 @@ def _escape(s: str) -> str:
class OboGraphToOboFormatConverter(DataModelConverter):
"""Converts from OboGraph to OBO Format."""

def dump(self, source: GraphDocument, target: str = None) -> None:
"""
Dump an OBO Graph Document to a FHIR CodeSystem

:param source:
:param target:
:return:
"""
obodoc = self.convert(source)
if target is None:
print(obodoc.dump())
else:
with open(target, "w", encoding="UTF-8") as f:
obodoc.dump(f)

def convert(self, source: GraphDocument, target: OboDocument = None) -> OboDocument:
"""
Convert an OBO Format Document.
Expand Down
14 changes: 14 additions & 0 deletions src/oaklib/converters/obo_graph_to_rdf_owl_converter.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,20 @@
class OboGraphToRdfOwlConverter(DataModelConverter):
"""Converts from OboGraph to OWL layered on RDF."""

def dump(self, source: GraphDocument, target: str = None) -> None:
"""
Dump an OBO Graph Document to a FHIR CodeSystem

:param source:
:param target:
:return:
"""
g = self.convert(source)
if target is None:
print(g.serialize(format="turtle"))
else:
g.serialize(format="turtle", destination=target)

def convert(self, source: GraphDocument, target: rdflib.Graph = None) -> rdflib.Graph:
"""
Convert an OBO GraphDocument.
Expand Down
Loading