Mercurial > code > home > repos > homeauto

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

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
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
comparison
equal deleted inserted replaced
1726:7d3797ed6681 1727:23e6154e6c11
1 """rdflib patches for prettier debug outut"""
2
3 import itertools
4
5 import rdflib
6 import rdflib.plugins.parsers.notation3
7 import rdflib.term
8 from rdflib import BNode, RDF
9
10 ROOM = rdflib.Namespace('http://projects.bigasterisk.com/room/')
11
12 ABBREVIATE = {
13 '': ROOM,
14 'rdf': RDF,
15 }
16
17
18 def patchSlimReprs():
19 """From: rdflib.term.URIRef('foo')
20 To: U('foo')
21 """
22
23 def ur(self):
24 clsName = "U" if self.__class__ is rdflib.term.URIRef else self.__class__.__name__
25 s = super(rdflib.term.URIRef, self).__str__()
26 for short, long in ABBREVIATE.items():
27 if s.startswith(str(long)):
28 s = short + ':' + s[len(str(long)):]
29 break
30
31 return """%s(%s)""" % (clsName, s)
32
33 rdflib.term.URIRef.__repr__ = ur
34
35 def br(self):
36 clsName = "BNode" if self.__class__ is rdflib.term.BNode else self.__class__.__name__
37 return """%s(%s)""" % (clsName, super(rdflib.term.BNode, self).__repr__())
38
39 rdflib.term.BNode.__repr__ = br
40
41 def vr(self):
42 clsName = "V" if self.__class__ is rdflib.term.Variable else self.__class__.__name__
43 return """%s(%s)""" % (clsName, '?' + super(rdflib.term.Variable, self).__str__())
44
45 rdflib.term.Variable.__repr__ = vr
46
47
48 def patchBnodeCounter(always=False):
49 """From: rdflib.terms.BNode('ne7bb4a51624993acdf51cc5d4e8add30e1'
50 To: BNode('f-6-1')
51
52 BNode creation can override this, which might matter when adding BNodes that
53 are known to be the same as each other. Set `always` to disregard this and
54 always get short ids.
55 """
56 serial = itertools.count()
57
58 def n(cls, value=None, _sn_gen='', _prefix='') -> BNode:
59 if always or value is None:
60 value = 'N-%s' % next(serial)
61 return rdflib.term.Identifier.__new__(cls, value)
62
63 rdflib.term.BNode.__new__ = n
64
65 def newBlankNode(self, uri=None, why=None):
66 if uri is None:
67 self.counter += 1
68 bn = BNode('f-%s-%s' % (self.number, self.counter))
69 else:
70 bn = BNode(uri.split('#').pop().replace('_', 'b'))
71 return bn
72
73 rdflib.plugins.parsers.notation3.Formula.newBlankNode = newBlankNode