From 6b997e9636a51d833bb7e7ca625c2bbca6d9ff65 Mon Sep 17 00:00:00 2001 From: Tom Jemmett Date: Tue, 22 Sep 2026 12:16:58 +0100 Subject: [PATCH 1/3] removes the tretspef_grouped column now only using tretspef --- src/nhp/data/aggregated_data/ecds.py | 1 - src/nhp/data/aggregated_data/outpatients.py | 1 - src/nhp/data/inputs_data/ip/__init__.py | 2 - src/nhp/data/inputs_data/op/__init__.py | 2 - .../model_data/generate_synthetic_data.py | 1 - src/nhp/data/raw_data/aae.py | 1 - src/nhp/data/raw_data/ecds.py | 1 - src/nhp/data/raw_data/helpers.py | 43 ------------------- src/nhp/data/raw_data/inpatients.py | 3 -- src/nhp/data/raw_data/outpatients.py | 4 -- 10 files changed, 59 deletions(-) diff --git a/src/nhp/data/aggregated_data/ecds.py b/src/nhp/data/aggregated_data/ecds.py index 7cbec1c2..fff6ad76 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 5f6a5ab7..9a06d83f 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/ip/__init__.py b/src/nhp/data/inputs_data/ip/__init__.py index f888a14b..aa898449 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/op/__init__.py b/src/nhp/data/inputs_data/op/__init__.py index d80d8177..6ecfb577 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 eae5acb5..d3459b65 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 fdc98f96..b529b433 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 da4d7837..cb855cb7 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 590b8776..95d77af3 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", diff --git a/src/nhp/data/raw_data/inpatients.py b/src/nhp/data/raw_data/inpatients.py index 5256bcd5..00ee32fb 100644 --- a/src/nhp/data/raw_data/inpatients.py +++ b/src/nhp/data/raw_data/inpatients.py @@ -8,7 +8,6 @@ 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, ) from nhp.data.table_names import table_names @@ -27,7 +26,6 @@ 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") @@ -119,7 +117,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 262e190c..710535be 100644 --- a/src/nhp/data/raw_data/outpatients.py +++ b/src/nhp/data/raw_data/outpatients.py @@ -9,7 +9,6 @@ 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, ) from nhp.data.table_names import table_names @@ -32,8 +31,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 +98,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"), From 145b9048d0d80db01dffd60c3d2e4cbbc9aa6076 Mon Sep 17 00:00:00 2001 From: Tom Jemmett Date: Wed, 30 Sep 2026 09:25:51 +0100 Subject: [PATCH 2/3] adds method for replacing invalid treatment function values with "Other" --- src/nhp/data/raw_data/helpers.py | 22 ++++++++++++++++++++++ src/nhp/data/raw_data/inpatients.py | 2 ++ src/nhp/data/raw_data/outpatients.py | 2 ++ 3 files changed, 26 insertions(+) diff --git a/src/nhp/data/raw_data/helpers.py b/src/nhp/data/raw_data/helpers.py index 95d77af3..5e08f6e8 100644 --- a/src/nhp/data/raw_data/helpers.py +++ b/src/nhp/data/raw_data/helpers.py @@ -51,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 00ee32fb..f500fab0 100644 --- a/src/nhp/data/raw_data/inpatients.py +++ b/src/nhp/data/raw_data/inpatients.py @@ -9,6 +9,7 @@ from nhp.data.raw_data.helpers import ( add_age_group_column, remove_mental_health_providers, + replace_invalid_tretspef_values, ) from nhp.data.table_names import table_names @@ -28,6 +29,7 @@ def get_inpatients_data(spark: SparkSession) -> DataFrame: df = add_main_icb(spark, hes_apc) 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 diff --git a/src/nhp/data/raw_data/outpatients.py b/src/nhp/data/raw_data/outpatients.py index 710535be..b10565dc 100644 --- a/src/nhp/data/raw_data/outpatients.py +++ b/src/nhp/data/raw_data/outpatients.py @@ -10,6 +10,7 @@ from nhp.data.raw_data.helpers import ( add_age_group_column, remove_mental_health_providers, + replace_invalid_tretspef_values, ) from nhp.data.table_names import table_names @@ -18,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( From 2bd7d06a8501df5c77654ac89d50262fed46da91 Mon Sep 17 00:00:00 2001 From: Tom Jemmett Date: Tue, 6 Oct 2026 10:32:24 +0100 Subject: [PATCH 3/3] removes old special cases for handling the tretspef in A&E/IP --- src/nhp/data/inputs_data/ae/baseline.py | 3 +-- src/nhp/data/inputs_data/ip/baseline.py | 6 ------ src/nhp/data/inputs_data/ip/expat_repat.py | 12 ------------ 3 files changed, 1 insertion(+), 20 deletions(-) diff --git a/src/nhp/data/inputs_data/ae/baseline.py b/src/nhp/data/inputs_data/ae/baseline.py index af257f0f..ac8d64bc 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/baseline.py b/src/nhp/data/inputs_data/ip/baseline.py index f49d62d1..c42932f3 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 720dd2de..1d35de00 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()