### Configure Project Paths and Environment Variables (Python)

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/project_config.ipynb

This Python snippet configures essential project paths, environment variables, and database names. It dynamically determines the project directory, parent directory, and current username. It also sets environment variables for `USERNAME`, `PROJECT_FOLDER`, and `PROJECT_NAME`, disables Spark caching, and defines database aliases. Finally, it copies a `table_mapping.json` file from a central location to the project's configuration directory and identifies archive folders from a monitoring path.

```python
import sys
import os
import shutil

# Project directory
project_directory = os.path.basename(os.path.dirname(dbutils.notebook.entry_point.getDbutils().notebook().getContext().notebookPath().get()))

# Parent dictory
parent_directory = (f'/Workspace{os.path.dirname(os.path.dirname(dbutils.notebook.entry_point.getDbutils().notebook().getContext().notebookPath().get()) )}')

# Username
username = spark.sql('select current_user() as user').collect()[0]['user'] #dbutils.notebook.entry_point.getDbutils().notebook().getContext().tags().apply('user')

# Set working directory to parent to read in functions
os.chdir(f'{parent_directory}')
project_name = f'{project_directory}'
project_folder = f'{parent_directory}/{project_name}/'

# Assign to environment 
os.environ['USERNAME'] = username
os.environ['PROJECT_FOLDER'] = project_folder
os.environ['PROJECT_NAME'] = project_name

# Disable cachespark.conf.set("spark.databricks.io.cache.enabled", "false")

# Database names
db = 'dars_nic_391419_j3w9t'
dbc = f'{db}_collab'
dsa = 'dsa_391419_j3w9t_collab'
dss = 'dss_corporate'


# Copy current table_mapping from table_management
hds_central = f'/Workspace/Shared/hds_team'
hds_central_management = f'{hds_central}/hds_table_management'
hds_central_monitoring = f'{hds_central}/hds_table_monitoring'
table_mapping_source = f'{hds_central_management}/config/table_mapping.json'
table_mapping_destination = f'{parent_directory}/{project_directory}/config/table_mapping.json'
shutil.copyfile(table_mapping_source, table_mapping_destination);
table_mapping_path = f'{parent_directory}/{project_directory}/config/table_mapping.json'

archived_base_path = f'{hds_central_monitoring}/outputs'
archive_folders = [f for f in os.listdir(archived_base_path) if os.path.isdir(os.path.join(archived_base_path, f))]

```

--------------------------------

### Prepare Cohort Dates

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D05-prior_mi.ipynb

Prepares the cohort demographics data by selecting 'person_id', 'date_of_birth' (aliased as 'min_date'), and calculating 'max_date' as one day before the cohort entry start date. This defines the window for historical event analysis.

```python
# This notebook is looking at history of a condition BEFORE the study start. Their history can only occur between being born and the study start date. This code prepares that criteria to be applied later

covartes_prepared = (
    cohort_demographics
    .select(
        'person_id',
        f.col('date_of_birth').alias('min_date'),
        f.date_sub('cohort_entry_start_date', 1).alias('max_date')
    )
)
```

--------------------------------

### Python Module for Environment Setup

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/README.md

Contains utilities for setting up and configuring the Databricks environment (`environment_utils.py`). This module helps in managing environment-specific settings.

```python
# functions/environment_utils.py
# Utilities for environment setup and configuration.
```

--------------------------------

### Prepare Cohort Study Dates

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D04-person_id_and_demographics.ipynb

Adds cohort entry start and end dates, follow-up end date, and production date to the 'cohort_person_id' DataFrame. These dates are converted to a date type and are crucial for time-sensitive data joins.

```python
# Preparing date columns based on those defined in the params notebook. This will be important for joining any time-sensitive eligibility criteria, or measurements later in the pipeline
cohort_study_dates = (
    cohort_person_id
    .withColumn('cohort_entry_start_date', f.to_date(f.lit(cohort_entry_start_date)))
    .withColumn('cohort_entry_end_date', f.to_date(f.lit(cohort_entry_end_date)))
    .withColumn('follow_up_end_date', f.to_date(f.lit(follow_up_end_date)))
    .withColumn('production_date', f.to_date(f.lit(production_date)))
)

```

--------------------------------

### Configure Study Parameters in Python

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/README.md

Sets key dates and parameters for cohort construction, including start and end dates for cohort entry, follow-up, production, and archiving. These Python variables are essential for defining the scope of the analysis.

```python
cohort_entry_start_date = '2020-01-01'
cohort_entry_end_date = '2020-12-31'
follow_up_end_date = '2021-12-31'
production_date = '2025-07-07'
archive_date = '2024-10-24'
```

--------------------------------

### Load Project Configuration and Parameters

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09a-measurements.ipynb

Loads project configuration and parameters using the %run magic command. This step is crucial for setting up the environment and variables needed for subsequent operations.

```python
%run ./project_config
```

```python
%run ./parameters
```

--------------------------------

### Get Latest Archive Version - Python

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D01-data_checks.ipynb

Identifies and returns the path or identifier for the most recent archive version available. This function is crucial for ensuring the analysis uses the latest data.

```python
get_latest_archive(archive_folders)
```

--------------------------------

### Load Project Configuration

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D04-person_id_and_demographics.ipynb

Executes the project configuration notebook, likely setting up environment variables and initial settings.

```python
%run ./project_config
```

--------------------------------

### Load Project Configuration and Parameters

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D10-outcomes.ipynb

This snippet executes external Python scripts to load project configuration and parameters. These are essential for setting up the environment and variables for subsequent operations.

