import abc
import typing
from hpotk.model import Identified, TermId
from hpotk.ontology import MinimalOntology
from hpotk.constants.hpo.base import PHENOTYPIC_ABNORMALITY
from hpotk.util import validate_instance
from ._model import ValidationResult, ValidationResults, ValidationLevel, RuleValidator
from ._util import SimpleFeature
T = typing.TypeVar('T', SimpleFeature, TermId)
class BaseOntologyRuleValidator(RuleValidator, metaclass=abc.ABCMeta):
def __init__(self, hpo: MinimalOntology):
self._hpo = validate_instance(hpo, MinimalOntology, 'hpo')
def _primary_term_id(self, feature: T) -> typing.Optional[T]:
"""
Update the term ID of the `feature` to the primary term ID in case the current term ID is obsolete.
"""
if isinstance(feature, SimpleFeature):
current_id = self._hpo.get_term(feature.identifier)
if current_id is None:
return None
if current_id.identifier != feature.identifier:
# Update to the primary term ID.
feature.identifier = current_id.identifier
return feature
elif isinstance(feature, TermId):
current = self._hpo.get_term(feature)
return None if current is None else current.identifier
else:
raise ValueError(f'feature must be either a `TermId` or `SimpleFeature` but was {type(feature)}')
[docs]
class AnnotationPropagationValidator(BaseOntologyRuleValidator):
"""
Validator to check that a sequence of terms, when interpreted together, does not violate the annotation propagation
rule.
More formally, the terms must not satisfy the following:
* terms must not contain a present term and its present or excluded ancestor
* terms must not contain an excluded term and its excluded ancestor
Violation of annotation propagation rule produces an :class:`ValidationLevel.ERROR`.
The validator replaces obsolete term IDs with the current term IDs before performing the validation.
:param ontology: HPO represented as :class:`hpotk.ontology.MinimalOntology`.
"""
def __init__(self, ontology: MinimalOntology):
super().__init__(ontology)
[docs]
def validate(self, items: typing.Sequence[typing.Union[Identified, TermId]]) -> ValidationResults:
stateful_features: typing.Collection[SimpleFeature] = {
self._primary_term_id(self._extract_stateful_feature(item))
for item in items
if item is not None
}
results = []
for feature in stateful_features:
if feature.is_present:
# A present feature cannot coexist with a present or excluded ancestor.
for anc in self._hpo.graph.get_ancestors(feature):
if any(anc == sf.identifier for sf in stateful_features):
current_term = self._hpo.get_term(feature.identifier)
term = self._hpo.get_term(anc)
results.append(
ValidationResult(level=ValidationLevel.ERROR,
category='annotation_propagation',
message=f'Terms should not contain both present '
f'{current_term.name} [{current_term.identifier.value}] '
f'and its present or excluded ancestor '
f'{term.name} [{term.identifier.value}]'))
else:
# An excluded feature cannot coexist with an excluded ancestor.
for anc in self._hpo.graph.get_ancestors(feature):
# We only check excluded ancestors here since presence of an ancestor is allowed
# for an excluded feature.
if any(anc == sf.identifier for sf in stateful_features if sf.is_excluded):
current_term = self._hpo.get_term(feature.identifier)
term = self._hpo.get_term(anc)
results.append(
ValidationResult(level=ValidationLevel.ERROR,
category='annotation_propagation',
message=f'Terms should not contain both excluded '
f'{current_term.name} [{current_term.identifier.value}] '
f'and its present or excluded ancestor '
f'{term.name} [{term.identifier.value}]'))
return ValidationResults(results)
[docs]
class PhenotypicAbnormalityValidator(BaseOntologyRuleValidator):
"""
Validator for checking that the term is a phenotypic abnormality
(a descendant of `Phenotypic abnormality <https://hpo.jax.org/app/browse/term/HP:0000118>`_ [HP:0000118]).
Presence of a term that is not a descendant of Phenotypic abnormality is a :class:`ValidationLevel.WARNING`.
The validator replaces obsolete term IDs with the current term IDs before performing the validation.
:param ontology: HPO represented as :class:`hpotk.ontology.MinimalOntology`.
"""
def __init__(self, ontology: MinimalOntology):
super().__init__(ontology)
[docs]
def validate(self, items: typing.Sequence[typing.Union[Identified, TermId]]) -> ValidationResults:
results = []
for item in items:
primary = self._primary_term_id(self._extract_stateful_feature(item))
if primary is None:
# Unable to get the primary term ID. Handling items with obsolete IDs is not the responsibility
# of this validator
continue
if not any(PHENOTYPIC_ABNORMALITY == anc for anc in self._hpo.graph.get_ancestors(primary)):
item = self._hpo.get_term(primary.identifier)
results.append(
ValidationResult(
level=ValidationLevel.WARNING,
category='phenotypic_abnormality_descendant',
message=f'{item.name} [{item.identifier.value}] '
f'is not a descendant of Phenotypic abnormality [{PHENOTYPIC_ABNORMALITY.value}]'
)
)
return ValidationResults(results)
[docs]
class ObsoleteTermIdsValidator(BaseOntologyRuleValidator):
"""
`ObsoleteTermIdsValidator` points out usage of obsolete term ids in `items`.
Presence of an obsolete term ID is a :class:`ValidationLevel.WARNING`.
:param ontology: HPO represented as :class:`hpotk.ontology.MinimalOntology`.
"""
def __init__(self, ontology: MinimalOntology):
super().__init__(ontology)
[docs]
def validate(self, items: typing.Sequence[typing.Union[Identified, TermId]]) -> ValidationResults:
results = []
for item in items:
sf = self._extract_stateful_feature(item)
current = sf.identifier # cache the ID since _primary_term_id can update in place.
primary = self._primary_term_id(sf)
if primary.identifier != current:
current_term = self._hpo.get_term(sf.identifier)
results.append(
ValidationResult(
level=ValidationLevel.WARNING,
category='obsolete_term_id_is_used',
message=f'Using the obsolete {current.value} instead of {primary.identifier.value} '
f'for {current_term.name}'
)
)
return ValidationResults(results)