ComputeSensorLStatistics.R

###############################################################################
# Title: Sensor L Left/Right Delta and Anomaly-Filtered Statistics
#
# Purpose:
#   Parallel sensor branch for left/right SensorL data. This script:
#     • Reads consolidated 60-second telemetry stores
#     • Pulls left and right Sensor L label streams within event windows
#     • Applies anomaly filtering
#     • Computes DeltaSensorL = RightSensorL - LeftSensorL
#     • Derives per-record statistics and rate-of-change metrics
#     • Outputs cleaned left/right streams and delta statistics
#
#   This advances the data by constructing an additional sensor family whose
#   cross-channel behavior may correlate with analysis-causing events.
#
# Skills shown:
#   • Dual-sensor delta computation and summary stats
#   • Domain-specific artifact removal (bounce filtering)
#   • Structured output for multi-sensor analysis
#
# Inputs:
#   • Consolidated telemetry stores:
#       $WORKDIR/Projects/telemetry_event_correlation/TelemetryStores/Consolidated_*.db
#       (tables: ConsolidatedRecords)
#   • HeaderData.RData
#        Contains file names for each record
#   • AnomalyLog.RData
#        Contains log of anomaly records
#
# Outputs:
#   • Delta and stats:
#       SensorL/SensorLDelta.db        (table StoresFiltered)
#       SensorL/SensorLDeltaStats.csv
#   • Cleaned left/right streams:
#       SensorL/LeftSensorL.db         (table StoresFiltered)
#       SensorL/RightSensorL.db        (table StoresFiltered)
#
# Author: Skylar Furey
###############################################################################



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


## Inputs ######################################################################
cmdArgs <- commandArgs()

output_root <- file.path(system("echo $HOME", intern = TRUE), "Projects/telemetry_event_correlation/")
data_root <- file.path(system("echo $WORKDIR", intern = TRUE), "Projects/telemetry_event_correlation/TelemetryStores/")

# Major events causing analysis
EventFilterTable <- data.frame(
  RecordID     = c(1001,1002,1003,1004),
  FilterTimestamp = c(1620030000,1621040000,1622050000,1623060000)
)

# Load lookup/reference data
# Must include: (RecordID, FileName)
load(file.path(data_root, "CACHED_USER_DATA/Lookup/HeaderData.RData"))

# Load anomaly log to remove anomalies
load(file.path(data_root, "CACHED_USER_DATA/telemetry_event_correlation/AnomalyLog.RData"))

# Identify consolidated telemetry stores built from earlier pipeline scripts
db_files <- list.files(
  path       = data_root,
  pattern    = paste0("^Consolidated_.*\\.db$"),
  full.names = TRUE
)


## Functions ###################################################################
# Remove windows associated with anomaly sequences
removeAnomalies <- function(df, anomaly_log) {
  
  merged <- merge(df, anomaly_log, by = "RecordID", all.x = TRUE, all.y = TRUE)
  
  flagged <- merged[
    merged$EventTimestamp >= merged$StartAnomaly &
      merged$EventTimestamp <= merged$EndAnomaly &
      !is.na(merged$AssetID) &
      !is.na(merged$StartAnomaly),
    c("RecordID", "AssetID")
  ]
  
  if (nrow(flagged) > 0) {
    flagged$IsAnomaly <- TRUE
    df <- merge(df, flagged, by = c("RecordID", "AssetID"), all.x = TRUE)
    df <- df[is.na(df$IsAnomaly), ]
    df$IsAnomaly <- NULL
  }
  
  return(df)
}


## Body ########################################################################

dfSensorLDelta      <- data.frame()
dfSensorLDeltaStats <- data.frame()
dfLeftSensorL       <- data.frame()
dfRightSensorL      <- data.frame()