```python
%run ./project_config

```

```python
%run ./parameters

```

--------------------------------

### Load Initial Datasets

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D06-index_mi.ipynb

Loads several key datasets ('cohort_prior_mi', 'hes_apc_diagnosis', 'gdppr') using the 'load_table' function. It then displays the first 50 records of each loaded table for initial inspection.

```python
cohort_prior_mi = load_table('cohort_prior_mi')
hes_apc_diagnosis = load_table('hes_apc_diagnosis')
gdppr = load_table('gdppr', method = 'gdppr')

display(cohort_prior_mi.limit(50))
display(hes_apc_diagnosis.limit(50))
display(gdppr.limit(50))

```

--------------------------------

### Load and Prepare for Window Function

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D05-prior_mi.ipynb

Loads the previously saved 'prior_mi_events' table and defines a window specification. The window partitions data by 'person_id' and orders it by 'date' and 'source_priority', preparing for time-based analysis or ranking within patient histories.

```python
prior_mi_events = load_table('prior_mi_events')

# We have previously defined primary care sourced records as priority '1' and hospital as priority '2'
_win = Window.partitionBy('person_id').orderBy('date', 'source_priority')
```

--------------------------------

### Prepare Cohort Entry Dates

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D06-index_mi.ipynb

Extracts 'min_date' and 'max_date' from the 'cohort_prior_mi' dataset to define the study window. This helps in filtering data to the relevant period, improving performance.

```python
# The study window will vary between projects and will be started in the parameters of a project ease of use, reference and update. Limiting the data to the eligible window early on in the code helps the code run quicker.

cohort_prepared = (
    cohort_prior_mi
    .select(
        'person_id',
        f.col('cohort_entry_start_date').alias('min_date'),
        f.col('cohort_entry_end_date').alias('max_date')
    )
)

```

--------------------------------

### Load Parameters Notebook

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D04-person_id_and_demographics.ipynb

Executes the parameters notebook, which is expected to define variables used throughout the pipeline, such as date ranges.

```python
%run ./parameters
```

--------------------------------

### Prepare Cohort Data with Date Ranges

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09c-medications.ipynb

Creates a 'cohort_prepared_meds' DataFrame from 'cohort_filtered'. It calculates 'min_date' and 'max_date' based on the 'date_of_birth' and 'index_mi_date' respectively, defining a relevant time window for medication analysis.

```python
cohort_prepared_meds = (
    cohort_filtered
    .select(
        'person_id',
        f.date_sub('date_of_birth', 2*365).alias('min_date'),
        f.date_sub('index_mi_date', 0).alias('max_date')
    )
)

```

--------------------------------

### Prepare Cohort Dates for Analysis (Python)

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D07-lsoa_region_imd.ipynb

Selects and transforms columns from the 'cohort_index_mi' DataFrame to create a 'cohort_prepared' DataFrame. It extracts 'person_id', 'index_mi_date' as 'target_date', and calculates 'min_date' (365 days before target_date) and 'max_date' (90 days after target_date). Rows with null 'target_date' are filtered out, and the first 100 rows are displayed. This prepares the cohort data for date-based filtering.

```python
cohort_prepared = (
    cohort_index_mi
    .select(
        'person_id',
        f.col('index_mi_date').alias('target_date'),
        f.date_sub('index_mi_date', 365).alias('min_date'),
        f.date_add('index_mi_date', 3*30).alias('max_date')
    )
    .filter("target_date IS NOT NULL")
)

display(cohort_prepared.limit(100))

```

--------------------------------

### Load Initial DataFrames (Python)

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D07-lsoa_region_imd.ipynb

Loads two critical DataFrames, 'cohort_index_mi' and 'lsoa_multisource', using the custom 'load_table' function. It then displays the first 50 rows of each DataFrame for initial inspection. This step is fundamental for accessing the raw data required for the pipeline.

```python
cohort_index_mi = load_table('cohort_index_mi')
lsoa_multisource = load_table('lsoa_multisource')

display(cohort_index_mi.limit(50))
display(lsoa_multisource.limit(50))

```

--------------------------------

### Load and Display Initial Data Tables

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09c-medications.ipynb

Loads the 'cohort_filtered' and 'primary_care_meds' tables using the custom 'load_table' function. It then displays the first 100 rows of each DataFrame to provide an initial view of the data.

```python
cohort_filtered = load_table('cohort_filtered')
display(cohort_filtered.limit(100))

primary_care_meds = load_table('primary_care_meds', method='primary_care_meds')
display(primary_care_meds.limit(100))

```

--------------------------------

### Prepare Cohort DataFrame for Joining

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09a-measurements.ipynb

Selects and transforms columns from the 'cohort_filtered' DataFrame to create 'cohort_prepared'. It calculates 'min_date' and 'max_date' based on 'index_mi_date' and selects 'person_id' and 'target_date'. This prepares the cohort data for matching with measures.

```python
cohort_prepared = (
    cohort_filtered
    .select(
        'person_id',
        f.date_sub('index_mi_date', 2*365).alias('min_date'),
        f.date_add('index_mi_date', 3*30).alias('max_date'),
        f.col('index_mi_date').alias('target_date')
    )
)
```

--------------------------------

### Load Initial Tables

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D05-prior_mi.ipynb

Loads core data tables including cohort demographics, HES APC diagnoses, and GDPPR records. These tables form the basis for subsequent analysis of medical history. Displays the first 50 rows of each loaded table.

