csg_datafusion_dicoms_extract.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# DICOMs infos extractor\n",
    "By Stephen Larroque @ Coma Science Group, GIGA Research, University of Liege\n",
    "Creation date: 2018-02-17\n",
    "License: MIT\n",
    "v1.4.1\n",
    "\n",
    "DESCRIPTION:\n",
    "This script extract metadata infos (patients names, scan date, etc) from all dicoms recursively from the specified folder.\n",
    "The script expects that the root folder contains either one folder per subject or one zip file per subject, because only one dicom file will be read. If there are multiple subjects in one folder or zip, then only the first will be included, the rest will be skipped (to save file walking time).\n",
    "Note: it is possible to provide a list of folders where all dicoms reside, each rootfolder will be processed one after the other.\n",
    "\n",
    "INSTALL NOTE:\n",
    "You need to pip install pandas before launching this script.\n",
    "Tested on Python 2.7.13\n",
    "\n",
    "USAGE:\n",
    "Modify the parameters below, the most important being the dicom root folder(s), where all your subjects' dicoms folders/zips can be found.\n",
    "\n",
    "TODO:\n",
    "* make a dicom checker script: each folder/zip should contain at least one readable dicom, else there is a problem + report if multiple patient names/date of acquisition in the same folder/zip!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Forcefully autoreload all python modules\n",
    "%load_ext autoreload\n",
    "%autoreload 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# IMPORT AUX FUNCTIONS\n",
    "\n",
    "import collections\n",
    "import os, sys\n",
    "import shutil\n",
    "import zipfile\n",
    "import re\n",
    "\n",
    "cur_path = os.path.realpath('.')\n",
    "sys.path.append(os.path.join(cur_path, 'csg_fileutil_libs'))  # for unidecode and cleanup_name, because it does not support relative paths (yet?)\n",
    "\n",
    "# For DB reorganization\n",
    "from csg_fileutil_libs.aux_funcs import save_dict_as_csv, save_df_as_csv, _tqdm, cleanup_name, recwalk, _StringIO\n",
    "\n",
    "# For Dicom reading\n",
    "import csg_fileutil_libs.pydicom as dicom\n",
    "from csg_fileutil_libs.pydicom.filereader import InvalidDicomError\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# PARAMETERS\n",
    "\n",
    "# Dicoms database\n",
    "# the rootpath_to_dicoms need to be a string of the folder containing one dicom folder or zip file per subject (so there can be lots of subjects, but one folder/zip per subject from the given path)\n",
    "# rootpath_to_dicoms can also be a list of strings/folders\n",
    "rootpath_to_dicoms = [r'H:\\ALLDICOMS\\PATIENTS\\NON_SEDATED', r'H:\\ALLDICOMS\\PATIENTS\\NON_SEDATED_TOORDER', r'H:\\ALLDICOMS\\PATIENTS\\SEDATED', r'H:\\ALLDICOMS\\PATIENTS\\UNKNOWN']  # rootpath where dicoms are stored (one subject per folder/zip). Can specify a list of paths to process multiple folders one after the other.\n",
    "#rootpath_to_dicoms = [r'G:\\Topreproc\\ReportsTun\\csg_fileutil_v2.2.0\\dicomtest\\a', r'G:\\Topreproc\\ReportsTun\\csg_fileutil_v2.2.0\\dicomtest\\b']\n",
    "csv_output = r'dicoms_db_subjects_new.csv'  # where to save the list of subjects\n",
    "csv_output2 = r'dicoms_db_infos_new.csv'  # where to save the list of subjects AND the extracted additional fields infos\n",
    "additional_fields = ['AcquisitionDate', 'PatientID', 'SeriesDescription', 'ProtocolName']  # additional fields to extract from the dicoms headers\n",
    "walk_all_dicoms = True  # if False, will extract infos from the first dicom found. If True, will recurse until all dicoms have been read and additional fields extracted for all dicoms (will be stored in set() so as to avoid duplication), this will ensure that you do not miss any info at the expense of (way) longer calculations.\n",
    "find_dicoms_matching = {'ProtocolName': ['dti', 'repos']}  # if you are looking for dicoms matching specific parameters, you can specify the matching here, the result will be saved in another csv. Format: {'DicomAttribute': ['first_attribute_to_match', 'second_attribute_to_match']}. It's an AND test, so it expects all the specified parameters to be found to return True for the current subject.\n",
    "csv_output3 = r'dicoms_db_infosmatch_new.csv'  # where to save the list of subjects matching the find_dicoms_matching patterns\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# MORE AUX FUNCTIONS\n",
    "def get_list_of_folders(rootpath):\n",
    "    return [item for item in os.listdir(rootpath) if os.path.isdir(os.path.join(rootpath, item))]\n",
    "\n",
    "def get_list_of_zip(rootpath):\n",
    "    return [item for item in os.listdir(rootpath) if os.path.isfile(os.path.join(rootpath, item)) and item.endswith('.zip')]\n",
    "\n",
    "def get_dcm_names_from_dir(rootpath, dcm_subj_list=None, folder_to_name=None, add_fields=None, walk_all_dicoms=False, verbose=False):\n",
    "    if dcm_subj_list is None:\n",
    "        dcm_subj_list = []  # store list of subjects names from dicom files (useful for csv filtering)\n",
    "    if folder_to_name is None:\n",
    "        folder_to_name = {}  # store the name of the patient stored in each root folder (useful for anonymization later on)\n",
    "    if add_fields is not None:\n",
    "        additional_infos = {}  # store all additional fields extracted from dicoms\n",
    "    for subject in _tqdm(get_list_of_folders(rootpath), desc='DIR'):\n",
    "        if verbose:\n",
    "            print('- Processing subject %s' % unicode(subject, 'latin1'))\n",
    "        fullpath = os.path.join(rootpath, subject)\n",
    "        if not isinstance(fullpath, unicode):\n",
    "            fullpath = unicode(fullpath, 'latin1')\n",
    "        pts_name = None\n",
    "        for dirpath, filename in recwalk(fullpath, filetype=['.dcm', '']):\n",
    "            try:\n",
    "                #print('* Try to read fields from dicom file: %s' % os.path.join(dirpath, filename))\n",
    "                # Read the dicom data in memory (via StringIO)\n",
    "                dcmdata = dicom.read_file(os.path.join(dirpath, filename), stop_before_pixels=True)  # stop_before_pixels allow for faster processing since we do not read the full dicom data, and here we can use it because we do not modify the dicom, we only read it to extract the dicom patient name\n",
    "                #print(dcmdata.PatientName)\n",
    "                # Extract and cleanup the patient's name\n",
    "                pts_name = cleanup_name(dcmdata.PatientName)\n",
    "                # Add to the list of names\n",
    "                if (not walk_all_dicoms or len(dcm_subj_list) == 0 or pts_name != dcm_subj_list[-1]):  # add only if the name is not already in the list\n",
    "                    dcm_subj_list.append( pts_name )\n",
    "                # Extract additional fields\n",
    "                if add_fields:\n",
    "                    additional_infos = add_dicom_fields(additional_infos, dcmdata, pts_name, add_fields, walk_all_dicoms)\n",
    "                # Stop here after the first valid dicom file found, except if we want to extract ALL data\n",
    "                if not walk_all_dicoms:\n",
    "                    break\n",
    "            except (InvalidDicomError, AttributeError, OverflowError) as exc:\n",
    "                pass\n",
    "        folder_to_name[subject] = pts_name\n",
    "    return dcm_subj_list, folder_to_name, additional_infos\n",
    "\n",
    "def get_dcm_names_from_zip(rootpath, dcm_subj_list=None, folder_to_name=None, add_fields=None, walk_all_dicoms=False, verbose=False):\n",
    "    if dcm_subj_list is None:\n",
    "        dcm_subj_list = []  # store list of subjects names from dicom files (useful for csv filtering)\n",
    "    if folder_to_name is None:\n",
    "        folder_to_name = {}  # store the name of the patient stored in each root folder (useful for anonymization later on)\n",
    "    if add_fields is not None:\n",
    "        additional_infos = {}  # store all additional fields extracted from dicoms\n",
    "    # Extract names from zipped dicom files (extract the first dicom file we can read and use its fields)\n",
    "    for zipfilename in _tqdm(get_list_of_zip(rootpath), desc='ZIP'):\n",
    "        zfilepath = os.path.join(rootpath, zipfilename)\n",
    "        if verbose:\n",
    "            print('- Processing file %s' % zipfilename)\n",
    "        try:\n",
    "            with zipfile.ZipFile(zfilepath, 'r') as zipfh:\n",
    "                # Extract only files, not directories (end with '/', this is standard detection in zipfile)\n",
    "                zfolder = (item for item in zipfh.namelist() if item.endswith('/'))\n",
    "                zfiles = (item for item in zipfh.namelist() if not item.endswith('/'))\n",
    "                # Get first top folder inside zip to extract folder name (because when we will extract the zip, we need the folder name)\n",
    "                try:\n",
    "                    folder_name = zfolder.next().strip('/')\n",
    "                except StopIteration:\n",
    "                    folder_name = re.search('^([^\\\\/]+)[\\\\/]', zipfh.namelist()[0]).group(1)\n",
    "                # Get first dicom file we can find\n",
    "                pts_name = None\n",
    "                for zf in zfiles:\n",
    "                    # Need to extract because pydicom does not support not having seek() (and zipfile in-memory does not provide seek())\n",
    "                    z = _StringIO(zipfh.read(zf)) # do not use .extract(), the path can be anything and it does not support unicode (so it can easily extract to the root instead of target folder!)\n",
    "                    # Try to open the extracted dicom\n",
    "                    try:\n",
    "                        if verbose:\n",
    "                            print('Try to decode dicom fields with file %s' % zf)\n",
    "                        # Read the dicom data in memory (via StringIO)\n",
    "                        dcmdata = dicom.read_file(z, stop_before_pixels=True)  # stop_before_pixels allow to only read headers, not the pixel infos (a lot faster)\n",
    "                        # Extract and cleanup the patient's name\n",
    "                        pts_name = cleanup_name(dcmdata.PatientName)\n",
    "                        # Add to the list of names\n",
    "                        if (not walk_all_dicoms or len(dcm_subj_list) == 0 or pts_name != dcm_subj_list[-1]):  # add only if the name is not already in the list\n",
    "                            dcm_subj_list.append( pts_name )\n",
    "                        # Extract additional fields\n",
    "                        if add_fields:\n",
    "                            additional_infos = add_dicom_fields(additional_infos, dcmdata, pts_name, add_fields, walk_all_dicoms)\n",
    "                        if not walk_all_dicoms:\n",
    "                            break\n",
    "                    except (InvalidDicomError, AttributeError, OverflowError) as exc:\n",
    "                        continue\n",
    "                    except IOError as exc:\n",
    "                        if 'no tag to read' in str(exc).lower():\n",
    "                            continue\n",
    "                        else:\n",
    "                            raise\n",
    "                # Add to the folder name -> dicom patient name mapping\n",
    "                folder_to_name[folder_name] = pts_name\n",
    "        except zipfile.BadZipfile as exc:\n",
    "            # If the zipfile is unreadable, just pass\n",
    "            continue\n",
    "    return dcm_subj_list, folder_to_name, additional_infos\n",
    "\n",
    "def add_dicom_fields(additional_infos, dcmdata, pts_name, add_fields, walk_all_dicoms=False):\n",
    "    \"\"\"Add dicom fields in the provided additional_infos dict (can be an empty dict)\"\"\"\n",
    "    dictid = '%s|%s' % (pts_name, dcmdata.AcquisitionDate)\n",
    "    for field in add_fields:\n",
    "        if field in dcmdata:\n",
    "            if walk_all_dicoms:\n",
    "                if not dictid in additional_infos:\n",
    "                    additional_infos[dictid] = {}\n",
    "                if not field in additional_infos[dictid]:\n",
    "                    additional_infos[dictid][field] = set()\n",
    "                if isinstance(field, str):\n",
    "                    additional_infos[dictid][field].add(dcmdata[dcmdata.data_element(field).tag].value)\n",
    "                else:\n",
    "                    additional_infos[dictid][field].add(dcmdata[field].value)\n",
    "            else:\n",
    "                if isinstance(field, str):\n",
    "                    additional_infos[dictid][field] = dcmdata[dcmdata.data_element(field).tag].value\n",
    "                else:\n",
    "                    additional_infos[dictid][field] = dcmdata[field].value\n",
    "    return additional_infos\n",
    "\n",
    "\n",
    "def dict_merge(dct, merge_dct, add_keys=True):\n",
    "    \"\"\" Recursive dict merge. Inspired by :meth:``dict.update()``, instead of\n",
    "    updating only top-level keys, dict_merge recurses down into dicts nested\n",
    "    to an arbitrary depth, updating keys. The ``merge_dct`` is merged into\n",
    "    ``dct``.\n",
    "\n",
    "    This version will return a copy of the dictionary and leave the original\n",
    "    arguments untouched.\n",
    "\n",
    "    The optional argument ``add_keys``, determines whether keys which are\n",
    "    present in ``merge_dict`` but not ``dct`` should be included in the\n",
    "    new dict.\n",
    "    \n",
    "    By DomWeldon: https://gist.github.com/angstwad/bf22d1822c38a92ec0a9#gistcomment-2622319\n",
    "\n",
    "    Args:\n",
    "        dct (dict) onto which the merge is executed\n",
    "        merge_dct (dict): dct merged into dct\n",
    "        add_keys (bool): whether to add new keys\n",
    "\n",
    "    Returns:\n",
    "        dict: updated dict\n",
    "    \"\"\"\n",
    "    try:\n",
    "        dct = dct.copy()\n",
    "    except Exception as exc:\n",
    "        print(dct)\n",
    "        raise\n",
    "    if not add_keys:\n",
    "        merge_dct = {\n",
    "            k: merge_dct[k]\n",
    "            for k in set(dct).intersection(set(merge_dct))\n",
    "        }\n",
    "\n",
    "    for k, v in merge_dct.items():\n",
    "        if (k in dct and isinstance(dct[k], dict)\n",
    "                and isinstance(merge_dct[k], collections.Mapping)):\n",
    "            dct[k] = dict_merge(dct[k], merge_dct[k], add_keys=add_keys)\n",
    "        else:\n",
    "            dct[k] = merge_dct[k]\n",
    "\n",
    "    return dct"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Extract subjects lists and infos from each rootpath\n",
    "if not isinstance(rootpath_to_dicoms, list):\n",
    "    # Convert to a list if it's a string\n",
    "    rootpath_to_dicoms = [rootpath_to_dicoms]\n",
    "# Initialize variables\n",
    "dcm_subj_list = []\n",
    "folder_to_name = {}\n",
    "additional_infos = {}\n",
    "paths_list = []\n",
    "# For each rootpath, extract infos\n",
    "for rootpath in rootpath_to_dicoms:\n",
    "    dcm_subj_list, folder_to_name, additional_infos1 = get_dcm_names_from_dir(rootpath, dcm_subj_list=dcm_subj_list, folder_to_name=folder_to_name, add_fields=additional_fields, walk_all_dicoms=walk_all_dicoms)\n",
    "    dcm_subj_list, folder_to_name, additional_infos2 = get_dcm_names_from_zip(rootpath, dcm_subj_list=dcm_subj_list, folder_to_name=folder_to_name, add_fields=additional_fields, walk_all_dicoms=walk_all_dicoms)\n",
    "    additional_infos = dict_merge(additional_infos, additional_infos1)\n",
    "    additional_infos = dict_merge(additional_infos, additional_infos2)\n",
    "    paths_list.extend([rootpath] * (len(additional_infos1)+len(additional_infos2)))\n",
    "# Save the names list\n",
    "save_dict_as_csv([{'name': x} for x in dcm_subj_list], csv_output, csv_order_by='name')\n",
    "# Display the result!\n",
    "dcm_subj_list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "additional_infos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "len(additional_infos.keys())\n",
    "len(paths_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Convert the additional infos to a pandas DataFrame and save it as csv\n",
    "# TODO: convert sets to lists\n",
    "import pandas as pd\n",
    "df_additional_infos = pd.DataFrame(additional_infos).transpose()\n",
    "df_additional_infos['rootpath'] = paths_list # add rootpath where each dicom was found, can then later be used for filtering\n",
    "df_additional_infos.index.set_names(['name']) # add a name to the index (so that the csv column is named)\n",
    "#df_additional_infos.drop_duplicates()\n",
    "save_df_as_csv(df_additional_infos, csv_output2, keep_index=True)\n",
    "df_additional_infos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "def find_in_dicoms(df, find={}, partialmatch=True):\n",
    "    \"\"\"Given a serie and a dictionary of lists, will return True for all records that match all the provided whitelist, and False otherwise\n",
    "    This allows to find what dicoms match the parameters you are looking for (eg, having both a dti and bold series).\n",
    "    to be used in a lambda apply: eg, df.apply(lambda x: find_in_dicoms(x, find={'ProtocolName': ['dti', 'bold']}))\"\"\"\n",
    "    for key, vals in find.items():\n",
    "        if not df[key] or pd.isnull(df[key]):\n",
    "            return False\n",
    "        for val in vals:\n",
    "            if partialmatch:\n",
    "                if not re.search(val, ' '.join(df[key]).lower(), re.I):\n",
    "                    return False\n",
    "            else:\n",
    "                if not val in df[key]:\n",
    "                    return False\n",
    "    return True\n",
    "\n",
    "if find_dicoms_matching:\n",
    "    df_match = df_additional_infos.apply(lambda df: find_in_dicoms(df, find=find_dicoms_matching), axis=1)\n",
    "    save_df_as_csv(df_match.to_frame().reset_index().rename(columns={0: 'name', 1: 'match'}), csv_output3, keep_index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.15"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}