example-editor.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pathlib import Path\n",
    "from yaml import load as loady, dump as dumpy, Loader, Dumper\n",
    "new_system = 'http://hl7.org/fhir/us/davinci-cdex/CodeSystem/cdex-temp'\n",
    "uscore_base = 'http://hl7.org/fhir/us/core'\n",
    "\n",
    "uscore_types = {\n",
    "'AllergyIntolerance': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-allergyintolerance|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-allergyintolerance|6.1.0'),\n",
    "'CarePlan': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-careplan|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-careplan|6.1.0'),\n",
    "'CareTeam': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-careteam|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-careteam|6.1.0'),\n",
    "'Condition': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-condition|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-condition-problems-health-concerns|6.1.0'),\n",
    "'Coverage': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-coverage|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-coverage|6.1.0'),\n",
    "'Device': (None, 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-implantable-device|6.1.0'),\n",
    "'DiagnosticReport': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-diagnosticreport-lab|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-diagnosticreport-lab|6.1.0'),\n",
    "'DocumentReference': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-documentreference|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-documentreference|6.1.0'),\n",
    "'Encounter': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-encounter|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-encounter|6.1.0'),\n",
    "'Goal': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-goal|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-goal|6.1.0'),\n",
    "'Immunization': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-immunization|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-immunization|6.1.0'),\n",
    "'Location': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-location|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-location|6.1.0'),\n",
    "'Medication': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-medication|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-medication|6.1.0'),\n",
    "'MedicationRequest': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-medicationrequest|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-medicationrequest|6.1.0'),\n",
    "'MedicationDispense': (None, 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-medicationdispense|6.1.0'),\n",
    "'Observation': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-observation-lab|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-observation-lab|6.1.0'),\n",
    "'Organization': (None, 'http://hl7.org/fhir/us/davinci-hrex/StructureDefinition/hrex-organization'),\n",
    "'Patient': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient|6.1.0'),\n",
    "'Practitioner': (None, 'http://hl7.org/fhir/us/davinci-hrex/StructureDefinition/hrex-practitioner'),\n",
    "'PractitionerRole': (None, 'http://hl7.org/fhir/us/davinci-hrex/StructureDefinition/hrex-practitionerrole'),\n",
    "'Procedure': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-procedure|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-procedure|6.1.0'),\n",
    "'Provenance': ('http://hl7.org/fhir/us/core/StructureDefinition/us-core-provenance|3.1.1', 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-provenance|6.1.0'),\n",
    "'RelatedPerson': (None, 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-relatedperson|6.1.0'),\n",
    "'ServiceRequest': (None, 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-servicerequest|6.1.0'),\n",
    "'Specimen': (None, 'http://hl7.org/fhir/us/core/StructureDefinition/us-core-specimen|6.1.0'),\n",
    "}\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "input/examples-yaml\n",
      "attachment-request-q found  in input/examples-yaml/cdex-task-inline-example22.yml and input = [{'system': 'http://hl7.org/fhir/us/davinci-pas/CodeSystem/PASTempCodes', 'code': 'questionnaires-needed'}]\n",
      "attachment-request-q found  in input/examples-yaml/cdex-task-example23.yml and input = [{'system': 'http://hl7.org/fhir/us/davinci-pas/CodeSystem/PASTempCodes', 'code': 'questionnaires-needed'}]\n",
      "attachment-request-q found  in input/examples-yaml/cdex-task-example22.yml and input = [{'system': 'http://hl7.org/fhir/us/davinci-pas/CodeSystem/PASTempCodes', 'code': 'questionnaires-needed'}]\n",
      "attachment-request-q found  in input/examples-yaml/cdex-task-example25.yml and input = [{'system': 'http://hl7.org/fhir/us/davinci-pas/CodeSystem/PASTempCodes', 'code': 'questionnaires-needed'}]\n",
      "attachment-request-q found  in input/examples-yaml/cdex-task-example24.yml and input = [{'system': 'http://hl7.org/fhir/us/davinci-pas/CodeSystem/PASTempCodes', 'code': 'questionnaires-needed'}]\n",
      "attachment-request-q found  in input/examples-yaml/cdex-task-example29.yml and input = [{'system': 'http://hl7.org/fhir/us/davinci-pas/CodeSystem/PASTempCodes', 'code': 'questionnaires-needed'}]\n",
      "attachment-request-q found  in input/examples-yaml/cdex-task-example28.yml and input = [{'system': 'http://hl7.org/fhir/us/davinci-pas/CodeSystem/PASTempCodes', 'code': 'questionnaires-needed'}]\n"
     ]
    }
   ],
   "source": [
    "from nested_lookup import nested_lookup\n",
    "# print(nested_lookup('severity', d , with_keys=True))\n",
    "\n",
    "def get_paths(d, my_key, value, current = ''):\n",
    "#   print(f'value = {value}')\n",
    "  for a, b in d.items():\n",
    "    # print(f'Key = {a}, Value = {b} b.type = {type(b)} b == value ?: {b == value}')\n",
    "\n",
    "    # print(f'Current = {current}')\n",
    "    if a==my_key and b == value:\n",
    "      current = current+f\"['{a}']\"\n",
    "      print(f'Found {value} at {current}')\n",
    "      yield current\n",
    "    elif isinstance(b, dict):\n",
    "    #   print(f'Key = {a}, Dict Value = {b}')\n",
    "      yield from get_paths(b, my_key, value, current+f\"['{a}']\",)\n",
    "    elif isinstance(b, list):\n",
    "      if value in b:\n",
    "        yield current\n",
    "      for i,j in enumerate(b):\n",
    "        # print(f'List item {i} = {j}')\n",
    "        try:\n",
    "          yield from get_paths(j, my_key, value, current+f\"['{a}'][{i}]\",)\n",
    "        except AttributeError:\n",
    "            pass\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "new_type = [{'system': 'http://hl7.org/fhir/us/davinci-pas/CodeSystem/PASTempCodes', 'code': 'questionnaires-needed'}]\n",
    "\n",
    "in_path = Path(r'input/examples-yaml')\n",
    "print(in_path)\n",
    "for counter,i in enumerate(in_path.glob('*.yml')):\n",
    "    # print(f'=== {counter}: {i} ===')\n",
    "    obj = loady(i.read_text(), Loader=Loader)\n",
    "    \n",
    "    if obj['resourceType'] == \"Task\":\n",
    "        # print (f'Found {i}')\n",
    "        # print (f'input = {obj[\"input\"][0][\"type\"]}')\n",
    "        if 'attachment-request-questionnaire' == obj['code'][\"coding\"][0]['code']:\n",
    "           print(f'attachment-request-q found  in {i} and input = {obj[\"input\"][0][\"type\"][\"coding\"]}')\n",
    "          # obj['input'][0]['type'][\"coding\"]= new_type\n",
    "          # print (f'output = {obj[\"input\"][0][\"type\"]}')\n",
    "          # i.write_text(dumpy(obj, Dumper=Dumper,sort_keys=False)) # update only if US Core type found\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "    \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "    example_types = set(nested_lookup('resourceType', obj))\n",
    "    if example_types & set(uscore_types): # only process if US Core type found\n",
    "      for example_type in example_types:\n",
    "          if example_type in uscore_types:\n",
    "            print(f'US Core type: {example_type}')\n",
    "            paths = list(get_paths(obj,'resourceType', example_type))\n",
    "            print(paths)\n",
    "            print()\n",
    "            for path in paths:\n",
    "              \n",
    "              print(f'Path = {path}')\n",
    "              exec(f\"print(obj{path})\")\n",
    "              if 'contained' in path:\n",
    "                add_meta = input(f'Add meta for {example_type} profile? Type \"Y\" or \"N\"')\n",
    "                if add_meta != 'Y':\n",
    "                  print(\"=========== Don't add meta profile to Contained resource ===========\")\n",
    "                  continue\n",
    "\n",
    "              try:\n",
    "                exec(f\"\"\"print(obj{path.replace(\"['resourceType']\",\"['meta']['profile']\")})\"\"\")\n",
    "              except KeyError:\n",
    "                print('====== No meta profile ======')\n",
    "              new_url = [url for url in uscore_types[example_type] if url]\n",
    "              print(f'new_url ={new_url}')\n",
    "\n",
    "              try:\n",
    "                  exec(f\"\"\"obj{path.replace(\"['resourceType']\",\"['meta']['profile']\")}={new_url}\"\"\")\n",
    "              except KeyError:\n",
    "                  exec(f\"\"\"obj{path.replace(\"['resourceType']\",\"['meta']\")} = {{'profile': new_url}}\"\"\")\n",
    "\n",
    "              # print(dumpy(obj, Dumper=Dumper,sort_keys=False))\n",
    "\n",
    "\n",
    "\n",
    "      i.write_text(dumpy(obj, Dumper=Dumper,sort_keys=False)) # update only if US Core type found"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.10.2 64-bit ('jupyter')",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.2"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "18672f2fb81dcbb023a0ac48a86c775a19488782369d96ffbd3382a22fd285cf"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}