```python
# This notebook establishes the prior history of a diagnosis/medical event - so begins with hes and gdppr tables where medical history is most complete

covariates = load_table('cohort_demographics')
hes_apc_diagnosis = load_table('hes_apc_diagnosis')
gdppr = load_table('gdppr', method = 'gdppr')

display(cohort_demographics.limit(50))
display(hes_apc_diagnosis.limit(50))
display(gdppr.limit(50))
```

--------------------------------

### Prepare GDPPR Data

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D05-prior_mi.ipynb

Formats the GDPPR dataset for analysis by selecting relevant columns ('person_id', 'date', 'code') and adding new columns 'data_source' and 'source_priority' to label the data origin and set its processing priority.

```python
gdppr_prepared = (
    gdppr
    .select(
        'person_id',
        'date',
        'code',
        f.lit('gdppr').alias('data_source'),
        f.lit(1).alias('source_priority')
    )
)
```

--------------------------------

### Load and Prepare LSOA to IMD Mapping (Python)

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D07-lsoa_region_imd.ipynb

Reads a CSV file containing LSOA to Index of Multiple Deprivation (IMD) decile mappings. It selects and renames relevant columns ('lsoa11cd' to 'lsoa', 'IMDDecil' to 'imd_decile') to standardize them for joining. The resulting DataFrame 'mapping_lsoa_imd' is then displayed. This mapping is crucial for enriching LSOA data with deprivation information.

```python
mapping_lsoa_imd = (
    read_csv_file('./mapping_files/lsoa11_imd19_OSGB1936.csv')
    .select(
        f.col('lsoa11cd').alias('lsoa'),
        f.col('IMDDecil').alias('imd_decile')
    )
)

display(mapping_lsoa_imd)

```

--------------------------------

### Prepare GDPPR Data for Merging

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D06-index_mi.ipynb

Selects and renames columns from the 'gdppr' dataset to align with other sources. It adds 'data_source' and 'source_priority' columns for easy identification and merging.

```python
# GDPPR can be a very long set so limiting to only the cols needed here is sensible. We'll also be sourcing data from other datasets and harmonising, so preparing flags to show where the records have come from helps with this.

gdppr_prepared = (
    gdppr
    .select(
        'person_id', 'date', 'code',
        f.lit('gdppr').alias('data_source'),
        f.lit(1).alias('source_priority')
    )
)

```

--------------------------------

### Load and Display Lookup and Demographics Tables

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D04-person_id_and_demographics.ipynb

Loads 'token_pseudo_id_lookup' and 'demographics' tables. The pseudo ID lookup table maps pseudo IDs to labels, while the demographics table contains harmonized personal information. The first 50 rows of each are displayed for initial inspection.

```python
# This table is a curated table that helps understand the pseudo id label used/assigned for each person in the datasets available
token_pseudo_id_lookup = load_table('token_pseudo_id_lookup')

# This curated table provides demographic information, harmonised from multiples sources. Some information (like address aka LSOA) might vary over time and different projects may want this curated in different ways. Please familiarise yourself with how these assets are curated in the HDS Documentation pages
demographics = load_table('demographics')

display(token_pseudo_id_lookup.limit(50))
display(demographics.limit(50))
```

--------------------------------

### Load Initial DataFrames

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09a-measurements.ipynb

Loads the 'cohort_filtered' and 'gdppr' DataFrames using the custom 'load_table' function. The 'gdppr' table is loaded with a specific method ('gdppr'). The first 100 rows of each DataFrame are displayed for inspection.

```python
cohort_filtered = load_table('cohort_filtered')
display(cohort_filtered.limit(100))

gdppr = load_table('gdppr', method='gdppr')
display(gdppr.limit(100))
```

--------------------------------

### Load and Display Final Cohort Table (Python)

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D07-lsoa_region_imd.ipynb

Loads the previously saved 'cohort_lsoa_imd_region' table using the 'load_table' function and displays the first 1000 rows. This step serves to verify the contents of the final processed table, confirming that the LSOA, IMD, and region information has been successfully appended to the cohort data.

```python
cohort_lsoa_imd_region = load_table('cohort_lsoa_imd_region')
display(cohort_lsoa_imd_region.limit(1000))

```

--------------------------------

### Prepare GDPPR DataFrame for Joining

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09a-measurements.ipynb

Selects and renames columns from the 'gdppr' DataFrame to create 'gdppr_prepared'. It specifically extracts 'person_id', 'date', 'code', and aliases 'value1_condition' to 'value'. This prepares the GDPPR data for further analysis.

```python
gdppr_prepared = (
    gdppr
    .select(
        'person_id', 'date', 'code',
        f.col('value1_condition').alias('value')
    )
)
```

--------------------------------

### Load and Display Initial Data Tables

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D10-outcomes.ipynb

Loads 'cohort_filtered' and 'hes_apc_procedure' tables into PySpark DataFrames and displays the first 100 rows of each. This step is for initial data inspection and verification.

```python
cohort_filtered = load_table('cohort_filtered')
display(cohort_filtered.limit(100))

hes_apc_procedure = load_table('hes_apc_procedure')
display(hes_apc_procedure.limit(100))

```

--------------------------------

### Append Earliest MI Events to Cohort

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D06-index_mi.ipynb

This section appends the earliest MI event information, processed in the previous step, to an existing cohort table. It loads both 'cohort_prior_mi' and 'index_mi_earliest' tables and performs a left join on 'person_id'. The resulting combined table, 'cohort_index_mi', containing prior cohort information and the earliest MI details, is then saved. This is crucial for enriching cohort data with specific event timelines.

