Merge pull request #31 from UNCDEPENdLab/codex/add-project-status-indexing-functions

Add project-wide processing status utilities

Author: michaelhallquist

NAMESPACE

@@ -1,6 +1,7 @@
 # Generated by roxygen2: do not edit by hand
 
 S3method(summary,bg_project_cfg)
+S3method(summary,bg_status_df)
 export(add_tracked_job_parent)
 export(butterworth_filter_4d)
 export(cluster_job_submit)
@@ -9,6 +10,8 @@ export(edit_project)
 export(extract_bids_info)
 export(filtfilt_cpp)
 export(get_fmriprep_outputs)
+export(get_project_status)
+export(get_subject_status)
 export(get_tracked_job_status)
 export(image_quantile)
 export(lmfit_residuals_4d)
@@ -76,6 +79,7 @@ importFrom(lubridate,day)
 importFrom(lubridate,dhours)
 importFrom(lubridate,hour)
 importFrom(lubridate,minute)
+importFrom(lubridate,parse_date_time)
 importFrom(lubridate,second)
 importFrom(lubridate,seconds_to_period)
 importFrom(reticulate,py_install)

R/status_functions.R

@@ -0,0 +1,97 @@
+#' Parse completion time from a .complete file
+#'
+#' @param file Path to .complete file.
+#' @return POSIXct time or NA if parsing fails.
+#' @keywords internal
+#' @importFrom lubridate parse_date_time
+parse_complete_time <- function(file) {
+  if (!file.exists(file)) return(as.POSIXct(NA))
+  tm <- tryCatch(readLines(file, n = 1L, warn = FALSE), error = function(e) NULL)
+  if (is.null(tm) || length(tm) == 0L) return(as.POSIXct(NA))
+  parsed <- suppressWarnings(lubridate::parse_date_time(tm[1L], orders = c("mdy@HM", "mdy@HMS", "ymd HMS", "ymd HM", "mdy HM", "mdy HMS")))
+  if (length(parsed) == 0L || is.na(parsed[1L])) return(as.POSIXct(NA))
+  as.POSIXct(parsed)
+}
+
+#' Get processing status for a single subject
+#'
+#' @param scfg Study configuration list.
+#' @param sub_id Subject identifier.
+#' @param ses_id Optional session identifier. When `NULL`, all sessions found in the log directory are returned.
+#' @return A data.frame with columns indicating completion status and times for each enabled step.
+#' @export
+#' @importFrom checkmate assert_class assert_string
+get_subject_status <- function(scfg, sub_id, ses_id = NULL) {
+  checkmate::assert_class(scfg, "bg_project_cfg")
+  checkmate::assert_string(sub_id)
+  checkmate::assert_string(ses_id, null.ok = TRUE)
+
+  steps <- c()
+  if (isTRUE(scfg$bids_conversion$enable)) steps <- c(steps, "bids_conversion")
+  if (isTRUE(scfg$mriqc$enable)) steps <- c(steps, "mriqc")
+  if (isTRUE(scfg$fmriprep$enable)) steps <- c(steps, "fmriprep")
+  if (isTRUE(scfg$aroma$enable)) steps <- c(steps, "aroma")
+  if (isTRUE(scfg$postprocess$enable)) steps <- c(steps, "postprocess")
+
+  pp_streams <- if ("postprocess" %in% steps) get_postprocess_stream_names(scfg) else character(0)
+
+  log_dir <- scfg$metadata$log_directory
+  sub_log_dir <- file.path(log_dir, paste0("sub-", sub_id))
+  comp_files <- list.files(sub_log_dir, pattern = "_complete$", full.names = FALSE)
+  ses_ids <- if (!is.null(ses_id)) ses_id else {
+    sids <- unique(sub("^.*_ses-([^_]+)_complete$", "\\1", comp_files[grepl("_ses-", comp_files)]))
+    sids <- sids[!is.na(sids) & sids != "^.*_ses-([^_]+)_complete$"]
+    if (length(sids) == 0) NA_character_ else sids
+  }
+
+  res <- lapply(ses_ids, function(ss) {
+    row <- list(sub_id = sub_id, ses_id = ifelse(is.na(ss), NA_character_, ss))
+    for (st in steps) {
+      if (st != "postprocess") {
+        chk <- is_step_complete(scfg, sub_id, ses_id = if (st == "bids_conversion" && !is.na(ss)) ss else NULL, step_name = st)
+        row[[paste0(st, "_complete")]] <- chk$complete
+        row[[paste0(st, "_time")]] <- if (chk$complete) parse_complete_time(chk$complete_file) else as.POSIXct(NA)
+      } else {
+        for (stream in pp_streams) {
+          chk <- is_step_complete(scfg, sub_id, ses_id = if (!is.na(ss)) ss else NULL, step_name = "postprocess", pp_stream = stream)
+          row[[paste0(stream, "_complete")]] <- chk$complete
+          row[[paste0(stream, "_time")]] <- if (chk$complete) parse_complete_time(chk$complete_file) else as.POSIXct(NA)
+        }
+      }
+    }
+    row
+  })
+
+  df <- do.call(rbind.data.frame, res)
+  class(df) <- c("bg_status_df", class(df))
+  df
+}
+
+#' Get processing status for all subjects
+#'
+#' @param scfg Study configuration list.
+#' @return Data.frame with one row per subject/session containing completion status columns.
+#' @export
+#' @importFrom checkmate assert_class
+get_project_status <- function(scfg) {
+  checkmate::assert_class(scfg, "bg_project_cfg")
+  log_dir <- scfg$metadata$log_directory
+  sub_dirs <- list.dirs(log_dir, recursive = FALSE, full.names = FALSE)
+  sub_ids <- sub("^sub-", "", sub_dirs[grepl("^sub-", sub_dirs)])
+  res <- lapply(sub_ids, function(id) get_subject_status(scfg, id))
+  df <- do.call(rbind.data.frame, res)
+  class(df) <- c("bg_status_df", class(df))
+  df
+}
+
+#' Summarize project status
+#'
+#' @param object A data.frame produced by `get_project_status()`.
+#' @return Data.frame summarizing number of subjects completed for each step.
+#' @export
+summary.bg_status_df <- function(object, ...) {
+  step_cols <- grep("_complete$", names(object), value = TRUE)
+  counts <- vapply(step_cols, function(x) sum(object[[x]], na.rm = TRUE), numeric(1))
+  data.frame(step = step_cols, n_complete = counts, row.names = NULL, stringsAsFactors = FALSE)
+}
+

