onconova.interoperability.fhir.schemas.tumor_marker

TumorMarkerProfile

Bases: OnconovaFhirBaseSchema, OnconovaTumorMarker

__model__ class-attribute instance-attribute

__schema__ class-attribute instance-attribute

fhir_to_onconova(obj) classmethod

Source code in onconova/interoperability/fhir/schemas/tumor_marker.py

@classmethod
def fhir_to_onconova(
    cls, obj: fhir.OnconovaTumorMarker
) -> schemas.TumorMarkerCreate:
    map = cls.map_to_internal(
        "tumorMarkerTestCode", obj.fhirpath_single("Observation.code.coding")
    )
    result_type = map.get("result_type")
    return schemas.TumorMarkerCreate(
        externalSource=None,
        externalSourceId=None,
        caseId=obj.fhirpath_single(
            "Observation.subject.reference.getValue()"
        ).replace("Patient/", ""),
        date=obj.fhirpath_single("Observation.effectiveDateTime.getValue()"),
        analyte=CodedConcept.model_validate(
            obj.fhirpath_single(
                "Observation.code.extension('http://onconova.github.io/fhir/StructureDefinition/onconova-ext-tumor-marker-analyte').valueCodeableConcept.coding"
            ).model_dump()
        ),
        relatedEntitiesIds=[
            ref.replace("Condition/", "")
            for ref in obj.fhirpath_values(
                "Observation.extension('http://hl7.org/fhir/us/mcode/StructureDefinition/mcode-related-condition').valueReference.reference.getValue()"
            )
        ],
        massConcentration=(
            Measure(
                value=obj.fhirpath_single(
                    "Observation.valueQuantity.value.getValue()"
                ),
                unit=ucum_to_internal(
                    obj.fhirpath_single("Observation.valueQuantity.code.getValue()")
                ),
            )
            if result_type == AnalyteResultType.mass_concentration
            else None
        ),
        arbitraryConcentration=(
            Measure(
                value=obj.fhirpath_single(
                    "Observation.valueQuantity.value.getValue()"
                ),
                unit=ucum_to_internal(
                    obj.fhirpath_single("Observation.valueQuantity.code.getValue()")
                ),
            )
            if result_type == AnalyteResultType.arbitary_concentration
            else None
        ),
        substanceConcentration=(
            Measure(
                value=obj.fhirpath_single(
                    "Observation.valueQuantity.value.getValue()"
                ),
                unit=ucum_to_internal(
                    obj.fhirpath_single("Observation.valueQuantity.code.getValue()")
                ),
            )
            if result_type == AnalyteResultType.substance_concentration
            else None
        ),
        fraction=(
            Measure(
                value=obj.fhirpath_single(
                    "Observation.valueQuantity.value.getValue()"
                ),
                unit=ucum_to_internal(
                    obj.fhirpath_single("Observation.valueQuantity.code.getValue()")
                ),
            )
            if result_type == AnalyteResultType.fraction
            else None
        ),
        multipleOfMedian=(
            Measure(
                value=obj.fhirpath_single(
                    "Observation.valueQuantity.value.getValue()"
                ),
                unit=ucum_to_internal(
                    obj.fhirpath_single("Observation.valueQuantity.code.getValue()")
                ),
            )
            if result_type == AnalyteResultType.multiple_of_median
            else None
        ),
        tumorProportionScore=(
            obj.fhirpath_single("Observation.valueString.getValue()")
            if result_type == AnalyteResultType.tumor_proportion_score
            else None
        ),
        immuneCellScore=(
            obj.fhirpath_single("Observation.valueString.getValue()")
            if result_type == AnalyteResultType.immmune_cells_score
            else None
        ),
        combinedPositiveScore=(
            Measure(
                value=obj.fhirpath_single(
                    "Observation.valueQuantity.value.getValue()"
                ),
                unit=ucum_to_internal(
                    obj.fhirpath_single("Observation.valueQuantity.code.getValue()")
                ),
            )
            if result_type == AnalyteResultType.combined_positive_score
            else None
        ),
        immunohistochemicalScore=(
            obj.fhirpath_single("Observation.valueString.getValue()")
            if result_type == AnalyteResultType.immunohistochemical_score
            else None
        ),
        presence=(
            obj.fhirpath_single("Observation.valueString.getValue()")
            if result_type == AnalyteResultType.presence
            else None
        ),
        nuclearExpressionStatus=(
            obj.fhirpath_single("Observation.valueString.getValue()")
            if result_type == AnalyteResultType.nuclear_expression_status
            else None
        ),
    )