```python
cohort_prior_mi = load_table('cohort_prior_mi')
index_mi_earliest = load_table('index_mi_earliest')

cohort_index_mi = (
    cohort_prior_mi
    .join(
        index_mi_earliest,
        on='person_id', how='left'
    )
)
    
save_table(cohort_index_mi, 'cohort_index_mi')
```

--------------------------------

### Display Sample of Combined Cohort Table in Python

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D11-combine_tables.ipynb

Loads the previously saved `cohort_final` table using the `load_table` function and then displays the first 1000 rows of this DataFrame using the `display` function. This is useful for inspecting the data after it has been processed and combined.

```python
cohort_final = load_table('cohort_final')
display(cohort_final.limit(1000))
```

--------------------------------

### Prepare LSOA Multisource Data with Priority (Python)

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D07-lsoa_region_imd.ipynb

Prepares the 'lsoa_multisource' DataFrame by filtering records to include only specified data sources and assigning a priority to each source. A dictionary 'lsoa_source_priority' defines the order, and the 'map_column_values' function is used to add a 'source_priority' column. The first 100 rows of the prepared 'lsoa_prepared' DataFrame are displayed. This step is crucial for resolving conflicting residence data.

```python
# Needed in case sources from around the same time disagree on the residence of the individual, then 
lsoa_source_priority = {
    'gdppr': 1,
    'hes_apc': 2,
    'hes_op': 3,
    'hes_ae': 4
}

lsoa_prepared = (
    lsoa_multisource
    .filter(f.col('data_source').isin(list(lsoa_source_priority.keys())))
    .transform(
        map_column_values,
        map_dict=lsoa_source_priority,
        column='data_source', new_column='source_priority'
    )
)

display(lsoa_prepared.limit(100))

```

--------------------------------

### Display Limited Cohort Prior MI Data

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D05-prior_mi.ipynb

This snippet loads the combined cohort and prior MI data and displays the first 250 records. It's useful for a quick inspection of the results. Dependencies include the 'cohort_prior_mi' table. The output is a displayed table.

```python
cohort_prior_mi = load_table('cohort_prior_mi')
display(cohort_prior_mi.limit(250))
```

--------------------------------

### Import Libraries and Functions

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D06-index_mi.ipynb

Imports necessary PySpark modules, functions for data loading and saving, and the reduce function from functools. These are foundational for subsequent data manipulation and processing.

```python
from pyspark.sql import functions as f, DataFrame
from pyspark.sql.window import Window
from functions import load_table, save_table, read_csv_file
from functools import reduce

```

--------------------------------

### List Available Datasets - Python

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D01-data_checks.ipynb

Retrieves and displays the keys from a JSON file that maps table names to their configurations. This helps users identify valid dataset names for their analysis.

```python
list((read_json_file(table_mapping_path)).keys())
```

--------------------------------

### Prepare GDPPR Data for Matching

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09b-comorbidities.ipynb

Selects and transforms columns from the 'gdppr' DataFrame to create 'gdppr_prepared'. It includes 'person_id', 'date', 'code', and adds two new columns: 'data_source' as 'gdppr' and 'source_priority' as 1. This prepares the data for joining with codelists and cohort information.

```python
gdppr_prepared = (
    gdppr
    .select(
        'person_id', 'date', 'code',
        f.lit('gdppr').alias('data_source'),
        f.lit(1).alias('source_priority')
    )
)

```

--------------------------------

### Import Libraries

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D05-prior_mi.ipynb

Imports necessary PySpark and standard Python libraries for data manipulation and pipeline operations. Includes functions for loading/saving tables and reading CSVs.

```python
from pyspark.sql import functions as f, DataFrame
from pyspark.sql.window import Window
from functions import load_table, save_table, read_csv_file
from functools import reduce
```

--------------------------------

### Join LSOA with IMD and Region Mappings (Python)

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D07-lsoa_region_imd.ipynb

Enriches the 'lsoa_selected' DataFrame by performing left joins with the 'mapping_lsoa_imd' and 'mapping_lsoa_region' DataFrames on the 'lsoa' column. Broadcast joins are used for efficiency with the mapping tables. The resulting 'lsoa_mapped' DataFrame contains LSOA, IMD decile, and region information.

```python
lsoa_mapped = (
    lsoa_selected
    .join(
        f.broadcast(mapping_lsoa_imd),
        on='lsoa', how='left' 
    )
    .join(
        f.broadcast(mapping_lsoa_region),
        on='lsoa', how='left' 
    )
)

```

--------------------------------

### Import PySpark Functions and Utilities

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D04-person_id_and_demographics.ipynb

Imports necessary functions from PySpark SQL for data manipulation and custom utility functions for table management and statistical analysis.

```python
from pyspark.sql import functions as f
from functions.table_management import load_table, save_table
from functions.functions import tabstat, count_var
```

--------------------------------

### Display Final Cohort Medications Table

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09c-medications.ipynb

Loads the final 'cohort_medications' DataFrame and displays the first 100 rows. This provides a preview of the result after all processing steps.

```python
cohort_medications = load_table('cohort_medications')
display(cohort_medications.limit(100))

```

--------------------------------

### Prepare Cohort Data with PySpark

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D10-outcomes.ipynb

Selects 'person_id' from 'cohort_filtered' and calculates 'min_date' and 'max_date' for each person based on index MI date and follow-up end dates. This defines the observation window for each individual.

