diff --git a/src/nhp/data/aggregated_data/ecds.py b/src/nhp/data/aggregated_data/ecds.py index 7cbec1c..fff6ad7 100644 --- a/src/nhp/data/aggregated_data/ecds.py +++ b/src/nhp/data/aggregated_data/ecds.py @@ -22,7 +22,6 @@ def get_ecds_data(spark: SparkSession) -> DataFrame: F.col("attendance_category"), F.col("acuity"), F.col("tretspef"), - F.col("tretspef_grouped"), F.col("group"), F.col("pod"), F.col("demog_type"), diff --git a/src/nhp/data/aggregated_data/outpatients.py b/src/nhp/data/aggregated_data/outpatients.py index 5f6a5ab..9a06d83 100644 --- a/src/nhp/data/aggregated_data/outpatients.py +++ b/src/nhp/data/aggregated_data/outpatients.py @@ -19,7 +19,6 @@ def get_outpatients_data(spark: SparkSession) -> DataFrame: F.col("sex"), F.col("imd_quintile"), F.col("tretspef"), - F.col("tretspef_grouped"), F.col("sitetret"), F.col("type"), F.col("group"), diff --git a/src/nhp/data/inputs_data/ae/baseline.py b/src/nhp/data/inputs_data/ae/baseline.py index af257f0..ac8d64b 100644 --- a/src/nhp/data/inputs_data/ae/baseline.py +++ b/src/nhp/data/inputs_data/ae/baseline.py @@ -18,8 +18,7 @@ def get_ae_baseline(spark: SparkSession, geography_column: str) -> DataFrame: """ return ( get_ae_df(spark) - .groupBy("fyear", geography_column, "group") + .groupBy("fyear", geography_column, "group", "tretspef") .agg(F.count("fyear").alias("count")) .withColumn("activity_type", F.lit("aae")) - .withColumn("tretspef", F.lit(None).cast("string")) ) diff --git a/src/nhp/data/inputs_data/ip/__init__.py b/src/nhp/data/inputs_data/ip/__init__.py index f888a14..aa89844 100644 --- a/src/nhp/data/inputs_data/ip/__init__.py +++ b/src/nhp/data/inputs_data/ip/__init__.py @@ -25,8 +25,6 @@ def get_ip_df(spark: SparkSession) -> DataFrame: .filter(F.isnotnull("age")) .drop("age_group") .join(inputs_age_group(spark), "age") - .drop("tretspef") - .withColumnRenamed("tretspef_grouped", "tretspef") ) diff --git a/src/nhp/data/inputs_data/ip/baseline.py b/src/nhp/data/inputs_data/ip/baseline.py index f49d62d..c42932f 100644 --- a/src/nhp/data/inputs_data/ip/baseline.py +++ b/src/nhp/data/inputs_data/ip/baseline.py @@ -18,12 +18,6 @@ def get_ip_baseline(spark: SparkSession, geography_column: str) -> DataFrame: """ return ( get_ip_df(spark) - .withColumn( - "tretspef", - F.when(F.col("group") == "maternity", "Other (Medical)").otherwise( - F.col("tretspef") - ), - ) .groupBy("fyear", geography_column, "group", "tretspef") .agg(F.count("fyear").alias("count")) .withColumn("activity_type", F.lit("ip")) diff --git a/src/nhp/data/inputs_data/ip/expat_repat.py b/src/nhp/data/inputs_data/ip/expat_repat.py index 720dd2d..1d35de0 100644 --- a/src/nhp/data/inputs_data/ip/expat_repat.py +++ b/src/nhp/data/inputs_data/ip/expat_repat.py @@ -18,12 +18,6 @@ def get_ip_expat_data(spark: SparkSession) -> DataFrame: """ return ( get_ip_df(spark) - .withColumn( - "tretspef", - F.when(F.col("group") == "maternity", "Other (Medical)").otherwise( - F.col("tretspef") - ), - ) .groupBy("fyear", "provider", "group", "tretspef") .count() .withColumn("activity_type", F.lit("ip")) @@ -34,12 +28,6 @@ def get_ip_expat_data(spark: SparkSession) -> DataFrame: def _get_icb_df(spark: SparkSession) -> DataFrame: return ( get_ip_df(spark) - .withColumn( - "tretspef", - F.when(F.col("group") == "maternity", "Other (Medical)").otherwise( - F.col("tretspef") - ), - ) .filter(F.isnotnull("icb")) .groupBy("fyear", "icb", "is_main_icb", "provider", "group", "tretspef") .count() diff --git a/src/nhp/data/inputs_data/op/__init__.py b/src/nhp/data/inputs_data/op/__init__.py index d80d817..6ecfb57 100644 --- a/src/nhp/data/inputs_data/op/__init__.py +++ b/src/nhp/data/inputs_data/op/__init__.py @@ -26,8 +26,6 @@ def get_op_df(spark: SparkSession) -> DataFrame: .filter(F.isnotnull("age")) .drop("age_group") .join(inputs_age_group(spark), "age") - .drop("tretspef") - .withColumnRenamed("tretspef_grouped", "tretspef") ) diff --git a/src/nhp/data/model_data/generate_synthetic_data.py b/src/nhp/data/model_data/generate_synthetic_data.py index eae5acb..d3459b6 100644 --- a/src/nhp/data/model_data/generate_synthetic_data.py +++ b/src/nhp/data/model_data/generate_synthetic_data.py @@ -127,7 +127,6 @@ def _ip(self) -> None: ) .withColumn("icb", F.when(F.col("is_main_icb"), "A").otherwise("B")) .withColumn("mainspef", F.col("tretspef")) - .filter(~F.col("tretspef_grouped").startswith("Other")) .orderBy("rn") .persist() ) diff --git a/src/nhp/data/raw_data/aae.py b/src/nhp/data/raw_data/aae.py index fdc98f9..b529b43 100644 --- a/src/nhp/data/raw_data/aae.py +++ b/src/nhp/data/raw_data/aae.py @@ -193,7 +193,6 @@ def get_aae_data(spark: SparkSession) -> DataFrame: ) .withColumn("hsagrp", F.concat(F.lit("aae_"), F.col("type"))) .withColumn("tretspef", F.lit("Other")) - .withColumn("tretspef_grouped", F.lit("Other")) .withColumn("pod", F.concat(F.lit("aae_type-"), F.col("aedepttype"))) .withColumn("demog_type", F.lit("demographics")) .withColumn("ndggrp", F.col("group")) diff --git a/src/nhp/data/raw_data/ecds.py b/src/nhp/data/raw_data/ecds.py index da4d783..cb855cb 100644 --- a/src/nhp/data/raw_data/ecds.py +++ b/src/nhp/data/raw_data/ecds.py @@ -246,7 +246,6 @@ def get_ecds_data(spark: SparkSession) -> DataFrame: ) .withColumn("hsagrp", F.concat(F.lit("aae_"), F.col("type"))) .withColumn("tretspef", F.lit("Other")) - .withColumn("tretspef_grouped", F.lit("Other")) .withColumn("pod", F.concat(F.lit("aae_type-"), F.col("aedepttype"))) .withColumn("demog_type", F.lit("demographics")) .withColumn("ndggrp", F.col("group")) diff --git a/src/nhp/data/raw_data/helpers.py b/src/nhp/data/raw_data/helpers.py index 590b877..5e08f6e 100644 --- a/src/nhp/data/raw_data/helpers.py +++ b/src/nhp/data/raw_data/helpers.py @@ -6,49 +6,6 @@ from nhp.data.table_names import table_names -def add_tretspef_grouped_column(self: DataFrame) -> DataFrame: - """Add tretspef grouped column to DataFrame - - :param self: The data frame to add the tretpsef grouped column to - :type df: DataFrame - :return: The data frame - :rtype: DataFrame - """ - - specialties = [ - "100", - "101", - "110", - "120", - "130", - "140", - "150", - "160", - "170", - "300", - "301", - "320", - "330", - "340", - "400", - "410", - "430", - "502", - ] - - tretspef_column = ( - F.when(F.col("tretspef").isin(specialties), F.col("tretspef")) - .when(F.expr("tretspef RLIKE '^1(?!80|9[02])'"), F.lit("Other (Surgical)")) - .when( - F.expr("tretspef RLIKE '^(1(80|9[02])|[2346]|5(?!60)|83[134])'"), - F.lit("Other (Medical)"), - ) - .otherwise(F.lit("Other")) - ) - - return self.withColumn("tretspef_grouped", tretspef_column) - - def add_age_group_column(self: DataFrame) -> DataFrame: return self.withColumn( "age_group", @@ -94,3 +51,25 @@ def remove_mental_health_providers( ) return df.join(mental_health_providers, provider_col, "anti") + + +def replace_invalid_tretspef_values( + spark: SparkSession, df: DataFrame, column: str = "tretspef" +) -> DataFrame: + """Replace invalid values in the tretspef column with 'Other' + + :param spark: The Spark session + :type spark: SparkSession + :param df: The data frame to process + :type df: DataFrame + :param column: The column name containing treatment specification values + :type column: str + :return: The data frame with invalid tretspef values replaced + :rtype: DataFrame + """ + tretspefs = spark.read.table(table_names.reference_tretspef_type).select("tretspef") + valid_values = [row[0] for row in tretspefs.collect()] + return df.withColumn( + column, + F.when(F.col(column).isin(valid_values), F.col(column)).otherwise("Other"), + ) diff --git a/src/nhp/data/raw_data/inpatients.py b/src/nhp/data/raw_data/inpatients.py index 5256bcd..f500fab 100644 --- a/src/nhp/data/raw_data/inpatients.py +++ b/src/nhp/data/raw_data/inpatients.py @@ -8,8 +8,8 @@ from nhp.data.nhp_datasets.icbs import add_main_icb from nhp.data.raw_data.helpers import ( add_age_group_column, - add_tretspef_grouped_column, remove_mental_health_providers, + replace_invalid_tretspef_values, ) from nhp.data.table_names import table_names @@ -27,9 +27,9 @@ def get_inpatients_data(spark: SparkSession) -> DataFrame: ) df = add_main_icb(spark, hes_apc) - df = add_tretspef_grouped_column(df) df = add_age_group_column(df) df = remove_mental_health_providers(spark, df, "provider") + df = replace_invalid_tretspef_values(spark, df, "tretspef") df_primary_diagnosis = spark.read.table(table_names.hes_apc_diagnoses).filter( F.col("diag_order") == 1 @@ -119,7 +119,6 @@ def get_inpatients_data(spark: SparkSession) -> DataFrame: F.col("classpat"), F.col("mainspef"), F.col("tretspef"), - F.col("tretspef_grouped"), F.col("hsagrp"), F.col("group"), F.col("admidate"), diff --git a/src/nhp/data/raw_data/outpatients.py b/src/nhp/data/raw_data/outpatients.py index 262e190..b10565d 100644 --- a/src/nhp/data/raw_data/outpatients.py +++ b/src/nhp/data/raw_data/outpatients.py @@ -9,8 +9,8 @@ from nhp.data.nhp_datasets.providers import read_data_with_provider from nhp.data.raw_data.helpers import ( add_age_group_column, - add_tretspef_grouped_column, remove_mental_health_providers, + replace_invalid_tretspef_values, ) from nhp.data.table_names import table_names @@ -19,6 +19,7 @@ def get_outpatients_data(spark: SparkSession) -> DataFrame: """Get Outpatients Data""" df = read_data_with_provider(spark, table_names.hes_opa) df = remove_mental_health_providers(spark, df, "provider") + df = replace_invalid_tretspef_values(spark, df, "tretspef") # Calculate icb column df = df.withColumn( @@ -32,8 +33,6 @@ def get_outpatients_data(spark: SparkSession) -> DataFrame: # add main icb column df = add_main_icb(spark, df) - # add the tretspef grouped column - df = add_tretspef_grouped_column(df) # add age and age_group columns df = df.withColumn( "age", @@ -101,7 +100,6 @@ def get_outpatients_data(spark: SparkSession) -> DataFrame: F.col("imd_quintile"), F.col("ethnos"), F.col("tretspef"), - F.col("tretspef_grouped"), F.col("sitetret"), F.col("has_procedures"), F.col("sushrg"),