[tests] More

bob/db/hci_tagging/models.py

@@ -4,7 +4,123 @@
 # Wed 30 Sep 2015 12:13:47 CEST
 
 import os
-import bob
+import numpy
+import bob.db.base
+import bob.io.base
+
+
+def bdf_load_signal(fn, name='EXG3', start=None, end=None):
+  """Loads a signal named ``name`` from the BDF filenamed ``fn``
+
+
+  Parameters:
+
+    fn (path): The full path to the file to read
+    name (str): The name of the channel to read.
+    start (int, option): Start time in seconds
+    end (int, optional): End time in seconds
+
+
+  List of physiological channels used (there are more available, but contain no
+  meaningful data) on the Mahnob HCI-Tagging database:
+
+    These are the 32 electrodes from the EEG cap (measurements in uV; for full
+    positioning details, see the full database description report, available on
+    the database website):
+
+      * AF3
+      * AF4
+      * C3
+      * C4
+      * CP1
+      * CP2
+      * CP5
+      * CP6
+      * Cz
+      * F3
+      * F4
+      * F7
+      * F8
+      * FC1
+      * FC2
+      * FC5
+      * FC6
+      * Fp1
+      * Fp2
+      * Fz
+      * O1
+      * O2
+      * Oz
+      * P3
+      * P4
+      * P7
+      * P8
+      * PO3
+      * PO4
+      * Pz
+      * T7
+      * T8
+
+    These are ECG sensors (measurements in uV):
+
+    * EXG1: Upper right corner of chest, under clavicle bone
+    * EXG2: Upper left corner of chest, under clavicle bone
+    * EXG3: Left side of abdomen (very clean)
+
+    Other sensors:
+
+    * GSR1: Galvanic skin response (in Ohm)
+    * Resp: Respiration belt (in uV)
+    * Status: Status channel containing markers (Boolean)
+    * Temp: Skin temperature on the left pinky (Celsius)
+
+  """
+
+  import edflib
+  e = edflib.EdfReader(fn)
+
+  # get the status information, so we how the video is synchronized
+  status_index = e.getSignalTextLabels().index('Status')
+  sample_frequency = e.samplefrequency(status_index)
+  status_size = e.samples_in_file(status_index)
+  status = numpy.zeros((status_size,), dtype='float64')
+  e.readsignal(status_index, 0, status_size, status)
+  status = status.round().astype('int')
+  nz_status = status.nonzero()[0]
+
+  # because we're interested in the video bits, make sure to get data
+  # from that period only
+  video_start = nz_status[0]
+  video_end = nz_status[-1]
+
+  # retrieve information from this rather chaotic API
+  index = e.getSignalTextLabels().index(name)
+  sample_frequency = e.samplefrequency(index)
+
+  video_start_seconds = video_start/sample_frequency
+
+  if start is not None:
+    start += video_start_seconds
+    start *= sample_frequency
+    if start < video_start: start = video_start
+    start = int(start)
+  else:
+    start = video_start
+
+  if end is not None:
+    end += video_start_seconds
+    end *= sample_frequency
+    if end > video_end: end = video_end
+    end = int(end)
+  else:
+    end = video_end
+
+  # now read the data into a numpy array (read everything or libedf crashes)
+  container = numpy.zeros((end-start,), dtype='float64')
+  e.readsignal(index, start, end-start, container)
+
+  return container, sample_frequency
+
 
 class File(object):
   """ Generic file container for HCI-Tagging files
@@ -26,7 +142,7 @@ class File(object):
     self.basedir = basedir
     self.stem = bdf
     self.video_stem = video
-    self.duration = duration
+    self.duration = int(duration)
 
 
   def __repr__(self):
@@ -80,5 +196,5 @@ class File(object):
     """
 
     path = self.make_path(directory, extension)
-    bob.db.utils.makedirs_safe(os.path.dirname(path))
-    bob.io.save(data, path)
+    bob.db.base.utils.makedirs_safe(os.path.dirname(path))
+    bob.io.base.save(data, path)

bob/db/hci_tagging/test.py

@@ -39,6 +39,25 @@ class HCITaggingTest(unittest.TestCase):
     self.assertEqual(len(self.db.objects()), 3490)
 
 
+  @db_available
+  def test02_can_read_bdf(self):
+
+    from .models import bdf_load_signal
+
+    for obj in self.db.objects():
+
+      path = obj.make_path(DATABASE_LOCATION, '.bdf')
+      self.assertTrue(os.path.exists(path))
+
+      signal, freq = bdf_load_signal(path)
+
+      time = len(signal)/freq
+
+      # correlation between video data and physiological signal
+      if abs(time-obj.duration) > 2:
+        print('Physiological signal (%d seconds) is very different in size from estimated video duration (%d seconds) on sample `%s/%s\'' % (time, obj.duration, obj.basedir, obj.stem))
+
+
 class CmdLineTest(unittest.TestCase):
   """Makes sure our command-line is working properly."""