```python
cohort_prepared = (
    cohort_filtered
    .select(
        'person_id',
        f.date_add('index_mi_date', 1).alias('min_date'),
        f.least(f.col('date_of_death'), f.col('follow_up_end_date')).alias('max_date')
    )
)

```

--------------------------------

### Join, Filter, and Find Closest LSOA Record (Python)

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D07-lsoa_region_imd.ipynb

Performs a series of transformations on cohort and LSOA data. It joins 'cohort_prepared' with 'lsoa_prepared' on 'person_id', filters records within a date range ('min_date' to 'max_date'), calculates date differences, and uses a window function to rank records by proximity to the 'target_date' and data source priority. It then filters to keep only the rank 1 record for each person and selects specific columns, aliasing 'data_source' to 'lsoa_source' and 'record_date' to 'lsoa_date'.

```python
lsoa_joined = (
    cohort_prepared
    .join(
        lsoa_prepared,
        on='person_id', how='inner'
    )
)

lsoa_filtered = (
    lsoa_joined
    .filter(f.col('record_date').between(f.col('min_date'), f.col('max_date')))
)

_window = Window.partitionBy('person_id').orderBy('abs_date_diff', 'source_priority')

lsoa_closest = (
    lsoa_filtered
    .withColumn('date_diff', f.datediff('record_date', 'target_date'))
    .withColumn('abs_date_diff', f.abs(f.col('date_diff')))
    .withColumn('rank', f.rank().over(_window))
    .filter(f.col('rank') == 1)
)

lsoa_selected = (
    lsoa_closest
    .select(
        'person_id', 'lsoa',
        f.col('data_source').alias('lsoa_source'),
        f.col('record_date').alias('lsoa_date')
    )
)

```

--------------------------------

### Aggregate and Pivot Medication Data

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09c-medications.ipynb

Loads 'medications_all_events', determines the last medication event for each person and phenotype using a window function, and then pivots the data. This creates a wider format where each phenotype becomes a column with details about the last event.

```python
medications_all_events = load_table('medications_all_events')

_win = Window.partitionBy('person_id', 'phenotype').orderBy(f.col('date').desc(), 'source_priority')

medications_last_event = (
    medications_all_events
    .withColumn('rank', f.row_number().over(_win))
    .filter('rank = 1')
    .withColumn('flag', f.lit(1))
)

medications_last_event = (
    medications_last_event
    .groupBy('person_id')
    .pivot('phenotype')
    .agg(
        f.first('flag').alias('flag'),
        f.first('date').alias('date'),
        f.first('code').alias('code'),
        f.first('data_source').alias('source')
    )
)

save_table(medications_last_event, 'medications_last_event')

```

--------------------------------

### Importing Pipeline Functions

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/README.md

Demonstrates how to import essential functions from the pipeline's utility modules. These functions cover table management, cohort construction, data privacy, and CSV utility operations.

```python
from functions.table_management import load_table, save_table
from functions.cohort_construction import apply_inclusion_criteria
from functions.data_privacy import round_counts
from functions.csv_utils import load_codelist
```

--------------------------------

### Loading and Using Codelists

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/README.md

Illustrates the process of loading a codelist from a CSV file and subsequently using it to filter a DataFrame. This is crucial for standardizing data based on predefined codes.

```python
from functions.csv_utils import load_codelist

# Load a codelist
diabetes_codes = load_codelist('codelists/diabetes_snomed.csv')

# Filter data using codelist
diabetes_patients = df.join(diabetes_codes, on='code', how='inner')
```

--------------------------------

### Load and Prepare LSOA to Region Mapping (Python)

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D07-lsoa_region_imd.ipynb

Reads a CSV file that maps LSOA codes to region names. It selects and renames columns ('LSOA11CD' to 'lsoa', 'RGN11NM' to 'region') for consistency. The 'mapping_lsoa_region' DataFrame, containing this mapping, is then displayed. This is used to add regional information to the LSOA data.

```python
mapping_lsoa_region = (
    read_csv_file('./mapping_files/lsoa11_buasd11_bua11_rgn11_best_fit_lookup_england_wales.csv')
    .select(
        f.col('LSOA11CD').alias('lsoa'),
        f.col('RGN11NM').alias('region')
    )
)

display(mapping_lsoa_region)

```

--------------------------------

### Load and Display Saved Cohort Demographics

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D04-person_id_and_demographics.ipynb

Reloads the 'cohort_demographics' table from the database and displays the first 50 rows. This step allows for querying and inspection of the previously saved data without rerunning the entire pipeline.

```python
cohort_demographics = load_table('cohort_demographics')
display(cohort_demographics.limit(50))
```

--------------------------------

### Filter and Join Medication Data with Cohort

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09c-medications.ipynb

Joins the processed 'primary_care_meds' with the 'codelist_bnf' and 'cohort_prepared_meds' DataFrames. It filters the results to include only medications prescribed within the calculated date range for each person.

```python
primary_care_meds_matched = (
    primary_care_meds
    .join(
        f.broadcast(codelist_bnf),
        on='code', how='inner'
    )
    .join(
        cohort_prepared_meds,
        on='person_id', how='inner'
    )
    .filter("(date >= min_date) AND (date <= max_date)")
)

```

--------------------------------

### Select Relevant Person ID Flags

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D04-person_id_and_demographics.ipynb

Creates a 'cohort_person_id' DataFrame by selecting the 'pseudo_id' (aliased as 'person_id') and 'valid_nhs_number' from the 'token_pseudo_id_lookup' table.

```python
# Select relevant person_id flags
cohort_person_id = (
    token_pseudo_id_lookup
    .select(f.col('pseudo_id').alias('person_id'), 'valid_nhs_number')
)

```

