tests/testthat/test-wfdb-structures.R

In EGM: Evaluating Cardiac Electrophysiology Signals

test_that("signal table class can be made", {

	skip_on_cran()
	skip_on_ci()

	expect_equal(signal_table(), new_signal_table())
	expect_output(print(signal_table()), "signal_table")

	# Reads in sample ECG data
	dat <- read_signal(
		record = "ecg",
		record_dir = test_path(),
		begin = 0,
		units = "digital"
	)

	# ECG data should be 12 signal columns and 1 index column
	x <- dat[, -1]
	expect_equal(dim(x), c(5000, 12))

})

test_that("header_table can be made using LSPro data", {

	# Initial values
	record_name = character() # Record line information
	number_of_channels = integer()
	frequency = 250.0
	samples = integer()
	start_time = Sys.time()
	ADC_saturation = integer()
	file_name = character() # Signal specific information
	storage_format = 16L
	ADC_gain = 200L
	ADC_units = "mV"
	ADC_resolution = 12L
	ADC_zero = 0L
	ADC_baseline = ADC_zero
	initial_value = ADC_zero
	checksum = 0L
	blocksize = 0L
	label = character()
	info_strings = list() # Secondary information
	additional_gain = 1.0
	low_pass = integer()
	high_pass = integer()
	color = character()
	scale = integer()

	file <- test_path("egm.txt")
	record_name <- deparse1(substitute(file))
	file_name <- paste0(deparse1(substitute(file)), ".dat")
	hea <-
		readLines(file, n = 13) |>
		tstrsplit(split = ":\ ", fill = NA) |>
		{\(.x)
			data.table(description = .x[[1]], value = .x[[2]])
		}() |>
		{\(.y)
			list(
				record_name = record_name,
				number_of_channels = as.numeric(.y$value[.y$description == "Channels exported"]),
				samples = {
					s <- .y$value[.y$description == "Samples per channel"]
					if (grepl(":", s)) {
						substr(s, start = 1, stop = nchar(s) - 8) |>
							as.numeric()
					} else {
						as.numeric(s)
					}
				},
				start_time = as.POSIXct(strptime(.y$value[.y$description == "Start time"], format = "%H:%M:%S")),
				end_time = as.POSIXct(strptime(.y$value[.y$description == "End time"], format = "%H:%M:%S")),
				frequency = {
					f <- .y$value[.y$description == "Sample Rate"]
					if (grepl("Hz", f)) {
						gsub("Hz", "", f) |>
							as.numeric()
					} else {
						as.numeric(f)
					}
				}
			)
		}()

	hea$ADC_saturation <- 32768

	ch_list <- list()
	for (i in 1:hea$number_of_channels) {
		ch_list[[i]] <-
			fread(
				file,
				skip = 13 + (i - 1) * 8,
				nrows = 8,
				sep = NULL,
				header = FALSE
			) |>
			unlist() |>
			tstrsplit(split = ":\ ") |>
			{
				\(.x)
				data.table(description = .x[[1]], value = .x[[2]])
			}() |>
			{
				\(.y)
				list(
					number = as.numeric(.y[1, 2]),
					label = as.character(.y[2, 2]),
					gain = as.numeric(gsub("mv", "", .y[3, 2])),
					low = as.numeric(gsub("Hz", "", .y[4, 2])),
					high = as.numeric(gsub("Hz", "", .y[5, 2])),
					frequency = as.numeric(gsub("Hz", "", .y[6, 2])),
					color = paste0("#", .y[7, 2]),
					scale = as.numeric(.y[8, 2])
				)
			}()
	}

	channels <- rbindlist(ch_list)

	h <- header_table(
		record_name = hea$record_name,
		number_of_channels = hea$number_of_channels,
		frequency = hea$frequency,
		samples = hea$samples,
		start_time = hea$start_time,
		ADC_saturation = hea$ADC_saturation,
		file_name = file_name,
		label = channels$label,
		additional_gain = channels$gain,
		low_pass = channels$low,
		high_pass = channels$high,
		color = channels$color
	)

	expect_length(h, 19)
	expect_true(length(unique(h$color)) > 1)
	expect_true(length(unique(h$low_pass)) == 2)
	expect_true(length(unique(h$high_pass)) == 2)
	rec <- attributes(h)$record_line
	info <- attributes(h)$info_string
	expect_type(rec, "list")
	expect_type(info, "list")
	expect_named(rec, c("record_name", "number_of_channels", "samples", "start_time", "frequency", "ADC_saturation"), ignore.order = TRUE)

})

test_that("minimal header_table can be made that can then be written to WFDB", {

	x <-
		header_table(record_name = "temp",
								 frequency = 500)

	expect_equal(attributes(x)$record_line$record_name, "temp")


})

test_that("an annotation table can be created", {

	fp <- system.file('extdata', 'muse-sinus.xml', package = 'EGM')
	ecg <- read_muse(fp)
	x <-
		ecg$signal$I |>
		abs()
	sample <- which(diff(sign(diff(x))) == -2) + 1
	sample <- sample[x[sample] > (mean(x, na.rm = TRUE) + 2 * sd(x, na.rm = TRUE))]


	ann <- annotation_table(
	  annotator = "test_peaks",
	  sample = sample,
	  type = "R",
	  frequency = attributes(ecg$header)$record_line$frequency,
	  channel = "I"
	)

	expect_s3_class(ann, "annotation_table")

})