<jupyter_start><jupyter_text>tciaclient> A TCIA (The Cancer Imaging Archive) download client for Python.<jupyter_code>#hide
from nbdev.showdoc import *
%load_ext autoreload
%autoreload 2
#export
import os
import urllib.request, urllib.error, urllib.parse
import urllib.request, urllib.parse, urllib.error
import sys
import math
import json
#export
class TCIAClient:
    GET_IMAGE = "getImage"
    GET_MANUFACTURER_VALUES = "getManufacturerValues"
    GET_MODALITY_VALUES = "getModalityValues"
    GET_COLLECTION_VALUES = "getCollectionValues"
    GET_BODY_PART_VALUES = "getBodyPartValues"
    GET_PATIENT_STUDY = "getPatientStudy"
    GET_SERIES = "getSeries"
    GET_PATIENT = "getPatient"
    GET_SERIES_SIZE = "getSeriesSize"
    CONTENTS_BY_NAME = "ContentsByName"

    def __init__(self, baseUrl="https://services.cancerimagingarchive.net/services/v4", resource="TCIA"):
        self.baseUrl = baseUrl + "/" + resource

    def execute(self, url, queryParameters={}):
        queryParameters = dict((k, v) for k, v in queryParameters.items() if v)
        queryString = "?%s" % urllib.parse.urlencode(queryParameters)
        requestUrl = url + queryString
        request = urllib.request.Request(url=requestUrl)
        resp = urllib.request.urlopen(request)
        return resp
    
    def jsonify(self, resp):
        return json.loads(resp.read())

    def get_modality_values(self,collection = None , bodyPartExamined = None , modality = None , outputFormat = "json" ):
        serviceUrl = self.baseUrl + "/query/" + self.GET_MODALITY_VALUES
        queryParameters = {"Collection" : collection , "BodyPartExamined" : bodyPartExamined , "Modality" : modality , "format" : outputFormat }
        resp = self.execute(serviceUrl , queryParameters)
        if outputFormat == "json" : resp = self.jsonify(resp)
        return resp

    def get_series_size(self, SeriesInstanceUID = None, outputFormat = "json"):
        serviceUrl = self.baseUrl + "/query/" + self.GET_SERIES_SIZE
        queryParameters = {"SeriesInstanceUID" : SeriesInstanceUID, "format" :
                           outputFormat}
        resp = self.execute(serviceUrl, queryParameters)
        if outputFormat == "json" : resp = self.jsonify(resp)
        return resp

    def contents_by_name(self, name = None):
        serviceUrl = self.baseUrl + "/query/" + self.CONTENTS_BY_NAME
        queryParameters = {"name" : name}
        print(serviceUrl)
        resp = self.execute(serviceUrl,queryParameters)
        return resp

    def get_manufacturer_values(self,collection = None , bodyPartExamined = None, modality = None , outputFormat = "json"):
        serviceUrl = self.baseUrl + "/query/" + self.GET_MANUFACTURER_VALUES
        queryParameters = {"Collection" : collection , "BodyPartExamined" : bodyPartExamined , "Modality" : modality , "format" : outputFormat }
        resp = self.execute(serviceUrl , queryParameters)
        if outputFormat == "json" : resp = self.jsonify(resp)
        return resp

    def get_collection_values(self,outputFormat = "json" ):
        serviceUrl = self.baseUrl + "/query/" + self.GET_COLLECTION_VALUES
        queryParameters = { "format" : outputFormat }
        resp = self.execute(serviceUrl , queryParameters)
        if outputFormat == "json" : resp = self.jsonify(resp)
        return resp

    def get_body_part_values(self,collection = None , bodyPartExamined = None , modality = None , outputFormat = "json" ):
        serviceUrl = self.baseUrl + "/query/" + self.GET_BODY_PART_VALUES
        queryParameters = {"Collection" : collection , "BodyPartExamined" : bodyPartExamined , "Modality" : modality , "format" : outputFormat }
        resp = self.execute(serviceUrl , queryParameters)
        if outputFormat == "json" : resp = self.jsonify(resp)
        return resp

    def get_patient_study(self,collection = None , patientId = None , studyInstanceUid = None , outputFormat = "json" ):
        serviceUrl = self.baseUrl + "/query/" + self.GET_PATIENT_STUDY
        queryParameters = {"Collection" : collection , "PatientID" : patientId , "StudyInstanceUID" : studyInstanceUid , "format" : outputFormat }
        resp = self.execute(serviceUrl , queryParameters)
        if outputFormat == "json" : resp = self.jsonify(resp)
        return resp

    def get_series(self, collection = None , modality = None , studyInstanceUid = None , outputFormat = "json" ):
        "Search for series by collection name and/or study name and/or modality."
        serviceUrl = self.baseUrl + "/query/" + self.GET_SERIES
        queryParameters = {"Collection" : collection , "StudyInstanceUID" : studyInstanceUid , "Modality" : modality , "format" : outputFormat }
        resp = self.execute(serviceUrl , queryParameters)
        if outputFormat == "json" : resp = self.jsonify(resp)
        return resp

    def get_patient(self,collection = None , outputFormat = "json" ):
        serviceUrl = self.baseUrl + "/query/" + self.GET_PATIENT
        queryParameters = {"Collection" : collection , "format" : outputFormat }
        resp = self.execute(serviceUrl , queryParameters)
        if outputFormat == "json" : resp = self.jsonify(resp)
        return resp

    def get_image(self , seriesInstanceUid , downloadPath, zipFileName):
        "Get a set of images as a zip file."
        serviceUrl = self.baseUrl + "/query/" + self.GET_IMAGE
        queryParameters = { "SeriesInstanceUID" : seriesInstanceUid }
        os.umask(0o002)
        try:
            if not os.path.exists(downloadPath):
                os.makedirs(downloadPath)
            file = os.path.join(downloadPath, zipFileName)
            resp = self.execute( serviceUrl , queryParameters)
            downloaded = 0
            CHUNK = 256 * 10240
            with open(file, 'wb') as fp:
                while True:
                    chunk = resp.read(CHUNK)
                    downloaded += len(chunk)
                    if not chunk: break
                    fp.write(chunk)
        except urllib.error.HTTPError as e:
            print("HTTP Error:",e.code , serviceUrl)
            return False
        except urllib.error.URLError as e:
            print("URL Error:",e.reason , serviceUrl)
            return False

        return True
show_doc(TCIAClient.get_series)
tc = TCIAClient()
tc.get_series(collection="TCGA-GBM")[:2]
show_doc(TCIAClient.get_image)
tc = TCIAClient()
tc.get_image(seriesInstanceUid = "1.3.6.1.4.1.14519.5.2.1.7695.4001.180700359927709468630440576839",
             downloadPath = "./", zipFileName = "images.zip")
#hide
from nbdev.export import notebook2script; notebook2script()<jupyter_output>Converted 00_core.ipynb.
Converted index.ipynb.
{'fraction_non_alphanumeric': 0.10605833956619297, 'fraction_numerical': 0.010770381451009723, 'mean_word_length': 2.9964136282127916, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 5, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '320424', 'license': 'Apache-2.0', 'n_tokens_mistral': 2044, 'n_tokens_neox': 1962, 'n_words': 449, 'path': '00_core.ipynb', 'repo_name': 'moritzschwyzer/tciaclient', 'size': '6685'}