R/write_parquet.R

In GGIR: Raw Accelerometer Data Analysis

write_dashboard_parquet = function(metadatadir = c(),
                                   params_output = c(),
                                   params_general = c(),
                                   params_phyact = c(),
                                   verbose = TRUE) {
  # Produce a single consolidated Parquet file from the CSV reports

  # generated by GGIR, ready for consumption by a DuckDB-WASM dashboard.
  #
  # This function is called at the end of GGIR() when

  # params_output$save_dashboard_parquet is TRUE.
  # It reads the already-generated CSVs, joins them, cleans column names,
  # casts types, embeds the variable dictionary as Parquet key-value metadata,
  # and writes results/parquet/ggir_results.parquet.

  results_dir = paste0(metadatadir, "/results")
  if (!dir.exists(results_dir)) {
    warning("\nNo results directory found. Skipping Parquet export.", call. = FALSE)
    return(invisible(NULL))
  }
  parquet_dir = file.path(results_dir, "parquet")
  if (!dir.exists(parquet_dir)) {
    dir.create(parquet_dir, recursive = TRUE, showWarnings = FALSE)
  }

  # ---------------------------------------------------------------
  # Helper: clean column names to be SQL-friendly
  # ---------------------------------------------------------------
  clean_colnames = function(x) {
    x = tolower(x)
    x = gsub("[^a-z0-9_]", "_", x)    # replace special chars with _
    x = gsub("_+", "_", x)             # collapse multiple underscores
    x = gsub("^_|_$", "", x)           # trim leading/trailing _
    x
  }

  # ---------------------------------------------------------------
  # Helper: safely read a CSV, return NULL if not found
  # ---------------------------------------------------------------
  safe_read_csv = function(filepath) {
    if (length(filepath) == 0 || !file.exists(filepath[1])) return(NULL)
    tryCatch(
      data.table::fread(filepath[1], data.table = FALSE),
      error = function(e) {
        if (verbose) warning(paste0("\nCould not read: ", filepath[1], " - ", e$message), call. = FALSE)
        NULL
      }
    )
  }

  # ---------------------------------------------------------------
  # 1. Read Part 5 day summary (the most granular level)
  #    Glob for all threshold combos (e.g. part5_daysummary_full_MM_*.csv)
  # ---------------------------------------------------------------
  p5_files = list.files(paste0(results_dir, "/QC"),
                        pattern = "^part5_daysummary_full_.*\\.csv$",
                        full.names = TRUE)
  if (length(p5_files) == 0) {
    warning("\nNo Part 5 day summary CSVs found. Skipping Parquet export.", call. = FALSE)
    return(invisible(NULL))
  }

  # Read and row-bind all Part 5 day summary files
  p5_list = lapply(p5_files, function(f) {
    df = safe_read_csv(f)
    if (!is.null(df) && nrow(df) > 0) df else NULL
  })
  p5_list = p5_list[!vapply(p5_list, is.null, logical(1))]
  if (length(p5_list) == 0) {
    warning("\nPart 5 day summary CSVs are empty. Skipping Parquet export.", call. = FALSE)
    return(invisible(NULL))
  }
  p5 = do.call(rbind, p5_list)

  # ---------------------------------------------------------------
  # 2. Read Part 4 night summary (per-night sleep data)
  # ---------------------------------------------------------------
  p4_file = paste0(results_dir, "/part4_nightsummary_sleep_cleaned.csv")
  p4 = safe_read_csv(p4_file)

  # ---------------------------------------------------------------
  # 3. Read Part 2 day summary (daily activity L5/M5/MVPA)
  # ---------------------------------------------------------------
  p2day_file = paste0(results_dir, "/part2_daysummary.csv")
  p2day = safe_read_csv(p2day_file)

  # ---------------------------------------------------------------
  # 4. Read Part 2 person summary (recording metadata)
  # ---------------------------------------------------------------
  p2_file = paste0(results_dir, "/part2_summary.csv")
  p2 = safe_read_csv(p2_file)

  # ---------------------------------------------------------------
  # 5. Read data quality report
  # ---------------------------------------------------------------
  qc_file = paste0(results_dir, "/QC/data_quality_report.csv")
  qc = safe_read_csv(qc_file)

  # ---------------------------------------------------------------
  # 6. Read variable dictionary for Parquet metadata
  # ---------------------------------------------------------------
  dict_file = paste0(results_dir, "/variableDictionary/part5_dictionary_daysummary_full.csv")
  dict = safe_read_csv(dict_file)

  # ---------------------------------------------------------------
  # 7. Build the consolidated data frame
  #    Start with Part 5 (most granular), then left-join others
  # ---------------------------------------------------------------
  consolidated = p5

  # Join Part 4 sleep data by ID + calendar_date
  if (!is.null(p4) && nrow(p4) > 0) {
    # Identify Part 4 columns that are NOT already in Part 5
    p4_unique_cols = setdiff(names(p4), names(consolidated))
    join_keys = c("ID", "calendar_date")
    p4_subset = p4[, c(intersect(join_keys, names(p4)), p4_unique_cols), drop = FALSE]
    if (all(join_keys %in% names(p4_subset)) && all(join_keys %in% names(consolidated))) {
      consolidated = merge(consolidated, p4_subset,
                           by = join_keys, all.x = TRUE, suffixes = c("", "_p4"))
    }
  }

  # Join Part 2 day-level data by ID + calendar_date
  if (!is.null(p2day) && nrow(p2day) > 0) {
    p2day_unique_cols = setdiff(names(p2day), names(consolidated))
    join_keys = c("ID", "calendar_date")
    p2day_subset = p2day[, c(intersect(join_keys, names(p2day)), p2day_unique_cols), drop = FALSE]
    if (all(join_keys %in% names(p2day_subset)) && all(join_keys %in% names(consolidated))) {
      consolidated = merge(consolidated, p2day_subset,
                           by = join_keys, all.x = TRUE, suffixes = c("", "_p2day"))
    }
  }

  # Join Part 2 person-level metadata by ID (one row per person, repeated)
  if (!is.null(p2) && nrow(p2) > 0) {
    p2_unique_cols = setdiff(names(p2), names(consolidated))
    join_key = "ID"
    p2_subset = p2[, c(join_key, p2_unique_cols), drop = FALSE]
    if (join_key %in% names(p2_subset) && join_key %in% names(consolidated)) {
      consolidated = merge(consolidated, p2_subset,
                           by = join_key, all.x = TRUE, suffixes = c("", "_p2"))
    }
  }

  # Join QC data by filename
  if (!is.null(qc) && nrow(qc) > 0) {
    qc_unique_cols = setdiff(names(qc), names(consolidated))
    join_key = "filename"
    qc_subset = qc[, c(join_key, qc_unique_cols), drop = FALSE]
    if (join_key %in% names(qc_subset) && join_key %in% names(consolidated)) {
      consolidated = merge(consolidated, qc_subset,
                           by = join_key, all.x = TRUE, suffixes = c("", "_qc"))
    }
  }

  if (nrow(consolidated) == 0) {
    warning("\nConsolidated data is empty. Skipping Parquet export.", call. = FALSE)
    return(invisible(NULL))
  }

  # ---------------------------------------------------------------
  # 8. Attach nested epoch-level time series as list-column
  #    (must happen before clean_colnames so join keys match)
  # ---------------------------------------------------------------
  epoch_by_day = tryCatch(
    build_epoch_lists_by_day(metadatadir = metadatadir,
                             params_general = params_general,
                             params_phyact = params_phyact,
                             results_dir = results_dir,
                             verbose = verbose),
    error = function(e) NULL
  )
  if (!is.null(epoch_by_day) && nrow(epoch_by_day) > 0) {
    join_keys = intersect(c("ID", "calendar_date"), names(consolidated))
    if (all(c("ID", "calendar_date") %in% join_keys) &&
        all(c("ID", "calendar_date") %in% names(epoch_by_day))) {
      consolidated = merge(consolidated, epoch_by_day,
                           by = c("ID", "calendar_date"),
                           all.x = TRUE, sort = FALSE)
    }
  }

  # Arrow expects every element of a list<struct> column to be a data.frame.
  # After a left merge, unmatched rows can carry NA in `epochs`; replace them
  # with a typed empty data.frame to keep nested serialization valid.
  if ("epochs" %in% names(consolidated)) {
    first_df_idx = which(vapply(consolidated$epochs, is.data.frame, logical(1)))
    if (length(first_df_idx) > 0) {
      empty_epoch_struct = consolidated$epochs[[first_df_idx[1]]][0, , drop = FALSE]
    } else if (!is.null(epoch_by_day) &&
               nrow(epoch_by_day) > 0 &&
               "epochs" %in% names(epoch_by_day) &&
               any(vapply(epoch_by_day$epochs, is.data.frame, logical(1)))) {
      first_epoch_idx = which(vapply(epoch_by_day$epochs, is.data.frame, logical(1)))[1]
      empty_epoch_struct = epoch_by_day$epochs[[first_epoch_idx]][0, , drop = FALSE]
    } else {
      empty_epoch_struct = data.frame(
        timenum = numeric(0),
        acc = numeric(0),
        class_id = integer(0),
        spt = logical(0),
        invalid = logical(0),
        window = integer(0)
      )
    }

    bad_epoch_cells = which(!vapply(consolidated$epochs, is.data.frame, logical(1)))
    if (length(bad_epoch_cells) > 0) {
      consolidated$epochs[bad_epoch_cells] = replicate(
        length(bad_epoch_cells),
        empty_epoch_struct,
        simplify = FALSE
      )
      if (verbose) {
        warning(
          paste0(
            "\nReplaced ",
            length(bad_epoch_cells),
            " unmatched epoch rows with empty nested epoch tables."
          ),
          call. = FALSE
        )
      }
    }
  }

  # ---------------------------------------------------------------
  # 9. Clean column names for SQL-friendly access
  # ---------------------------------------------------------------
  names(consolidated) = clean_colnames(names(consolidated))

  # Deduplicate any identical column names that may have arisen from merges
  dupes = duplicated(names(consolidated))
  if (any(dupes)) {
    names(consolidated)[dupes] = paste0(names(consolidated)[dupes], "_dup",
                                        seq_len(sum(dupes)))
  }

  # ---------------------------------------------------------------
  # 10. Cast types for Parquet
  # ---------------------------------------------------------------

  # calendar_date -> Date
  if ("calendar_date" %in% names(consolidated)) {
    consolidated$calendar_date = tryCatch(
      as.Date(consolidated$calendar_date),
      error = function(e) consolidated$calendar_date
    )
  }

  # Boolean columns
  bool_cols = c("daysleeper", "sleeplog_used", "acc_available",
                "file_corrupt", "file_too_short", "use_temperature")
  for (col in bool_cols) {
    if (col %in% names(consolidated)) {
      consolidated[[col]] = as.logical(as.numeric(consolidated[[col]]))
    }
  }

  # Integer columns (counts, codes, nights, bouts)
  int_patterns = c("^n_", "^nbouts_", "^nblocks_", "cleaning_?code",
                    "night_number", "window_number", "measurement_?day",
                    "page", "n_hours_ignored")
  for (pat in int_patterns) {
    matching = grep(pat, names(consolidated), value = TRUE)
    for (col in matching) {
      if (is.character(consolidated[[col]]) || is.numeric(consolidated[[col]])) {
        consolidated[[col]] = tryCatch(
          as.integer(consolidated[[col]]),
          warning = function(w) consolidated[[col]],
          error = function(e) consolidated[[col]]
        )
      }
    }
  }

  # ---------------------------------------------------------------
  # 11. Build Parquet key-value metadata from dictionary
  # ---------------------------------------------------------------
  kv_metadata = list(
    ggir_export = "nested_dashboard",
    created_at = as.character(Sys.time())
  )
  if (!is.null(dict) && nrow(dict) > 0 &&
      "Variable" %in% names(dict) && "Definition" %in% names(dict)) {
    for (i in seq_len(nrow(dict))) {
      clean_name = clean_colnames(dict$Variable[i])
      kv_metadata[[clean_name]] = dict$Definition[i]
    }
  }

  # ---------------------------------------------------------------
  # 11b. Embed behavioral codes (class_id <-> class_name mapping)
  # ---------------------------------------------------------------
  ms5_root = paste0(metadatadir, "/meta/ms5.outraw")
  if (dir.exists(ms5_root)) {
    bc_files = list.files(ms5_root, pattern = "^behavioralcodes.*\\.csv$",
                          full.names = TRUE)
    if (length(bc_files) > 0) {
      bc = tryCatch(
        data.table::fread(bc_files[length(bc_files)], data.table = FALSE),
        error = function(e) NULL
      )
      if (!is.null(bc) && nrow(bc) > 0 &&
          all(c("class_name", "class_id") %in% names(bc))) {
        bc_json = paste0("{",
          paste(vapply(seq_len(nrow(bc)), function(i) {
            paste0("\"", bc$class_id[i], "\":\"", bc$class_name[i], "\"")
          }, character(1)), collapse = ","),
        "}")
        kv_metadata[["behavioral_codes"]] = bc_json
      }
    }
  }

  # ---------------------------------------------------------------
  # 11c. Embed threshold configuration and acc metric
  # ---------------------------------------------------------------
  threshold_lig = params_phyact[["threshold.lig"]]
  threshold_mod = params_phyact[["threshold.mod"]]
  threshold_vig = params_phyact[["threshold.vig"]]
  if (!is.null(threshold_lig)) {
    kv_metadata[["threshold_lig"]] = as.character(threshold_lig)
  }
  if (!is.null(threshold_mod)) {
    kv_metadata[["threshold_mod"]] = as.character(threshold_mod)
  }
  if (!is.null(threshold_vig)) {
    kv_metadata[["threshold_vig"]] = as.character(threshold_vig)
  }
  if (!is.null(params_phyact[["part6_threshold_combi"]])) {
    kv_metadata[["threshold_combi"]] = params_phyact[["part6_threshold_combi"]]
  }
  if (!is.null(params_general[["acc.metric"]])) {
    kv_metadata[["acc_metric"]] = params_general[["acc.metric"]]
  }

  # Estimate epoch length from the nested epoch time series
  if (!is.null(epoch_by_day) && nrow(epoch_by_day) > 0 &&
      "epochs" %in% names(epoch_by_day)) {
    first_epoch = epoch_by_day$epochs[[1]]
    if (!is.null(first_epoch) && is.data.frame(first_epoch) &&
        nrow(first_epoch) > 1 && "timenum" %in% names(first_epoch)) {
      diffs = diff(first_epoch$timenum[1:min(nrow(first_epoch), 100)])
      diffs = diffs[is.finite(diffs) & diffs > 0]
      if (length(diffs) > 0) {
        kv_metadata[["epoch_length_seconds"]] = as.character(stats::median(diffs))
      }
    }
  }

  # ---------------------------------------------------------------
  # 12. Write Parquet file
  # ---------------------------------------------------------------
  # Name parquet after participant ID when possible.
  # If multiple IDs exist, use a stable fallback filename.
  parquet_basename = "ggir_results"
  if ("id" %in% names(consolidated)) {
    ids = unique(as.character(consolidated$id))
  } else if ("ID" %in% names(consolidated)) {
    ids = unique(as.character(consolidated$ID))
  } else {
    ids = character(0)
  }
  ids = ids[!is.na(ids) & nzchar(ids)]
  if (length(ids) == 1) {
    parquet_basename = ids[1]
  } else if (length(ids) > 1) {
    parquet_basename = "multiple_participants"
    if (verbose) {
      warning("\nMultiple participant IDs found. Using 'multiple_participants.parquet'.", call. = FALSE)
    }
  }
  parquet_basename = gsub("[^A-Za-z0-9._-]", "_", parquet_basename)
  parquet_path = file.path(parquet_dir, paste0(parquet_basename, ".parquet"))

  # Convert to Arrow table so we can attach key-value metadata
  tbl = arrow::arrow_table(consolidated)
  tbl$metadata = kv_metadata

  arrow::write_parquet(tbl, parquet_path)

  if (verbose) {
    cat(paste0("\n  Parquet file written: ", parquet_path))
    cat(paste0("\n  Rows: ", nrow(consolidated),
               ", Columns: ", ncol(consolidated), "\n"))
  }

  invisible(parquet_path)
}

