Fixed sub questionnaires and inline choice extensions

app/aidbox/operations.py

@@ -22,16 +22,21 @@
 @aidbox_operation(["GET"], ["Questionnaire", {"name": "id"}, "$assemble"])
 @prepare_args
 async def assemble_op(request: AidboxSdcRequest):
+    aidbox_client = request.aidbox_client
+
     questionnaire = (
         await request.aidbox_client.resources("Questionnaire")
         .search(_id=request.route_params["id"])
         .get()
     )
 
-    assembled_questionnaire_lazy = await assemble(request.fhir_client, questionnaire)
-    assembled_questionnaire = json.loads(
-        json.dumps(assembled_questionnaire_lazy, default=list)
+    def get_to_first_class_extension(fhir_resource):
+        return to_first_class_extension(fhir_resource, aidbox_client)
+
+    assembled_questionnaire_lazy = await assemble(
+        request.fhir_client, questionnaire, get_to_first_class_extension
     )
+    assembled_questionnaire = json.loads(json.dumps(assembled_questionnaire_lazy, default=list))
     if request.is_fhir:
         assembled_questionnaire = await from_first_class_extension(
             assembled_questionnaire, request.aidbox_client
@@ -102,9 +107,7 @@ async def extract_questionnaire_operation(request: AidboxSdcRequest):
         if "Questionnaire" not in env:
             raise MissingParamOperationOutcome("`Questionnaire` parameter is required")
         if "QuestionnaireResponse" not in env:
-            raise MissingParamOperationOutcome(
-                "`QuestionnaireResponse` parameter is required"
-            )
+            raise MissingParamOperationOutcome("`QuestionnaireResponse` parameter is required")
 
         fce_questionnaire = (
             await to_first_class_extension(env["Questionnaire"], request.aidbox_client)
@@ -155,9 +158,7 @@ async def extract_questionnaire_instance_operation(request: AidboxSdcRequest):
     )
     as_root = fce_questionnaire.get("runOnBehalfOfRoot")
     extract_client = (
-        request.client
-        if as_root
-        else get_user_sdk_client(request.request, request.client)
+        request.client if as_root else get_user_sdk_client(request.request, request.client)
     )
     return web.json_response(
         await extract_questionnaire_instance(
@@ -181,15 +182,11 @@ async def extract_questionnaire_instance(
 ):
     if resource["resourceType"] == "QuestionnaireResponse":
         env = {}
-        env_questionnaire_response = extract_client.resource(
-            "QuestionnaireResponse", **resource
-        )
+        env_questionnaire_response = extract_client.resource("QuestionnaireResponse", **resource)
     elif resource["resourceType"] == "Parameters":
         env = parameter_to_env(resource)
         if "QuestionnaireResponse" not in env:
-            raise MissingParamOperationOutcome(
-                "`QuestionnaireResponse` parameter is required"
-            )
+            raise MissingParamOperationOutcome("`QuestionnaireResponse` parameter is required")
 
         env_questionnaire_response = env["QuestionnaireResponse"]
     else:
@@ -232,17 +229,11 @@ async def populate_questionnaire(request: AidboxSdcRequest):
         else env["Questionnaire"]
     )
     as_root = fce_questionnaire.get("runOnBehalfOfRoot")
-    client = (
-        request.client
-        if as_root
-        else get_user_sdk_client(request.request, request.client)
-    )
+    client = request.client if as_root else get_user_sdk_client(request.request, request.client)
 
     fce_populated_qr = await populate(client, fce_questionnaire, env)
     if request.is_fhir:
-        fce_populated_qr = await from_first_class_extension(
-            fce_populated_qr, request.aidbox_client
-        )
+        fce_populated_qr = await from_first_class_extension(fce_populated_qr, request.aidbox_client)
     return web.json_response(fce_populated_qr)
 
 
@@ -265,9 +256,7 @@ async def populate_questionnaire_instance(request: AidboxSdcRequest):
 
     fce_populated_qr = await populate(client, fce_questionnaire, env)
     if request.is_fhir:
-        fce_populated_qr = await from_first_class_extension(
-            fce_populated_qr, request.aidbox_client
-        )
+        fce_populated_qr = await from_first_class_extension(fce_populated_qr, request.aidbox_client)
     return web.json_response(fce_populated_qr)
 
 

app/converter/fce_to_fhir.py

@@ -94,6 +94,15 @@ def process_items(items):
             item["extension"].append(macro_extension)
             del item["macro"]
 
+        if item.get("subQuestionnaire"):
+            sub_questionnaire_extension = {
+                "url": "http://hl7.org/fhir/uv/sdc/StructureDefinition/sdc-questionnaire-subQuestionnaire",
+                "valueCanonical": item["subQuestionnaire"],
+            }
+            item["extension"] = item.get("extension", [])
+            item["extension"].append(sub_questionnaire_extension)
+            del item["subQuestionnaire"]
+
         if item.get("itemControl"):
             item_control_extension = {
                 "url": "http://hl7.org/fhir/StructureDefinition/questionnaire-itemControl",
@@ -105,6 +114,15 @@ def process_items(items):
             item["extension"].append(item_control_extension)
             del item["itemControl"]
 
+        if item.get("inlineChoiceDirection"):
+            inline_choice_direction_extension = {
+                "url": "https://beda.software/fhir-emr-questionnaire/inline-choice-direction",
+                "valueString": item["inlineChoiceDirection"],
+            }
+            item["extension"] = item.get("extension", [])
+            item["extension"].append(inline_choice_direction_extension)
+            del item["inlineChoiceDirection"]
+
         if item.get("start") is not None:
             start_extension = {
                 "url": "https://beda.software/fhir-emr-questionnaire/slider-start",
@@ -257,9 +275,11 @@ def process_items(items):
 
         if item.get("initial"):
             item["initial"] = [
-                {"valueBoolean": entry["value"]["boolean"]}
-                if "boolean" in entry["value"]
-                else {"valueCoding": entry["value"]["Coding"]}
+                (
+                    {"valueBoolean": entry["value"]["boolean"]}
+                    if "boolean" in entry["value"]
+                    else {"valueCoding": entry["value"]["Coding"]}
+                )
                 for entry in item["initial"]
             ]
 

app/converter/fhir_to_fce.py

@@ -162,9 +162,9 @@ def process_extension(fhirQuestionnaire):
         "mapping": mapping if mapping else None,
         "sourceQueries": source_queries if source_queries else None,
         "targetStructureMap": target_structure_map if target_structure_map else None,
-        "itemPopulationContext": item_population_context["valueExpression"]
-        if item_population_context
-        else None,
+        "itemPopulationContext": (
+            item_population_context["valueExpression"] if item_population_context else None
+        ),
         "assembleContext": assemble_context["valueString"] if assemble_context else None,
     }
 
@@ -288,27 +288,44 @@ def get_updated_properties_from_item(item):
     hidden = find_extension(item, "http://hl7.org/fhir/StructureDefinition/questionnaire-hidden")
     if hidden is not None:
         hidden = hidden["valueBoolean"]
+
     initial_expression = find_extension(
         item, "http://hl7.org/fhir/uv/sdc/StructureDefinition/sdc-questionnaire-initialExpression"
     )
     if initial_expression is not None:
         initial_expression = initial_expression["valueExpression"]
+
     item_population_context = find_extension(
         item,
         "http://hl7.org/fhir/uv/sdc/StructureDefinition/sdc-questionnaire-itemPopulationContext",
     )
     if item_population_context is not None:
         item_population_context = item_population_context["valueExpression"]
+
+    sub_questionnaire = find_extension(
+        item, "http://hl7.org/fhir/uv/sdc/StructureDefinition/sdc-questionnaire-subQuestionnaire"
+    )
+    if sub_questionnaire is not None:
+        sub_questionnaire = sub_questionnaire["valueCanonical"]
+
     item_control = find_extension(
         item, "http://hl7.org/fhir/StructureDefinition/questionnaire-itemControl"
     )
     if item_control is not None:
         item_control = item_control["valueCodeableConcept"]
 
+    item_inline_choice_direction = find_extension(
+        item, "https://beda.software/fhir-emr-questionnaire/inline-choice-direction"
+    )
+    if item_inline_choice_direction is not None:
+        item_inline_choice_direction = item_inline_choice_direction["valueString"]
+
     updated_properties["hidden"] = hidden
     updated_properties["initialExpression"] = initial_expression
     updated_properties["itemPopulationContext"] = item_population_context
+    updated_properties["subQuestionnaire"] = sub_questionnaire
     updated_properties["itemControl"] = item_control
+    updated_properties["inlineChoiceDirection"] = item_inline_choice_direction
 
     item_type = item.get("type", "")
 

app/fhir_server/operations.py

@@ -24,11 +24,18 @@
 @routes.get("/Questionnaire/{id}/$assemble")
 async def assemble_handler(request: web.BaseRequest):
     client: AsyncFHIRClient = request.app["client"]
+    aidbox_client = request.aidbox_client
+
     questionnaire = (
         await client.resources("Questionnaire").search(_id=request.match_info["id"]).get()
     )
 
-    assembled_questionnaire_lazy = await assemble(client, to_first_class_extension(questionnaire))
+    async def get_to_first_class_extension(fhir_resource):
+        return to_first_class_extension(fhir_resource, aidbox_client)
+
+    assembled_questionnaire_lazy = await assemble(
+        client, to_first_class_extension(questionnaire), get_to_first_class_extension
+    )
     assembled_questionnaire = json.loads(json.dumps(assembled_questionnaire_lazy, default=list))
 
     return web.json_response(from_first_class_extension(assembled_questionnaire))
@@ -52,11 +59,7 @@ async def get_questionnaire_context_handler(request: web.BaseRequest):
     client = request.app["client"]
 
     return web.json_response(
-        await get_questionnaire_context(
-            client, 
-            to_first_class_extension(env["Questionnaire"]),
-            env
-        )
+        await get_questionnaire_context(client, to_first_class_extension(env["Questionnaire"]), env)
     )
 
 

app/sdc/assemble.py

@@ -3,8 +3,6 @@
 from funcy.colls import project
 from funcy.seqs import concat, distinct, flatten
 
-from app.converter.fhir_to_fce import to_first_class_extension
-
 from .utils import prepare_link_ids, prepare_variables, validate_context
 
 WHITELISTED_ROOT_ELEMENTS = {
@@ -19,23 +17,29 @@
 PROPAGATE_ELEMENTS = ["itemContext", "itemPopulationContext"]
 
 
-async def assemble(client, fce_questionnaire):
+async def assemble(client, fce_questionnaire, to_first_class_extension_async):
     root_elements = project(dict(fce_questionnaire), WHITELISTED_ROOT_ELEMENTS.keys())
     fce_questionnaire["item"] = await _assemble_questionnaire(
-        client, fce_questionnaire, fce_questionnaire["item"], root_elements
+        client,
+        fce_questionnaire,
+        fce_questionnaire["item"],
+        root_elements,
+        to_first_class_extension_async,
     )
     dict.update(fce_questionnaire, root_elements)
     fce_questionnaire["assembledFrom"] = fce_questionnaire["id"]
     del fce_questionnaire["id"]
     return fce_questionnaire
 
 
-async def _load_sub_questionnaire(client, root_elements, parent_item, item):
+async def _load_sub_questionnaire(
+    client, root_elements, parent_item, item, to_first_class_extension_async
+):
     if "subQuestionnaire" in item:
         fhir_subq = (
             await client.resources("Questionnaire").search(_id=item["subQuestionnaire"]).get()
         )
-        fce_subq = to_first_class_extension(fhir_subq)
+        fce_subq = await to_first_class_extension_async(fhir_subq)
 
         variables = prepare_variables(item)
         if _validate_assemble_context(fce_subq, variables):
@@ -55,18 +59,25 @@ async def _load_sub_questionnaire(client, root_elements, parent_item, item):
     return item
 
 
-async def _assemble_questionnaire(client, parent, questionnaire_items, root_elements):
+async def _assemble_questionnaire(
+    client, parent, questionnaire_items, root_elements, to_first_class_extension_async
+):
     with_sub_items = questionnaire_items
     while len([i for i in with_sub_items if "subQuestionnaire" in i]) > 0:
         with_sub_items_futures = (
-            _load_sub_questionnaire(client, root_elements, parent, i) for i in with_sub_items
+            _load_sub_questionnaire(
+                client, root_elements, parent, i, to_first_class_extension_async
+            )
+            for i in with_sub_items
         )
         with_sub_items = list(flatten(await asyncio.gather(*with_sub_items_futures)))
 
     resp = []
     for i in with_sub_items:
         if "item" in i:
-            i["item"] = await _assemble_questionnaire(client, i, i["item"], root_elements)
+            i["item"] = await _assemble_questionnaire(
+                client, i, i["item"], root_elements, to_first_class_extension_async
+            )
         resp.append(i)
     return resp