Merge pull request #784 from G-Node/nixio-datetime-conversions

[NIXIO] Save and load date, time, and datetime annotations

Author: JuliaSprenger

neo/io/nixio.py

@@ -22,8 +22,7 @@
 
 from __future__ import absolute_import
 
-import time
-from datetime import datetime
+from datetime import date, time, datetime
 try:
     from collections.abc import Iterable
 except ImportError:
@@ -55,10 +54,19 @@
 except NameError:
     string_types = str
 
+datetime_types = (date, time, datetime)
+
 EMPTYANNOTATION = "EMPTYLIST"
 ARRAYANNOTATION = "ARRAYANNOTATION"
+DATETIMEANNOTATION = "DATETIME"
+DATEANNOTATION = "DATE"
+TIMEANNOTATION = "TIME"
 MIN_NIX_VER = Version("1.5.0")
 
+datefmt = "%Y-%m-%d"
+timefmt = "%H:%M:%S.%f"
+datetimefmt = datefmt + "T" + timefmt
+
 
 def stringify(value):
     if value is None:
@@ -83,10 +91,40 @@ def units_to_string(pqunit):
     return dim
 
 
-def calculate_timestamp(dt):
+def dt_to_nix(dt):
+    """
+    Converts date, time, and datetime objects to an ISO string representation
+    appropriate for storing in NIX. Returns the converted value and the
+    annotation type definition for converting back to the original value
+    type.
+    """
     if isinstance(dt, datetime):
-        return int(time.mktime(dt.timetuple()))
-    return int(dt)
+        return dt.strftime(datetimefmt), DATETIMEANNOTATION
+    if isinstance(dt, date):
+        return dt.strftime(datefmt), DATEANNOTATION
+    if isinstance(dt, time):
+        return dt.strftime(timefmt), TIMEANNOTATION
+    # Unknown: returning as is
+    return dt
+
+
+def dt_from_nix(nixdt, annotype):
+    """
+    Inverse function of 'dt_to_nix()'. Requires the stored annotation type to
+    distinguish between the three source types (date, time, and datetime).
+    """
+    if annotype == DATEANNOTATION:
+        dt = datetime.strptime(nixdt, datefmt)
+        return dt.date()
+    if annotype == TIMEANNOTATION:
+        dt = datetime.strptime(nixdt, timefmt)
+        return dt.time()
+    if annotype == DATETIMEANNOTATION:
+        dt = datetime.strptime(nixdt, datetimefmt)
+        return dt
+    # Unknown type: older (or newer) IO version?
+    # Returning as is to avoid data loss.
+    return nixdt
 
 
 def check_nix_version():
@@ -307,7 +345,6 @@ def _nix_to_neo_segment(self, nix_group):
         neo_segment.rec_datetime = datetime.fromtimestamp(
             nix_group.created_at
         )
-
         self._neo_map[nix_group.name] = neo_segment
 
         # this will probably get all the DAs anyway, but if we change any part
@@ -582,12 +619,12 @@ def write_block(self, block, use_obj_names=False):
         metadata["neo_name"] = neoname
         nixblock.definition = block.description
         if block.rec_datetime:
-            nixblock.force_created_at(
-                calculate_timestamp(block.rec_datetime)
-            )
+            nix_rec_dt = int(block.rec_datetime.strftime("%s"))
+            nixblock.force_created_at(nix_rec_dt)
         if block.file_datetime:
-            fdt = calculate_timestamp(block.file_datetime)
-            metadata["file_datetime"] = fdt
+            fdt, annotype = dt_to_nix(block.file_datetime)
+            fdtprop = metadata.create_property("file_datetime", fdt)
+            fdtprop.definition = annotype
         if block.annotations:
             for k, v in block.annotations.items():
                 self._write_property(metadata, k, v)
@@ -683,12 +720,12 @@ def _write_segment(self, segment, nixblock):
         metadata["neo_name"] = neoname
         nixgroup.definition = segment.description
         if segment.rec_datetime:
-            nixgroup.force_created_at(
-                calculate_timestamp(segment.rec_datetime)
-            )
+            nix_rec_dt = int(segment.rec_datetime.strftime("%s"))
+            nixgroup.force_created_at(nix_rec_dt)
         if segment.file_datetime:
-            fdt = calculate_timestamp(segment.file_datetime)
-            metadata["file_datetime"] = fdt
+            fdt, annotype = dt_to_nix(segment.file_datetime)
+            fdtprop = metadata.create_property("file_datetime", fdt)
+            fdtprop.definition = annotype
         if segment.annotations:
             for k, v in segment.annotations.items():
                 self._write_property(metadata, k, v)
@@ -1161,8 +1198,10 @@ def _write_property(self, section, name, v):
             else:
                 section.create_property(name, v.magnitude.item())
             section.props[name].unit = str(v.dimensionality)
-        elif isinstance(v, datetime):
-            section.create_property(name, calculate_timestamp(v))
+        elif isinstance(v, datetime_types):
+            value, annotype = dt_to_nix(v)
+            prop = section.create_property(name, value)
+            prop.definition = annotype
         elif isinstance(v, string_types):
             if len(v):
                 section.create_property(name, v)