build_epoch_table = function(metadatadir = c(),
                              params_general = c(),
                              params_phyact = c(),
                              results_dir = NULL,
                              verbose = TRUE) {
  # Build a flat epoch-level data.frame from ms5.outraw time series.
  #
  # Returns a list with:
  #   epoch_df   - data.frame with one row per epoch, including ID, filename,
  #                timenum, ACC, SleepPeriodTime, invalidepoch, class_id,
  #                guider, window, calendar_date, and weekday.
  #   config_name - basename of the ms5.outraw configuration used.
  # Returns NULL if no usable data is found.

  if (is.null(results_dir) || !length(results_dir)) {
    results_dir = paste0(metadatadir, "/results")
  }

  ms5_root = paste0(metadatadir, "/meta/ms5.outraw")
  if (!dir.exists(ms5_root)) {
    if (verbose) warning("\nNo ms5.outraw directory found. Cannot build epoch table.", call. = FALSE)
    return(NULL)
  }

  # Determine which ms5.outraw configuration folder to use
  config_dir = character(0)
  if (!is.null(params_phyact[["part6_threshold_combi"]]) &&
      length(params_phyact[["part6_threshold_combi"]]) > 0 &&
      nchar(params_phyact[["part6_threshold_combi"]][1]) > 0) {
    cfg_name = params_phyact[["part6_threshold_combi"]][1]
    cfg_path = paste0(ms5_root, "/", cfg_name)
    if (dir.exists(cfg_path)) {
      config_dir = cfg_path
    }
  }
  if (!length(config_dir)) {
    subdirs = list.dirs(ms5_root, recursive = FALSE, full.names = TRUE)
    if (length(subdirs) > 0) {
      has_ts = vapply(
        subdirs,
        function(d) {
          any(file.exists(list.files(d, pattern = "\\.(csv|RData)$", full.names = TRUE)))
        },
        logical(1)
      )
      subdirs = subdirs[has_ts]
    }
    if (length(subdirs) > 0) {
      config_dir = subdirs[1]
      if (verbose) {
        cat(paste0("\n  Using ms5.outraw configuration: ", basename(config_dir)))
      }
    }
  }
  if (!length(config_dir) || !dir.exists(config_dir)) {
    if (verbose) warning("\nNo ms5.outraw configuration with time series found.", call. = FALSE)
    return(NULL)
  }

  # Prefer CSV over RData when both are available
  csv_files = list.files(config_dir, pattern = "[.]csv$", full.names = TRUE)
  rda_files = list.files(config_dir, pattern = "[.]RData$", full.names = TRUE)
  files = if (length(csv_files) > 0) csv_files else rda_files
  if (length(files) == 0) {
    if (verbose) warning("\nNo ms5.outraw time series files found in configuration folder.", call. = FALSE)
    return(NULL)
  }

  # Map filename -> ID from part2_summary.csv if available
  id_map = NULL
  p2_file = paste0(results_dir, "/part2_summary.csv")
  if (file.exists(p2_file)) {
    p2 = tryCatch(
      data.table::fread(p2_file, data.table = FALSE),
      error = function(e) NULL
    )
    if (!is.null(p2) && all(c("filename", "ID") %in% names(p2))) {
      id_map = unique(p2[, c("filename", "ID")])
    }
  }

  desiredtz_global = ""
  if (!is.null(params_general[["desiredtz"]])) desiredtz_global = params_general[["desiredtz"]]

  epoch_list = vector("list", length(files))

  for (i in seq_along(files)) {
    f = files[i]
    df = NULL
    filename_val = NA_character_
    tz_file = desiredtz_global

    if (grepl("[.]csv$", f, ignore.case = TRUE)) {
      df = tryCatch(
        data.table::fread(f, data.table = FALSE),
        error = function(e) {
          if (verbose) warning(paste0("\nCould not read ms5 CSV: ", f, " - ", e$message), call. = FALSE)
          NULL
        }
      )
    } else {
      env = new.env()
      ok = tryCatch(
        {
          load(f, envir = env)
          TRUE
        },
        error = function(e) {
          if (verbose) warning(paste0("\nCould not load ms5 RData: ", f, " - ", e$message), call. = FALSE)
          FALSE
        }
      )
      if (ok && exists("mdat", envir = env, inherits = FALSE)) {
        df = get("mdat", envir = env)
      }
      if (exists("desiredtz_part1", envir = env, inherits = FALSE)) {
        tz_candidate = get("desiredtz_part1", envir = env)
        if (!is.null(tz_candidate) && nchar(tz_candidate) > 0) {
          tz_file = tz_candidate
        }
      }
      if (exists("filename", envir = env, inherits = FALSE)) {
        filename_val = as.character(get("filename", envir = env))
      }
    }

    if (is.null(df) || nrow(df) == 0) next
    if (!("timenum" %in% names(df))) next

    # Derive filename if not present on data
    if ("filename" %in% names(df)) {
      filename_val = as.character(df$filename[1])
    } else {
      if (is.na(filename_val)) {
        filename_val = basename(f)
      }
      df$filename = filename_val
    }

    # Attach ID via filename if mapping is available
    if (!("ID" %in% names(df)) && !is.null(id_map) && !is.na(filename_val)) {
      id_match = id_map[id_map$filename == filename_val, , drop = FALSE]
      if (nrow(id_match) > 0) {
        df$ID = id_match$ID[1]
      }
    }

    # Derive calendar_date and weekday from timenum and timezone
    if (is.null(tz_file) || nchar(tz_file) == 0) tz_file = "UTC"
    ts_posix = tryCatch(
      as.POSIXct(df$timenum, origin = "1970-01-01", tz = tz_file),
      error = function(e) NULL
    )
    if (!is.null(ts_posix)) {
      df$calendar_date = as.Date(ts_posix)
      df$weekday = weekdays(ts_posix)
    }

    epoch_list[[i]] = df
  }

  epoch_list = epoch_list[!vapply(epoch_list, is.null, logical(1))]
  if (length(epoch_list) == 0) {
    if (verbose) warning("\nNo valid epoch-level time series found.", call. = FALSE)
    return(NULL)
  }

  # Enforce consistent column set across recordings (fill missing with NA)
  all_cols = unique(unlist(lapply(epoch_list, names)))
  epoch_list = lapply(epoch_list, function(df) {
    missing = setdiff(all_cols, names(df))
    for (col in missing) df[[col]] = NA
    df[, all_cols, drop = FALSE]
  })

  epoch_df = do.call(rbind, epoch_list)

  list(epoch_df = epoch_df, config_name = basename(config_dir))
}

