Remove tretspef grouped

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"),

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"),

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"))
     )

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")
     )
 
 

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"))

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()

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")
     )
 
 

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()
         )

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"))

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"))

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"),
+    )

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"),

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"),