--------------------------------

### Display First 500 Rows of Combined Cohort

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D06-index_mi.ipynb

This snippet displays the first 500 rows of the 'cohort_index_mi' table, which combines prior cohort data with the earliest MI event information. It's a utility function primarily used for data inspection and verification after the join operation. The 'display' function is commonly used in environments like Databricks notebooks for interactive data exploration.

```python
cohort_index_mi = load_table('cohort_index_mi')
display(cohort_index_mi.limit(500))
```

--------------------------------

### Load Initial Data Tables

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09b-comorbidities.ipynb

Loads three key data tables: 'cohort_filtered', 'gdppr', and 'hes_apc_diagnosis'. The 'gdppr' table is loaded with a specific 'gdppr' method. The `display` function is used to show the first 100 rows of each loaded table, allowing for initial data inspection.

```python
cohort_filtered = load_table('cohort_filtered')
display(cohort_filtered.limit(100))

gdppr = load_table('gdppr', method='gdppr')
display(gdppr.limit(100))

hes_apc_diagnosis = load_table('hes_apc_diagnosis')
display(hes_apc_diagnosis.limit(100))

```

--------------------------------

### Combine Cohort with Pivoted Medication Data

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09c-medications.ipynb

Loads 'cohort_filtered' and the pivoted 'medications_last_event' DataFrames. It performs a left join to associate each person in the cohort with their last medication event details, creating the final 'cohort_medications' table.

```python
cohort_filtered = load_table('cohort_filtered')
medications_last_event = load_table('medications_last_event')

cohort_medications = (
    cohort_filtered
    .select('person_id')
    .join(
        medications_last_event,
        on='person_id', how='left'
    )
)
    
save_table(cohort_medications, 'cohort_medications')

```

--------------------------------

### Process and Combine Codelists from CSV Files

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09c-medications.ipynb

Reads multiple codelist CSV files, assigns a 'phenotype' to each based on the filename, and combines them into a single Spark DataFrame. This step prepares medication codes for filtering and analysis.

```python
dict_codelists_bnf = {
    "antihypertensives": "./codelists/antihypertensives_bnf.csv",
    "insulin": "./codelists/insulin_bnf.csv",
}

list_codelists_bnf = [
    read_csv_file(codelist_path)
    .withColumn('phenotype', f.lit(phenotype))
    for phenotype, codelist_path in dict_codelists_bnf.items()
]

codeli_bnf = spark.createDataFrame(union_dataframe_list(list_codelists_bnf))

display(codelist_bnf)

```

--------------------------------

### Match Cohort with Prior MI Information

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D05-prior_mi.ipynb

This Python code joins the main cohort demographics with the previously processed prior MI information. It performs a left join on 'person_id' to ensure all cohort members are retained. Dependencies include 'cohort_demographics' and 'prior_mi_earliest' tables. The output is a table named 'cohort_prior_mi'.

```python
# We now retrieve the cohort from earlier to add the relevant prior_MI information columns to the respective individuals

cohort_demographics = load_table('cohort_demographics')
prior_mi_earliest = load_table('prior_mi_earliest')

cohort_prior_mi = (
    cohort_demographics
    .join(
        prior_mi_earliest,
        on='person_id', how='left'
    )
)
    
save_table(cohort_prior_mi, 'cohort_prior_mi')
```

--------------------------------

### Python Script for Table Directory Creation

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/README.md

A Python script (`D03-create_table_directory.py`) used for table configuration mapping. It likely defines how different data tables are organized and referenced within the pipeline, ensuring consistent data access.

```python
# D03-create_table_directory.py
# This script handles table configuration mapping.
```

--------------------------------

### Append LSOA, IMD, and Region to Cohort and Save (Python)

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D07-lsoa_region_imd.ipynb

Appends the prepared LSOA, IMD, and region information from 'lsoa_mapped' to the original 'cohort_index_mi' DataFrame using a left join on 'person_id'. The combined DataFrame, named 'cohort_lsoa_imd_region', is then saved using the custom 'save_table' function. This consolidates all relevant LSOA-related attributes with the cohort data.

```python
cohort_lsoa_imd_region = (
    cohort_index_mi
    .join(
        lsoa_mapped,
        on='person_id', how='left'
    )
)

save_table(cohort_lsoa_imd_region, 'cohort_lsoa_imd_region')

```

--------------------------------

### Import PySpark Libraries and Custom Functions

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D10-outcomes.ipynb

Imports necessary PySpark SQL functions and custom utility functions for data loading, saving, CSV reading, column mapping, and DataFrame union operations. These functions facilitate efficient data manipulation.

```python
from pyspark.sql import functions as f, DataFrame
from pyspark.sql.window import Window
from functions import load_table, save_table, read_csv_file, map_column_values
from functions.functions import union_dataframe_list

```

--------------------------------

### Extract Earliest MI Event per Patient

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D06-index_mi.ipynb

This code snippet processes the 'index_mi_events' table to find the earliest Myocardial Infarction (MI) event for each patient. It uses PySpark's window functions to partition data by 'person_id' and order by 'date' and 'source_priority' to determine the chronological order. The earliest event is then filtered, and the data is pivoted to aggregate flags, dates, codes, and sources for each MI phenotype per patient. The output is saved to 'index_mi_earliest'.

