HighPriSensorCodeStatistics.R

###############################################################################
# Title: High-Priority Sensor Code Window Statistics
#
# Purpose:
#   Exploratory sensor-code branch of the pipeline. For each high-priority
#   sanitized sensor code and each source SQLite database, this script:
#     • Extracts Telemetry rows where SensorValue exceeds a threshold
#     • Computes consecutive high-value windows (durations < 10 minutes)
#     • Aggregates per-window statistics (min/max/delta sensor value, total time)
#     • Writes per-sensor summary tables to SQLite
#
#   This advances the data by surfacing which sensors exhibit meaningful
#   excursions, providing an additional signal for sensor/feature selection.
#
# Skills shown:
#   • Parallelized scanning of many SQLite databases
#   • Sliding-window and duration calculations for time-series data
#   • Programmatic table creation and aggregation per sensor code
#
# Inputs:
#   • All telemetry SQLite databases under <DATALAKEDIR> (recursive)
#   • Sanitized list of sensor codes (SensorC–SensorJ defined in script)
#
# Outputs:
#   • Per-DB sensor-window SQLite files:
#       <PID>_SensorWindow.sqlite
#     containing:
#       - SensorWindowStats_<SensorCode> table
#
# Author: Skylar Furey
###############################################################################


# Libraries -------------------------------------------------------------------
packages <- c("RSQLite", "dplyr", "doParallel")
installed <- packages %in% rownames(installed.packages())
if (any(!installed)) install.packages(packages[!installed])
invisible(lapply(packages, library, character.only = TRUE))


# Inputs ----------------------------------------------------------------------
cmdArgs <- commandArgs()
data_root  <- cmdArgs[grep("DATALAKEDIR", cmdArgs) + 1]
output_root <- file.path(system("echo $HOME", intern = TRUE), "Projects/telemetry_event_correlation/HiPriSensors")


db_files <- list.files(
  path       = file.path(data_root, "SQLITE_FILES"),
  pattern    = "\\.db$",
  full.names = TRUE,
  recursive  = TRUE
)

# Sanitized sensor label codes
sensor_codes <- c(
  "SensorC", "SensorD", "SensorE", "SensorF", 
  "SensorG", "SensorH", "SensorI", "SensorJ"
)


# Parallel Setup ---------------------------------------------------------------
cl <- makeCluster(4, type = "FORK")
registerDoParallel(cl)

foreach(db = db_files, .packages = c("RSQLite", "dplyr")) %dopar% {
  
  log_file <- file.path(system("echo $WORKDIR", intern = TRUE), 
                       paste0("Projects/telemetry_event_correlation/", basename(db), "_HiPri_log.txt")),
  
  cat("Processing:", db, file = log_file, append = TRUE, sep = "\n")
  
  con  <- dbConnect(SQLite(), db, flags = SQLITE_RO)
  out_db_file <- paste0(output_root,  Sys.getpid(), "_SensorWindow.db")
  out_con <- dbConnect(SQLite(), out_db_file)
  
  cat("Output DB initialized", file = log_file, append = TRUE, sep = "\n")
  
  # Process each sanitized sensor code ----------------------------------------
  for (sensor_code in sensor_codes) {
    
    cat("Sensor:", sensor_code, file = log_file, append = TRUE, sep = "\n")
    
    # Extract high-value sensor readings
    query <- sprintf(
      'SELECT RecordID, Timestamp, SensorValue
       FROM Telemetry
       WHERE SensorName = "%s"
         AND CAST(SensorValue AS REAL) > 25
       ORDER BY RecordID, Timestamp;',
      sensor_code
    )
    
    sensor_raw <- dbGetQuery(con, query)
    
    if (nrow(sensor_raw) < 2) next
    
    # Create lagged dataset for consecutive timing
    lagged <- rbind(NA, sensor_raw[1:(nrow(sensor_raw) - 1), ])
    colnames(lagged) <- paste0("Lag_", colnames(sensor_raw))
    combined <- cbind(sensor_raw, lagged)
    
    combined$ValidPair <- combined$Lag_RecordID == combined$RecordID
    combined$Duration  <- combined$Timestamp - combined$Lag_Timestamp
    
    # Filter to consecutive readings within 10 minutes
    sensor_windows <- combined[combined$ValidPair & combined$Duration < 600, ]
    sensor_windows <- sensor_windows[complete.cases(sensor_windows), ]
    
    # Create working temp table
    dbExecute(out_con,
        'CREATE TEMP TABLE SensorTemp (
         RecordID INTEGER,
         Timestamp REAL,
         SensorValue REAL, 
         Lag_RecordID INTEGER,
         Lag_Timestamp REAL,
         Lag_SensorValue REAL,
         Duration REAL
       );'
    )
    
    dbWriteTable(out_con, "SensorTemp", sensor_windows, append = TRUE)
    
    # Create final destination table for this sensor
    stats_table <- paste0("SensorWindowStats_", sensor_code)
    dbExecute(
      out_con,
      sprintf(
           'CREATE TABLE IF NOT EXISTS %s (
            RecordID INTEGER,
            MinSensorValue REAL,
            MaxSensorValue REAL,
            DeltaSensorValue REAL,
            TotalDuration REAL,
            SourceDB TEXT
         );',
        stats_table
      )
    )
    
    # Insert aggregate statistics
    insert_query <- sprintf(
      'INSERT INTO %s
       SELECT
         RecordID,
         Timestamp AS StartTimeStamp,
         MIN(SensorValue) AS MinSensorValue,
         MAX(SensorValue) AS MaxSensorValue,
         (Lag_SensorValue - SensorValue) AS DeltaSensorValue,
         SUM(Duration) AS TotalDuration,
         "%s" AS SourceDB
       FROM SensorTemp
       GROUP BY RecordID;',
      stats_table,
      basename(db)
    )
    
    dbExecute(out_con, insert_query)
    
    # Cleanup temp table
    dbRemoveTable(out_con, "SensorTemp")
    
    cat("Completed sensor:", sensor_code,
        file = log_file, append = TRUE, sep = "\n")
  }
  
  dbDisconnect(con)
  dbDisconnect(out_con)
}

stopCluster(cl)