write_epoch_parquet = function(metadatadir = c(),
                               params_output = c(),
                               params_general = c(),
                               params_phyact = c(),
                               verbose = TRUE) {
  # Produce an epoch-level Parquet file from the ms5.outraw time series.
  # The resulting table is flat with one row per epoch.

  results_dir = paste0(metadatadir, "/results")
  if (!dir.exists(results_dir)) {
    warning("\nNo results directory found. Skipping epoch Parquet export.", call. = FALSE)
    return(invisible(NULL))
  }
  parquet_dir = file.path(results_dir, "parquet")
  if (!dir.exists(parquet_dir)) {
    dir.create(parquet_dir, recursive = TRUE, showWarnings = FALSE)
  }

  result = build_epoch_table(metadatadir = metadatadir,
                              params_general = params_general,
                              params_phyact = params_phyact,
                              results_dir = results_dir,
                              verbose = verbose)
  if (is.null(result)) {
    return(invisible(NULL))
  }

  epoch_df = result$epoch_df
  config_name = result$config_name

  # ---------------------------------------------------------------
  # Type coercions
  # ---------------------------------------------------------------
  if ("calendar_date" %in% names(epoch_df)) {
    epoch_df$calendar_date = as.Date(epoch_df$calendar_date)
  }

  bool_cols = c("SleepPeriodTime", "invalidepoch")
  for (col in bool_cols) {
    if (col %in% names(epoch_df)) {
      epoch_df[[col]] = as.logical(epoch_df[[col]])
    }
  }

  int_cols = c("class_id", "window", "guider")
  for (col in int_cols) {
    if (col %in% names(epoch_df)) {
      epoch_df[[col]] = tryCatch(
        as.integer(epoch_df[[col]]),
        warning = function(w) epoch_df[[col]],
        error = function(e) epoch_df[[col]]
      )
    }
  }

  # ---------------------------------------------------------------
  # Build Arrow table and attach Parquet key-value metadata
  # ---------------------------------------------------------------
  parquet_path = file.path(parquet_dir, "ggir_epochs.parquet")

  tbl = arrow::arrow_table(epoch_df)

  kv_metadata = list(
    ggir_export = "epoch_timeseries",
    created_at = as.character(Sys.time()),
    ms5_configuration = config_name
  )

  if (!is.null(params_general[["acc.metric"]])) {
    kv_metadata[["acc_metric"]] = params_general[["acc.metric"]]
  }

  # Estimate epoch length (in seconds) from timenum differences
  if ("timenum" %in% names(epoch_df)) {
    ord = order(epoch_df$timenum)
    max_n = min(length(ord), 1001L)
    if (max_n > 1L) {
      diffs = diff(epoch_df$timenum[ord][seq_len(max_n)])
      diffs = diffs[is.finite(diffs) & diffs > 0]
      if (length(diffs) > 0) {
        epoch_len = stats::median(diffs)
        kv_metadata[["epoch_length_seconds"]] = as.character(epoch_len)
      }
    }
  }

  tbl$metadata = kv_metadata

  arrow::write_parquet(tbl, parquet_path)

  if (verbose) {
    cat(paste0("\n  Epoch Parquet file written: ", parquet_path))
    cat(paste0("\n  Rows: ", nrow(epoch_df),
               ", Columns: ", ncol(epoch_df), "\n"))
  }

  invisible(parquet_path)
}

