--- /dev/null
+import RDF
+from pyld import jsonld
+
+def load_into_model(model, json_data):
+ '''Given a PyLD dictionary, load its statements into our Redland model
+ '''
+ json_graphs = jsonld.to_rdf(json_data)
+ for graph in json_graphs:
+ for triple in json_graphs[graph]:
+ stmt = triple_to_statement(triple)
+ model.add_statement(stmt) #, graph_context)
+
+def triple_to_statement(triple):
+ '''Convert PyLD triple dictionary to a librdf statement
+ '''
+ s = to_node(triple['subject'])
+ p = to_node(triple['predicate'])
+ o = to_node(triple['object'])
+ return RDF.Statement(s, p, o)
+
+def to_node(item):
+ '''Convert a PyLD node to a Redland node'''
+ nodetype = item['type']
+ value = item['value']
+ datatype = item.get('datatype', None)
+
+ if nodetype == 'blank node':
+ return RDF.Node(blank=value)
+ elif nodetype == 'IRI':
+ return RDF.Node(uri_string=str(value))
+ else:
+ return RDF.Node(literal=unicode(value).encode('utf-8'),
+ datatype=RDF.Uri(datatype))
--- /dev/null
+from unittest2 import TestCase, TestSuite, defaultTestLoader, skip
+
+from htsworkflow.util.rdfjsonld import load_into_model, to_node, triple_to_statement
+from htsworkflow.util.rdfhelp import get_model
+
+jstatement = {
+ 'object': {'datatype': u'http://www.w3.org/2001/XMLSchema#dateTime',
+ 'type': 'literal',
+ 'value': '1940-10-09'},
+ 'predicate': {'type': 'IRI',
+ 'value': u'http://schema.org/birthDate'},
+ 'subject': {'type': 'blank node',
+ 'value': '_:a'}
+}
+doc = {
+ "@context": "http://json-ld.org/contexts/person.jsonld",
+ "@id": "http://dbpedia.org/resource/John_Lennon",
+ "name": "John Lennon",
+ "born": "1940-10-09",
+ "spouse": "http://dbpedia.org/resource/Cynthia_Lennon"
+}
+
+class TestJsonLD(TestCase):
+ def test_to_node(self):
+ obj = to_node(jstatement['object'])
+ self.assertTrue(obj.is_literal())
+ self.assertEqual(str(obj), '1940-10-09')
+ pred = to_node(jstatement['predicate'])
+ self.assertTrue(pred.is_resource())
+ self.assertEqual(str(pred.uri), jstatement['predicate']['value'])
+ subj = to_node(jstatement['subject'])
+ self.assertTrue(subj.is_blank())
+
+ def test_to_statement(self):
+ stmt = triple_to_statement(jstatement)
+ self.assertTrue(stmt.object.is_literal())
+ self.assertEqual(str(stmt.object), '1940-10-09')
+ self.assertTrue(stmt.predicate.is_resource())
+ self.assertEqual(str(stmt.predicate.uri), jstatement['predicate']['value'])
+ self.assertTrue(stmt.subject.is_blank())
+
+ def test_load_model(self):
+ model = get_model(use_contexts=False)
+ self.assertEqual(len(model), 0)
+ load_into_model(model, doc)
+ self.assertEqual(len(model), 3)
+
+def suite():
+ suite = TestSuite()
+ suite.addTests(
+ defaultTestLoader.loadTestsFromTestCase(TestJsonLD))
+ return suite
+
+if __name__ == "__main__":
+ from unittest2 import main
+ main(defaultTest='suite')