Skip to content

Commit

Permalink
Merge branch 'issue-407' into split-gks-common-models
Browse files Browse the repository at this point in the history
  • Loading branch information
korikuzma committed Jul 10, 2024
2 parents d925b45 + 3992a0a commit 66fd5ab
Show file tree
Hide file tree
Showing 2 changed files with 107 additions and 116 deletions.
221 changes: 106 additions & 115 deletions src/ga4gh/core/entity_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,22 +10,14 @@
* `import ga4gh.core`, and refer to models using the fully-qualified
module name, e.g., `ga4gh.core.entity_models.Method`
"""
from __future__ import annotations
import datetime
from typing import Any, Dict, Literal, Annotated, Optional, Union, List
from enum import Enum

from pydantic import BaseModel, Field, RootModel, StringConstraints, constr, field_validator, model_serializer, model_validator
from pydantic import BaseModel, Field, RootModel, StringConstraints, model_serializer

from ga4gh.core import GA4GH_IR_REGEXP


class AgentSubtype(str, Enum):
"""Define constraints for agent subtype"""

PERSON = "person"
ORGANIZATION = "organization"
COMPUTER = "computer"

class Relation(str, Enum):
"""A mapping relation between concepts as defined by the Simple Knowledge
Expand Down Expand Up @@ -99,18 +91,6 @@ def ga4gh_serialize(self):
)


class RecordMetadata(BaseModel):
"""A reusable structure that encapsulates provenance metadata about a serialized
data record or object in a particular dataset (as opposed to provenance about the
real world entity this record or object represents).
"""

recordIdentifier: Optional[str] = Field(None, description="The identifier of the data record or object described in this RecordMetadata object.")
recordVersion: Optional[str] = Field(None, description="The version number of the record-level artifact the object describes.")
derivedFrom: Optional[str] = Field(None, description="Another data record from which the record described here was derived, through a data ingest and/or transformation process. Value should be a string representing the identifier of the source record.")
dateRecordCreated: Optional[str] = Field(None, description="The date the record was initially created.")
contributions: Optional[List[Contribution]] = Field(None, description="Describes specific contributions made by an human or software agent to the creation, modification, or administrative management of a data record or object.")


class Coding(BaseModel):
"""A structured representation of a code for a defined concept in a terminology or
Expand Down Expand Up @@ -173,15 +153,6 @@ class Expression(BaseModel):
#########################################


class CommonEntityType(str, Enum):
"""Define GKS Common Entity types"""

AGENT = "Agent"
CONTRIBUTION = "Contribution"
DOCUMENT = "Document"
METHOD = "Method"


class _Entity(BaseModel):
"""Entity is the root class of the 'gks-common' core information model classes -
those that have identifiers and other general metadata like labels, xrefs, urls,
Expand Down Expand Up @@ -216,110 +187,130 @@ class _DomainEntity(_Entity):
mappings: Optional[List[ConceptMapping]] = Field(None, description="A list of mappings to concepts in terminologies or code systems. Each mapping should include a coding and a relation.")


class Agent(_Entity):
"""An autonomous actor (person, organization, or computational agent) that bears
some form of responsibility for an activity taking place, for the existence of an
entity, or for another agent's activity.
"""
#########################################
# GKS Common Domain Entities
#########################################

type: Literal[CommonEntityType.AGENT] = Field(CommonEntityType.AGENT, description=f"MUST be '{CommonEntityType.AGENT.value}'.")
name: Optional[str] = Field(None, description="The descriptive name of the agent.")
subtype: Optional[AgentSubtype] = Field(None, description="A more specific type of agent the agent represents.")

class CommonDomainType(str, Enum):
"""Define GKS Common Domain Entity types"""

class Activity(_Entity):
"""An action or set of actions performed by an agent, that occurs over a period of
time. Activities may use, generate, modify, move, or destroy one or more entities.
"""
PHENOTYPE = "Phenotype"
DISEASE = "Disease"
TRAIT_SET = "TraitSet"
TR_ACTION = "TherapeuticAction"
TR_AGENT = "TherapeuticAgent"
TR_SUB = "TherapeuticSubstituteGroup"
TR_COMB = "CombinationTherapy"
GENE = "Gene"