for (db in db_files) {
  
  message("Processing: ", db)
  con <- dbConnect(SQLite(), dbname = db, flags = SQLITE_RO)
  
  # Pull left/right SensorL label streams
  dfLeftDB <- dbGetQuery(con,
    "SELECT RecordID, AssetID, EventTimestamp, Timestamp, CAST(SensorValue AS DECIMAL) AS LeftSensorL
     FROM ConsolidatedRecords
     WHERE SensorName = 'LeftSensorL'
     ORDER BY RecordID, EventTimestamp;"
  )
  
  dfRightDB <- dbGetQuery(con,
    "SELECT RecordID, AssetID, EventTimestamp, Timestamp, CAST(SensorValue AS DECIMAL) AS RightSensorL
     FROM ConsolidatedRecords
     WHERE T.SensorName = 'RightSensorL'
     ORDER BY RecordID, EventTimestamp;"
  )
  
  dbDisconnect(con)
  rm(con)
  
  ## Delta Computation --------------------------------------------------------
  dfLeftDB  <- distinct(dfLeftDB)
  dfRightDB <- distinct(dfRightDB)
  
  dfDelta <- merge(dfLeftDB, dfRightDB,  by = c("RecordID", "AssetID", "EventTimestamp", "Timestamp"), all = TRUE)
  
  dfDelta$DeltaSensorL <- dfDelta$RightSensorL - dfDelta$LeftSensorL
  
  dfDelta <- merge(
    dfDelta[, c("RecordID", "AssetID", "EventTimestamp", "Timestamp", "DeltaSensorL")],
    HeaderData[, c("RecordID", "Filename")],
    by = "RecordID",
    all.x = TRUE
  )
  
  ## Summary Stats ------------------------------------------------------------
  dfStats <- dfDelta %>%
    group_by(RecordID, AssetID, EventTimestamp, Filename) %>%
    summarise(
      minDeltaSensorL  = min(DeltaSensorL),
      maxDeltaSensorL  = max(DeltaSensorL),
      medianDeltaSensorL = median(DeltaSensorL),
      meanDeltaSensorL   = mean(DeltaSensorL),
      firstDeltaSensorL  = first(DeltaSensorL),
      lastDeltaSensorL   = last(DeltaSensorL),
      firstTS        = first(EventTimestamp),
      lastTS         = last(EventTimestamp),
      .groups = "drop"
    )
  
  dfStats$RateOfChange <- (dfStats$lastDeltaSensorL - dfStats$firstDeltaSensorL) /
    (dfStats$lastTS - dfStats$firstTS)
  
  # Remove anomalies ---------------------------------------------------------
  dfStats  <- removeAnomalies(dfStats, anomaly_log)
  dfDelta  <- removeAnomalies(dfDelta, anomaly_log)
  dfLeftDB  <- removeAnomalies(dfLeftDB,  anomaly_log)
  dfRightDB <- removeAnomalies(dfRightDB, anomaly_log)
  
  # Store results
  dfSensorLDelta   <- bind_rows(dfSensorLDelta, dfDelta)
  dfSensorLDeltaStats <- bind_rows(dfSensorLDeltaStats, dfStats)
  dfLeftSensorL  <- bind_rows(dfLeftSensorL,  dfLeftDB)
  dfRightSensorL <- bind_rows(dfRightSensorL, dfRightDB)
  
  rm(dfDelta, dfStats, dfLeftDB, dfRightDB, anomaly_log)
}


## Deduplicate ###############################################################
dfSensorLDelta      <- distinct(dfSensorLDelta)
dfSensorLDeltaStats <- distinct(dfSensorLDeltaStats)
dfLeftSensorL       <- distinct(dfLeftSensorL)
dfRightSensorL      <- distinct(dfRightSensorL)


## Outputs ###################################################################
con <- dbConnect(SQLite(), file.path(output_root, "SensorL/SensorLDelta.db"))
dbWriteTable(con, "StoresFiltered", dfSensorLDelta, overwrite = TRUE)
dbDisconnect(con)

write.table(
  dfSensorLDeltaStats,
  file = file.path(output_root, "SensorL/SensorLDeltaStats.csv"),
  sep = ",", row.names = FALSE
)

con <- dbConnect(SQLite(), file.path(output_root, "SensorL/LeftSensorL.db"))
dbWriteTable(con, "StoresFiltered", dfLeftSensorL, overwrite = TRUE)
dbDisconnect(con)

con <- dbConnect(SQLite(), file.path(output_root, "SensorL/RightSensorL.db"))
dbWriteTable(con, "StoresFiltered", dfRightSensorL, overwrite = TRUE)
dbDisconnect(con)