{"cells":[{"metadata":{"trusted":true},"cell_type":"code","source":"# TODO\n# Első fontos dolog:\n# User-szintű statisztikákat csinálni.\n#  Percent átlag, medián,\n#  Delta átlag, medián, tendencia (egyértelműen növekvő, egyértelműen csökkenő, hullámzó, random - ebben az esetben end - start érték)\n#  Egyéni szintre a populációtól való eltérés legyen - pl.intenzitásban, változékonyság, ebből lehet esetleg klasztert csinálni\n#  Esetleg kitalálni a hiányzó weekek értékeit - a két meglévőből számított heti átlag eltérés alapján csak be-beszúrni.\n#  Három oszlopban az utolsó, utolsó előtti és azelőtti nap értéke, meg ezekhez való statisztikák.","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"import os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        os.path.join(dirname, filename)\n\n# You can write up to 5GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session\n\n!pip install pyspark\nfrom pyspark.sql import SparkSession\n\nspark = SparkSession.builder.appName('pulmonary_fibrosis').getOrCreate()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"from pyspark.sql import Window\nfrom pyspark.sql.functions import expr, col, lag, avg as sparkavg, min as sparkmin, max as sparkmax, stddev, skewness, count, kurtosis, round as sparkround, first, last, when","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"df_train = spark.read.options(header='true', inferSchema='true').csv(\"/kaggle/input/osic-pulmonary-fibrosis-progression/train.csv\")\ndf_test = spark.read.options(header='true', inferSchema='true').csv(\"/kaggle/input/osic-pulmonary-fibrosis-progression/test.csv\")","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"weeksWindow = Window.partitionBy('Patient').orderBy('Weeks')\nweeksDescWindow = Window.partitionBy('Patient').orderBy(col('Weeks').desc())\n\ndef enrichWithWindow(dataframe):\n    return dataframe \\\n    .withColumn('weekLag', lag('Weeks').over(weeksWindow)) \\\n    .withColumn('percentLag', lag('Percent').over(weeksWindow)) \\\n    .withColumn('weekDiff', col('Weeks') - col('weekLag')) \\\n    .withColumn('percentDiff', col('Percent') - col('percentLag')) \\\n    .drop('weekLag').drop('percentLag') \\\n    .withColumn('percentDeltaPerWeek', col('percentDiff')/col('weekDiff'))\n\ndfTrain = enrichWithWindow(df_train)\ndfTrain.persist()\ndfTrain.show(1)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"from pyspark.sql import DataFrameStatFunctions as sparkstat\n\npercentMedian = dfTrain.select('Percent').approxQuantile('Percent', [0.5], 0)\navgOfPercents = dfTrain.selectExpr('avg(Percent) avgOfPercents').collect()\n\nprint(f'Median of percents is: {percentMedian[0]}')\nprint(f'Average of percents is: {avgOfPercents[0].avgOfPercents}')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Calculate statistics about the average percent delta per week value.\n# These statistics show something about the trends of the deltas.\n\ndef generateCaseWhenFromList(qList, qDict, colName):\n    _tempList = []\n    for counter, element in enumerate(qList):\n        if counter == 0:\n            _tempList.append(f'case when {colName} <= {qDict[qList[counter]]} then {qList[counter]} ')\n        elif counter < len(qList) - 1:\n            _tempList.append(f'when {colName} > {qDict[qList[counter -1 ]]} and {colName} <= {qDict[qList[counter]]} then {qList[counter]} ')\n        elif counter == len(qList) - 1:\n            _tempList.append(f'else {qList[counter]} end as whichDeltaQuantile')\n    return _tempList\n        \n\ndfDeltaStatsTemp = dfTrain.select('Patient', 'Age', 'Sex', 'Smokingstatus').distinct() \\\n.join(\n    dfTrain.select('Patient','percentDeltaPerWeek') \\\n    .groupBy('Patient') \\\n    .agg(\n        sparkavg('percentDeltaPerWeek').alias('avgDelta'),\n        stddev('percentDeltaPerWeek').alias('stddevDelta'),\n        skewness('percentDeltaPerWeek').alias('skewnessDelta'),\n        kurtosis('percentDeltaPerWeek').alias('kurtosisDelta')\n    ),\n    ['Patient'],\n    'inner'\n)\n\ndeltaQuantilesList = [0.1, 0.3, 0.5, 0.7, 0.9]\ndeltaQuantiles_ = dfDeltaStatsTemp.approxQuantile('avgDelta', deltaQuantilesList , 0)\ndeltaQuantiles = dict(zip(deltaQuantilesList, deltaQuantiles_))\n\ndfDeltaStats = dfDeltaStatsTemp.withColumn('whichDeltaQuantile',\n            expr(''.join(generateCaseWhenFromList(deltaQuantilesList, deltaQuantiles, 'avgDelta')))\n            )\n\ndfDeltaStats.show()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"dfAgeBucket.printSchema()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"from pyspark.ml import Pipeline\nfrom pyspark.ml.feature import OneHotEncoder,VectorAssembler, StringIndexer, QuantileDiscretizer\n\nquantileDiscretizer = QuantileDiscretizer(inputCol=\"Age\", outputCol=\"AgeBucket\") \\\n.setNumBuckets(5)\ndfAgeBucket = quantileDiscretizer.fit(dfDeltaStats).transform(dfDeltaStats)\n\nstages = list()\n\ndef indexColumn(df, column):\n  return StringIndexer(inputCol=column, outputCol=column + '_index')\n\nstringColumns = ['Sex', 'Smokingstatus', 'whichDeltaQuantile']\nnominalColumns = ['AgeBucket']\n\nfor column in stringColumns:\n  stages.append(indexColumn(dfAgeBucket, column))\n\noneHotBaby = OneHotEncoder(inputCols=['AgeBucket'] + [f'{column}_index' for column in stringColumns], \n                                 outputCols=['oneHotAgeBucket'] + [f'{column}_encoded' for column in stringColumns])\n\nstages.append(oneHotBaby)\n\nfeatureColumns = [f'{column}_encoded' for column in stringColumns] + ['oneHotAgeBucket'] + ['avgDelta', 'stddevDelta', 'skewnessDelta', 'kurtosisDelta']\n\nassembler = VectorAssembler(\n  inputCols=featureColumns,\n  outputCol=\"features\")\nstages.append(assembler)\n\ndfTrendStabilityFeatures = Pipeline(stages=stages).fit(dfAgeBucket).transform(dfAgeBucket)\ndfTrendStabilityFeatures.persist()\ndfTrendStabilityFeatures.show(1)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# from pyspark.ml.clustering import KMeans, BisectingKMeans\n# from pyspark.ml.evaluation import ClusteringEvaluator\n\n# Trains a k-means model.\n\ndef trainBisectingKmeans(df, k):\n    kmeans = BisectingKMeans().setK(i).setSeed(1)\n    kmeansModel = kmeans.fit(df)\n\n    # Make predictions\n    _dfTrendStabilityClusters = kmeansModel.transform(df)\n\n    # Evaluate clustering by computing Silhouette score\n    evaluator = ClusteringEvaluator()\n\n    silhouette = evaluator.evaluate(_dfTrendStabilityClusters)\n    print(\"Silhouette with squared euclidean distance = \" + str(silhouette))\n    return silhouette\n    \n    \nbisectModelSilhouettes = dict()\nfor i in [4, 5, 6, 7, 8, 9, 10]:\n    bisectModelSilhouettes.update({f'{i}' : str(trainBisectingKmeans(dfTrendStabilityFeatures, i))})\nbisectModelSilhouettes","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"bisectModelSilhouettes","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Not an elegant way. C'est la vie\nkmeans = BisectingKMeans().setK(i).setSeed(1)\nkmeansModel = kmeans.fit(dfTrendStabilityFeatures)\n\ndfTrendStabilityClusters = kmeansModel.transform(dfTrendStabilityFeatures)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"dfTrendStabilityClusters.groupBy(\"prediction\").count().show()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"dfTrendStabilityClusters.select('Patient', 'Age', 'Sex', 'Smokingstatus', 'avgDelta','stddevDelta', 'prediction').show(20)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# See if age correlates with the measure of change of FVC/percent\ndfDeltaStats.corr('Age', 'avgDelta')\n\n# Well, no","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# I would like to know that how many weeks the patients' have in the data\n\ndfNumWeeks = dfTrain.select('Patient', 'Weeks').groupBy('Patient').agg(count(col('Weeks')).alias('numWeeks'))\n\ndfNumWeeks.selectExpr('max(numWeeks)').union(\ndfNumWeeks.selectExpr('min(numWeeks)')).union(\ndfNumWeeks.selectExpr('avg(numWeeks)')\n).show()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# And also the range of time their examinations took place\n\ndfPatientTimeRange = dfTrain.select('Patient', 'Weeks') \\\n.withColumn('minWeek', first('Weeks').over(weeksWindow)) \\\n.withColumn('maxWeek', first('Weeks').over(weeksDescWindow)) \\\n.withColumn('examinedTimeRange', col('maxWeek') - col('minWeek')) \\\n.drop('Weeks').distinct()\n\ndfPatientTimeRange.show(10, truncate=False)\n\ndfPatientTimeRange.groupBy('examinedTimeRange').agg(count('Patient').alias('cnt')) \\\n.toPandas().plot.scatter(x='examinedTimeRange', y='cnt')\n# .show(truncate=False)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"dfPercentAvg = dfTrain.select(col('Patient'), col('percentDeltaPerWeek')) \\\n    .groupBy('Patient') \\\n    .agg(sparkavg(col('percentDeltaPerWeek')).alias('avgPercentDelta'))\n\ndfPercentKurtosis = dfTrain.select(col('Patient'), col('percentDeltaPerWeek')) \\\n    .groupBy('Patient') \\\n    .agg(kurtosis(col('percentDeltaPerWeek')).alias('kurtosisPercentDelta'))\n\n\ndfPercentSkewness = dfTrain.select(col('Patient'), col('percentDeltaPerWeek')) \\\n    .groupBy('Patient') \\\n    .agg(skewness(col('percentDeltaPerWeek')).alias('skewnessPercentDelta'))\n\ndfPatientTendencies = dfPercentAvg \\\n    .join(dfPercentSkewness, ['Patient'], 'inner') \\\n    .join(dfPercentKurtosis, ['Patient'], 'inner')\n\n# dfPatientTendencies.show(truncate=False)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"dfPercentSkewness.select('skewnessPercentDelta') \\\n    .withColumn('roundedSkew', sparkround(col('skewnessPercentDelta'), 1)) \\\n    .drop('skewnessPercentDelta') \\\n    .groupBy('roundedSkew').agg(count(col('roundedSkew')).alias('cnt')) \\\n    .toPandas().plot.scatter(x='roundedSkew', y='cnt')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"dfTrain.filter(col('Patient') == 'ID00309637202282195513787').toPandas().plot.scatter(x='Weeks', y='percentDeltaPerWeek')\ndfTrain.filter(col('Patient') == 'ID00309637202282195513787').toPandas().plot.scatter(x='Weeks', y='Percent')\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"dfAgg = dfTrain.groupBy('Patient').agg(sparkavg('percentDeltaPerWeek').alias('percentDeltaPerWeek')) \\\n.toPandas().plot.scatter(x='percentDeltaPerWeek', y='Patient')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"dfTrain.filter(col('Patient') == 'ID00323637202285211956970').sort(col('Weeks').asc()).show()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"import matplotlib.pyplot as plt\nimport pydicom\ndataset1 = pydicom.dcmread(\"/kaggle/input/osic-pulmonary-fibrosis-progression/train/ID00323637202285211956970/99.dcm\")","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"print(dataset.__str__)\nplt.imshow(dataset.pixel_array, cmap=plt.cm.bone)\nplt.show()\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"","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}