Merge pull request #11 from FHIR-Aggregator/development

release

fhir_query/cli.py

@@ -3,10 +3,12 @@
 import logging
 import sys
 from collections import defaultdict
+from typing import Any
 
 import pandas as pd
 
 import click
+import requests
 from click_default_group import DefaultGroup
 import yaml
 from fhir.resources.graphdefinition import GraphDefinition
@@ -15,6 +17,7 @@
 from fhir_query import GraphDefinitionRunner, setup_logging, VocabularyRunner
 from fhir_query.dataframer import Dataframer
 from fhir_query.visualizer import visualize_aggregation
+from fhir_query.vocabulary import vocabulary_simplifier
 
 
 @click.group(cls=DefaultGroup, default="main")
@@ -96,33 +99,49 @@ async def run_runner() -> None:
 
 @cli.command()
 @click.option("--fhir-base-url", required=True, help="Base URL of the FHIR server.")
+@click.option("--raw", is_flag=True, default=False, help="Do not create a dataframe. default=False")
+@click.option("--tsv", is_flag=True, default=True, help="Render dataframe as tsv. default=True")
 @click.option("--debug", is_flag=True, help="Enable debug mode.")
-@click.option("--log-file", default="app.log", help="Path to the log file.")
-def vocabularies(
+@click.option("--log-file", default="./fq.log", help='Path to the log file. default="./fq.log"')
+@click.argument("output_path", type=click.File("w"), required=False, default=sys.stdout)
+def vocabulary(
     fhir_base_url: str,
+    output_path: click.File,
     debug: bool,
     log_file: str,
+    raw: bool,
+    tsv: bool,
 ) -> None:
-    """Collect vocabularies (CodeableConcepts) from key resources."""
+    """Retrieve Vocabulary Observation and ResearchStudy resources from the FHIR server.
+    \b
+
+    OUTPUT_PATH: Path to the output file. If not provided, the output will be printed to stdout.
+    """
 
     setup_logging(debug, log_file)
 
     if fhir_base_url.endswith("/"):
         fhir_base_url = fhir_base_url[:-1]
 
-    async def collect_vocabularies(_runner: VocabularyRunner, _spinner: Halo) -> list:
-        _counts = await _runner.collect(
-            resource_types=["Observation", "Condition", "Procedure", "Medication", "Specimen", "Encounter", "DocumentReference"],
-            spinner=_spinner,
-        )  #
-        return _counts
+    output_stream: Any = output_path
 
     try:
         with Halo(text="Collecting vocabularies", spinner="dots", stream=sys.stderr) as spinner:
-            runner = VocabularyRunner(fhir_base_url)
-            counts = asyncio.run(collect_vocabularies(runner, spinner))
-            yaml_results = yaml.dump(counts, default_flow_style=False)
-        print(yaml_results)
+            query_url = f"{fhir_base_url}/Observation?code=vocabulary&_include=Observation:focus"
+            response = requests.get(query_url, timeout=300)
+            response.raise_for_status()
+            bundle = response.json()
+            results = bundle
+            if not raw:
+                results = vocabulary_simplifier(bundle)
+            if not tsv:
+                yaml_results = yaml.dump(results, default_flow_style=False, sort_keys=False)
+                print(yaml_results, file=output_stream)
+            else:
+                df = pd.DataFrame(results)
+                df.to_csv(output_stream, sep="\t", index=False)
+            spinner.succeed(f"Wrote {len(results)} vocabularies to {output_stream.name}")
+
     except Exception as e:
         logging.error(f"Error: {e}", exc_info=True)
         click.echo(f"Error: {e}", file=sys.stderr)

fhir_query/vocabulary.py

@@ -0,0 +1,72 @@
+def _get_path(component):
+    """Get the vocabulary path from the component."""
+    for coding in component.get("code", {}).get("coding", []):
+        if coding.get("system", "") == "http://fhir-aggregator.org/fhir/CodeSystem/vocabulary/path":
+            return coding.get("code", None)
+
+
+def _get_coding(component):
+    """Get the vocabulary codeable from the component."""
+    for coding in component.get("code", {}).get("coding", []):
+        if coding.get("system", "") != "http://fhir-aggregator.org/fhir/CodeSystem/vocabulary/path":
+            return coding
+
+
+def vocabulary_simplifier(bundle) -> list[dict]:
+    """Simplify the vocabulary bundle."""
+    df = []
+    resources = {f'{r["resource"]["resourceType"]}/{r["resource"]["id"]}': r["resource"] for r in bundle.get("entry", [])}
+    for _id, resource in resources.items():
+        if resource["resourceType"] != "Observation":
+            continue
+        focus = next(iter(resource.get("focus", [])), None)
+        assert focus, f"No focus found for Observation {resource['id']}"
+        focus_reference = focus.get("reference", None)
+        research_study = resources.get(focus_reference, None)
+        assert research_study, f"No research_study reference found for Observation {resource['id']} {focus_reference}"
+        for component in resource.get("component", []):
+            path = _get_path(component)
+            coding = _get_coding(component)
+            item = {
+                "research_study_identifiers": ",".join([i.get("value", "") for i in research_study.get("identifier", [])]),
+                "path": path,
+            }
+            if path.endswith(".extension"):
+                item.update(
+                    {
+                        "code": coding.get("code", None) if coding.get("code", None) != "range" else None,
+                        "display": coding.get("display", None) if coding.get("display", None) != "range" else None,
+                        "system": None,
+                        "extension_url": coding.get("system", None),
+                    }
+                )
+            else:
+                item.update(
+                    {
+                        "code": coding.get("code", None),
+                        "display": coding.get("display", None),
+                        "system": coding.get("system", None),
+                        "extension_url": None,
+                    }
+                )
+
+            if "valueInteger" in component:
+                item["count"] = component["valueInteger"]
+            else:
+                item["count"] = None
+            if "valueRange" in component:
+                item["low"] = component["valueRange"].get("low", {}).get("value", None)
+                item["high"] = component["valueRange"].get("high", {}).get("value", None)
+            else:
+                item["low"] = None
+                item["high"] = None
+            item.update(
+                {
+                    "research_study_title": research_study.get("title", None),
+                    "research_study_description": research_study.get("description", None),
+                    "observation": f'Observation/{resource["id"]}',
+                    "research_study": f'ResearchStudy/{research_study["id"]}',
+                }
+            )
+            df.append(item)
+    return df