build_epoch_lists_by_day = function(metadatadir = c(),
                                    params_general = c(),
                                    params_phyact = c(),
                                    results_dir = c(),
                                    verbose = TRUE) {
  # Helper to construct a per-day list-column of epoch-level time series
  # from ms5.outraw time series. Returns a data.frame with columns:
  #   ID, calendar_date, epochs (list of data.frames).

  if (!length(results_dir)) {
    results_dir = paste0(metadatadir, "/results")
  }
  ms5_root = paste0(metadatadir, "/meta/ms5.outraw")
  if (!dir.exists(ms5_root)) {
    if (verbose == TRUE) {
      warning("\nNo ms5.outraw directory found. Cannot build epoch lists.", call. = FALSE)
    }
    return(NULL)
  }

  # Determine which ms5.outraw configuration folder to use
  config_dir = character(0)
  if (!is.null(params_phyact[["part6_threshold_combi"]]) &&
      length(params_phyact[["part6_threshold_combi"]]) > 0 &&
      nchar(params_phyact[["part6_threshold_combi"]][1]) > 0) {
    cfg_name = params_phyact[["part6_threshold_combi"]][1]
    cfg_path = paste0(ms5_root, "/", cfg_name)
    if (dir.exists(cfg_path)) {
      config_dir = cfg_path
    }
  }
  if (!length(config_dir)) {
    subdirs = list.dirs(ms5_root, recursive = FALSE, full.names = TRUE)
    if (length(subdirs) > 0) {
      has_ts = vapply(
        subdirs,
        function(d) {
          any(file.exists(list.files(d, pattern = "\\.(csv|RData)$", full.names = TRUE)))
        },
        logical(1)
      )
      subdirs = subdirs[has_ts]
    }
    if (length(subdirs) > 0) {
      config_dir = subdirs[1]
      if (verbose == TRUE) {
        cat(paste0("\n  Using ms5.outraw configuration for nesting: ", basename(config_dir)))
      }
    }
  }
  if (!length(config_dir) || !dir.exists(config_dir)) {
    if (verbose == TRUE) {
      warning("\nNo ms5.outraw configuration with time series found. Cannot build epoch lists.", call. = FALSE)
    }
    return(NULL)
  }

  # Prefer CSV over RData when both are available
  csv_files = list.files(config_dir, pattern = "[.]csv$", full.names = TRUE)
  rda_files = list.files(config_dir, pattern = "[.]RData$", full.names = TRUE)
  files = if (length(csv_files) > 0) csv_files else rda_files
  if (length(files) == 0) {
    if (verbose == TRUE) {
      warning("\nNo ms5.outraw time series files found in configuration folder. Cannot build epoch lists.", call. = FALSE)
    }
    return(NULL)
  }

  # Map filename -> ID from part2_summary.csv if available
  id_map = NULL
  normalize_filename = function(x) {
    x = tolower(as.character(x))
    x = gsub("\\\\", "/", x)
    basename(x)
  }
  p2_file = paste0(results_dir, "/part2_summary.csv")
  if (file.exists(p2_file)) {
    p2 = tryCatch(
      data.table::fread(p2_file, data.table = FALSE),
      error = function(e) NULL
    )
    if (!is.null(p2) && all(c("filename", "ID") %in% names(p2))) {
      id_map = unique(p2[, c("filename", "ID")])
      id_map$filename_norm = normalize_filename(id_map$filename)
    }
  }

  desiredtz_global = ""
  if (!is.null(params_general[["desiredtz"]])) desiredtz_global = params_general[["desiredtz"]]

  epoch_list = vector("list", length(files))

  for (i in seq_along(files)) {
    f = files[i]
    df = NULL
    filename_val = NA_character_
    tz_file = desiredtz_global

    if (grepl("[.]csv$", f, ignore.case = TRUE)) {
      df = tryCatch(
        data.table::fread(f, data.table = FALSE),
        error = function(e) {
          if (verbose == TRUE) warning(paste0("\nCould not read ms5 CSV: ", f, " - ", e$message), call. = FALSE)
          NULL
        }
      )
    } else {
      env = new.env()
      ok = tryCatch(
        {
          load(f, envir = env)
          TRUE
        },
        error = function(e) {
          if (verbose == TRUE) warning(paste0("\nCould not load ms5 RData: ", f, " - ", e$message), call. = FALSE)
          FALSE
        }
      )
      if (ok && exists("mdat", envir = env, inherits = FALSE)) {
        df = get("mdat", envir = env)
      }
      if (exists("desiredtz_part1", envir = env, inherits = FALSE)) {
        tz_candidate = get("desiredtz_part1", envir = env)
        if (!is.null(tz_candidate) && nchar(tz_candidate) > 0) {
          tz_file = tz_candidate
        }
      }
      if (exists("filename", envir = env, inherits = FALSE)) {
        filename_val = as.character(get("filename", envir = env))
      }
    }

    if (is.null(df) || nrow(df) == 0) next
    if (!("timenum" %in% names(df))) next

    # Ensure filename column
    if ("filename" %in% names(df)) {
      filename_val = as.character(df$filename[1])
    } else {
      if (is.na(filename_val)) {
        filename_val = basename(f)
      }
      df$filename = filename_val
    }

    # Attach ID via filename if mapping is available
    if (!("ID" %in% names(df)) && !is.null(id_map) && !is.na(filename_val)) {
      filename_norm = normalize_filename(filename_val)
      id_match = id_map[id_map$filename_norm == filename_norm, , drop = FALSE]
      if (nrow(id_match) > 0) {
        df$ID = id_match$ID[1]
      }
    }

    # Derive calendar_date from timenum and timezone
    if (is.null(tz_file) || nchar(tz_file) == 0) tz_file = "UTC"
    ts_posix = tryCatch(
      as.POSIXct(df$timenum, origin = "1970-01-01", tz = tz_file),
      error = function(e) NULL
    )
    if (!is.null(ts_posix)) {
      df$calendar_date = as.Date(ts_posix)
    }

    epoch_list[[i]] = df
  }

  epoch_list = epoch_list[!vapply(epoch_list, is.null, logical(1))]
  if (length(epoch_list) == 0) {
    if (verbose == TRUE) {
      warning("\nNo valid epoch-level time series found for nesting.", call. = FALSE)
    }
    return(NULL)
  }

  epoch_df = do.call(rbind, epoch_list)

  # Ensure join keys and ordering columns exist
  if (!all(c("ID", "calendar_date", "timenum") %in% names(epoch_df))) {
    if (verbose == TRUE) {
      warning("\nEpoch data lacks ID, calendar_date or timenum. Cannot build nested epochs.", call. = FALSE)
    }
    return(NULL)
  }

  # Decide optional fields globally
  has_acc = "ACC" %in% names(epoch_df)
  has_anglez = "anglez" %in% names(epoch_df)
  has_lux = "lux" %in% names(epoch_df)
  has_temperature = "temperature" %in% names(epoch_df)
  has_steps = "steps" %in% names(epoch_df)
  has_sibdetection = "sibdetection" %in% names(epoch_df)
  has_guider = "guider" %in% names(epoch_df)

  keys = unique(epoch_df[, c("ID", "calendar_date")])
  keys = keys[order(keys$ID, keys$calendar_date), , drop = FALSE]
  epochs_col = vector("list", nrow(keys))

  for (i in seq_len(nrow(keys))) {
    sel = which(epoch_df$ID == keys$ID[i] &
                  epoch_df$calendar_date == keys$calendar_date[i])
    sub = epoch_df[sel, , drop = FALSE]
    if (nrow(sub) == 0) {
      epochs_col[[i]] = data.frame()
      next
    }
    sub = sub[order(sub$timenum), , drop = FALSE]

    acc_col = if (has_acc) as.numeric(sub$ACC) else rep(NA_real_, nrow(sub))
    class_col = if ("class_id" %in% names(sub)) as.integer(sub$class_id) else rep(NA_integer_, nrow(sub))
    spt_col = if ("SleepPeriodTime" %in% names(sub)) as.logical(sub$SleepPeriodTime) else rep(NA, nrow(sub))
    invalid_col = if ("invalidepoch" %in% names(sub)) as.logical(sub$invalidepoch) else rep(NA, nrow(sub))
    window_col = if ("window" %in% names(sub)) as.integer(sub$window) else rep(NA_integer_, nrow(sub))

    struct_df = data.frame(
      timenum = as.numeric(sub$timenum),
      acc = acc_col,
      class_id = class_col,
      spt = spt_col,
      invalid = invalid_col,
      window = window_col
    )

    if (has_anglez) {
      anglez_col = if ("anglez" %in% names(sub)) as.numeric(sub$anglez) else rep(NA_real_, nrow(sub))
      struct_df$anglez = anglez_col
    }
    if (has_lux) {
      lux_col = if ("lux" %in% names(sub)) as.numeric(sub$lux) else rep(NA_real_, nrow(sub))
      struct_df$lux = lux_col
    }
    if (has_temperature) {
      temp_col = if ("temperature" %in% names(sub)) as.numeric(sub$temperature) else rep(NA_real_, nrow(sub))
      struct_df$temperature = temp_col
    }
    if (has_steps) {
      steps_col = if ("steps" %in% names(sub)) as.integer(sub$steps) else rep(NA_integer_, nrow(sub))
      struct_df$steps = steps_col
    }
    if (has_sibdetection) {
      sib_col = if ("sibdetection" %in% names(sub)) as.integer(sub$sibdetection) else rep(NA_integer_, nrow(sub))
      struct_df$sibdetection = sib_col
    }
    if (has_guider) {
      guider_col = if ("guider" %in% names(sub)) as.integer(sub$guider) else rep(NA_integer_, nrow(sub))
      struct_df$guider = guider_col
    }

    epochs_col[[i]] = struct_df
  }

  out = keys
  out$epochs = epochs_col
  out
}