man/get_project_status.Rd

@@ -0,0 +1,8 @@
+\name{get_project_status}
+\alias{get_project_status}
+\title{Get processing status for all subjects}
+\usage{get_project_status(scfg)}
+\arguments{\item{scfg}{Study configuration list.}}
+\value{Data.frame with one row per subject/session containing completion status columns.}
+\description{Uses \code{is_step_complete()} to check whether pipeline steps have completed for each subject.}
+\seealso{\code{\link{get_subject_status}}}

man/get_subject_status.Rd

@@ -0,0 +1,10 @@
+\name{get_subject_status}
+\alias{get_subject_status}
+\title{Get processing status for a subject}
+\usage{get_subject_status(scfg, sub_id, ses_id = NULL)}
+\arguments{\item{scfg}{Study configuration list.}
+\item{sub_id}{Subject identifier.}
+\item{ses_id}{Optional session identifier.}}
+\value{Data.frame summarizing completion status and times for the requested subject/session.}
+\description{Checks the log directory for \code{.complete} files and returns completion information for each enabled processing step.}
+\seealso{\code{\link{get_project_status}}}

man/summary.bg_status_df.Rd

@@ -0,0 +1,8 @@
+\name{summary.bg_status_df}
+\alias{summary.bg_status_df}
+\title{Summarize project status}
+\usage{\method{summary}{bg_status_df}(object, ...)}
+\arguments{\item{object}{A data.frame from \code{get_project_status}.}\item{...}{Additional arguments (unused).}}
+\value{Data.frame with counts of subjects that have completed each processing step.}
+\description{Provides a tabular summary of completion counts for each step in the pipeline.}
+\seealso{\code{\link{get_project_status}}}

tests/testthat/test-status_functions.R

@@ -0,0 +1,36 @@
+test_that("get_project_status reports completion", {
+  root <- tempdir()
+  log_dir <- file.path(root, "logs"); dir.create(log_dir)
+  bids_dir <- file.path(root, "bids"); dir.create(bids_dir)
+  fmriprep_dir <- file.path(root, "fmriprep"); dir.create(fmriprep_dir)
+  mriqc_dir <- file.path(root, "mriqc"); dir.create(mriqc_dir)
+
+  sub <- "01"; ses <- "A"
+  dir.create(file.path(log_dir, paste0("sub-", sub)))
+  dir.create(file.path(bids_dir, paste0("sub-", sub), paste0("ses-", ses)), recursive = TRUE)
+  dir.create(file.path(fmriprep_dir, paste0("sub-", sub)), recursive = TRUE)
+  dir.create(file.path(fmriprep_dir, paste0("sub-", sub), paste0("ses-", ses)), recursive = TRUE)
+
+  cat("2024-05-04 10:00:00", file = file.path(log_dir, paste0("sub-", sub), paste0(".bids_conversion_sub-", sub, "_ses-", ses, "_complete")))
+  cat("2024-05-04 11:00:00", file = file.path(log_dir, paste0("sub-", sub), paste0(".fmriprep_sub-", sub, "_complete")))
+  cat("2024-05-04 12:00:00", file = file.path(log_dir, paste0("sub-", sub), paste0(".postprocess_stream1_sub-", sub, "_ses-", ses, "_complete")))
+
+  scfg <- list(
+    metadata = list(log_directory = log_dir, bids_directory = bids_dir, fmriprep_directory = fmriprep_dir, mriqc_directory = mriqc_dir),
+    bids_conversion = list(enable = TRUE),
+    mriqc = list(enable = FALSE),
+    fmriprep = list(enable = TRUE),
+    aroma = list(enable = FALSE),
+    postprocess = list(enable = TRUE, stream1 = list())
+  )
+  class(scfg) <- "bg_project_cfg"
+
+  res <- get_project_status(scfg)
+  expect_equal(nrow(res), 1)
+  expect_true(res$bids_conversion_complete)
+  expect_true(res$fmriprep_complete)
+  expect_true(res$stream1_complete)
+
+  sm <- summary(res)
+  expect_equal(sm$n_complete[sm$step == "bids_conversion_complete"], 1)
+})