R/pipeline-glassbox.R

In eyeris: Flexible, Extensible, & Reproducible Pupillometry Preprocessing

#' The opinionated "glass box" `eyeris` pipeline
#'
#' This `glassbox` function (in contrast to a "black box" function where you run
#' it and get a result but have no (or little) idea as to how you got from input
#' to output) has a few primary benefits over calling each exported function
#' from `eyeris` separately.
#'
#' First, this `glassbox` function provides a highly opinionated prescription of
#' steps and starting parameters we believe any pupillometry researcher should
#' use as their defaults when preprocessing pupillometry data.
#'
#' Second, and not mutually exclusive from the first point, using this function
#' should ideally reduce the probability of accidental mishaps when
#' "reimplementing" the steps from the preprocessing pipeline both within and
#' across projects. We hope to streamline the process in such a way that you
#' could collect a pupillometry dataset and within a few minutes assess the
#' quality of those data while simultaneously running a full preprocessing
#' pipeline in 1-ish line of code!
#'
#' Third, `glassbox` provides an "interactive" framework where you can evaluate
#' the consequences of the parameters within each step on your data in real
#' time, facilitating a fairly easy-to-use workflow for parameter optimization
#' on your particular dataset. This process essentially takes each of the
#' opinionated steps and provides a pre-/post-plot of the timeseries data for
#' each step so you can adjust parameters and re-run the pipeline until you are
#' satisfied with the choices of your paramters and their consequences on your
#' pupil timeseries data.
#'
#' @param file An SR Research EyeLink `.asc` file generated by the official
#' EyeLink `edf2asc` command
#' @param interactive_preview A flag to indicate whether to run the `glassbox`
#' pipeline autonomously all the way through (set to `FALSE` by default), or to
#' interactively provide a visualization after each pipeline step, where you
#' must also indicate "(y)es" or "(n)o" to either proceed or cancel the
#' current `glassbox` pipeline operation (set to `TRUE`)
#' @param preview_n Number of random example "epochs" to generate for
#' previewing the effect of each preprocessing step on the pupil timeseries
#' @param preview_duration Time in seconds of each randomly selected preview
#' @param preview_window The start and stop raw timestamps used to subset the
#' preprocessed data from each step of the `eyeris` workflow for visualization.
#' Defaults to NULL, meaning random epochs as defined by `preview_n` and
#' `preview_duration` will be plotted. To override the random epochs, set
#' `preview_window` here to a vector with relative start and stop times (in
#' seconds), for example -- `c(5,6)` -- to indicate the raw data from 5-6 secs
#' on data that were recorded at 1000 Hz). Note, the start/stop time values
#' indicated here are in seconds because `eyeris` automatically computes the
#' indices for the supplied range of seconds using the `$info$sample.rate`
#' metadata in the `eyeris` S3 class object
#' @param verbose A logical flag to indicate whether to print status messages to
#' the console. Defaults to `TRUE`. Set to `FALSE` to suppress messages about
#' the current processing step and run silently
#' @param ... Additional arguments to override the default, prescribed settings
#' @param confirm **(Deprecated)** Use `interactive_preview` instead
#' @param num_previews **(Deprecated)** Use `preview_n` instead
#' @param detrend_data **(Deprecated)** A flag to indicate whether to run the
#' `detrend` step (set to `FALSE` by default). Detrending your pupil timeseries
#' can have unintended consequences; we thus recommend that users understand the
#' implications of detrending -- in addition to whether detrending is
#' appropriate for the research design and question(s) -- before using this
#' function
#' @param skip_detransient **(Deprecated)** A flag to indicate whether to skip
#' the `detransient` step (set to `FALSE` by default). In most cases, this
#' should remain `FALSE`. For a more detailed description about likely edge
#' cases that would prompt you to set this to `TRUE`, see the docs for
#' [eyeris::detransient()]
#'
#' @return Preprocessed pupil data contained within an object of class `eyeris`
#'
#' @seealso [lifecycle::deprecate_warn()]
#'
#' @examples
#' demo_data <- eyelink_asc_demo_dataset()
#'
#' # (1) examples using the default prescribed parameters and pipeline recipe
#'
#' ## (a) run an automated pipeline with no real-time inspection of parameters
#' output <- eyeris::glassbox(demo_data)
#'
#' start_time <- min(output$timeseries$block_1$time_secs)
#' end_time <- max(output$timeseries$block_1$time_secs)
#'
#' # by default, verbose = TRUE. To suppress messages, set verbose = FALSE.
#' plot(
#'   output,
#'   steps = c(1, 5),
#'   preview_window = c(start_time, end_time),
#'   seed = 0
#' )
#'
#' ## (b) run a interactive workflow (with confirmation prompts after each step)
#' \donttest{
#' output <- eyeris::glassbox(demo_data, interactive_preview = TRUE, seed = 0)
#' }
#'
#' # (2) examples of overriding the default parameters
#' output <- eyeris::glassbox(
#'   demo_data,
#'   interactive_preview = FALSE, # TRUE to visualize each step in real-time
#'   deblink = list(extend = 40),
#'   lpfilt = list(plot_freqz = TRUE) # overrides verbose parameter
#' )
#'
#' # to suppress messages, set verbose = FALSE in plot():
#' plot(output, seed = 0, verbose = FALSE)
#'
#' # (3) examples of disabling certain steps
#' output <- eyeris::glassbox(
#'   demo_data,
#'   detransient = FALSE,
#'   detrend = FALSE,
#'   zscore = FALSE
#' )
#'
#' plot(output, seed = 0)
#'
#' @export
glassbox <- function(file,
                     interactive_preview = FALSE,
                     preview_n = 3,
                     preview_duration = 5,
                     preview_window = NULL,
                     verbose = TRUE,
                     ...,
                     confirm = deprecated(),
                     num_previews = deprecated(),
                     detrend_data = deprecated(),
                     skip_detransient = deprecated()) {
  original_call <- match.call()

  # handle deprecated parameters
  if (is_present(confirm)) {
    deprecate_warn(
      "1.1.0",
      "glassbox(confirm)",
      "glassbox(interactive_preview)"
    )
    interactive_preview <- confirm
  }

  if (is_present(num_previews)) {
    deprecate_warn(
      "1.1.0",
      "glassbox(num_previews)",
      "glassbox(preview_n)"
    )
    preview_n <- num_previews
  }

  if (is_present(detrend_data)) {
    deprecate_warn(
      "1.1.0",
      "glassbox(detrend_data)",
      details = paste(
        "The `detrend_data` argument is no longer used",
        "and will be ignored."
      )
    )

    detrend_data <- NULL
  }

  if (is_present(skip_detransient)) {
    deprecate_warn(
      "1.1.0",
      "glassbox(skip_detransient)",
      details = paste(
        "The `skip_detransient` argument is no longer used",
        "and will be ignored."
      )
    )

    skip_detransient <- NULL
  }

  # the default glassbox pipeline parameters
  default_params <- list(
    load_asc = list(block = "auto"),
    deblink = list(extend = 50),
    detransient = list(n = 16, mad_thresh = NULL),
    interpolate = TRUE,
    lpfilt = list(wp = 4, ws = 8, rp = 1, rs = 35, plot_freqz = verbose),
    downsample = FALSE,
    bin = FALSE,
    detrend = FALSE,
    zscore = TRUE,
    seed = 123
  )

  # override defaults
  params <- utils::modifyList(default_params, list(...))

  # handle method parameter for bin operation
  if (
    "method" %in% names(list(...)) &&
      !is.null(params$bin) &&
      is.list(params$bin)
  ) {
    params$bin$method <- list(...)$method
  }

  # guard params that accept lists in the event a boolean is supplied
  if ("load_asc" %in% names(list(...)) && isTRUE(list(...)$load_asc)) {
    cli::cli_alert_warning(
      paste(
        "[ WARN ] - `load_asc` expects a list of args (not a boolean)...",
        "using default: `list(block = \"auto\")`"
      )
    )
    params$load_asc <- default_params$load_asc
  }

  if ("deblink" %in% names(list(...)) && isTRUE(list(...)$deblink)) {
    cli::cli_alert_warning(
      paste(
        "[ WARN ] - `deblink` expects a list of args (not a boolean)...",
        "using default: `list(extend = 50)`"
      )
    )
    params$deblink <- default_params$deblink
  }

  if ("detransient" %in% names(list(...)) && isTRUE(list(...)$detransient)) {
    cli::cli_alert_warning(
      paste(
        "[ WARN ] - `detransient` expects a list of args (not a boolean)...",
        "using default: `list(n = 16, mad_thresh = NULL)`"
      )
    )
    params$detransient <- default_params$detransient
  }

  if ("lpfilt" %in% names(list(...)) && isTRUE(list(...)$lpfilt)) {
    cli::cli_alert_warning(paste(
      "[ WARN ] - `lpfilt` expects a list of args (not a boolean)...",
      "using default:",
      "`list(wp = 4, ws = 8, rp = 1, rs = 35, plot_freqz = verbose)`"
    ))
    params$lpfilt <- default_params$lpfilt
  }

  if ("downsample" %in% names(list(...)) && isTRUE(list(...)$downsample)) {
    cli::cli_alert_warning(paste(
      "[ WARN ] - `downsample` expects a list of args (not a boolean)...",
      "using default: `list(target_fs = 100, plot_freqz = verbose)`"
    ))
    params$downsample <- default_params$downsample
  }

  if ("bin" %in% names(list(...)) && isTRUE(list(...)$bin)) {
    cli::cli_alert_warning(paste(
      "[ WARN ] - `bin` expects a list of args (not a boolean)...",
      "using default: `list(bins_per_second = 10, method = \"mean\")`"
    ))
    params$bin <- default_params$bin
  }

  # abort if both downsample and bin are enabled
  step_status <- evaluate_pipeline_step_params(
    list(downsample = params$downsample, bin = params$bin)
  )

  if (
    !is.null(params$downsample) &&
      !is.null(params$bin) &&
      step_status[1] &&
      step_status[2]
  ) {
    cli::cli_abort(
      c(
        "[ ABORT ] - Both 'downsample' and 'bin' steps are enabled.",
        "x You cannot use both downsampling and binning in the same glassbox.",
        "i Please enable only one (or neither) of these steps."
      )
    )
  }

  # evaluate which steps of pipeline to run
  which_steps <- evaluate_pipeline_step_params(params)

  if (which_steps[["detrend"]] &&
    !any(which_steps[c(
      "deblink", "detransient",
      "interpolate", "lpfilt"
    )])) {
    cli::cli_alert_warning(
      paste(
        "[ WARN ] - Detrend is enabled but no other preprocessing steps are",
        "enabled. This may cause plotting issues since there will be no pupil",
        "columns to detrend against. Consider enabling at least one",
        "preprocessing step before detrending, or disable detrending if you",
        "want to work with raw data."
      )
    )
  }

  # eyeris workflow data structure
  pipeline <- list(
    load_asc = function(data, params, original_call) {
      if (which_steps[["load_asc"]]) {
        call_info <- list(
          call = original_call,
          parameters = list(block = params$load_asc$block)
        )
        result <- eyeris::load_asc(data, block = params$load_asc$block)
        if (!is.list(result$params)) result$params <- list()
        result$params[["load_asc"]] <- call_info
        result
      } else {
        stop("No data loaded... the glassbox pipeline cannot proceed.")
      }
    },
    deblink = function(data, params, original_call) {
      if (which_steps[["deblink"]]) {
        call_info <- list(
          call = original_call,
          parameters = list(extend = params$deblink$extend)
        )
        eyeris::deblink(
          data,
          extend = params$deblink$extend,
          call_info = call_info
        )
      } else {
        data
      }
    },
    detransient = function(data, params, original_call) {
      if (which_steps[["detransient"]]) {
        call_info <- list(
          call = original_call,
          parameters = list(
            n = params$detransient$n,
            mad_thresh = params$detransient$mad_thresh
          )
        )
        eyeris::detransient(
          data,
          n = params$detransient$n,
          mad_thresh = params$detransient$mad_thresh,
          call_info = call_info
        )
      } else {
        data
      }
    },
    interpolate = function(data, params, original_call) {
      if (which_steps[["interpolate"]]) {
        call_info <- list(
          call = original_call,
          parameters = list(verbose = verbose)
        )
        eyeris::interpolate(data, verbose = verbose, call_info = call_info)
      } else {
        data
      }
    },
    lpfilt = function(data, params, original_call) {
      if (which_steps[["lpfilt"]]) {
        call_info <- list(
          call = original_call,
          parameters = list(
            wp = params$lpfilt$wp,
            ws = params$lpfilt$ws,
            rp = params$lpfilt$rp,
            rs = params$lpfilt$rs,
            plot_freqz = params$lpfilt$plot_freqz
          )
        )
        eyeris::lpfilt(
          data,
          wp = params$lpfilt$wp,
          ws = params$lpfilt$ws,
          rp = params$lpfilt$rp,
          rs = params$lpfilt$rs,
          plot_freqz = params$lpfilt$plot_freqz,
          call_info = call_info
        )
      } else {
        data
      }
    },
    downsample = function(data, params, original_call) {
      if (which_steps[["downsample"]]) {
        if (is.null(params$downsample$plot_freqz)) {
          params$downsample$plot_freqz <- verbose
        }
        if (is.null(params$downsample$rp)) params$downsample$rp <- 1
        if (is.null(params$downsample$rs)) params$downsample$rs <- 35
        call_info <- list(
          call = original_call,
          parameters = list(
            target_fs = params$downsample$target_fs,
            plot_freqz = params$downsample$plot_freqz,
            rp = params$downsample$rp,
            rs = params$downsample$rs
          )
        )
        eyeris::downsample(
          data,
          target_fs = params$downsample$target_fs,
          plot_freqz = params$downsample$plot_freqz,
          rp = params$downsample$rp,
          rs = params$downsample$rs,
          call_info = call_info
        )
      } else {
        data
      }
    },
    bin = function(data, params, original_call) {
      if (which_steps[["bin"]]) {
        call_info <- list(
          call = original_call,
          parameters = list(
            bins_per_second = params$bin$bins_per_second,
            method = params$bin$method
          )
        )
        eyeris::bin(
          data,
          bins_per_second = params$bin$bins_per_second,
          method = params$bin$method,
          call_info = call_info
        )
      } else {
        data
      }
    },
    detrend = function(data, params, original_call) {
      if (which_steps[["detrend"]]) {
        call_info <- list(
          call = original_call,
          parameters = list()
        )
        eyeris::detrend(data, call_info = call_info)
      } else {
        data
      }
    },
    zscore = function(data, params, original_call) {
      if (which_steps[["zscore"]]) {
        call_info <- list(
          call = original_call,
          parameters = list()
        )
        eyeris::zscore(data, call_info = call_info)
      } else {
        data
      }
    }
  )

  seed <- params$seed
  step_counter <- 1
  only_linear_trend <- FALSE
  next_step <- c()

  if (which_steps[["load_asc"]]) {
    if (verbose) {
      cli::cli_alert_success("[  OK  ] - Running eyeris::load_asc()")
    }
    file <- pipeline[["load_asc"]](file, params, original_call)

    if (interactive_preview) {
      plot_with_seed(
        file = file,
        step_counter = 1,
        seed = seed,
        preview_n = preview_n,
        preview_duration = preview_duration,
        preview_window = preview_window,
        only_linear_trend = only_linear_trend,
        next_step = NULL,
        verbose = verbose
      )

      if (!prompt_user()) {
        if (verbose) {
          cli::cli_alert_info(
            paste(
              "[ INFO ] - Process cancelled after loading data.",
              "Adjust your parameters and re-run!\n"
            )
          )
        }
        return(file)
      }
    }
  }

  has_multiple_blocks <- is.list(file$timeseries) && length(file$timeseries) > 0

  # process each block individually through all steps (except load_asc)
  if (has_multiple_blocks) {
    block_names <- names(file$timeseries)
    processed_blocks <- list()

    # store orig latest pointer to restore it later
    original_latest <- file$latest
    final_latest <- NULL
    block_states <- list()
    # collect params from all blocks
    all_params <- list()

    for (block_name in block_names) {
      if (verbose) {
        cli::cli_alert_info(paste0("[ INFO ] - Processing block: ", block_name))
      }

      temp_file <- file
      temp_file$timeseries <- list(file$timeseries[[block_name]])
      names(temp_file$timeseries) <- block_name

      # set latest pointer for current block
      if (is.list(original_latest)) {
        # multiblock: use pointer for current block
        temp_file$latest <- list()
        temp_file$latest[[block_name]] <- original_latest[[block_name]]
      } else {
        # single block converted to multiblock: use original pointer
        temp_file$latest <- list()
        temp_file$latest[[block_name]] <- original_latest
      }

      # init block state
      block_states[[block_name]] <- list(
        latest_pointer = temp_file$latest[[block_name]],
        steps_completed = 0,
        has_errors = FALSE
      )

      block_step_counter <- 2

      for (step_name in names(pipeline)[-1]) {
        action <- "Running "
        skip_plot <- FALSE

        if (!which_steps[[step_name]]) {
          action <- "Skipping "
          block_step_counter <- block_step_counter - 1
          skip_plot <- TRUE

          if (!is.null(temp_file$latest[[block_name]])) {
            expected_col <- paste0(
              temp_file$latest[[block_name]], "_", step_name
            )
            block_data <- temp_file$timeseries[[block_name]]
            if (expected_col %in% colnames(block_data)) {
              temp_file$latest[[block_name]] <- expected_col
            }
          }
        } else {
          if (step_name == "detrend") {
            only_linear_trend <- TRUE
          }
        }

        if (verbose) {
          if (action == "Running ") {
            cli::cli_alert_success(
              paste0(
                "[  OK  ] - ", action, "eyeris::",
                step_name, "() for ", block_name
              )
            )
          } else {
            cli::cli_alert_warning(
              paste0(
                "[ SKIP ] - ", action, "eyeris::",
                step_name, "() for ", block_name
              )
            )
          }
        }

        step_to_run <- pipeline[[step_name]]
        err_thrown <- FALSE

        temp_file <- tryCatch(
          {
            step_to_run(temp_file, params, original_call)
          },
          error = function(e) {
            if (!which_steps[["interpolate"]] && which_steps[["detrend"]]) {
              cli::cli_alert_danger(
                paste0(
                  "[ WARN ] - ", "Because missing pupil samples were not ",
                  "interpolated, there is a mismatch in the number of samples ",
                  "in the detrended data. Please set `interpolate` to `TRUE` ",
                  "before detrending data OR disable detrending by setting ",
                  "`detrend` to `FALSE`."
                )
              )
            }

            if (verbose) {
              cli::cli_alert_info(
                paste0(
                  "[ SKIP ] - ", "Skipping eyeris::",
                  step_name, "() for ", block_name, ": ",
                  e$message
                )
              )
            }
            err_thrown <<- TRUE
            block_step_counter <<- block_step_counter - 1

            # mark current block as having errors
            block_states[[block_name]]$has_errors <- TRUE

            # reset latest pointer to prevent corruption from propagating
            # find last valid column name in current block
            block_data <- temp_file$timeseries[[block_name]]
            pupil_cols <- grep("^pupil_", colnames(block_data), value = TRUE)
            if (length(pupil_cols) > 0) {
              # use last valid pupil column
              temp_file$latest[[block_name]] <- pupil_cols[length(pupil_cols)]
              block_states[[block_name]]$latest_pointer <-
                temp_file$latest[[block_name]]
            } else {
              # fallback to original pointer for this block
              if (is.list(original_latest)) {
                temp_file$latest[[block_name]] <- original_latest[[block_name]]
              } else {
                temp_file$latest[[block_name]] <- original_latest
              }
              block_states[[block_name]]$latest_pointer <-
                temp_file$latest[[block_name]]
            }

            temp_file
          }
        )

        if (
          verbose &&
            action == "Running " &&
            (step_name == "downsample" || step_name == "bin")
        ) {
          cli::cli_alert_success(
            paste(
              "[ INFO ] - Decimating sampling rate from",
              temp_file$info$sample.rate, "Hz -->",
              temp_file$decimated.sample.rate, "Hz..."
            )
          )
        }

        if (interactive_preview && !err_thrown && !skip_plot) {
          pupil_steps <- grep("^pupil_",
            colnames(temp_file$timeseries[[block_name]]),
            value = TRUE
          )

          if (block_step_counter + 1 <= length(names(pipeline))) {
            next_step <- c(next_step, pupil_steps[block_step_counter])
          } else {
            next_step <- NULL
          }

          plot_with_seed(
            file = temp_file,
            step_counter = block_step_counter,
            seed = seed,
            preview_n = preview_n,
            preview_duration = preview_duration,
            preview_window = preview_window,
            only_linear_trend = only_linear_trend,
            next_step = next_step,
            block_name = block_name,
            verbose = verbose
          )

          if (step_name == "detrend") {
            only_linear_trend <- FALSE
          }

          if (step_name != "zscore") {
            if (!prompt_user()) {
              if (verbose) {
                cli::cli_alert_info(
                  paste(
                    "Process cancelled after running the",
                    step_name, "step for", block_name, ".",
                    "Adjust your parameters and re-run!\n"
                  )
                )
              }
              break
            }
          }
        }

        block_step_counter <- block_step_counter + 1
      }

      processed_blocks[[block_name]] <- temp_file$timeseries[[block_name]]

      # preserve decimated.sample.rate from processed blocks
      if (!is.null(temp_file$decimated.sample.rate)) {
        file$decimated.sample.rate <- temp_file$decimated.sample.rate
      }

      # track latest pointer from successfully processed blocks
      if (!is.null(temp_file$latest[[block_name]]) &&
            !grepl("_([^_]+)_\\1", temp_file$latest[[block_name]])) {
        final_latest <- temp_file$latest[[block_name]]
      }

      # update block state with final state
      block_states[[block_name]]$latest_pointer <-
        temp_file$latest[[block_name]]
      block_states[[block_name]]$steps_completed <- block_step_counter - 1

      # update main file's latest pointer for current block
      if (is.list(file$latest)) {
        file$latest[[block_name]] <- temp_file$latest[[block_name]]
      } else {
        # Convert to list if it wasn't already
        file$latest <- list()
        file$latest[[block_name]] <- temp_file$latest[[block_name]]
      }

      # collect params from this block
      if (!is.null(temp_file$params) && is.list(temp_file$params)) {
        all_params <- modifyList(all_params, temp_file$params)
      }
    }

    # recombine processed blocks
    file$timeseries <- processed_blocks

    # preserve params from processed blocks
    if (length(all_params) > 0) {
      file$params <- all_params
    }

    if (verbose) {
      cat("\nBlock processing summary:\n")
      for (block_name in names(block_states)) {
        state <- block_states[[block_name]]
        status <- if (state$has_errors) "ERRORS" else "OK"
        cat(sprintf(
          "  %s: %s (steps: %d, latest: %s)\n",
          block_name, status, state$steps_completed, state$latest_pointer
        ))
      }
      cat("\n")
    }
  } else {
    cli::cli_abort("No data blocks found error.")
  }

  # generate confounds after all other steps
  if (verbose) {
    cli::cli_alert_success("[  OK  ] - Running eyeris::summarize_confounds()")
  }

  file <- eyeris::summarize_confounds(file)

  return(file)
}

