{"cells":[{"metadata":{},"cell_type":"markdown","source":"<h1>Preprocessing Medical Images (DICOM files) with Apache Beam</h1>\n\n<strong>This notebook demonstrates how to use Apache Beam to convert medical images (in the DICOM format) into TFRecords. </strong>\n\nAn interactive version of the notebook is available on Kaggle at:<br>\nhttps://www.kaggle.com/spiroganas/preprocessing-medical-images-with-apache-beam\n<hr>\n\nThe dataset for Kaggle's [\"RSNA STR Pulmonary Embolism Detection\" competition](https://www.kaggle.com/c/rsna-str-pulmonary-embolism-detection/) includes approximately 1.94 million medical images, which consume about 912 gigabytes of hard drive space.  <strong><em>You may not believe me, but this is \"small data\"</em></strong> (you can easily store the entire dataset on a hard drive that costs less than $200). When you start dealing with data at a healthcare system like England's NHS, which serves 56 million people, you simply can't run your analysis on a single computer.  If your gonna work with real <strong>BIG DATA</strong>, you need Apache Beam.\n\n[Apache Beam](https://beam.apache.org/get-started/beam-overview/) is \"particularly useful for [Embarrassingly Parallel](https://en.wikipedia.org/wiki/Embarrassingly_parallel) data processing tasks\".  Basically, it let's you run your code on hundreds or thousands of computers at once.\n\nThis is a fairly basic example:\n* The files are stored on a single, local hard drive.\n* The Apache Beam pipeline is running on [Direct Runner](https://beam.apache.org/documentation/runners/direct/), a simple, local execution tool designed for testing.\n* The output is TFRecord files\n\n\"[The TFRecord format is a simple format for storing a sequence of binary records.](https://www.tensorflow.org/tutorials/load_data/tfrecord)\"  TensorFlow can quickly read data stored in the TFRecord format.  This is especially important when the model is being trained on fast GPUs or [TPUs](https://cloud.google.com/tpu) (where IO is frequently the training bottleneck).\n\n\n\nIf you were running this on real-world data, the data would probably be stored in Google Cloud Storage or AWS S3, and the pipeline would run on [Apache Spark](https://beam.apache.org/documentation/runners/spark/) or [Google DataFlow](https://cloud.google.com/dataflow). \n\nA lot of the image preprocessing code is based on: https://www.kaggle.com/gzuidhof/full-preprocessing-tutorial\n\n\n\n"},{"metadata":{"trusted":true},"cell_type":"code","source":"# define a few constants to keep track of the folders we will be using\nWORKING_FOLDER = '/kaggle/working/'\nTEMP_FOLDER = '/kaggle/temp/'\n\nTRAIN_FOLDER = '/kaggle/input/rsna-str-pulmonary-embolism-detection/train'\nTEST_FOLDER = '/kaggle/input/rsna-str-pulmonary-embolism-detection/test'\n\nLABELS_FILE = '/kaggle/input/rsna-str-pulmonary-embolism-detection/train.csv'\n\n# Change this to True if you want to print all the arrays (which make the notebook really hard to read...)\nVERBOSE = False","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Step 1: Install and import the libraries we will be using\n\n* [Pydicom](https://pydicom.github.io/) is a library that reads DICOM files\n* Some DICOM files contain images that are compressed.  I used [GDCM](https://en.wikipedia.org/wiki/GDCM) to decompress images in this Kaggle notesbook.  If GDCM won't work on your computer, you can also use the pylibjpeg, pylibjpeg-libjpeg, and pylibjpeg-openjpeg libraries.\n* opencv is a computer vision library.  We will us it to resize the image and to change it from grayscale to RGB color.\n* apache-beam makes it easy to run your pipeline on a cluster of computers.\n* The Kaggle Notebook already has some popular data science libraries installed (numpy, tensorflow 2, and pandas).  So if you're runnning this on your own machine, you will need to install those.\n\n\n"},{"metadata":{"trusted":true},"cell_type":"code","source":"# Install the packages we will be using to read and process the DICOM Files\n!apt-get -qq update && apt-get -qq install -y  libgdcm-tools python-gdcm\n!conda install -y -q -c conda-forge opencv gdcm pydicom apache-beam -y","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"import numpy as np\nimport tensorflow as tf\nimport apache_beam\nimport pydicom\nimport gdcm\nimport cv2 as cv\nimport matplotlib.pyplot as plt\nimport os\nimport sys\nimport csv\nimport pickle\nimport random\nfrom datetime import datetime\nprint('Libraries Imported!')\n\n%matplotlib inline\nnp.set_printoptions(threshold=sys.maxsize)\nprint('Settings Set!')","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Step 2:  Get a list of the DICOM files you want to convert to TFRecords\n\nThere are several ways you can do this:\n1. Hardcode the filenames as a list (only really useful when you're developing/testing).\n2. Store the filenames in a text or Excel file and read them into a list.\n3. Use os.walk or glob. \n4. If your files are on AWS, use the boto3 Python library.\n5. If your files are on Google Cloud Storage, use the google-cloud-storage Python library.\n\nHowever you get/create your list of files, they will be the input to the pipeline.  Apache Beam will split the list and send a sub-list to each of the worker computers in your cluster.\n\n(In this Kaggle Notebook example, we are using direct runner, so the entire list of files will be processed on a single machine.  But Apache Beam's raison d'être is to farm out tasks to dozens/hundreds/thousands of computer!)"},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"# This will get a list of all the Kaggle Data\n\n\ndef list_dicoms_in_folder(folder):\n    '''Lists the full path to all the DICOM files in a given folder.'''\n    my_list = []\n    for dirname, _, filenames in os.walk(folder):\n        for filename in filenames:\n            if filename[-4:]=='.dcm':\n                my_list.append(os.path.join(dirname, filename))\n    return my_list\n\n\n# Change this to True if you want to do something with all the data\n# For testing purposes, you can just use the small list of files defined in the next cell\nif False:\n    Train_Data = list_dicoms_in_folder(TRAIN_FOLDER)\n    Test_Data = list_dicoms_in_folder(TEST_FOLDER)\n\n\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# This creates a very small dataset for testing\nSMALL_DATA = ['/kaggle/input/rsna-str-pulmonary-embolism-detection/train/0003b3d648eb/d2b2960c2bbf/0787742383e4.dcm',\n                        '/kaggle/input/rsna-str-pulmonary-embolism-detection/train/0003b3d648eb/d2b2960c2bbf/10454bf652e0.dcm',\n                        '/kaggle/input/rsna-str-pulmonary-embolism-detection/train/00cf4b2b751b/540a03df1d81/04a5bfe1bf00.dcm',]\nprint(SMALL_DATA)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Step 3:  Create a dictionary that maps medical images to their labels.\n\nWe need to link each medical image to the label that we want to predict.\n\nMedical images are uniquely identified by the SOP Instance UID field.  So that will be the key in our dictionary.\n\n1. The labels are stored in a csv file.  \n2. The labels will be passed to the Apache Beam pipeline as a [Side Input](https://beam.apache.org/documentation/patterns/side-inputs/).\n\n"},{"metadata":{"trusted":true},"cell_type":"code","source":"def create_labels_dict(labels_csv):\n    \"\"\"Reads in a csv file and outputs a dictionary of the SOP Instance UID\n    and a list of the other columns.\"\"\"\n    labels_dict = {}\n    with open(labels_csv, newline=\"\") as csvfile:\n        my_reader = csv.reader(csvfile, delimiter=\",\", quotechar=\"|\")\n        next(my_reader)\n        for row in my_reader:\n            labels_dict[row[2]] = {\n                \"pe_present_on_image\": int(row[3]),\n                \"negative_exam_for_pe\": int(row[4]),\n                \"qa_motion\": int(row[5]),\n                \"qa_contrast\": int(row[6]),\n                \"flow_artifact\": int(row[7]),\n                \"rv_lv_ratio_gte_1\": int(row[8]),\n                \"rv_lv_ratio_lt_1\": int(row[9]),\n                \"leftsided_pe\": int(row[10]),\n                \"chronic_pe\": int(row[11]),\n                \"true_filling_defect_not_pe\": int(row[12]),\n                \"rightsided_pe\": int(row[13]),\n                \"acute_and_chronic_pe\": int(row[14]),\n                \"central_pe\": int(row[15]),\n                \"indeterminate\": int(row[16]),\n            }\n\n    return labels_dict\n    \n    \nlabels_dict = create_labels_dict('/kaggle/input/rsna-str-pulmonary-embolism-detection/train.csv')","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Step 4: An example of reading a DICOM Medical Image\n\nIn this step, we are going to show how to extract the numpy array containing the medical image from the DICOM file.\n\nSo you can think of a DICOM file as if it were a dictionary of key/value pairs.  The keys are called \"Headers\" or \"Tags\" and they contain information about the image.  These tags will include things like:\n* The machine used to generate the image.\n* The date the image was taken.\n* The part of the body in the image.\n* Patient Identifiers\n\nThe pixel_array tag stores a numpy array that represents the actual medical image.  \n\nSometimes the image data is compressed.  There are a lot of valid ways to compress the image (see the TransferSyntaxUID_dict in the code below).  The TransferSyntaxUID tag will tell you how an image was compressed.  \n\nPydicom wasn't designed to decompress images, which is why we had to install GDCM."},{"metadata":{"trusted":true},"cell_type":"code","source":"element = SMALL_DATA[0]\nds = pydicom.dcmread(element)\n\n# The kaggle images are compressed, so pydicom needs to use the GDCM software: https://pydicom.github.io/pydicom/stable/old/image_data_handlers.html\n# The TransferSyntaxUID tells you what image compression method was applied to the image\n# Dictinary that translates UID to a text description is from: http://dicom.nema.org/medical/dicom/2018a/output/chtml/part06/chapter_A.html\n\ndef TransferSyntaxDescription(ds):\n    '''Translates the TransferSyntaxUID into a text description of the image compression method used.\n    '''\n\n    TransferSyntaxUID_dict = {\n        '1.2.840.10008.1.2' : 'Implicit VR Little Endian: Default Transfer Syntax for DICOM',\n        '1.2.840.10008.1.2.1' : 'Explicit VR Little Endian',\n        '1.2.840.10008.1.2.1.99' : 'Deflated Explicit VR Little Endian',\n        '1.2.840.10008.1.2.2' : 'Explicit VR Big Endian (Retired)',\n        '1.2.840.10008.1.2.4.50' : 'JPEG Baseline (Process 1): Default Transfer Syntax for Lossy JPEG 8 Bit Image Compression',\n        '1.2.840.10008.1.2.4.51' : 'JPEG Extended (Process 2 & 4): Default Transfer Syntax for Lossy JPEG 12 Bit Image Compression (Process 4 only)',\n        '1.2.840.10008.1.2.4.52' : 'JPEG Extended (Process 3 & 5) (Retired)',\n        '1.2.840.10008.1.2.4.53' : 'JPEG Spectral Selection, Non-Hierarchical (Process 6 & 8) (Retired)',\n        '1.2.840.10008.1.2.4.54' : 'JPEG Spectral Selection, Non-Hierarchical (Process 7 & 9) (Retired)',\n        '1.2.840.10008.1.2.4.55' : 'JPEG Full Progression, Non-Hierarchical (Process 10 & 12) (Retired)',\n        '1.2.840.10008.1.2.4.56' : 'JPEG Full Progression, Non-Hierarchical (Process 11 & 13) (Retired)',\n        '1.2.840.10008.1.2.4.57' : 'JPEG Lossless, Non-Hierarchical (Process 14)',\n        '1.2.840.10008.1.2.4.58' : 'JPEG Lossless, Non-Hierarchical (Process 15) (Retired)',\n        '1.2.840.10008.1.2.4.59' : 'JPEG Extended, Hierarchical (Process 16 & 18) (Retired)',\n        '1.2.840.10008.1.2.4.60' : 'JPEG Extended, Hierarchical (Process 17 & 19) (Retired)',\n        '1.2.840.10008.1.2.4.61' : 'JPEG Spectral Selection, Hierarchical (Process 20 & 22) (Retired)',\n        '1.2.840.10008.1.2.4.62' : 'JPEG Spectral Selection, Hierarchical (Process 21 & 23) (Retired)',\n        '1.2.840.10008.1.2.4.63' : 'JPEG Full Progression, Hierarchical (Process 24 & 26) (Retired)',\n        '1.2.840.10008.1.2.4.64' : 'JPEG Full Progression, Hierarchical (Process 25 & 27) (Retired)',\n        '1.2.840.10008.1.2.4.65' : 'JPEG Lossless, Hierarchical (Process 28) (Retired)',\n        '1.2.840.10008.1.2.4.66' : 'JPEG Lossless, Hierarchical (Process 29) (Retired)',\n        '1.2.840.10008.1.2.4.70' : 'JPEG Lossless, Non-Hierarchical, First-Order Prediction (Process 14 [Selection Value 1]): Default Transfer Syntax for Lossless JPEG Image Compression',\n        '1.2.840.10008.1.2.4.80' : 'JPEG-LS Lossless Image Compression',\n        '1.2.840.10008.1.2.4.81' : 'JPEG-LS Lossy (Near-Lossless) Image Compression',\n        '1.2.840.10008.1.2.4.90' : 'JPEG 2000 Image Compression (Lossless Only)',\n        '1.2.840.10008.1.2.4.91' : 'JPEG 2000 Image Compression',\n        '1.2.840.10008.1.2.4.92' : 'JPEG 2000 Part 2 Multi-component Image Compression (Lossless Only)',\n        '1.2.840.10008.1.2.4.93' : 'JPEG 2000 Part 2 Multi-component Image Compression',\n        '1.2.840.10008.1.2.4.94' : 'JPIP Referenced',\n        '1.2.840.10008.1.2.4.95' : 'JPIP Referenced Deflate',\n        '1.2.840.10008.1.2.4.100' : 'MPEG2 Main Profile / Main Level',\n        '1.2.840.10008.1.2.4.101' : 'MPEG2 Main Profile / High Level',\n        '1.2.840.10008.1.2.4.102' : 'MPEG-4 AVC/H.264 High Profile / Level 4.1',\n        '1.2.840.10008.1.2.4.103' : 'MPEG-4 AVC/H.264 BD-compatible High Profile / Level 4.1',\n        '1.2.840.10008.1.2.4.104' : 'MPEG-4 AVC/H.264 High Profile / Level 4.2 For 2D Video',\n        '1.2.840.10008.1.2.4.105' : 'MPEG-4 AVC/H.264 High Profile / Level 4.2 For 3D Video',\n        '1.2.840.10008.1.2.4.106' : 'MPEG-4 AVC/H.264 Stereo High Profile / Level 4.2',\n        '1.2.840.10008.1.2.4.107' : 'HEVC/H.265 Main Profile / Level 5.1',\n        '1.2.840.10008.1.2.4.108' : 'HEVC/H.265 Main 10 Profile / Level 5.1',\n        '1.2.840.10008.1.2.5' : 'RLE Lossless',\n        '1.2.840.10008.1.2.6.1' : 'RFC 2557 MIME encapsulation',\n        '1.2.840.10008.1.2.6.2' : 'XML Encoding',\n        '1.2.840.10008.1.20' : 'Papyrus 3 Implicit VR Little Endian (Retired)',\n    }\n    try:\n        return TransferSyntaxUID_dict.get(ds.file_meta.TransferSyntaxUID, \"Unknown TransferSyntaxUID\")\n    except KeyError:\n        return \"Missing TransferSyntaxUID\"\n        \n    \n    \n# Let's take a look at the decompressed image\nprint('Image Compression Mode: ', TransferSyntaxDescription(ds))\nplt.imshow(ds.pixel_array , cmap=plt.cm.gray)\nplt.show()\n\nif VERBOSE:\n    print(\"pixel_array: \")\n    print()\n    print(ds.pixel_array )\n\n\n\n\n","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Step 5:  Create a Beam function to read the DICOM files\n\nWe can add custom code to an Apache Beam pipeline by creating a class that inherits from apache_beam.DoFn.  We overide the process() function with our custom code.\n\nThe process function requires an element argument.  element is the output of the previous step in the pipeline (i.e. the input to this step).  You can also feed the process function \"side inputs\", which is data that doesn't come from the previous step in the pipeline.  In this example, the side input is our dictionary that maps images to their labels."},{"metadata":{"trusted":true},"cell_type":"code","source":"\nclass Read_Dicom_File(apache_beam.DoFn):\n    \"\"\"Input is a list of dictionaries that contains a \"Dicom_Filename\" key.\n    This function reads the DICOM file and extract certain data tags.\n    It returns a list of dictionaries that includes the input data\n    and the data extracted from the Dicom file\"\"\"\n\n    def process(self, element, labels_dict, include_label=True):\n\n        ds = pydicom.dcmread(element)\n\n        label = (\n            int(labels_dict[ds[\"SOPInstanceUID\"].value][\"pe_present_on_image\"])\n            if include_label\n            else -9999999  # For prediction/test data sets, you don't have a label\n        )\n\n        return [\n            {\n                \"FileName\": element,\n                \"SOPInstanceUID\": ds[\"SOPInstanceUID\"].value,\n                \"Label\": label,\n                \"RescaleSlope\": float(ds[\"RescaleSlope\"].value),\n                \"RescaleIntercept\": float(ds[\"RescaleIntercept\"].value),\n                # \"PatientAge\": ds[\"PatientAge\"].value,\n                # \"PatientSex\": ds[\"PatientSex\"].value,\n                # \"BodyPartExamined\": ds[\"BodyPartExamined\"].value,\n                \"PixelData\": ds.pixel_array,\n                \n            }\n        ]","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Step 6: Standardizing CT Images\nIf the DICOM file is from a CT scan, the PixelData needs to be converted to the [Hounsfield Scale](https://en.wikipedia.org/wiki/Hounsfield_scale).\n\nBy converting pixels to the Hounsfield Units,  we can compare CT scans that were taken on different scanners.\n\nThis Beam function is an example of applying arbitrary Python code to an element as it flows through the pipeline.  We will be adding the \"PixelData_PreProcessed\" key to our element and using that to hold our modified image.  \n"},{"metadata":{"trusted":true},"cell_type":"code","source":"\nclass Convert_to_Hounsfield_Units(apache_beam.DoFn):\n    \"\"\"Input is a dictionary containing at least the keys PixelData, RescaleSlope and RescaleIntercept.\n    Output is the same as input, plus the key PixelData_PreProcessed.\"\"\"\n\n    def process(self, element):\n\n        # If the image has already been pre-processed, use that.\n        # Otherwise use the original image.\n        image = element.get(\"PixelData_PreProcessed\", element[\"PixelData\"])\n\n        image = element[\"PixelData\"].astype(dtype=np.float32)\n        slope = np.array(element[\"RescaleSlope\"], dtype=np.float32)\n        intercept = np.array(element[\"RescaleIntercept\"], dtype=np.float32)\n\n        # Convert the image to Hounsfield Units\n        image = (slope * image) + intercept\n\n        # Very small values represent area that is outside the scanning bound of the\n        # CT scanner.  We replace those values with the value for air (which by definition is -1000)\n        image[image < -1000.0] = -1000.0\n\n        element[\"PixelData_PreProcessed\"] = image.astype(dtype=np.float32)\n\n        return [element]\n","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Step 7:  Resize the image\n\nYou may want your medical images to be a certain size.\n\nThis may be especially helpful if you're planning to use transfer learning (i.e. using one of the pre-trained models availble at https://www.tensorflow.org/api_docs/python/tf/keras/applications).  For example, the Resnet50 model has a default input_size of (224, 224, 3).\n\n\nYou can also apply other opencv transformation, like converting the image from grayscale to RGB color."},{"metadata":{"trusted":true},"cell_type":"code","source":"class Resize_for_Machine_Learning(apache_beam.DoFn):\n    \"\"\"Input is a dictionary containing the key PixelData_PreProcessed or PixelData.\n       \"\"\"\n\n    def process(self, element, new_image_size=(224, 224), grayscale=True):\n\n        # The default imput size for the VGG16 CNN model is (224, 224)\n        # but you can use this to make your images any size.\n\n        # If the image has already been pre-processed, use that.\n        # Otherwise use the original image.\n        image = element.get(\"PixelData_PreProcessed\", element[\"PixelData\"]).astype(\n            dtype=np.float32\n        )\n\n        image = cv.resize(image, new_image_size)\n\n        # Convert from greyscale to RGB (RGB is required by some machine learning models)\n        if grayscale:\n            image = cv.cvtColor(image, cv.COLOR_GRAY2RGB)\n\n        element[\"PixelData_PreProcessed\"] = image.astype(dtype=np.float32)\n\n        return [element]\n\n        \n        ","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Step 8:  Rescale the image for Machine Learning\n\nThe Kaggle Tutorial https://www.kaggle.com/gzuidhof/full-preprocessing-tutorial has a lot of good ideas about how to prepare your images for use in a machine learning model.  Basically, you should:\n* Filter out anything you're not interested in (for example, filter out bones if you're looking at a soft tissue disease).\n* Rescale the data so the values fall between 0 and 1.\n* Zero-center the data, so the mean value is 0 and all values fall between -1 and 1.\n\n"},{"metadata":{"trusted":true},"cell_type":"code","source":"\nclass Rescale_for_Machine_Learning(apache_beam.DoFn):\n    \"\"\"Input is a dictionary containing at least the key PixelData or PixelData_PreProcessed.\n    Output is the same as input, plus the key PixelData_PreProcessed.\"\"\"\n\n    def process(self, element):\n        # This logic comes from the Kaggle Tutorial:  https://www.kaggle.com/gzuidhof/full-preprocessing-tutorial\n\n        # Set the Min and Max range for Hounsfield Units\n        # For lung CT scans, we are not interested in bones or other very dense.  So our max cut-off is set to 400.\n        MIN_BOUND = -1000.0\n        MAX_BOUND = 400.0\n\n        # If the image has already been pre-processed, use that.\n        # Otherwise use the original image.\n        image = element.get(\"PixelData_PreProcessed\", element[\"PixelData\"]).astype(\n            dtype=np.float32\n        )\n\n        def normalize(image, MIN_BOUND=-1000.0, MAX_BOUND=400.0):\n            \"\"\"This logic comes from the Kaggle Tutorial:  https://www.kaggle.com/gzuidhof/full-preprocessing-tutorial\n            Our values currently range from -1024 to around 2000. Anything above 400 is not interesting\n            to us, as these are simply bones with different radiodensity. A commonly used set of\n            thresholds in the LUNA16 competition to normalize between are -1000 and 400.\n            Here's some code you can use:\n            \"\"\"\n            image = (image - MIN_BOUND) / (MAX_BOUND - MIN_BOUND)\n            image[image > 1] = 1.0\n            image[image < 0] = 0.0\n            return image\n\n        image = normalize(image)\n\n        # Zero centering (from https://www.kaggle.com/gzuidhof/full-preprocessing-tutorial)\n        # As a final preprocessing step, it is advisory to zero center your data so that\n        # your mean value is 0. To do this you simply subtract the mean pixel value from all pixels.\n        #\n        # To determine this mean you simply average all images in the whole dataset. If that sounds\n        # like a lot of work, we found this to be around 0.25 in the LUNA16 competition.\n        PIXEL_MEAN = 0.25\n        image = image - PIXEL_MEAN\n\n        element[\"PixelData_PreProcessed\"] = image.astype(dtype=np.float32)\n\n        return [element]","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Step 9:  A Beam function that writes data to a TFRecord file\n\nTFRecords are a format that can speed up Data IO.  \n\nMore details are here:  https://www.tensorflow.org/tutorials/load_data/tfrecord\n\n\n"},{"metadata":{"trusted":true},"cell_type":"code","source":"\n# This is how we write to a TFRecord file\n\n#  https://www.tensorflow.org/tutorials/load_data/tfrecord\n# The following functions can be used to convert a value to a type compatible\n# with tf.train.Example.\n\n\ndef _bytes_feature(value):\n    \"\"\"Returns a bytes_list from a string / byte.\"\"\"\n    if isinstance(value, type(tf.constant(0))):\n        value = value.numpy()  # BytesList won't unpack a string from an EagerTensor.\n    return tf.train.Feature(bytes_list=tf.train.BytesList(value=[value]))\n\n\ndef _float_feature(value):\n    \"\"\"Returns a float_list from a float / double.\"\"\"\n    return tf.train.Feature(float_list=tf.train.FloatList(value=[value]))\n\n\ndef _int64_feature(value):\n    \"\"\"Returns an int64_list from a bool / enum / int / uint.\"\"\"\n    return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))\n\n\ndef serialize_example(image, label):\n    \"\"\"\n    Creates a tf.train.Example message ready to be written to a file.\n    \"\"\"\n    # Create a dictionary mapping the feature name to the tf.train.Example-compatible\n    # data type.\n    feature = {\n        \"image\": _bytes_feature(image.tostring()),\n        \"label\": _int64_feature(label),\n    }\n\n    # Create a Features message using tf.train.Example.\n    example_proto = tf.train.Example(features=tf.train.Features(feature=feature))\n    return example_proto.SerializeToString()\n\n\nclass Write_to_TFRecord(apache_beam.DoFn):\n    \"\"\"Input is a dictionary containing at least the key PixelData_HU.\n    Output is the same as input, plus the key PixelData_HU_Rescaled.\"\"\"\n\n    def process(self, element, OUTPUT_FOLDER):\n\n        filename = os.path.join(OUTPUT_FOLDER, element[\"SOPInstanceUID\"] + \".TFRecord\")\n\n        image = element.get(\"PixelData_PreProcessed\", element[\"PixelData\"]).astype(\n            dtype=np.float32\n        )\n        label = element.get(\"Label\")\n\n        # Write the `tf.train.Example` observations to the file.\n        with tf.io.TFRecordWriter(filename) as writer:\n            example = serialize_example(image, label)\n            writer.write(example)\n\n\n\n","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Step 10: Balance the data set\n\nThis data set has about 20 negative cases for each positive case.  We want our training data set to be closer to 50-50, so we create a function that \"throws away\" some of the negative cases."},{"metadata":{"trusted":true},"cell_type":"code","source":"def balanceDataset(element):\n    \"\"\"Balances the dataset by including all the positive cases\n    and 5% of the negative cases\"\"\"\n    Percent_of_Negative_Cases_to_include = 0.05\n    if element[\"Label\"] == 1:\n        return True\n    elif random.random() < Percent_of_Negative_Cases_to_include:\n        return True\n    else:\n        return False","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Step 11:  Run the Apache Beam pipeline\n\nNormally, you would run an Apache Beam pipeline on a cluster of computers.\n\nThis example will use the \"directrunner\" to run the code within this notebook."},{"metadata":{"trusted":true},"cell_type":"code","source":"list_of_Dicoms_to_process = SMALL_DATA\nlabels_dict = create_labels_dict(LABELS_FILE)\n\n\n\nstartTime = datetime.now()\nprint(\"Now running your Apache Beam pipeline!!!  Start Time: \", startTime)\n\nwith apache_beam.Pipeline() as p:\n    rows = (\n        p\n        | apache_beam.Create(list_of_Dicoms_to_process, reshuffle=True)\n        | apache_beam.ParDo(\n            Read_Dicom_File(), labels_dict, include_label=True\n        )\n      #  | \"Balance Dataset\" >> apache_beam.Filter(balanceDataset)  # Comment this out for testing so it doesn't filter out any elements\n        | apache_beam.ParDo(Convert_to_Hounsfield_Units())\n        | apache_beam.ParDo(Resize_for_Machine_Learning())\n        | apache_beam.ParDo(Rescale_for_Machine_Learning())\n        | apache_beam.ParDo(Write_to_TFRecord(), OUTPUT_FOLDER=WORKING_FOLDER)\n        # | \"print\" >> apache_beam.Map(print)\n    )\n\nprint(\"Finished running your Apache Beam pipeline!!!\")\nprint(\"Total Time:\", datetime.now() - startTime)\n\nprint(\"Finished running all the code!!!!!!!!!!!!!!!!!!!!!!!!!!\")\n\n\n\n\n\n\n","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Step 12:  List the files created by the Apache Beam pipeline"},{"metadata":{"trusted":true},"cell_type":"code","source":"#!ls /kaggle/working/\n\n# Print the number of TFRecord files\n!ls -l /kaggle/working/*.TFRecord | wc -l\n! echo --------------------------------------------------------------------------------------------------\n\n# See how big the TFRecord files are\n!ls /kaggle/working/ -l #--block-size=M","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Step 13:  Read a TFRecord file and display the medical image\n\nTo prove that this worked, we will load the TFRecord files into a Tensorflow Dataset (which could easily be used to train a mode).\nThen we will parse the TFRecords back into a numpy and print the image.\n\nNote the axes of the image... They prove that the image has been shrunk from around 500 by 500 to the requested 300 by 300."},{"metadata":{"trusted":true},"cell_type":"code","source":"# Get a list of the TFRecord files\nTFRecordFiles = []\nfor dirname, _, filenames in os.walk(WORKING_FOLDER):\n    for filename in filenames:\n        if filename[-9:]=='.TFRecord':\n            TFRecordFiles.append(os.path.join(dirname, filename))\n          \n        \n        \n        \n# Define two parsing functions that will turn the TFRecord back into an array and a label        \ndef _parse_function(example, image_shape=(224, 224, 1)):\n    # Parse the input `tf.train.Example` proto using the feature_dictionary.\n    # Create a description of the features.\n    feature_description = {\n        \"image\": tf.io.FixedLenFeature([], tf.string, default_value=\"\"),\n        \"label\": tf.io.FixedLenFeature([], tf.int64, default_value=0),\n    }\n\n    parsed_example = tf.io.parse_single_example(example, feature_description)\n    return parsed_example\n\n\n\ndef _parse_function2(example):\n    label = example[\"label\"]\n    image = tf.io.decode_raw(\n        example[\"image\"], float, little_endian=True, fixed_length=None, name=None\n    )\n    image = tf.reshape(image, (224, 224, 3))\n    # apache beam is now producing an rgb image\n    # image = tf.image.grayscale_to_rgb(image)  # I moved this into the apache beam pipeline\n    return image, label        \n            \n            \n# Create a dataset         \nTF_dataset = tf.data.TFRecordDataset(TFRecordFiles)\n\n# Use map to apply the parsing functions to the data\nTF_dataset = TF_dataset.map(_parse_function)\nTF_dataset = TF_dataset.map(_parse_function2)\n            \n            \nfor images, labels in TF_dataset.take(1):  # only take first element of dataset\n    image = images.numpy()\n    label = labels       \n      \nprint(\"Label: \", label)        \nprint()        \nplt.imshow(image , cmap=plt.cm.gray)\nplt.show()\n           \n    \n","execution_count":null,"outputs":[]}],"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat":4,"nbformat_minor":4}