Handle dateTime timestamps in RDF.Nodes

my encode_find test code needed it to test parsing timestamps
from the submission page.

diff --git a/encode_submission/test_encode_find.py b/encode_submission/test_encode_find.py
index 98bdb46d11f6e200b7a35567a16b4477cb9b7a69..9beb10239d184cc6021bb5bd08e35c75e2cea978 100644 (file)
--- a/encode_submission/test_encode_find.py
+++ b/encode_submission/test_encode_find.py
@@ -1,4 +1,5 @@
 #!/usr/bin/env python
+from datetime import datetime
 import os
 import unittest
 
@@ -6,7 +7,7 @@ import RDF
 
 import encode_find
 from htsworkflow.submission.ucsc import submission_view_url
-from htsworkflow.util.rdfhelp import dump_model, get_model
+from htsworkflow.util.rdfhelp import dump_model, get_model, fromTypedNode
 
 SOURCE_PATH = os.path.split(os.path.abspath(__file__))[0]
 print SOURCE_PATH
@@ -40,7 +41,9 @@ class TestEncodeFind(unittest.TestCase):
         encode_find.parse_submission_page(model, tree, subNode)
         dates = encode_find.get_creation_dates(model, subNode)
         self.assertEqual(len(dates), 1)
-        self.assertEqual(str(dates[0].object), '2011-12-07T15:23:00')
+        object_date = fromTypedNode(dates[0].object)
+        self.assertEqual(object_date, datetime(2011,12,7,15,23,0))
+
 
 def suite():
     return unittest.makeSuite(TestEncodeFind, "test")

diff --git a/htsworkflow/util/rdfhelp.py b/htsworkflow/util/rdfhelp.py
index 8fb1424960571d4ab327ab2c4261b293eb1efb56..3b2bfe0ce3ccafacde089eabf34b2d03c5844356 100644 (file)
--- a/htsworkflow/util/rdfhelp.py
+++ b/htsworkflow/util/rdfhelp.py
@@ -1,5 +1,6 @@
 """Helper features for working with librdf
 """
+from datetime import datetime
 import logging
 import os
 import types
@@ -22,6 +23,9 @@ libraryOntology = RDF.NS("http://jumpgate.caltech.edu/wiki/LibraryOntology#")
 inventoryOntology = RDF.NS("http://jumpgate.caltech.edu/wiki/InventoryOntology#")
 submissionLog = RDF.NS("http://jumpgate.caltech.edu/wiki/SubmissionsLog/")
 
+ISOFORMAT_MS = "%Y-%m-%dT%H:%M:%S.%f"
+ISOFORMAT_SHORT = "%Y-%m-%dT%H:%M:%S"
+
 def sparql_query(model, query_filename):
     """Execute sparql query from file
     """
@@ -61,6 +65,12 @@ def toTypedNode(value):
     elif type(value) == types.FloatType:
         value_type = xsdNS['float'].uri
         value = unicode(value)
+    elif isinstance(value, datetime):
+        value_type = xsdNS['dateTime'].uri
+        if value.microsecond == 0:
+            value = value.strftime(ISOFORMAT_SHORT)
+        else:
+            value = value.strftime(ISOFORMAT_MS)
     else:
         value_type = None
         value = unicode(value)
@@ -95,8 +105,10 @@ def fromTypedNode(node):
     elif value_type in ('string'):
         return literal
     elif value_type in ('dateTime'):
-        raise NotImplemented('need to parse isoformat date-time')
-
+        try:
+            return datetime.strptime(literal, ISOFORMAT_MS)
+        except ValueError, e:
+            return datetime.strptime(literal, ISOFORMAT_SHORT)
     return literal
 
 

diff --git a/htsworkflow/util/test/test_rdfhelp.py b/htsworkflow/util/test/test_rdfhelp.py
index b173d5aef696be5bc971ba814d7d0fcb89dcaaaa..df275edd095b818301308816589000405e67833d 100644 (file)
--- a/htsworkflow/util/test/test_rdfhelp.py
+++ b/htsworkflow/util/test/test_rdfhelp.py
@@ -1,32 +1,39 @@
 import unittest
 import types
 
-from htsworkflow.util.rdfhelp import blankOrUri, toTypedNode, fromTypedNode
+from datetime import datetime
+
+from htsworkflow.util.rdfhelp import \
+     blankOrUri, \
+     toTypedNode, \
+     fromTypedNode, \
+     xsdNS
+
 try:
   import RDF
-  
+
   class TestRDFHelp(unittest.TestCase):
       def test_typed_node_boolean(self):
           node = toTypedNode(True)
           self.failUnlessEqual(node.literal_value['string'], u'1')
           self.failUnlessEqual(str(node.literal_value['datatype']),
                                'http://www.w3.org/2001/XMLSchema#boolean')
-  
+
       def test_typed_node_string(self):
           node = toTypedNode('hello')
           self.failUnlessEqual(node.literal_value['string'], u'hello')
           self.failUnless(node.literal_value['datatype'] is None)
-  
+
       def test_blank_or_uri_blank(self):
           node = blankOrUri()
           self.failUnlessEqual(node.is_blank(), True)
-  
+
       def test_blank_or_uri_url(self):
           s = 'http://google.com'
           node = blankOrUri(s)
           self.failUnlessEqual(node.is_resource(), True)
           self.failUnlessEqual(str(node.uri), s)
-  
+
       def test_blank_or_uri_node(self):
           s = RDF.Node(RDF.Uri('http://google.com'))
           node = blankOrUri(s)
@@ -34,18 +41,42 @@ try:
           self.failUnlessEqual(node, s)
 
       def test_unicode_node_roundtrip(self):
-        literal = u'\u5927'
-        roundtrip = fromTypedNode(toTypedNode(literal))
-        self.failUnlessEqual(roundtrip, literal)
-        self.failUnlessEqual(type(roundtrip), types.UnicodeType)
-          
+          literal = u'\u5927'
+          roundtrip = fromTypedNode(toTypedNode(literal))
+          self.failUnlessEqual(roundtrip, literal)
+          self.failUnlessEqual(type(roundtrip), types.UnicodeType)
+
+      def test_datetime_no_microsecond(self):
+          dateTimeType = xsdNS['dateTime'].uri
+          short_isostamp = '2011-12-20T11:44:25'
+          short_node = RDF.Node(literal=short_isostamp,
+                               datatype=dateTimeType)
+          short_datetime = datetime(2011,12,20,11,44,25)
+
+          self.assertEqual(fromTypedNode(short_node), short_datetime)
+          self.assertEqual(toTypedNode(short_datetime), short_node)
+          self.assertEqual(fromTypedNode(toTypedNode(short_datetime)),
+                           short_datetime)
+
+      def test_datetime_with_microsecond(self):
+          dateTimeType = xsdNS['dateTime'].uri
+          long_isostamp = '2011-12-20T11:44:25.081776'
+          long_node = RDF.Node(literal=long_isostamp,
+                               datatype=dateTimeType)
+          long_datetime = datetime(2011,12,20,11,44,25,81776)
+
+          self.assertEqual(fromTypedNode(long_node), long_datetime)
+          self.assertEqual(toTypedNode(long_datetime), long_node)
+          self.assertEqual(fromTypedNode(toTypedNode(long_datetime)),
+                           long_datetime)
+
   def suite():
       return unittest.makeSuite(testRdfHelp, 'test')
 except ImportError, e:
     print "Unable to test rdfhelp"
-    
+
     def suite():
         return None
-    
+
 if __name__ == "__main__":
     unittest.main(defaultTest='suite')