Merge pull request #186 from MIT-LCP/fix/empatica-validation

Recognize Empatica E4 exports in dataset validation

Author: tompollard

README.md

@@ -142,6 +142,10 @@ dataset directory.
 - **privacy** - PHI patterns (SSN, email, phone numbers), date patterns, and
   sensitive file detection.
 
+CSV exports from the Empatica E4 wearable are recognized, so their headerless
+layout, empty `tags.csv` files, and Unix timestamps don't produce false errors
+or privacy warnings. This doesn't check that timestamps are de-identified.
+
 **Examples:**
 
 ```bash

physionet/validate/checks/csv_formats.py

@@ -0,0 +1,29 @@
+"""Choose the CSV layout used by all validation checks.
+
+Additional formats can provide a detector returning CsvLayout or None and add
+it to _FORMAT_DETECTORS. Ordinary headered CSV remains the default.
+"""
+
+from pathlib import Path
+from typing import List
+
+from physionet.validate.checks.csv_layout import CsvLayout
+from physionet.validate.checks.empatica import get_e4_csv_layout
+
+
+_FORMAT_DETECTORS = (get_e4_csv_layout,)
+
+
+def find_files_by_suffix(path: Path, suffix: str) -> List[Path]:
+    """Find files under path whose extension matches suffix in any letter case."""
+    suffix = suffix.lower()
+    return [p for p in path.rglob("*") if p.suffix.lower() == suffix and p.is_file()]
+
+
+def get_csv_layout(path: Path) -> CsvLayout:
+    """Use a recognized layout or ordinary headered CSV rules."""
+    for detect in _FORMAT_DETECTORS:
+        layout = detect(path)
+        if layout is not None:
+            return layout
+    return CsvLayout()

physionet/validate/checks/csv_layout.py

@@ -0,0 +1,41 @@
+"""Shared CSV layout information for validation checks."""
+
+import csv
+from dataclasses import dataclass
+from typing import Optional, TextIO, Tuple
+
+
+@dataclass(frozen=True)
+class CsvLayout:
+    """Describe which CSV rows are metadata and how data columns are named.
+
+    With no explicit columns, the first row is treated as a header. Explicit
+    columns describe a headerless file; metadata_rows precede its data rows.
+    A file with no data rows gets a warning unless no_data_message is set, in
+    which case that message is reported as information.
+    """
+
+    columns: Optional[Tuple[str, ...]] = None
+    metadata_rows: int = 0
+    allow_empty: bool = False
+    empty_message: Optional[str] = None
+    no_data_message: Optional[str] = None
+    # Columns whose Unix timestamps are not reported as phone numbers.
+    unix_timestamp_columns: Tuple[str, ...] = ()
+
+    def dict_reader(self, stream: TextIO) -> csv.DictReader:
+        """Read data rows under the layout's column names."""
+        reader = csv.DictReader(stream, fieldnames=self.columns)
+        for _ in range(self.metadata_rows):
+            next(reader, None)
+        return reader
+
+
+def is_unix_timestamp(value: str) -> bool:
+    """Recognize nine- or ten-digit Unix seconds, including shifted dates."""
+    whole, separator, fraction = value.strip().partition(".")
+    if len(whole) not in (9, 10) or not whole.isdigit() or whole[0] == "0":
+        return False
+    if separator and (not fraction or not fraction.isdigit()):
+        return False
+    return True

physionet/validate/checks/empatica.py

@@ -0,0 +1,118 @@
+"""Recognize the headerless CSV layout exported by Empatica E4 Connect.
+
+Format: https://www.empatica.com/blog/decoding-wearable-sensor-signals-what-to-expect-from-your-e4-data/
+
+- ACC, BVP, EDA, HR, TEMP: a row of Unix start times, a row of sample rates,
+  then samples. Data columns are named acc_x/acc_y/acc_z, bvp, eda, hr, temp,
+  which is what ValidationConfig.value_ranges keys should use.
+- IBI: a "<start time>, IBI" row, then time_offset,ibi rows. A file with only
+  the first row means no beats were detected and is reported as information.
+- tags: one event time per row, no header; empty means no events. It is only
+  treated as E4 when a recognized sensor or IBI file sits beside it, and only
+  its event_time column is exempt from the phone-number pattern.
+
+File names are matched in any letter case. tests/fixtures/e4_session holds a
+small synthetic example.
+"""
+
+import csv
+from pathlib import Path
+from typing import Optional, Tuple
+
+from physionet.validate.checks.csv_layout import CsvLayout, is_unix_timestamp
+
+
+# Keyed by lowercase file name; E4 Connect writes upper-case names, but
+# re-exported or hand-tidied datasets sometimes lowercase them.
+_SENSOR_LAYOUTS = {
+    "acc.csv": (("acc_x", "acc_y", "acc_z"), 32),
+    "bvp.csv": (("bvp",), 64),
+    "eda.csv": (("eda",), 4),
+    "hr.csv": (("hr",), 1),
+    "temp.csv": (("temp",), 4),
+}
+
+_IBI_COLUMNS = ("time_offset", "ibi")
+
+
+def _read_first_rows(file_path: Path, count: int) -> Optional[list]:
+    """Return the first rows of a CSV file, or None if it cannot supply them."""
+    try:
+        with file_path.open(newline="", encoding="utf-8") as stream:
+            reader = csv.reader(stream)
+            return [next(reader) for _ in range(count)]
+    except (OSError, UnicodeError, StopIteration, csv.Error):
+        return None
+
+
+def is_e4_sensor_file(file_path: Path) -> bool:
+    """Match a sensor's timestamp and sample-rate rows, without reading its data."""
+    layout = _SENSOR_LAYOUTS.get(file_path.name.lower())
+    if not layout or not file_path.is_file():
+        return False
+
+    rows = _read_first_rows(file_path, 2)
+    if rows is None:
+        return False
+    timestamps, rates = rows
+    columns, sample_rate = layout
+    channels = len(columns)
+    try:
+        return (
+            len(timestamps) == channels
+            and len(rates) == channels
+            and all(is_unix_timestamp(value) for value in timestamps)
+            and all(float(value) == sample_rate for value in rates)
+        )
+    except ValueError:
+        return False
+
+
+def e4_sensor_columns(file_path: Path) -> Optional[Tuple[str, ...]]:
+    """Return positional channel names for a recognized E4 sensor file."""
+    if is_e4_sensor_file(file_path):
+        return _SENSOR_LAYOUTS[file_path.name.lower()][0]
+    return None
+
+
+def is_e4_ibi_file(file_path: Path) -> bool:
+    """Match IBI.csv, whose first row is the start time followed by 'IBI'."""
+    if file_path.name.lower() != "ibi.csv" or not file_path.is_file():
+        return False
+    rows = _read_first_rows(file_path, 1)
+    if rows is None:
+        return False
+    first = rows[0]
+    return len(first) == 2 and is_unix_timestamp(first[0]) and first[1].strip() == "IBI"
+
+
+def is_e4_tags_file(file_path: Path) -> bool:
+    """Identify event tags only when a neighboring E4 sensor or IBI file confirms the format."""
+    if file_path.name.lower() != "tags.csv":
+        return False
+    try:
+        siblings = list(file_path.parent.iterdir())
+    except OSError:
+        return False
+    return any(is_e4_sensor_file(sibling) or is_e4_ibi_file(sibling) for sibling in siblings)
+
+
+def get_e4_csv_layout(file_path: Path) -> Optional[CsvLayout]:
+    """Describe recognized E4 sensor, IBI, and event-tag files for CSV checks."""
+    columns = e4_sensor_columns(file_path)
+    if columns is not None:
+        return CsvLayout(columns=columns, metadata_rows=2)
+    if is_e4_ibi_file(file_path):
+        return CsvLayout(
+            columns=_IBI_COLUMNS,
+            metadata_rows=1,
+            no_data_message="No Empatica E4 inter-beat intervals detected (IBI.csv has only its start-time row)",
+        )
+    if is_e4_tags_file(file_path):
+        return CsvLayout(
+            columns=("event_time",),
+            allow_empty=True,
+            empty_message="No Empatica E4 event markers recorded (empty tags.csv)",
+            unix_timestamp_columns=("event_time",),
+        )
+    return None

physionet/validate/checks/filesystem.py

@@ -5,6 +5,7 @@
 
 from physionet.validate.models import CheckResult, ValidationIssue, CheckCategory, Severity
 from physionet.validate.config import ValidationConfig
+from physionet.validate.checks.csv_formats import get_csv_layout
 
 # Proprietary file formats and their recommended open alternatives
 PROPRIETARY_FORMATS = {
@@ -93,6 +94,17 @@ def check_filesystem(path: Path, config: ValidationConfig) -> CheckResult:
             try:
                 size = file_path.stat().st_size
                 if size == 0:
+                    layout = get_csv_layout(file_path)
+                    if layout.allow_empty:
+                        result.issues.append(
+                            ValidationIssue(
+                                severity=Severity.INFO,
+                                category=CheckCategory.FILESYSTEM,
+                                file=str(file_path.relative_to(path)),
+                                message=layout.empty_message or "Empty file is valid for this CSV layout",
+                            )
+                        )
+                        continue
                     result.issues.append(
                         ValidationIssue(
                             severity=Severity.WARNING,

physionet/validate/checks/integrity.py

@@ -6,6 +6,7 @@
 
 from physionet.validate.models import CheckResult, ValidationIssue, CheckCategory, Severity
 from physionet.validate.config import ValidationConfig
+from physionet.validate.checks.csv_formats import find_files_by_suffix, get_csv_layout
 
 
 def check_integrity(path: Path, config: ValidationConfig, progress_callback: Optional[Callable[[str], None]] = None) -> CheckResult:
@@ -28,7 +29,7 @@ def check_integrity(path: Path, config: ValidationConfig, progress_callback: Opt
     result = CheckResult(category=CheckCategory.INTEGRITY)
 
     # Find and validate CSV files
-    csv_files = list(path.rglob("*.csv"))
+    csv_files = find_files_by_suffix(path, ".csv")
     for i, csv_file in enumerate(csv_files):
         if progress_callback:
             progress_callback(f"Checking {csv_file.name} ({i+1}/{len(csv_files)} CSV files)")
@@ -43,24 +44,27 @@ def check_integrity(path: Path, config: ValidationConfig, progress_callback: Opt
 
 def _validate_csv_structure(csv_file: Path, base_path: Path, result: CheckResult) -> None:
     """Validate CSV file structure."""
+    layout = get_csv_layout(csv_file)
     try:
         with open(csv_file, "r", encoding="utf-8") as f:
             # Try to detect dialect
             sample = f.read(1024)
             f.seek(0)
 
             try:
-                dialect = csv.Sniffer().sniff(sample)
+                dialect = csv.Sniffer().sniff(sample, delimiters=",;\t|")
             except csv.Error:
                 # Use default dialect if detection fails
                 dialect = csv.excel
 
             reader = csv.reader(f, dialect)
 
-            # Read header
+            # Read the first row, which is a header for ordinary CSV files.
             try:
-                header = next(reader)
+                first_row = next(reader)
             except StopIteration:
+                if layout.allow_empty:
+                    return
                 result.issues.append(
                     ValidationIssue(
                         severity=Severity.ERROR,
@@ -71,7 +75,7 @@ def _validate_csv_structure(csv_file: Path, base_path: Path, result: CheckResult
                 )
                 return
 
-            if not header:
+            if not first_row:
                 result.issues.append(
                     ValidationIssue(
                         severity=Severity.ERROR,
@@ -82,33 +86,41 @@ def _validate_csv_structure(csv_file: Path, base_path: Path, result: CheckResult
                 )
                 return
 
-            # Check for duplicate column names
-            if len(header) != len(set(header)):
-                duplicates = [col for col in header if header.count(col) > 1]
-                result.issues.append(
-                    ValidationIssue(
-                        severity=Severity.ERROR,
-                        category=CheckCategory.INTEGRITY,
-                        file=str(csv_file.relative_to(base_path)),
-                        message=f"Duplicate column names found: {', '.join(set(duplicates))}",
+            if layout.columns is None:
+                header = first_row
+                if len(header) != len(set(header)):
+                    duplicates = [col for col in header if header.count(col) > 1]
+                    result.issues.append(
+                        ValidationIssue(
+                            severity=Severity.ERROR,
+                            category=CheckCategory.INTEGRITY,
+                            file=str(csv_file.relative_to(base_path)),
+                            message=f"Duplicate column names found: {', '.join(set(duplicates))}",
+                        )
                     )
-                )
-
-            # Check for empty column names
-            if any(not col.strip() for col in header):
-                result.issues.append(
-                    ValidationIssue(
-                        severity=Severity.ERROR,
-                        category=CheckCategory.INTEGRITY,
-                        file=str(csv_file.relative_to(base_path)),
-                        message="CSV contains empty column names",
+                if any(not col.strip() for col in header):
+                    result.issues.append(
+                        ValidationIssue(
+                            severity=Severity.ERROR,
+                            category=CheckCategory.INTEGRITY,
+                            file=str(csv_file.relative_to(base_path)),
+                            message="CSV contains empty column names",
+                        )
                     )
-                )
+                expected_cols = len(header)
+                first_data_line = 2
+            else:
+                # Headerless files have explicit names; validate their first data row.
+                f.seek(0)
+                reader = csv.reader(f, dialect)
+                for _ in range(layout.metadata_rows):
+                    next(reader, None)
+                expected_cols = len(layout.columns)
+                first_data_line = layout.metadata_rows + 1
 
             # Validate row consistency
-            expected_cols = len(header)
             row_count = 0
-            for line_num, row in enumerate(reader, start=2):  # Start at 2 (after header)
+            for line_num, row in enumerate(reader, start=first_data_line):
                 row_count += 1
                 if len(row) != expected_cols:
                     result.issues.append(
@@ -132,13 +144,13 @@ def _validate_csv_structure(csv_file: Path, base_path: Path, result: CheckResult
                         )
                         break
 
-            if row_count == 0:
+            if row_count == 0 and not layout.allow_empty:
                 result.issues.append(
                     ValidationIssue(
-                        severity=Severity.WARNING,
+                        severity=Severity.INFO if layout.no_data_message else Severity.WARNING,
                         category=CheckCategory.INTEGRITY,
                         file=str(csv_file.relative_to(base_path)),
-                        message="CSV file contains only header row (no data)",
+                        message=layout.no_data_message or "CSV file contains only header row (no data)",
                     )
                 )