_targets.R

# _targets.R file
library(targets)
library(tarchetypes)
future::plan(future.callr::callr)

options(readr.read_lazy = TRUE)

tar_option_set(
  imports = "createslf",
  packages = "createslf",
  garbage_collection = TRUE,
  format = "parquet",
  resources = tar_resources(
    parquet = tar_resources_parquet(compression = "zstd"),
    qs = tar_resources_qs(preset = "high")
  ),
  error = "continue",
  storage = "worker",
  memory = "persistent" # default option
)

years_to_run <- createslf::years_to_run()

list(
  ## Phase I, all years ----
  tar_rds(test_mode, TRUE),
  tar_rds(write_to_disk, TRUE),
  tar_rds(
    file_path_ext_clean,
    make_lowercase_ext(),
    priority = 1.0,
    cue = tar_cue_age(
      name = file_path_ext_clean,
      age = as.difftime(7.0, units = "days")
    )
  ),
  ## Lookup data ##
  tar_target(gpprac_opendata, get_gpprac_opendata()),
  tar_target(la_code_opendata, get_la_code_opendata_lookup()),
  tar_target(gpprac_ref_path, get_gpprac_ref_path(), format = "file"),
  tar_target(locality_path, get_locality_path(), format = "file"),
  tar_target(simd_path, get_simd_path(), format = "file"),
  tar_target(spd_path, get_spd_path(), format = "file"),
  tar_file_read(it_chi_deaths_extract,
    command = get_it_deaths_path(),
    read = read_it_chi_deaths(!!.x)
  ),
  tar_file_read(dd_data, get_dd_path(), read_extract_delayed_discharges(!!.x)),
  tar_file_read(ltc_data, get_it_ltc_path(), read_lookup_ltc(!!.x)),
  tar_target(
    slf_ch_name_lookup_path,
    get_slf_ch_name_lookup_path(),
    format = "file"
  ),
  ## Process Lookups ##
  tar_target(
    sc_demog_data,
    read_lookup_sc_demographics(),
    cue = tar_cue_age(
      name = sc_demog_data,
      age = as.difftime(28.0, units = "days")
    )
  ),
  tar_target(
    sc_demog_lookup,
    process_lookup_sc_demographics(
      sc_demog_data,
      write_to_disk = write_to_disk
    ),
    priority = 0.9
  ),
  tar_target(
    # Tests, LOOKUP series
    tests_sc_demog_lookup,
    process_tests_sc_demographics(sc_demog_lookup)
  ),
  tar_target(
    it_chi_deaths_data,
    process_it_chi_deaths(
      data = it_chi_deaths_extract,
      write_to_disk = write_to_disk
    ),
    priority = 0.9
  ),
  tar_target(
    # Tests, LOOKUP series
    tests_it_chi_deaths,
    # use anonymous function to sequence writing tests to excel
    # to avoid conflicts of writing to the same file at the same time
    (\(it_chi_deaths_data, tests_sc_demog_lookup) {
      process_tests_it_chi_deaths(it_chi_deaths_data)
    })(it_chi_deaths_data, tests_sc_demog_lookup)
  ),
  tar_target(
    source_gp_lookup,
    process_lookup_gpprac(
      open_data = gpprac_opendata,
      gpprac_ref_path = gpprac_ref_path,
      spd_path = spd_path,
      write_to_disk = write_to_disk
    ),
    priority = 0.9
  ),
  tar_target(
    # Tests, LOOKUP series
    tests_source_gp_lookup,
    (\(source_gp_lookup, tests_it_chi_deaths) {
      process_tests_lookup_gpprac(source_gp_lookup)
    })(source_gp_lookup, tests_it_chi_deaths)
  ),
  tar_target(
    source_pc_lookup,
    process_lookup_postcode(
      spd_path = spd_path,
      simd_path = simd_path,
      locality_path = locality_path,
      write_to_disk = write_to_disk
    ),
    priority = 0.9
  ),
  tar_target(
    # Tests, LOOKUP series
    tests_source_pc_lookup,
    (\(source_pc_lookup, tests_source_gp_lookup) {
      process_tests_lookup_pc(source_pc_lookup)
    })(source_pc_lookup, tests_source_gp_lookup)
  ),
  ### Cost Lookups ----
  tar_target(ch_cost_lookup, process_costs_ch_rmd(), priority = 0.8),
  tar_target(dn_cost_lookup, process_costs_dn_rmd(), priority = 0.8),
  tar_target(hc_cost_lookup, process_costs_hc_rmd(), priority = 0.8),
  tar_target(gp_ooh_cost_lookup, process_costs_gp_ooh_rmd()),
  ### Social Care - 'All' data ----
  tar_target(
    all_at_extract,
    read_sc_all_alarms_telecare(),
    cue = tar_cue_age(
      name = all_at_extract,
      age = as.difftime(28.0, units = "days")
    )
  ),
  tar_target(
    all_at,
    process_sc_all_alarms_telecare(
      all_at_extract,
      sc_demog_lookup = sc_demog_lookup %>% slfhelper::get_chi(),
      write_to_disk = write_to_disk
    ),
    priority = 0.5
  ),
  tar_target(
    # Tests, LOOKUP series
    tests_sc_all_at,
    (\(all_at, tests_source_pc_lookup) {
      process_tests_sc_all_at_episodes(all_at)
    })(all_at, tests_source_pc_lookup)
  ),
  tar_target(
    all_home_care_extract,
    read_sc_all_home_care(),
    cue = tar_cue_age(
      name = all_home_care_extract,
      age = as.difftime(28.0, units = "days")
    )
  ),
  tar_target(
    all_home_care,
    process_sc_all_home_care(
      all_home_care_extract,
      sc_demog_lookup = sc_demog_lookup %>% slfhelper::get_chi(),
      write_to_disk = write_to_disk
    ),
    priority = 0.5
  ),
  tar_target(
    # Tests, LOOKUP series
    tests_sc_all_home_care,
    (\(all_home_care, tests_sc_all_at) {
      process_tests_sc_all_hc_episodes(all_home_care)
    })(all_home_care, tests_sc_all_at)
  ),
  tar_target(
    all_care_home_extract,
    read_sc_all_care_home(),
    cue = tar_cue_age(
      name = all_care_home_extract,
      age = as.difftime(28.0, units = "days")
    )
  ),
  tar_target(
    refined_death_data,
    process_refined_death(
      it_chi_deaths = it_chi_deaths_data,
      write_to_disk = write_to_disk
    )
  ),
  tar_target(
    all_care_home,
    process_sc_all_care_home(
      all_care_home_extract,
      sc_demog_lookup = sc_demog_lookup %>% slfhelper::get_chi(),
      refined_death = refined_death_data %>% slfhelper::get_chi(),
      ch_name_lookup_path = slf_ch_name_lookup_path,
      spd_path = spd_path,
      write_to_disk = write_to_disk
    ),
    priority = 0.5
  ),
  tar_target(
    # Tests, LOOKUP series
    tests_all_care_home,
    (\(all_care_home, tests_sc_all_home_care) {
      process_tests_sc_all_ch_episodes(all_care_home)
    })(all_care_home, tests_sc_all_home_care)
  ),
  tar_target(
    all_sds_extract,
    read_sc_all_sds(),
    cue = tar_cue_age(
      name = all_sds_extract,
      age = as.difftime(28.0, units = "days")
    )
  ),
  tar_target(
    all_sds,
    process_sc_all_sds(
      all_sds_extract,
      sc_demog_lookup = sc_demog_lookup %>% slfhelper::get_chi(),
      write_to_disk = write_to_disk
    ),
    priority = 0.5
  ),
  tar_target(
    # Tests, LOOKUP series
    tests_sc_all_sds,
    (\(all_sds, tests_all_care_home) {
      process_tests_sc_all_sds_episodes(all_sds)
    })(all_sds, tests_all_care_home)
  ),

  ## Phase II, year specific ----
  tar_map(
    list(year = years_to_run),
    tar_rds(
      compress_extracts,
      gzip_files(year),
      priority = 1.0,
      cue = tar_cue_age(
        name = compress_extracts,
        age = as.difftime(7.0, units = "days")
      )
    ),
    ### target data extracts ###
    tar_file_read(
      acute_data,
      get_boxi_extract_path(year, type = "acute"),
      read_extract_acute(year, !!.x)
    ),
    tar_file_read(
      ae_data,
      get_boxi_extract_path(year, type = "ae"),
      read_extract_ae(year, !!.x)
    ),
    tar_file_read(
      cmh_data,
      get_boxi_extract_path(year, type = "cmh"),
      read_extract_cmh(year, !!.x)
    ),
    tar_file_read(
      dn_data,
      get_boxi_extract_path(year, type = "dn"),
      read_extract_district_nursing(year, !!.x)
    ),
    tar_file_read(
      homelessness_data,
      get_boxi_extract_path(year, type = "homelessness"),
      read_extract_homelessness(year, !!.x)
    ),
    tar_file_read(
      maternity_data,
      get_boxi_extract_path(year, type = "maternity"),
      read_extract_maternity(year, !!.x)
    ),
    tar_file_read(
      mental_health_data,
      get_boxi_extract_path(year, type = "mh"),
      read_extract_mental_health(year, !!.x)
    ),
    # tar_file_read(
    #   nrs_deaths_data,
    #   get_boxi_extract_path(year, type = "deaths"),
    #   read_extract_nrs_deaths(year, !!.x)
    # ),
    tar_file_read(
      outpatients_data,
      get_boxi_extract_path(year, type = "outpatient"),
      read_extract_outpatients(year, !!.x)
    ),
    tar_file_read(
      prescribing_data,
      get_it_prescribing_path(year),
      read_extract_prescribing(year, !!.x)
    ),
    tar_target(
      diagnosis_data_path,
      get_boxi_extract_path(year = year, type = "gp_ooh-d"),
      format = "file"
    ),
    tar_target(
      outcomes_data_path,
      get_boxi_extract_path(year = year, type = "gp_ooh-o"),
      format = "file"
    ),
    tar_target(
      consultations_data_path,
      get_boxi_extract_path(year = year, type = "gp_ooh-c"),
      format = "file"
    ),
    tar_target(
      acute_cup_path,
      get_boxi_extract_path(year, type = "acute_cup"),
      format = "file"
    ),
    tar_target(
      gp_ooh_cup_path,
      get_boxi_extract_path(year, type = "gp_ooh_cup"),
      format = "file"
    ),
    tar_qs(
      ooh_data,
      read_extract_gp_ooh(
        year,
        diagnosis_data_path,
        outcomes_data_path,
        consultations_data_path
      )
    ),
    ### Target source processed extracts ----
    tar_target(source_acute_extract, process_extract_acute(
      acute_data,
      year,
      acute_cup_path,
      write_to_disk = write_to_disk
    )),
    tar_target(
      # Tests, EXTRACT series
      tests_source_acute_extract,
      process_tests_acute(
        source_acute_extract,
        year
      )
    ),
    tar_target(source_ae_extract, process_extract_ae(
      ae_data,
      year,
      write_to_disk = write_to_disk
    )),
    tar_target(
      # Tests, EXTRACT series
      tests_source_ae_extract,
      (\(source_ae_extract, year, tests_source_acute_extract) {
        process_tests_ae(
          source_ae_extract,
          year
        )
      })(source_ae_extract, year, tests_source_acute_extract)
    ),
    tar_target(source_cmh_extract, process_extract_cmh(
      cmh_data,
      year,
      write_to_disk = write_to_disk
    )),
    tar_target(
      # Tests, EXTRACT series
      tests_source_cmh_extract,
      (\(source_cmh_extract, year, tests_source_ae_extract){
        process_tests_cmh(
          source_cmh_extract,
          year
        )
      })(source_cmh_extract, year, tests_source_ae_extract)
    ),
    tar_target(source_dd_extract, process_extract_delayed_discharges(
      dd_data,
      year,
      write_to_disk = write_to_disk
    )),
    tar_target(
      # Tests, EXTRACT series
      tests_source_dd_extract,
      (\(source_dd_extract, year, tests_source_cmh_extract) {
        process_tests_delayed_discharges(
          source_dd_extract,
          year
        )
      })(source_dd_extract, year, tests_source_cmh_extract)
    ),
    tar_target(source_dn_extract, process_extract_district_nursing(
      dn_data,
      year,
      costs = dn_cost_lookup,
      write_to_disk = write_to_disk
    )),
    tar_target(
      # Tests, EXTRACT series
      tests_source_dn_extract,
      (\(source_dn_extract, year, tests_source_dd_extract){
        process_tests_district_nursing(
          source_dn_extract,
          year
        )
      })(source_dn_extract, year, tests_source_dd_extract)
    ),
    tar_target(
      source_homelessness_extract,
      process_extract_homelessness(
        data = homelessness_data,
        year = year,
        write_to_disk = write_to_disk,
        la_code_lookup = la_code_opendata
      )
    ),
    tar_target(
      # Tests, EXTRACT series
      tests_source_homelessness_extract,
      (\(source_homelessness_extract, year, tests_source_dn_extract){
        process_tests_homelessness(
          source_homelessness_extract,
          year
        )
      })(source_homelessness_extract, year, tests_source_dn_extract)
    ),
    tar_target(source_ltc_lookup, process_lookup_ltc(
      ltc_data,
      year,
      write_to_disk = write_to_disk
    )),
    tar_target(
      # Tests, EXTRACT series
      tests_ltc,
      (\(source_ltc_lookup, year, tests_source_homelessness_extract){
        process_tests_ltcs(
          source_ltc_lookup,
          year
        )
      })(source_ltc_lookup, year, tests_source_homelessness_extract)
    ),
    tar_target(source_maternity_extract, process_extract_maternity(
      maternity_data,
      year,
      write_to_disk = write_to_disk
    )),
    tar_target(
      # Tests, EXTRACT series
      tests_source_maternity_extract,
      (\(source_maternity_extract, year, tests_ltc){
        process_tests_maternity(
          source_maternity_extract,
          year
        )
      })(source_maternity_extract, year, tests_ltc)
    ),
    tar_target(
      source_mental_health_extract,
      process_extract_mental_health(
        mental_health_data,
        year,
        write_to_disk = write_to_disk
      )
    ),
    tar_target(
      # Tests, EXTRACT series
      tests_source_mental_health_extract,
      (\(source_mental_health_extract, year, tests_source_maternity_extract){
        process_tests_mental_health(
          source_mental_health_extract,
          year
        )
      })(source_mental_health_extract, year, tests_source_maternity_extract)
    ),
    # tar_target(source_nrs_deaths_extract, process_extract_nrs_deaths(
    #   nrs_deaths_data,
    #   year,
    #   write_to_disk = write_to_disk
    # )),
    tar_target(
      source_nrs_deaths_extract,
      # use this anonymous function with redundant but necessary refined_death
      # to make sure reading year-specific nrs deaths extracts after it is produced
      (\(year, refined_death_data) {
        read_file(get_source_extract_path(year, "deaths")) %>%
          as.data.frame()
      })(year, refined_death_data)
    ),
    tar_target(
      # Tests, EXTRACT series
      tests_source_nrs_deaths_extract,
      (\(source_nrs_deaths_extract, year, tests_source_mental_health_extract){
        process_tests_nrs_deaths(
          source_nrs_deaths_extract,
          year
        )
      })(source_nrs_deaths_extract, year, tests_source_mental_health_extract)
    ),
    tar_target(source_ooh_extract, process_extract_gp_ooh(
      year,
      ooh_data,
      gp_ooh_cup_path,
      write_to_disk = write_to_disk
    )),
    tar_target(
      # Tests, EXTRACT series
      tests_source_ooh_extract,
      (\(source_ooh_extract, year, tests_source_nrs_deaths_extract){
        process_tests_gp_ooh(
          source_ooh_extract,
          year
        )
      })(source_ooh_extract, year, tests_source_nrs_deaths_extract)
    ),
    tar_target(source_outpatients_extract, process_extract_outpatients(
      outpatients_data,
      year,
      write_to_disk = write_to_disk
    )),
    tar_target(
      # Tests, EXTRACT series
      tests_source_outpatients_extract,
      (\(source_outpatients_extract, year, tests_source_ooh_extract){
        process_tests_outpatients(
          source_outpatients_extract,
          year
        )
      })(source_outpatients_extract, year, tests_source_ooh_extract)
    ),
    tar_target(source_prescribing_extract, process_extract_prescribing(
      prescribing_data,
      year,
      write_to_disk = write_to_disk
    )),
    tar_target(
      # Tests, EXTRACT series
      tests_prescribing,
      (\(source_prescribing_extract, year, tests_source_outpatients_extract){
        process_tests_prescribing(
          source_prescribing_extract,
          year
        )
      })(source_prescribing_extract, year, tests_source_outpatients_extract)
    ),
    ### Target process year specific social care ----
    tar_target(
      sc_client_data,
      read_lookup_sc_client(fyyear = year)
    ),
    tar_target(
      sc_client_lookup,
      process_lookup_sc_client(
        data = sc_client_data,
        year = year,
        sc_demographics = sc_demog_lookup %>%
          slfhelper::get_chi() %>%
          dplyr::select(c("sending_location", "social_care_id", "chi", "latest_flag")),
        write_to_disk = write_to_disk
      )
    ),
    tar_target(
      # Tests, EXTRACT series
      tests_sc_client_lookup,
      (\(sc_client_lookup, year, tests_prescribing){
        process_tests_sc_client_lookup(sc_client_lookup, year = year)
      })(sc_client_lookup, year, tests_prescribing)
    ),
    tar_target(
      source_sc_alarms_tele,
      process_extract_alarms_telecare(
        data = all_at %>% slfhelper::get_chi(),
        year = year,
        write_to_disk = write_to_disk
      )
    ),
    tar_target(
      # Tests, EXTRACT series
      tests_alarms_telecare,
      (\(source_sc_alarms_tele, year, tests_sc_client_lookup){
        process_tests_alarms_telecare(
          data = source_sc_alarms_tele,
          year = year
        )
      })(source_sc_alarms_tele, year, tests_sc_client_lookup)
    ),
    tar_target(
      source_sc_care_home,
      process_extract_care_home(
        data = all_care_home %>% slfhelper::get_chi(),
        year = year,
        ch_costs = ch_cost_lookup,
        write_to_disk = write_to_disk
      )
    ),
    tar_target(
      # Tests, EXTRACT series
      tests_care_home,
      (\(source_sc_care_home, year, tests_alarms_telecare){
        process_tests_care_home(
          data = source_sc_care_home,
          year = year
        )
      })(source_sc_care_home, year, tests_alarms_telecare)
    ),
    tar_target(
      source_sc_home_care,
      process_extract_home_care(
        data = all_home_care %>% slfhelper::get_chi(),
        year = year,
        write_to_disk = write_to_disk
      )
    ),
    tar_target(
      # Tests, EXTRACT series
      tests_home_care,
      (\(source_sc_home_care, year, tests_care_home){
        process_tests_home_care(
          data = source_sc_home_care,
          year = year
        )
      })(source_sc_home_care, year, tests_care_home)
    ),
    tar_target(
      source_sc_sds,
      process_extract_sds(
        data = all_sds %>% slfhelper::get_chi(),
        year = year,
        write_to_disk = write_to_disk
      )
    ),
    tar_target(
      # Tests, EXTRACT series
      tests_sds,
      (\(source_sc_sds, year, tests_home_care){
        process_tests_sds(
          data = source_sc_sds,
          year = year
        )
      })(source_sc_sds, year, tests_home_care)
    ),
    tar_target(
      slf_deaths_lookup,
      process_slf_deaths_lookup(
        year = year,
        refined_death = refined_death_data,
        write_to_disk = write_to_disk
      )
    ),
    tar_file_read(nsu_cohort, get_nsu_path(year), read_file(!!.x)),
    tar_target(
      homelessness_lookup,
      create_homelessness_lookup(
        year,
        homelessness_data = source_homelessness_extract %>% slfhelper::get_chi()
      )
    )
  ) # ,
  # tar_target(
  #   combined_deaths_lookup,
  #   process_combined_deaths_lookup()
  # )
)
## End of Targets pipeline ##

