digitaltwins-on-fhir 1.3.3

An adapter for transfer DigitalTWIN Clinic Description to FHIR

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: ResearchStudy (Assay) reference
• focus: ActivityDefinition (workflow tool) reference
• basedOn: ResearchSubject (patient research subject) reference
• requester (Optional): Practitioner (researcher) reference
• references in input
  • ImagingStudy
  • Observation
  • DocumentReference
• references in output
  • ImagingStudy
  • Observation
  • DocumentReference

Example

• Find a specific workflow process
  • If known: patient, assay, and workflow tool uuids

client = adapter.async_client

# Step 1: find the patient
patient = await client.resources("Patient").search(
                                    identifier="patient-xxxx").first()
# Step 2: find the assay
assay = 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 (cohort in assay)
research_subject = await client.resources("ResearchSubject").search(
                                    patient=patient.to_reference().reference,
                                    study=assay.to_reference().reference).first()
workflow_tool_process = await client.resources("Task").search(
                                    subject=assay.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_uuid = composition.get_by_path([
        'identifier',
        {'system':'https://www.auckland.ac.nz/en/abi.html'},
        'value'
    ], '')
dataset = await client.resources("Composition").search(identifier=dataset_uuid).fetch_all()

• 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

    assay = await client.resources("ResearchStudy").search(
                                    identifier="dataset-xxxx").first()
    workflow_tool_processes = await client.resources("Task").search(
                                        subject=assay.to_reference()).fetch_all()
    

Search in DigitalTWINS on FHIR methods

search = adapter.search()

• Finding all primary measurements for a patient

measurements = await self.search.get_patient_measurements("xxx-xxxx")

• Find which workflow, tool, and primary data was used to generate a specific derived measurement observation

res = await self.search.get_workflow_details_by_derived_data("Observation", "xxxx-xxxx")

• Find all inputs and their dataset uuid for generating the Observation

res = await self.search.get_all_inputs_by_derived_data("Observation","xxx-xxxx")

• Find all tools and models used by a workflow and their workflow tool processes

res = await self.search.get_all_workflow_tools_by_workflow(
                    name="Automated torso model generation - script")

• Find inputs and outputs of a given tool in a workflow

res = await self.search.get_all_inputs_outputs_of_workflow_tool(
                    name="Tumour Position Correction (Manual) Tool")

Reference in resource

• ResearchStudy - Study
  • principalInvestigator: Practitioner reference
• ResearchStudy - Assay
  • protocol: [ PlanDefinition(Workflow) reference ]
  • partOf: [ ResearchStudy(Study) reference ]
• ResearchSubject - Assay cohort
  • individual(patient): Patient reference
  • study: ResearchStudy(Assay) reference
  • consent: Consent reference
• ResearchSubject - dataset cohort
  • individual(patient): Patient reference
  • consent: Consent reference
• Composition - primary measurements
  • author: [ Patient reference, Practitioner reference ]
  • subject: ResearchSubject reference
  • entry: [ Observation reference, ImagingStudy reference, DocumentReference reference ]
• ImagingStudy
  • subject: Patient reference
  • endpoint: [ Endpoint Reference ]
  • referrer: Practitioner reference
• Observation - primary measurements
  • subject: Patient reference
• DocumentRefernce
  • subject: Patient reference
• PlanDefinition:
  • action.definitionCanonical: ActivityDefinition reference string
• ActivityDefinition:
  • participant: [ software uuid, model uuid ]
• Task:
  • owner: patient reference
  • for(subject): ResearchSubject(Assay) reference
  • focus: ActivityDefinition(workflow) tool reference
  • basedOn: research subject reference
  • requester (Optional): practitioner reference
  • input: [ Observation reference, ImagingStudy reference ]
  • output: [ Observation reference, ImagingStudy reference ]

Work steps

• Upload measurements dataset (primary measurements)
• Upload workflow / workflow tools
• Create Assay (get practitioner, study, and workflow process information)

DigitalTWIN on FHIR Diagram

Contributors

Linkun Gao

Chinchien Lin

Ayah Elsayed

Jiali Xu

Gregory Sands

David Nickerson

Thiranja Prasad Babarenda Gamage

Publications

1. Paper Title One, Author1, Author2. Journal Name, Year.
2. Paper Title Two, Author1, Author2. Conference Name, Year.

Please cite the corresponding paper if you use this project in your research.