From: Diane Trout Date: Tue, 20 Dec 2011 20:33:16 +0000 (-0800) Subject: Handle dateTime timestamps in RDF.Nodes X-Git-Tag: v0.5.5~75 X-Git-Url: http://woldlab.caltech.edu/gitweb/?p=htsworkflow.git;a=commitdiff_plain;h=6d1e0c92678f15a366ef6f39f2697fcb8eb44628 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 98bdb46..9beb102 100644 --- 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 8fb1424..3b2bfe0 100644 --- 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 b173d5a..df275ed 100644 --- 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')