diff service/mqtt_to_rdf/inference/rdflib_debug_patches.py @ 1727:23e6154e6c11

file moves
author drewp@bigasterisk.com
date Tue, 20 Jun 2023 23:26:24 -0700
parents service/mqtt_to_rdf/rdflib_debug_patches.py@37710d28890b
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/service/mqtt_to_rdf/inference/rdflib_debug_patches.py	Tue Jun 20 23:26:24 2023 -0700
@@ -0,0 +1,73 @@
+"""rdflib patches for prettier debug outut"""
+
+import itertools
+
+import rdflib
+import rdflib.plugins.parsers.notation3
+import rdflib.term
+from rdflib import BNode, RDF
+
+ROOM = rdflib.Namespace('http://projects.bigasterisk.com/room/')
+
+ABBREVIATE = {
+    '': ROOM,
+    'rdf': RDF,
+}
+
+
+def patchSlimReprs():
+    """From: rdflib.term.URIRef('foo')
+         To: U('foo')
+    """
+
+    def ur(self):
+        clsName = "U" if self.__class__ is rdflib.term.URIRef else self.__class__.__name__
+        s = super(rdflib.term.URIRef, self).__str__()
+        for short, long in ABBREVIATE.items():
+            if s.startswith(str(long)):
+                s = short + ':' + s[len(str(long)):]
+                break
+
+        return """%s(%s)""" % (clsName, s)
+
+    rdflib.term.URIRef.__repr__ = ur
+
+    def br(self):
+        clsName = "BNode" if self.__class__ is rdflib.term.BNode else self.__class__.__name__
+        return """%s(%s)""" % (clsName, super(rdflib.term.BNode, self).__repr__())
+
+    rdflib.term.BNode.__repr__ = br
+
+    def vr(self):
+        clsName = "V" if self.__class__ is rdflib.term.Variable else self.__class__.__name__
+        return """%s(%s)""" % (clsName, '?' + super(rdflib.term.Variable, self).__str__())
+
+    rdflib.term.Variable.__repr__ = vr
+
+
+def patchBnodeCounter(always=False):
+    """From: rdflib.terms.BNode('ne7bb4a51624993acdf51cc5d4e8add30e1'
+         To: BNode('f-6-1')
+
+    BNode creation can override this, which might matter when adding BNodes that
+    are known to be the same as each other. Set `always` to disregard this and
+    always get short ids.
+    """
+    serial = itertools.count()
+
+    def n(cls, value=None, _sn_gen='', _prefix='') -> BNode:
+        if always or value is None:
+            value = 'N-%s' % next(serial)
+        return rdflib.term.Identifier.__new__(cls, value)
+
+    rdflib.term.BNode.__new__ = n
+
+    def newBlankNode(self, uri=None, why=None):
+        if uri is None:
+            self.counter += 1
+            bn = BNode('f-%s-%s' % (self.number, self.counter))
+        else:
+            bn = BNode(uri.split('#').pop().replace('_', 'b'))
+        return bn
+
+    rdflib.plugins.parsers.notation3.Formula.newBlankNode = newBlankNode