class Phenotype(_DomainEntity):
"""An observable characteristic or trait of an organism."""

subtype: Optional[Coding] = Field(None, description="A more specific type of activity that an Activity object may represent.")
date: Optional[str] = Field(None, description="The date that the Activity was completed. The date SHOULD be formatted as a date string in ISO format 'YYYY-MM-DD'.")
performedBy: Optional[List[Agent]] = Field(None, description="An Agent who contributed to executing the Activity.")
specifiedBy: Optional[List[Method]] = Field(None, description="A method that was followed in performing an Activity, that describes how it was executed.")

@field_validator("date")
@classmethod
def date_format(cls, v: Optional[str]) -> Optional[str]:
"""Check that date is YYYY-MM-DD format"""
if v:
valid_format = "%Y-%m-%d"

try:
datetime.datetime.strptime(v, valid_format).replace(
tzinfo=datetime.timezone.utc
).strftime(valid_format)
except ValueError as e:
msg = "`date` must use YYYY-MM-DD format"
raise ValueError(msg) from e
return v


class Contribution(Activity):
"""An action taken by an agent in contributing to the creation, modification,
assessment, or deprecation of a particular entity (e.g. a Statement, EvidenceLine,
DataItem, Publication, etc.)
type: Literal[CommonDomainType.PHENOTYPE] = Field(
CommonDomainType.PHENOTYPE,
description=f'MUST be "{CommonDomainType.PHENOTYPE.value}".'
)


class Disease(_DomainEntity):
"""A particular abnormal condition that negatively affects the structure or function
of all or part of an organism and is not immediately due to any external injury.
"""

type: Literal[CommonEntityType.CONTRIBUTION] = CommonEntityType.CONTRIBUTION
contributor: Optional[Agent] = Field(None, description="The agent that made the contribution.")
contributionMadeTo: Optional[_InformationEntity] = Field(None, description="The artifact toward which the contribution was made.") # noqa: N815
activityType: Optional[Coding] = Field(None, description="SHOULD describe a concept descending from the Contributor Role Ontology.")
type: Literal[CommonDomainType.DISEASE] = Field(
CommonDomainType.DISEASE,
description=f'MUST be "{CommonDomainType.DISEASE.value}".'
)

@model_validator(mode="before")
def handle_extends_prop(cls, values: Dict[str, Any]) -> Dict[str, Any]:
"""Handle extends properties by renaming fields

