An adapter for transfer DigitalTWIN Clinic Description to FHIR
Project description
Digitaltwins on FHIR
Usage
- Setup and connect to FHIR server
from digitaltwins_on_fhir.core import Adapter
adapter = Adapter("http://localhost:8080/fhir/")
Load data to FHIR server
Primary measurements
- Load FHIR bundle
await adapter.loader().load_fhir_bundle('./dataset/dataset-fhir-bundles')
- Load DigitalTWIN Clinical Description (primary measurements)
measurements = adapter.loader().load_sparc_dataset_primary_measurements()
with open('./dataset/measurements.json', 'r') as file:
data = json.load(file)
await measurements.add_measurements_description(data).generate_resources()
- Add Practitioner (researcher) to FHIR server
from digitaltwins_on_fhir.core.resource import Identifier, Code, HumanName, Practitioner
await measurements.add_practitioner(researcher=Practitioner(
active=True,
identifier=[
Identifier(use=Code("official"), system="sparc.org",
value='sparc-d557ac68-f365-0718-c945-8722ec')],
name=[HumanName(use="usual", text="Xiaoming Li", family="Li", given=["Xiaoming"])],
gender="male"
))
Workflow
Search
References in Task (workflow tool process) resource
- owner:
Patient
reference - for:
PlanDefinition
(workflow) reference - focus:
ActivityDefinition
(workflow tool) reference - basedOn:
ResearchSubject
reference - requester (Optional):
Practitioner
(researcher) reference - references in input
- ImagingStudy
- Observation
- references in output
- Observation
Example
- Find a specific workflow process
- If known: patient, dataset, workflow tool and workflow uuids
client = adapter.async_client
# Step 1: find the patient
patient = await client.resources("Patient").search(
identifier="patient-xxxx").first()
# Step 2: find the dataset
dataset = await client.resources("ResearchStudy").search(
identifier="dataset-xxxx").first()
# Step 3: find the workflow tool
workflow_tool = await client.resources("ActivityDefinition").search(
identifier="workflow-tool-xxxx").first()
# Step 4: find the research subject
research_subject = await client.resources("ResearchSubject").search(
patient=patient.to_reference().reference,
study=dataset.to_reference().reference).first()
# Step 5: find workflow
workflow = await client.resources("PlanDefinition").search(
identifier="sparc-workflow-uuid-001").first()
workflow_tool_process = await client.resources("Task").search(
subject=workflow.to_reference(),
focus=workflow_tool.to_reference(),
based_on=research_subject.to_reference(),
owner=patient.to_reference()).first()
- Find all input resources of the workflow tool process
inputs = workflow_tool_process.get("input")
for i in inputs:
input_reference = i.get("valueReference")
input_resource = await input_reference.to_resource()
- Find the input data comes from with dataset
- Assume we don't know the dataset and patient uuids at this stage
composition = await client.resources("Composition").search(
title="primary measurements",
entry=input_reference).first()
dataset = await composition.get("subject").to_resource()
- Find all output resources of the workflow tool process
outputs = workflow_tool_process.get("output")
for output in outputs:
output_reference = output.get("valueReference")
output_resource = await output_reference.to_resource()
References in PlanDefinition (workflow) resource
- action
- definition_canonical: ActivityDefinition (workflow tool) reference
Example
- If known workflow uuid
- Find all related workflow tools
workflow = await client.resources("PlanDefinition").search( identifier="sparc-workflow-uuid-001").first() actions = workflow.get("action") for a in actions: if a.get("definitionCanonical") is None: continue resource_type, _id = a.get("definitionCanonical").split("/") workflow_tool = await client.reference(resource_type, _id).to_resource()
- Find all related workflow processes
workflow_tool_processes = await client.resources("Task").search( subject=workflow.to_reference()).fetch_all()
- Find all related workflow tools
DigitalTWIN on FHIR Diagram
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
digitaltwins_on_fhir-1.2.1.tar.gz
(63.7 kB
view hashes)
Built Distribution
Close
Hashes for digitaltwins_on_fhir-1.2.1.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 57b6996065a70ae6bc78764ca5dc0f4a729868cd6296273dd55054df46dff350 |
|
MD5 | b9d1e4ead4cca9cc7b1dca40d51b3e6c |
|
BLAKE2b-256 | 4ea365af94a16634641b50c93c2530ef13005d82e146094a9b82565142a4c8a8 |
Close
Hashes for digitaltwins_on_fhir-1.2.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | b2ea95750ae99b33caffdc43957594668abdb1d6c370b99c5c46a76312c637c0 |
|
MD5 | 7618338561645cb96700731d5715d752 |
|
BLAKE2b-256 | 6890aa47cd8107267b7c827dc20dce0e3c2b158a373a0e5a87347b15758ef214 |