onconova_to_fhir(obj) classmethod

Source code in onconova/interoperability/fhir/schemas/tumor_marker.py

@classmethod
def onconova_to_fhir(cls, obj: schemas.TumorMarker) -> fhir.OnconovaTumorMarker:
    resource: fhir.OnconovaTumorMarker = fhir.OnconovaTumorMarker.model_construct()
    resource.id = str(obj.id)
    resource.text = Narrative(
        status="generated",
        div=f'<div xmlns="http://www.w3.org/1999/xhtml">{obj.description}</div>',
    )
    resource.subject = Reference(
        reference=f"Patient/{obj.caseId}",
    )
    resource.effectiveDateTime = obj.date.isoformat()
    resource.extension = resource.extension or []
    for cond_id in obj.relatedEntitiesIds or []:
        resource.extension.append(
            fhir.RelatedCondition(
                valueReference=Reference(reference=f"Condition/{cond_id}")
            )
        )

    if obj.massConcentration:
        result_type = AnalyteResultType.mass_concentration
        resource.valueQuantity = fhir.Quantity(
            value=obj.massConcentration.value,
            code=internal_to_ucum(obj.massConcentration.unit),
            system="http://unitsofmeasure.org",
        )
    elif obj.arbitraryConcentration:
        result_type = AnalyteResultType.arbitary_concentration
        resource.valueQuantity = fhir.Quantity(
            value=obj.arbitraryConcentration.value,
            code=internal_to_ucum(obj.arbitraryConcentration.unit),
            system="http://unitsofmeasure.org",
        )
    elif obj.substanceConcentration:
        result_type = AnalyteResultType.substance_concentration
        resource.valueQuantity = fhir.Quantity(
            value=obj.substanceConcentration.value,
            code=internal_to_ucum(obj.substanceConcentration.unit),
            system="http://unitsofmeasure.org",
        )
    elif obj.multipleOfMedian:
        result_type = AnalyteResultType.multiple_of_median
        resource.valueQuantity = fhir.Quantity(
            value=obj.multipleOfMedian.value,
            code=internal_to_ucum(obj.multipleOfMedian.unit),
            system="http://unitsofmeasure.org",
        )
    elif obj.fraction:
        result_type = AnalyteResultType.fraction
        resource.valueQuantity = fhir.Quantity(
            value=obj.fraction.value,
            code=internal_to_ucum(obj.fraction.unit),
            system="http://unitsofmeasure.org",
        )
    elif obj.presence:
        result_type = AnalyteResultType.presence
        resource.valueString = obj.presence
    elif obj.combinedPositiveScore:
        result_type = AnalyteResultType.combined_positive_score
        resource.valueQuantity = fhir.Quantity(
            value=obj.combinedPositiveScore.value,
            code=internal_to_ucum(obj.combinedPositiveScore.unit),
            system="http://unitsofmeasure.org",
        )
    elif obj.immuneCellScore:
        result_type = AnalyteResultType.immmune_cells_score
        resource.valueString = obj.immuneCellScore
    elif obj.tumorProportionScore:
        result_type = AnalyteResultType.tumor_proportion_score
        resource.valueString = obj.tumorProportionScore
    elif obj.immunohistochemicalScore:
        result_type = AnalyteResultType.immunohistochemical_score
        resource.valueString = obj.immunohistochemicalScore
    elif obj.nuclearExpressionStatus:
        result_type = AnalyteResultType.nuclear_expression_status
        resource.valueString = obj.nuclearExpressionStatus
    else:
        result_type = None

    resource.code = fhir.OnconovaTumorMarkerCode(
        coding=[
            (
                cls.map_to_fhir(
                    "tumorMarkerTestCode",
                    dict(code=obj.analyte.code, result_type=result_type),
                )
                if result_type
                else Coding(
                    code="",
                    display="Tumor marker Cancer",
                    system="http://loinc.org",
                )
            )
        ]
    )
    resource.code.extension = [
        fhir.TumorMarkerAnalyte(
            valueCodeableConcept=construct_fhir_codeable_concept(obj.analyte)
        )
    ]

    return resource

2026-04-262026-04-26runner