loysharosen/arithmedic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added fhir mapping

Browse Source
This commit is contained in:
loysharosen
2026-04-24 11:12:06 +02:00
parent 8b57002549
commit 5e7dca5c85
10 changed files with 231 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1 -1
View File
@@ -1,4 +1,4 @@
.git/
.venv/
__pycache__/
arithmetic.egg-info
arithmedic.egg-info/
api/egfr/ckd_epi_2021.py
+92 -20
View File
@@ -1,11 +1,13 @@
from typing import Annotated, Literal
from fastapi import APIRouter, Form, Request
from fastapi import APIRouter, Form, HTTPException, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fhir.resources.parameters import Parameters
from calculators.egfr.ckd_epi_2021 import ckd_epi_2021
from schemas.egfr.ckd_epi_2021 import CkdEpi2021Input, CkdEpi2021Output
from fhir_mapping.egfr.ckd_epi_2021 import build_output, parse_input
from schemas.egfr.ckd_epi_2021 import CkdEpi2021Output
FORMULA = "2021 CKD-EPI Creatinine"
REFERENCE = (
@@ -17,6 +19,32 @@ REFERENCE_URL = "https://doi.org/10.1056/NEJMoa2102953"
router = APIRouter()
templates = Jinja2Templates(directory="templates")
_EXAMPLE_REQUEST = {
"resourceType": "Parameters",
"parameter": [
{"name": "serum-creatinine", "valueQuantity": {"value": 90, "unit": "umol/L"}},
{"name": "age", "valueInteger": 65},
{"name": "sex", "valueCode": "female"},
],
}
_EXAMPLE_RESPONSE = {
"resourceType": "Parameters",
"parameter": [
{
"name": "egfr",
"valueQuantity": {
"value": 61,
"unit": "mL/min/1.73m2",
"system": "http://unitsofmeasure.org",
"code": "mL/min/{1.73_m2}",
},
},
{"name": "formula", "valueString": "2021 CKD-EPI Creatinine"},
{"name": "reference", "valueUri": "https://doi.org/10.1056/NEJMoa2102953"},
],
}
@router.get("/ckd-epi-2021", response_class=HTMLResponse, include_in_schema=False)
def egfr_form(request: Request) -> HTMLResponse:
@@ -27,29 +55,74 @@ def egfr_form(request: Request) -> HTMLResponse:
)
@router.post("/ckd-epi-2021", response_model=CkdEpi2021Output)
def calculate_ckd_epi_2021(payload: CkdEpi2021Input) -> CkdEpi2021Output:
"""
Calculate estimated GFR using the 2021 CKD-EPI Creatinine equation.
@router.post(
"/ckd-epi-2021",
response_model=None,
summary="Calculate eGFR (2021 CKD-EPI Creatinine)",
description="""
Calculate estimated GFR using the 2021 CKD-EPI Creatinine equation.
Accepts and returns a FHIR **Parameters** resource.
**Input parameters**
| Name | Type | Description |
|---|---|---|
| `serum-creatinine` | valueQuantity | Creatinine with unit (`umol/L` or `mg/dL`) |
| `age` | valueInteger | Age in years (18–120) |
| `sex` | valueCode | FHIR administrative-gender (`male` or `female`) |
**Output parameters**
| Name | Type | Description |
|---|---|---|
| `egfr` | valueQuantity | eGFR in mL/min/1.73m² |
| `formula` | valueString | Formula name |
| `reference` | valueUri | DOI of source publication |
""",
openapi_extra={
"requestBody": {
"required": True,
"content": {
"application/json": {
"example": _EXAMPLE_REQUEST,
}
},
},
"responses": {
"200": {
"description": "FHIR Parameters with eGFR result",
"content": {
"application/json": {
"example": _EXAMPLE_RESPONSE,
}
},
}
},
},
)
def calculate_ckd_epi_2021(params: Parameters) -> dict:
"""FHIR Parameters in, FHIR Parameters out."""
try:
inputs = parse_input(params)
except ValueError as e:
raise HTTPException(status_code=422, detail=str(e))
- **serum_creatinine**: creatinine value in the specified unit
- **unit**: "umol/L" (default) or "mg/dL"
- **age**: years (18–120)
- **sex**: "male" or "female"
"""
egfr = ckd_epi_2021(
serum_creatinine=payload.serum_creatinine,
unit=payload.unit,
age=payload.age,
sex=payload.sex,
serum_creatinine=inputs.serum_creatinine,
unit=inputs.unit,
age=inputs.age,
sex=inputs.sex,
)
return CkdEpi2021Output(
output = CkdEpi2021Output(
egfr=round(egfr),
unit="mL/min/1.73m²",
formula=FORMULA,
reference=REFERENCE,
reference_url=REFERENCE_URL,
)
return build_output(output).model_dump()
@router.post("/ckd-epi-2021/calculate", response_class=HTMLResponse, include_in_schema=False)
def calculate_ckd_epi_2021_form(
@@ -68,9 +141,8 @@ def calculate_ckd_epi_2021_form(
)
output = CkdEpi2021Output(
egfr=round(egfr),
unit="mL/min/1.73m²",
formula=FORMULA,
reference=REFERENCE,
reference_url=REFERENCE_URL,
)
return templates.TemplateResponse(
request,
arithmedic.egg-info/PKG-INFO
-3
View File
@@ -1,3 +0,0 @@
Metadata-Version: 2.4
Name: arithmedic
Version: 0.1.0
arithmedic.egg-info/SOURCES.txt
-15
View File
@@ -1,15 +0,0 @@
README.md
pyproject.toml
api/__init__.py
api/egfr/__init__.py
api/egfr/ckd_epi_2021.py
arithmedic.egg-info/PKG-INFO
arithmedic.egg-info/SOURCES.txt
arithmedic.egg-info/dependency_links.txt
arithmedic.egg-info/top_level.txt
calculators/__init__.py
calculators/egfr/__init__.py
calculators/egfr/ckd_epi_2021.py
schemas/__init__.py
schemas/egfr/__init__.py
schemas/egfr/ckd_epi_2021.py
arithmedic.egg-info/dependency_links.txt
-1
View File
@@ -1 +0,0 @@
arithmedic.egg-info/top_level.txt
-3
View File
@@ -1,3 +0,0 @@
api
calculators
schemas
fhir_mapping/egfr/__init__.py
View File
fhir_mapping/egfr/ckd_epi_2021.py
+119
View File
@@ -0,0 +1,119 @@
# FHIR Parameters translation layer for the 2021 CKD-EPI Creatinine calculator.
#
# Responsibilities:
# - parse_input: FHIR Parameters → CkdEpi2021Input (validated internal type)
# - build_output: CkdEpi2021Output → FHIR Parameters
#
# This module is the only place in the codebase that knows about FHIR.
# The calculator and schemas remain FHIR-agnostic.
from fhir.resources.parameters import Parameters
from fhir.resources.quantity import Quantity
from schemas.egfr.ckd_epi_2021 import CkdEpi2021Input, CkdEpi2021Output
# Accepted UCUM unit codes → internal unit strings
_UNIT_MAP: dict[str, str] = {
"umol/L": "umol/L",
"umol/l": "umol/L",
"mmol/L": "umol/L", # not correct clinically but guard against it
"mg/dL": "mg/dL",
"mg/dl": "mg/dL",
}
# FHIR administrative-gender → internal sex strings
_SEX_MAP: dict[str, str] = {
"male": "male",
"female": "female",
}
def parse_input(params: Parameters) -> CkdEpi2021Input:
"""
Extract and validate CKD-EPI inputs from a FHIR Parameters resource.
Expected parameters:
serum-creatinine valueQuantity Creatinine with unit (umol/L or mg/dL)
age valueInteger Age in years (18–120)
sex valueCode FHIR administrative-gender code
Raises ValueError for missing or unrecognised parameters — FastAPI
will catch this and return a 422.
"""
indexed = {p.name: p for p in (params.parameter or [])}
# ── serum-creatinine ──────────────────────────────────────────────
creatinine_param = indexed.get("serum-creatinine")
if creatinine_param is None:
raise ValueError("Missing required parameter: serum-creatinine")
quantity: Quantity = creatinine_param.valueQuantity
if quantity is None:
raise ValueError("serum-creatinine must be a valueQuantity")
raw_unit = quantity.unit or quantity.code or ""
unit = _UNIT_MAP.get(raw_unit)
if unit is None:
raise ValueError(
f"Unrecognised creatinine unit '{raw_unit}'. "
f"Use 'umol/L' or 'mg/dL'."
)
# ── age ───────────────────────────────────────────────────────────
age_param = indexed.get("age")
if age_param is None:
raise ValueError("Missing required parameter: age")
age = age_param.valueInteger
if age is None:
raise ValueError("age must be a valueInteger")
# ── sex ───────────────────────────────────────────────────────────
sex_param = indexed.get("sex")
if sex_param is None:
raise ValueError("Missing required parameter: sex")
raw_sex = sex_param.valueCode
sex = _SEX_MAP.get(raw_sex or "")
if sex is None:
raise ValueError(
f"Unrecognised sex code '{raw_sex}'. Use 'male' or 'female'."
)
# Delegate remaining validation (range checks etc.) to the Pydantic schema
return CkdEpi2021Input(
serum_creatinine=float(quantity.value),
unit=unit,
age=age,
sex=sex,
)
def build_output(result: CkdEpi2021Output) -> Parameters:
"""
Serialize CKD-EPI output as a FHIR Parameters resource.
Returned parameters:
egfr valueQuantity eGFR in mL/min/1.73m² (UCUM)
formula valueString Human-readable formula name
reference valueUri DOI of the source publication
"""
return Parameters(**{
"parameter": [
{
"name": "egfr",
"valueQuantity": {
"value": result.egfr,
"unit": "mL/min/1.73m2",
"system": "http://unitsofmeasure.org",
"code": "mL/min/{1.73_m2}",
},
},
{
"name": "formula",
"valueString": result.formula,
},
{
"name": "reference",
"valueUri": result.reference_url,
},
]
})
requirements.txt
+3 -1
View File
@@ -1,7 +1,7 @@
annotated-doc==0.0.4
annotated-types==0.7.0
anyio==4.13.0
-e git+ssh://git@github.com/loysharosen/arithmedic.git@e70bdd6496b98bb1b8202b8452ca8d68b80552a3#egg=arithmedic
-e git+ssh://git@github.com/loysharosen/arithmedic.git@8b570025493bc59886a237dacc69cca05eeed595#egg=arithmedic
certifi==2026.2.25
click==8.3.2
dnspython==2.8.0
@@ -10,6 +10,8 @@ fastapi==0.135.3
fastapi-cli==0.0.24
fastapi-cloud-cli==0.16.1
fastar==0.10.0
fhir.resources==8.2.0
fhir_core==1.1.7
greenlet==3.4.0
h11==0.16.0
httpcore==1.0.9
schemas/egfr/ckd_epi_2021.py
+16 -13
View File
@@ -1,28 +1,31 @@
""""Pydantic schemas"""
from pydantic import BaseModel, ConfigDict, Field
"""Pydantic schemas — internal types, FHIR-agnostic."""
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
class CkdEpi2021Input(BaseModel):
model_config = ConfigDict(
json_schema_extra={
"example": {
"serum_creatinine": 90,
"unit": "umol/L",
"age": 65,
"sex": "female"
}
}
)
json_schema_extra={
"example": {
"serum_creatinine": 90,
"unit": "umol/L",
"age": 65,
"sex": "female",
}
}
)
serum_creatinine: float = Field(..., gt=0, description="Serum creatinine in umol/L or mg/dL")
unit: Literal["umol/L", "mg/dL"] = Field(default="umol/L", description="Unit of serum creatinine")
age: int = Field(..., ge=18, le=120, description="Age in years")
sex: Literal["male", "female"]
class CkdEpi2021Output(BaseModel):
model_config = ConfigDict(frozen=True)
egfr: int = Field(..., description="Estimated GFR rounded to nearest integer")
unit: str = Field(default="mL/min/1.73m²")
formula: str = Field(default="2021 CKD-EPI Creatinine")
reference: str
reference_url: str = Field(..., description="DOI of the source publication")