SQL_script.R

library(dplyr)
library(dbplyr)
library(dotenv)
library(DBI)
library(RPostgres)

# Load connection params into variables
PGSQL_HOSTNAME=("spinup-db003025.c9ukc6s0rmbg.us-east-1.rds.amazonaws.com")
PGSQL_USERNAME=("REDACTED")
PGSQL_PASSWORD=("REDACTED")
PGSQL_DATABASE=("postgres")

# Load connection params into variables
host <- PGSQL_HOSTNAME
username <- PGSQL_USERNAME
password <- PGSQL_PASSWORD
database_name <- PGSQL_DATABASE




# Connect to the database
con <- DBI::dbConnect(
  RPostgres::Postgres(),
  host = host,
  dbname = database_name,
  user = username,
  password = password,
  options = "-c search_path=omop"
)

# Set start/end dates and relevant concept IDs
cohort_start  <- as.Date("2021-01-01")
cohort_end    <- as.Date("2025-06-30")
obs_end       <- as.Date("2025-06-30")

# Checking for correct concept IDs
# T2DM 
tbl(con, sql("
  SELECT DISTINCT c.concept_id, c.concept_name
  FROM omop.condition_occurrence co
  JOIN omop.concept c ON co.condition_concept_id = c.concept_id
  WHERE LOWER(c.concept_name) LIKE '%type 2%diabetes%'
     OR LOWER(c.concept_name) LIKE '%diabetes mellitus type 2%'
  LIMIT 20
")) %>% collect() %>% print()

# T1DM 
tbl(con, sql("
  SELECT DISTINCT c.concept_id, c.concept_name
  FROM omop.condition_occurrence co
  JOIN omop.concept c ON co.condition_concept_id = c.concept_id
  WHERE LOWER(c.concept_name) LIKE '%type 1diabetes%'
     OR LOWER(c.concept_name) LIKE '%diabetes mellitus type 1%'
  LIMIT 20
")) %>% collect() %>% print()

#  HTN 
tbl(con, sql("
  SELECT DISTINCT c.concept_id, c.concept_name
  FROM omop.condition_occurrence co
  JOIN omop.concept c ON co.condition_concept_id = c.concept_id
  WHERE LOWER(c.concept_name) LIKE '%hypertension%'
  LIMIT 20
")) %>% collect() %>% print()

# HbA1c 
tbl(con, sql("
  SELECT DISTINCT c.concept_id, c.concept_name
  FROM omop.measurement m
  JOIN omop.concept c ON m.measurement_concept_id = c.concept_id
  WHERE LOWER(c.concept_name) LIKE '%hemoglobin a1c%'
     OR LOWER(c.concept_name) LIKE '%hba1c%'
     OR LOWER(c.concept_name) LIKE '%glycated hemoglobin%'
  LIMIT 20
")) %>% collect() %>% print()

#  SBP 
tbl(con, sql("
  SELECT DISTINCT c.concept_id, c.concept_name
  FROM omop.measurement m
  JOIN omop.concept c ON m.measurement_concept_id = c.concept_id
  WHERE LOWER(c.concept_name) LIKE '%systolic%'
  LIMIT 20
")) %>% collect() %>% print()

# Set concept IDs or load concept IDs from relationships in vocabulary tables
t1dm_concept_ids  <- c(201254)        # T1DM diagnoses
t2dm_concept_ids  <- c(201826)        # T2DM diagnoses
htn_concept_ids   <- c(320128)         # Hypertension diagnoses
hba1c_concept_ids <- c(3004410)  # HbA1c measurements
sbp_concept_ids   <- c(3004249)                  # Systolic BP

# Identify and load needed visits
# Note: with dbplyr, query results use lazy loading
#  so binding the query: tbl(con, sql("SQL QUERY HERE"))
#  will not actually read/return records until you collect()
visit_df <- tbl(con, sql("
  SELECT
    v.visit_occurrence_id,
    v.person_id,
    v.visit_concept_id,
    v.visit_start_date,
    v.visit_end_date,
    p.gender_concept_id,
    p.birth_datetime
  FROM omop.visit_occurrence v
  JOIN omop.person p ON v.person_id = p.person_id
  WHERE v.visit_start_date >= '2021-01-01'
    AND v.visit_end_date   <= '2025-06-30'
")) %>% collect()


# Identify and load needed measurements for cohort

# age at each visit
visit_df <- visit_df %>%
  mutate(
    birth_date        = as.Date(birth_datetime),
    visit_start_date  = as.Date(visit_start_date),
    visit_end_date    = as.Date(visit_end_date),
    age_years_at_visit = as.integer(
      floor(as.numeric(difftime(visit_start_date, birth_date, units = "days")) / 365.25)
    )
  )

# Apply inclusion criteria:
#    age >= 18 at first visit within cohort period
#    >= 2 visits during cohort period
first_visit_age <- visit_df %>%
  group_by(person_id) %>%
  slice_min(visit_start_date, n = 1, with_ties = FALSE) %>%
  summarise(age_at_first_visit = first(age_years_at_visit))

visit_counts <- visit_df %>%
  group_by(person_id) %>%
  summarise(n_visits = n())

eligible_patients <- first_visit_age %>%
  inner_join(visit_counts, by = "person_id") %>%
  filter(age_at_first_visit >= 18, n_visits >= 2) %>%
  pull(person_id)

visit_df <- visit_df %>%
  filter(person_id %in% eligible_patients)

# Joining gender concept name
gender_concepts <- tbl(con, sql("
  SELECT concept_id, concept_name AS gender_concept_name
  FROM omop.concept
  WHERE domain_id = 'Gender'
")) %>% collect()

visit_df <- visit_df %>%
  left_join(gender_concepts, by = c("gender_concept_id" = "concept_id"))

# Identify and load needed conditions for cohort

# blood measurements 
meas_concept_ids_str <- paste(c(hba1c_concept_ids, sbp_concept_ids), collapse = ",")

meas_df <- tbl(con, sql(paste0("
  SELECT
    person_id,
    visit_occurrence_id,
    measurement_concept_id,
    measurement_date,
    value_as_number
  FROM omop.measurement
  WHERE measurement_concept_id IN (", meas_concept_ids_str, ")
    AND measurement_date <= '2025-06-30'
    AND value_as_number IS NOT NULL
"))) %>% collect() %>%
  mutate(measurement_date = as.Date(measurement_date))
# SBP, visit level aggregate
sbp_visit <- meas_df %>%
  filter(measurement_concept_id %in% sbp_concept_ids) %>%
  inner_join(visit_df %>% select(visit_occurrence_id, visit_start_date, visit_end_date),
             by = "visit_occurrence_id") %>%
  group_by(visit_occurrence_id) %>%
  summarise(
    visit_min_sbp  = as.integer(floor(min(value_as_number,  na.rm = TRUE))),
    visit_mean_sbp = as.integer(floor(mean(value_as_number, na.rm = TRUE))),
    visit_max_sbp  = as.integer(floor(max(value_as_number,  na.rm = TRUE)))
  )
# SBP, prior max (before first visit)
sbp_all <- meas_df %>%
  filter(measurement_concept_id %in% sbp_concept_ids) %>%
  select(person_id, measurement_date, value_as_number)

sbp_prior <- visit_df %>%
  select(person_id, visit_occurrence_id, visit_start_date) %>%
  left_join(sbp_all, by = "person_id") %>%
  filter(measurement_date < visit_start_date) %>%
  group_by(visit_occurrence_id) %>%
  summarise(prior_max_sbp = as.integer(floor(max(value_as_number, na.rm = TRUE)))) %>%
  mutate(prior_max_sbp = if_else(is.infinite(prior_max_sbp), NA_integer_, prior_max_sbp))

# HbA1c max
hba1c_visit <- meas_df %>%
  filter(measurement_concept_id %in% hba1c_concept_ids) %>%
  inner_join(visit_df %>% select(visit_occurrence_id, visit_start_date, visit_end_date),
             by = "visit_occurrence_id") %>%
  group_by(visit_occurrence_id) %>%
  summarise(visit_max_hba1c = max(value_as_number, na.rm = TRUE)) %>%
  mutate(visit_max_hba1c = if_else(is.infinite(visit_max_hba1c), NA_real_, visit_max_hba1c))
# HbA1c prior max
hba1c_all <- meas_df %>%
  filter(measurement_concept_id %in% hba1c_concept_ids) %>%
  select(person_id, measurement_date, value_as_number)

hba1c_prior <- visit_df %>%
  select(person_id, visit_occurrence_id, visit_start_date) %>%
  left_join(hba1c_all, by = "person_id") %>%
  filter(measurement_date < visit_start_date) %>%
  group_by(visit_occurrence_id) %>%
  summarise(prior_max_hba1c = max(value_as_number, na.rm = TRUE)) %>%
  mutate(prior_max_hba1c = if_else(is.infinite(prior_max_hba1c), NA_real_, prior_max_hba1c))

# DIAGNOSTIC CODES
all_dx_ids_str <- paste(c(t1dm_concept_ids, t2dm_concept_ids, htn_concept_ids), collapse = ",")

cond_df <- tbl(con, sql(paste0("
  SELECT
    person_id,
    condition_concept_id,
    condition_start_date
  FROM omop.condition_occurrence
  WHERE condition_concept_id IN (", all_dx_ids_str, ")
    AND condition_start_date <= '2025-06-30'
"))) %>% collect() %>%
  mutate(condition_start_date = as.Date(condition_start_date))

# Diagnosis date for each patient for each condition
first_t1dm <- cond_df %>%
  filter(condition_concept_id %in% t1dm_concept_ids) %>%
  group_by(person_id) %>%
  summarise(first_date_t1dm = min(condition_start_date))

first_t2dm <- cond_df %>%
  filter(condition_concept_id %in% t2dm_concept_ids) %>%
  group_by(person_id) %>%
  summarise(first_date_t2dm = min(condition_start_date))

first_htn <- cond_df %>%
  filter(condition_concept_id %in% htn_concept_ids) %>%
  group_by(person_id) %>%
  summarise(first_date_htn = min(condition_start_date))

# Joins
features_df <- visit_df %>%
  left_join(sbp_visit,    by = "visit_occurrence_id") %>%
  left_join(sbp_prior,    by = "visit_occurrence_id") %>%
  left_join(hba1c_visit,  by = "visit_occurrence_id") %>%
  left_join(hba1c_prior,  by = "visit_occurrence_id") %>%
  left_join(first_t1dm,   by = "person_id") %>%
  left_join(first_t2dm,   by = "person_id") %>%
  left_join(first_htn,    by = "person_id") %>%
  mutate(
    prior_dx_t1dm = !is.na(first_date_t1dm) & first_date_t1dm < visit_start_date,
    prior_dx_t2dm = !is.na(first_date_t2dm) & first_date_t2dm < visit_start_date,
    prior_hx_htn  = !is.na(first_date_htn)  & first_date_htn  < visit_start_date,
    # prior_hx_dm: dx T1DM, dx T2DM, OR prior HbA1c >= 6.5
    prior_hx_dm   = prior_dx_t1dm | prior_dx_t2dm |
      (!is.na(prior_max_hba1c) & prior_max_hba1c >= 6.5)
  )

# Filtering the necessary columns
features_df <- features_df %>%
  filter(
    visit_start_date >= cohort_start,
    visit_end_date   <= cohort_end
  ) %>%
  select(
    visit_occurrence_id,
    person_id,
    gender_concept_id,
    gender_concept_name,
    visit_start_date,
    visit_end_date,
    visit_concept_id,
    age_years_at_visit,
    visit_min_sbp,
    visit_mean_sbp,
    visit_max_sbp,
    prior_max_sbp,
    visit_max_hba1c,
    prior_max_hba1c,
    prior_dx_t1dm,
    prior_dx_t2dm,
    prior_hx_dm,
    prior_hx_htn
  )

# Write final features to CSV file without an index column
dir.create("output", showWarnings = FALSE)
write.csv(features_df, "output/features.csv", row.names = FALSE)
cat("features.csv written:", nrow(features_df), "rows\n")

# Generate metrics using final features data frame
first_visits <- features_df %>%
  group_by(person_id) %>%
  slice_min(visit_start_date, n = 1, with_ties = FALSE) %>%
  ungroup()

total_patients       <- n_distinct(features_df$person_id)
total_visits         <- n_distinct(features_df$visit_occurrence_id)
total_hx_dm_only     <- first_visits %>% filter( prior_hx_dm & !prior_hx_htn) %>% nrow()
total_hx_htn_only    <- first_visits %>% filter(!prior_hx_dm &  prior_hx_htn) %>% nrow()
total_hx_dm_htn      <- first_visits %>% filter( prior_hx_dm &  prior_hx_htn) %>% nrow()
total_control_cohort <- first_visits %>% filter(!prior_hx_dm & !prior_hx_htn) %>% nrow()

writeLines(
  paste0('{
  "total_patients": ', total_patients, ',
  "total_visits": ', total_visits, ',
  "t1dm_concept_ids": [', paste(t1dm_concept_ids, collapse = ", "), '],
  "t2dm_concept_ids": [', paste(t2dm_concept_ids, collapse = ", "), '],
  "hba1c_concept_ids": [', paste(hba1c_concept_ids, collapse = ", "), '],
  "total_hx_dm_only": ', total_hx_dm_only, ',
  "htn_concept_ids": [', paste(htn_concept_ids, collapse = ", "), '],
  "sbp_concept_ids": [', paste(sbp_concept_ids, collapse = ", "), '],
  "total_hx_htn_only": ', total_hx_htn_only, ',
  "total_hx_dm_htn": ', total_hx_dm_htn, ',
  "total_control_cohort": ', total_control_cohort, '
}'),
  "output/metrics.json"
)
cat("metrics.json written\n")
# Write metrics to JSON file