Mercurial > code > home > repos > homeauto

comparison service/mqtt_to_rdf/candidate_binding.py @ 1618:48bf62008c82

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
attempted to rewrite with CandidateTermMatches but it broke
author drewp@bigasterisk.com
date Wed, 08 Sep 2021 18:32:11 -0700
parents bcfa368e5498
children
comparison
equal deleted inserted replaced
1617:e105032b0e3d 1618:48bf62008c82
1 from dataclasses import dataclass 1 from dataclasses import dataclass
2 from typing import Dict, Iterable, Iterator, Union 2 from typing import Dict, Iterable, Iterator, Optional, Union
3 3
4 from prometheus_client import Summary 4 from prometheus_client import Summary
5 from rdflib import BNode, Graph 5 from rdflib import BNode, Graph
6 from rdflib.term import Node, Variable 6 from rdflib.term import Node, Variable
7 7
14 14
15 def __repr__(self): 15 def __repr__(self):
16 b = " ".join("%s=%s" % (k, v) for k, v in sorted(self.binding.items())) 16 b = " ".join("%s=%s" % (k, v) for k, v in sorted(self.binding.items()))
17 return f'CandidateBinding({b})' 17 return f'CandidateBinding({b})'
18 18
19 def apply(self, g: Union[Graph, Iterable[Triple]]) -> Iterator[Triple]: 19 def apply(self, g: Union[Graph, Iterable[Triple]], returnBoundStatementsOnly=True) -> Iterator[Triple]:
20 for stmt in g: 20 for stmt in g:
21 try: 21 try:
22 bound = (self._applyTerm(stmt[0]), self._applyTerm(stmt[1]), self._applyTerm(stmt[2])) 22 bound = (
23 self._applyTerm(stmt[0], returnBoundStatementsOnly),
24 self._applyTerm(stmt[1], returnBoundStatementsOnly),
25 self._applyTerm(stmt[2], returnBoundStatementsOnly))
23 except BindingUnknown: 26 except BindingUnknown:
24 continue 27 continue
25 yield bound 28 yield bound
26 29
27 def _applyTerm(self, term: Node): 30 def _applyTerm(self, term: Node, failUnbound=True):
28 if isinstance(term, (Variable, BNode)): 31 if isinstance(term, (Variable, BNode)):
29 if term in self.binding: 32 if term in self.binding:
30 return self.binding[term] 33 return self.binding[term]
31 else: 34 else:
32 raise BindingUnknown() 35 if failUnbound:
36 raise BindingUnknown()
33 return term 37 return term
34 38
35 def addNewBindings(self, newBindings: 'CandidateBinding'): 39 def addNewBindings(self, newBindings: 'CandidateBinding'):
36 for k, v in newBindings.binding.items(): 40 for k, v in newBindings.binding.items():
37 if k in self.binding and self.binding[k] != v: 41 # if k in self.binding and self.binding[k] != v:
38 raise ValueError(f'conflict- thought {k} would be {self.binding[k]} but another Evaluation said it should be {v}') 42 # raise ValueError(f'conflict- thought {k} would be {self.binding[k]} but another Evaluation said it should be {v}')
39 self.binding[k] = v 43 self.binding[k] = v