```python
index_mi_events = load_table('index_mi_events')

_win = Window.partitionBy('person_id').orderBy('date', 'source_priority')

index_mi_earliest = (
    index_mi_events
    .withColumn('rank', f.row_number().over(_win))
    .filter('rank = 1')
    .withColumn('flag', f.lit(1))
)

index_mi_earliest = (
    index_mi_earliest
    .groupBy('person_id')
    .pivot('phenotype')
    .agg(
        f.first('flag').alias('flag'),
        f.first('date').alias('date'),
        f.first('code').alias('code'),
        f.first('data_source').alias('source')
    )
)

save_table(index_mi_earliest, 'index_mi_earliest')
```

--------------------------------

### Import Data Quality Functions - Python

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D01-data_checks.ipynb

Imports necessary functions for data quality reporting and retrieving archive information from the project's utility modules. Relies on the 'functions' directory containing 'table_monitoring' and other helper modules.

```python
from functions.table_monitoring import data_quality_report, get_latest_archive
from functions import read_json_file
```

--------------------------------

### Select and Alias Columns for Primary Care Medications

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09c-medications.ipynb

Transforms the 'primary_care_meds' DataFrame by selecting specific columns ('person_id', 'prescribedbnfcode', 'processingperioddate') and aliasing them to more general names ('code', 'date', 'data_source', 'source_priority'). This standardizes the medication data.

```python
primary_care_meds = (
    primary_care_meds
    .select(
        'person_id',
        f.col('prescribedbnfcode').alias('code'),
        f.col('processingperioddate').alias('date'),
        f.lit('primary_care_meds').alias('data_source'),
        f.lit(1).alias('source_priority')
    )
)

```

--------------------------------

### Databricks Notebook for Measurement Extraction

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/README.md

Extracts key patient measurements, including BMI, height, and weight (`D09a-measurements.ipynb`). This notebook is essential for analyzing anthropometric data related to the cohort.

```python
# D09a-measurements.ipynb
# Extracts measurements like BMI, height, and weight.
```

--------------------------------

### Import PySpark Functions and Custom Functions

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09a-measurements.ipynb

Imports necessary modules from PySpark's SQL functions and windowing capabilities, along with custom functions for data loading, saving, CSV reading, and data mapping. These functions encapsulate common data manipulation tasks.

```python
from pyspark.sql import functions as f, DataFrame
from pyspark.sql.window import Window
from functions import load_table, save_table, read_csv_file, apply_inclusion_criteria, map_column_values
from functions.functions import union_dataframe_list
```

--------------------------------

### Databricks Notebook for Final Cohort Assembly

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/README.md

Combines all previously processed tables and data to assemble the final cohort (`D11-combine_tables.ipynb`). This notebook serves as the concluding step in the cohort construction process.

```python
# D11-combine_tables.ipynb
# Assembles the final cohort by combining processed tables.
```

--------------------------------

### Process Prior MI Events to Patient Level

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D05-prior_mi.ipynb

This code processes prior MI events, prioritizing the earliest record per patient. It then aggregates and pivots the data to create a patient-level summary table. Dependencies include the 'prior_mi_events' table and Spark SQL functions. The output is a table named 'prior_mi_earliest'.

```python
prior_mi_earliest = (
    prior_mi_events
    .withColumn('rank', f.row_number().over(_win))
    .filter('rank = 1')
    .withColumn('flag', f.lit(1))
)

prior_mi_earliest = (
    prior_mi_earliest
    .groupBy('person_id')
    .pivot('phenotype')
    .agg(
        f.first('flag').alias('flag'),
        f.first('date').alias('date'),
        f.first('code').alias('code'),
        f.first('data_source').alias('source')
    )
)

save_table(prior_mi_earliest, 'prior_mi_earliest')
```

--------------------------------

### Databricks Notebook for Cohort and Demographics

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/README.md

A Databricks notebook (`D04-person_id_and_demographics.ipynb`) focused on creating the main cohort and extracting demographic information. It processes person identifiers and associated demographic data.

```python
# D04-person_id_and_demographics.ipynb
# Handles cohort creation and demographics extraction.
```

--------------------------------

### Prepare Cohort Filtered Data

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09b-comorbidities.ipynb

Selects and renames columns from 'cohort_filtered' to create 'cohort_prepared'. It takes 'person_id', aliases 'date_of_birth' to 'min_date', and calculates 'max_date' by subtracting one day from 'index_mi_date'. This defines the date range for matching comorbidities.

```python
cohort_prepared = (
    cohort_filtered
    .select(
        'person_id',
        f.col('date_of_birth').alias('min_date'),
        f.date_sub('index_mi_date', 1).alias('max_date')
    )
)

```

--------------------------------

### Configure Databricks Database Names in Python

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/README.md

Specifies the names of the Databricks databases to be used for raw data, collaboration, curated assets, and reference data. These configurations are crucial for connecting the pipeline to the correct data sources within the Databricks environment.

```python
db = 'your_raw_data_database'
dbc = 'your_collab_database'
dsa = 'your_curated_assets_database'
dss = 'your_reference_database'
```

--------------------------------

### Load Saved Cohort Measures

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09a-measurements.ipynb

Loads the previously saved 'cohort_measures' DataFrame using the 'load_table' function. The first 100 rows of the loaded DataFrame are then displayed for verification.

```python
cohort_measures = load_table('cohort_measures')
display(cohort_measures.limit(100))
```

--------------------------------

### Python Module for Cohort Construction Logic

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/README.md

Encapsulates the logic for cohort construction, including inclusion and exclusion criteria (`cohort_construction.py`). This module defines the rules for selecting individuals into the study cohort.