@@ -1201,8 +1240,7 @@ def _write_property(self, section, name, v):
                     values.append(item)
             section.create_property(name, values)
             section.props[name].unit = unit
-            if definition:
-                section.props[name].definition = definition
+            section.props[name].definition = definition
         elif type(v).__module__ == "numpy":
             section.create_property(name, v.item())
         else:
@@ -1237,6 +1275,9 @@ def _nix_attr_to_neo(nix_obj):
                         values = ""
                 elif len(values) == 1:
                     values = values[0]
+                if prop.definition in (DATEANNOTATION, TIMEANNOTATION,
+                                       DATETIMEANNOTATION):
+                    values = dt_from_nix(values, prop.definition)
                 if prop.definition == ARRAYANNOTATION:
                     if 'array_annotations' in neo_attrs:
                         neo_attrs['array_annotations'][prop.name] = values
@@ -1248,10 +1289,6 @@ def _nix_attr_to_neo(nix_obj):
         # there's no reason to keep it in the annotations
         neo_attrs["name"] = stringify(neo_attrs.pop("neo_name", None))
 
-        if "file_datetime" in neo_attrs:
-            neo_attrs["file_datetime"] = datetime.fromtimestamp(
-                neo_attrs["file_datetime"]
-            )
         return neo_attrs
 
     @staticmethod
@@ -1283,9 +1320,7 @@ def _get_time_dimension(obj):
         return None
 
     def _use_obj_names(self, blocks):
-
         errmsg = "use_obj_names enabled: found conflict or anonymous object"
-
         allobjs = []
 
         def check_unique(objs):

neo/test/iotest/test_nixio.py

@@ -17,7 +17,7 @@
     from collections.abc import Iterable
 except ImportError:
     from collections import Iterable
-from datetime import datetime
+from datetime import date, time, datetime
 
 from tempfile import mkdtemp
 
@@ -29,7 +29,8 @@
 from neo.core import (Block, Segment, ChannelIndex, AnalogSignal,
                       IrregularlySampledSignal, Unit, SpikeTrain, Event, Epoch)
 from neo.test.iotest.common_io_test import BaseTestIO
-from neo.io.nixio import NixIO, create_quantity, units_to_string, neover
+from neo.io.nixio import (NixIO, create_quantity, units_to_string, neover,
+                          dt_from_nix, dt_to_nix, DATETIMEANNOTATION)
 from neo.io.nixio_fr import NixIO as NixIO_lazy
 from neo.io.proxyobjects import AnalogSignalProxy, SpikeTrainProxy, EventProxy, EpochProxy
 
@@ -319,9 +320,10 @@ def compare_attr(self, neoobj, nixobj):
             self.assertEqual(neoobj.rec_datetime,
                              datetime.fromtimestamp(nixobj.created_at))
         if hasattr(neoobj, "file_datetime") and neoobj.file_datetime:
-            self.assertEqual(neoobj.file_datetime,
-                             datetime.fromtimestamp(
-                                 nixobj.metadata["file_datetime"]))
+            nixdt = dt_from_nix(nixobj.metadata["file_datetime"],
+                                DATETIMEANNOTATION)
+            assert neoobj.file_datetime == nixdt
+            self.assertEqual(neoobj.file_datetime, nixdt)
         if neoobj.annotations:
             nixmd = nixobj.metadata
             for k, v, in neoobj.annotations.items():
@@ -1317,7 +1319,7 @@ def test_to_value(self):
         # datetime
         dt = self.rdate()
         writeprop(section, "dt", dt)
-        self.assertEqual(datetime.fromtimestamp(section["dt"]), dt)
+        self.assertEqual(section["dt"], dt_to_nix(dt)[0])
 
         # string
         randstr = self.rsentence()
@@ -1446,6 +1448,29 @@ def generate_complete_block():
 
             self.write_and_compare([block_lazy])
 
+    def test_annotation_types(self):
+        annotations = {
+            "somedate": self.rdate(),
+            "now": datetime.now(),
+            "today": date.today(),
+            "sometime": time(13, 37, 42),
+            "somequantity": self.rquant(10, pq.ms),
+            "somestring": self.rsentence(3),
+            "npfloat": np.float(10),
+            "nparray": np.array([1, 2, 400]),
+            "emptystr": "",
+        }
+        wblock = Block("annotation_block", **annotations)
+        self.writer.write_block(wblock)
+        rblock = self.writer.read_block(neoname="annotation_block")
+        for k in annotations:
+            orig = annotations[k]
+            readval = rblock.annotations[k]
+            if isinstance(orig, np.ndarray):
+                np.testing.assert_almost_equal(orig, readval)
+            else:
+                self.assertEqual(annotations[k], rblock.annotations[k])
+
 
 @unittest.skipUnless(HAVE_NIX, "Requires NIX")
 class NixIOReadTest(NixIOTest):