#' Plot with seed handling for glassbox pipeline
#'
#' Internal function to handle plotting with consistent seed management
#' for the glassbox pipeline interactive previews.
#'
#' @param file The `eyeris` object to plot
#' @param step_counter Current step counter
#' @param seed A random seed for reproducible plotting
#' @param preview_n Number of preview epochs
#' @param preview_duration Duration of each preview in seconds
#' @param preview_window Preview window specification
#' @param only_linear_trend A flag to indicate whether to show only linear
#' trend
#' @param next_step Next step information
#' @param block_name Block name (optional, for multi-block processing)
#' @param verbose A flag to indicate whether to show verbose output
#'
#' @keywords internal
plot_with_seed <- function(file,
                           step_counter,
                           seed,
                           preview_n,
                           preview_duration,
                           preview_window,
                           only_linear_trend,
                           next_step,
                           block_name = NULL,
                           verbose = TRUE) {
  if (is.null(seed)) {
    seed <- rlang::`%||%`(seed, sample.int(.Machine$integer.max, 1))
  }

  withr::with_seed(
    seed,
    {
      if (!is.null(block_name)) {
        bn <- get_block_numbers(block_name)
        plot(
          file,
          steps = step_counter,
          preview_n = preview_n,
          seed = seed,
          preview_duration = preview_duration,
          preview_window = preview_window,
          only_linear_trend = only_linear_trend,
          next_step = next_step,
          block = bn,
          suppress_prompt = FALSE,
          verbose = verbose
        )
      } else {
        plot(
          file,
          steps = step_counter,
          preview_n = preview_n,
          seed = seed,
          preview_duration = preview_duration,
          preview_window = preview_window,
          only_linear_trend = only_linear_trend,
          next_step = next_step,
          suppress_prompt = FALSE,
          verbose = verbose
        )
      }
    }
  )
}

#' Prompt user for continuation
#'
#' Prompts the user to continue or cancel the current operation.
#'
#' @return A logical flag indicating whether the user chose to continue
#'
#' @keywords internal
prompt_user <- function() {
  resp <- readline(prompt = "Continue? [Yes/No]: ")
  tolower(resp) == "yes" | tolower(resp) == "y"
}

#' Evaluate pipeline step parameters
#'
#' Converts pipeline step parameters to logical values for evaluation.
#'
#' @param params A list of pipeline step parameters
#'
#' @return A logical vector indicating which steps should be executed
#'
#' @keywords internal
evaluate_pipeline_step_params <- function(params) {
  sapply(params, function(x) {
    if (is.logical(x)) {
      isTRUE(x)
    } else {
      !identical(x, FALSE)
    }
  })
}