## Phase III, create ep file and ind file----
## Redundant code which may still be useful for including ep/indiv files.
# tar_qs(
#   processed_data_list,
#   list(
#     source_acute_extract,
#     source_ae_extract,
#     source_cmh_extract,
#     source_dn_extract,
#     source_homelessness_extract,
#     source_maternity_extract,
#     source_mental_health_extract,
#     source_nrs_deaths_extract,
#     source_ooh_extract,
#     source_outpatients_extract,
#     source_prescribing_extract,
#     source_sc_care_home,
#     source_sc_home_care,
#     source_sc_sds,
#     source_sc_alarms_tele
#   )
# ),
# tar_target(
#   episode_file,
#   create_episode_file(
#     processed_data_list,
#     year,
#     homelessness_lookup = homelessness_lookup,
#     dd_data = source_dd_extract %>% slfhelper::get_chi(),
#     nsu_cohort = nsu_cohort %>% slfhelper::get_chi(),
#     ltc_data = source_ltc_lookup %>% slfhelper::get_chi(),
#     slf_pc_lookup = source_pc_lookup,
#     slf_gpprac_lookup = source_gp_lookup,
#     slf_deaths_lookup = slf_deaths_lookup %>% slfhelper::get_chi(),
#     sc_client = sc_client_lookup %>% slfhelper::get_chi(),
#     write_to_disk
#   )
# ),
# tar_target(
#   episode_file_tests,
#   process_tests_episode_file(
#     data = episode_file,
#     year = year
#   )
# ) # ,
# tar_target(
#   cross_year_tests,
#   process_tests_cross_year(year = year)
# ), # ,
# tar_target(
#   individual_file,
#   create_individual_file(
#     episode_file = episode_file,
#     year = year,
#     homelessness_lookup = homelessness_lookup,
#     write_to_disk = write_to_disk
#   )
# ),
# tar_target(
#   individual_file_tests,
#   process_tests_individual_file(
#     data = individual_file,
#     year = year
#   )
# ) # ,
# tar_target(
#   episode_file_dataset,
#   arrow::write_dataset(
#     dataset = episode_file,
#     path = fs::path(
#       get_year_dir(year),
#       stringr::str_glue("source-episode-file-{year}")
#     ),
#     format = "parquet",
#     # Should correspond to the available slfhelper filters
#     partitioning = c("recid", "hscp2018"),
#     compression = "zstd",
#     version = "latest"
#   )
# ),
# tar_target(
#   individual_file_dataset,
#   arrow::write_dataset(
#     dataset = individual_file,
#     path = fs::path(
#       get_year_dir(year),
#       stringr::str_glue("source-individual-file-{year}")
#     ),
#     format = "parquet",
#     # Should correspond to the available slfhelper filters
#     partitioning = c("hscp2018"),
#     compression = "zstd",
#     version = "latest"
#   )
# )