```python
# functions/cohort_construction.py
# Implements cohort inclusion and exclusion criteria.
```

--------------------------------

### Join Measures with Codelists and Cohort

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09a-measurements.ipynb

Performs two join operations. First, it joins 'gdppr_prepared' with 'codelist_snomed' on 'code' and filters by value ranges. Second, it joins the result with 'cohort_prepared' on 'person_id' and filters by date ranges, creating 'measures_cohort_matched'.

```python
measures_code_matched = (
    gdppr_prepared
    .join(
        f.broadcast(codelist_snomed),
        on='code', how='inner'
    )
    .filter(f.col('value').between(f.col('min_value'), f.col('max_value')))
)

measures_cohort_matched = (
    measures_code_matched
    .join(
        cohort_prepared,
        on='person_id', how='inner'
    )
    .filter(f.col('date').between(f.col('min_date'), f.col('max_date')))
)
```

--------------------------------

### Display Final Cohort Outcomes Data

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D10-outcomes.ipynb

Loads the 'cohort_outcomes' table and displays the first 100 rows. This serves as a final check and display of the processed cohort outcomes data.

```python
cohort_outcomes = load_table('cohort_outcomes')
display(cohort_outcomes.limit(100))

```

--------------------------------

### Match Procedures to Cohort within Date Range using PySpark

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D10-outcomes.ipynb

Joins prepared procedure data with the codelist and cohort information, filtering for procedures that occurred within the defined observation window ('min_date' to 'max_date'). This identifies relevant procedures for each cohort member.

```python
hes_apc_procedure_matched = (
    hes_apc_procedure_prepared
    .join(
        f.broadcast(codelist_opcs),
        on='code', how='inner'
    )
    .join(
        cohort_prepared,
        on='person_id', how='inner'
    )
    .filter("(date >= min_date) AND (date <= max_date)")
)

```

--------------------------------

### Identify First Event per Phenotype and Save with PySpark

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D10-outcomes.ipynb

Loads 'outcomes_all_events', ranks events by date for each person and phenotype, filters for the first event, and then pivots the data to create flags and dates for the first occurrence of each phenotype. The result is saved as 'outcomes_first_event'.

```python
outcomes_all_events = load_table('outcomes_all_events')

_win = Window.partitionBy('person_id', 'phenotype').orderBy(f.col('date').asc(), 'source_priority')

outcomes_first_event = (
    outcomes_all_events
    .withColumn('rank', f.row_number().over(_win))
    .filter('rank = 1')
    .withColumn('flag', f.lit(1))
)

outcomes_first_event = (
    outcomes_first_event
    .groupBy('person_id')
    .pivot('phenotype')
    .agg(
        f.first('flag').alias('flag'),
        f.first('date').alias('date'),
        f.first('code').alias('code'),
        f.first('data_source').alias('source')
    )
)

save_table(outcomes_first_event, 'outcomes_first_event')

```

--------------------------------

### Save Matched Medication Events

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09c-medications.ipynb

Assigns the 'primary_care_meds_matched' DataFrame to 'medications_all_events' and then saves this DataFrame to a persistent table. This step stores the filtered and joined medication records.

```python
medications_all_events = (
    primary_care_meds_matched
)

save_table(medications_all_events, 'medications_all_events')

```

--------------------------------

### Databricks Notebook for Medication History

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/README.md

Extracts and analyzes medication history for the cohort members (`D09c-medications.ipynb`). This involves processing records related to prescribed and administered drugs.

```python
# D09c-medications.ipynb
# Analyzes patient medication history.
```

--------------------------------

### Match GDPPR Events to Cohort and Codelist

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D06-index_mi.ipynb

Joins the prepared GDPPR data with the SNOMED codelist and the prepared cohort data. It filters the results to include only events within the defined study window ('min_date' to 'max_date').

```python
# Here we take the codelists of codes we are interested in for this study, find only the records in primary care (gdppr) that match, join this to the cohort we've already defined thus far and a limit to only the study window.

gdppr_matched = (
    gdppr_prepared
    .join(
        f.broadcast(codelist_snomed),
        on='code', how='inner'
    )
    .join(
        cohort_prepared,
        on='person_id', how='inner'
    )
    .filter("(date >= min_date) AND (date <= max_date)")
)

```

--------------------------------

### Prepare HES APC Diagnosis Data

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D05-prior_mi.ipynb

Formats the HES APC diagnosis dataset by selecting 'person_id', 'code', and 'epistart' (aliased as 'date'). It also adds 'data_source' and 'source_priority' columns to identify the data origin and set its processing priority.

```python
hes_apc_prepared = (
    hes_apc_diagnosis
    .select(
        'person_id',
        'code',
        f.col('epistart').alias('date'),
        f.lit('hes_apc').alias('data_source'),
        f.lit(2).alias('source_priority')
    )
)
```

--------------------------------

### Prepare HES APC Diagnosis Data for Matching

Source: https://github.com/bhfdsc/standard-pipeline/blob/main/D09b-comorbidities.ipynb

Transforms the 'hes_apc_diagnosis' DataFrame into 'hes_apc_prepared'. It selects 'person_id' and 'code', aliases 'epistart' to 'date', and adds 'data_source' as 'hes_apc' with 'source_priority' set to 2. This step standardizes the HES APC data for subsequent analysis.

```python
hes_apc_prepared = (
    hes_apc_diagnosis
    .select(
        'person_id', 'code',
        f.col('epistart').alias('date'),
        f.lit('hes_apc').alias('data_source'),
        f.lit(2).alias('source_priority')
    )
)

```