:param values: Input values to process
:return: Processed values with extended properties renamed
"""
if "performedBy" in values:
values["contributor"] = values.pop("performedBy")
return values
class TraitSet(_DomainEntity):
"""A set of phenotype and/or disease concepts that together constitute a condition."""

type: Literal[CommonDomainType.TRAIT_SET] = Field(
CommonDomainType.TRAIT_SET,
description=f'MUST be "{CommonDomainType.TRAIT_SET.value}".'
)
traits: List[Union[Disease, Phenotype]] = Field(
...,
min_length=2
)

class _InformationEntity(_Entity):
"""Information Entities are abstract (non-physical) entities that are about
something (i.e. they carry information about things in the real world).
"""

id: str
specifiedBy: Optional[Union[Method, IRI]] = Field(None, description="A `Method` that describes all or part of the process through which the information was generated.")
contributions: Optional[List[Contribution]] = Field(None, description="A list of `Contribution` objects that describe the activities performed by agents upon this entity.")
isReportedIn: Optional[List[Union[Document, IRI]]] = Field(None, description="A document in which the information content is expressed.")
dateAuthored: Optional[str] = Field(None, description="Indicates when the information content expressed in the Information Entity was generated.")
derivedFrom: Optional[List[_InformationEntity]] = Field(None, description="Another Information Entity from which this Information Entity is derived, in whole or in part.")
recordMetadata: Optional[RecordMetadata] = Field(None, description="Metadata that applies to a specific concrete record of information as encoded in a particular system.")
class Condition(RootModel):
"""A disease or other medical disorder."""

root: Union[TraitSet, Disease, Phenotype] = Field(
...,
json_schema_extra={'description': 'A disease or other medical disorder.'},
discriminator='type',
)


class Document(_InformationEntity):
"""a representation of a physical or digital document"""
class TherapeuticAction(_DomainEntity):
"""A therapeutic action taken that is intended to alter or stop a pathologic process."""

type: Literal[CommonEntityType.DOCUMENT] = CommonEntityType.DOCUMENT
subtype: Optional[Coding] = Field(
None, description="A more specific type for the document (e.g. a publication, patent, pathology report)"
type: Literal[CommonDomainType.TR_ACTION] = Field(
CommonDomainType.TR_ACTION,
description=f'MUST be "{CommonDomainType.TR_ACTION.value}".'
)
title: Optional[str] = Field(None, description="The title of the Document")
url: Optional[constr(pattern="^(https?|s?ftp)://")] = Field(
None, description="A URL at which the document may be retrieved."


class TherapeuticAgent(_DomainEntity):
"""An administered therapeutic agent that is intended to alter or stop a pathologic process."""

type: Literal[CommonDomainType.TR_AGENT] = Field(
CommonDomainType.TR_AGENT,
description=f'MUST be "{CommonDomainType.TR_AGENT.value}".'
)
doi: Optional[constr(pattern="^10.(\\d+)(\\.\\d+)*\\/[\\w\\-\\.]+")] = Field(
None,
description="A `Digital Object Identifier <https://www.doi.org/the-identifier/what-is-a-doi/>_` for the document.",


class TherapeuticSubstituteGroup(_DomainEntity):
"""A group of therapeutic procedures that may be treated as substitutes for one another."""

type: Literal[CommonDomainType.TR_SUB] = Field(
CommonDomainType.TR_SUB,
description=f'MUST be "{CommonDomainType.TR_SUB.value}".'
)
pmid: Optional[int] = Field(
None,
description="A `PubMed unique identifier <https://en.wikipedia.org/wiki/PubMed#PubMed_identifier>`_.",
substitutes: List[Union[TherapeuticAction, TherapeuticAgent]] = Field(
...,
description='The individual therapeutic procedures that may be treated as substitutes.',
min_length=2
)


class Method(_InformationEntity):
"""A set of instructions that specify how to achieve some objective (e.g.
experimental protocols, curation guidelines, rule sets, etc.)
class CombinationTherapy(_DomainEntity):
"""A therapeutic procedure that involves multiple different therapeutic procedures
performed in combination.
"""

type: Literal[CommonEntityType.METHOD] = Field(CommonEntityType.METHOD, description=f"MUST be '{CommonEntityType.METHOD.value}'.")
isReportedIn: Optional[Union[Document, IRI]] = None # noqa: N815
subtype: Optional[Coding] = Field(
None,
description="A more specific type of entity the method represents (e.g. Variant Interpretation Guideline, Experimental Protocol)",
type: Literal[CommonDomainType.TR_COMB] = Field(
CommonDomainType.TR_COMB,
description=f'MUST be "{CommonDomainType.TR_COMB.value}".'
)
components: List[Union[TherapeuticSubstituteGroup, TherapeuticAction, TherapeuticAgent]] = Field(
...,
description='The individual therapeutic procedure components that constitute the combination therapy.',
min_length=2
)


class TherapeuticProcedure(RootModel):
"""An action or administration of therapeutic agents to produce an effect that is
intended to alter or stop a pathologic process.
"""

root: Union[CombinationTherapy, TherapeuticSubstituteGroup, TherapeuticAction, TherapeuticAgent] = Field(
...,
json_schema_extra={'description': 'An action or administration of therapeutic agents to produce an effect that is intended to alter or stop a pathologic process.'},
discriminator='type',
)


class Gene(_DomainEntity):
"""A basic physical and functional unit of heredity."""

type: Literal[CommonDomainType.GENE] = Field(
CommonDomainType.GENE,
description=f'MUST be "{CommonDomainType.GENE.value}".'
)
license: Optional[str] = Field(None, description="A particular license that dictates legal permissions for how a published method (e.g. an experimental protocol, workflow specification, curation guideline) can be used.")
2 changes: 1 addition & 1 deletion submodules/vrs

0 comments on commit 66fd5ab

Please sign in to comment.