Mercurial > code > home > repos > homeauto

comparison service/mqtt_to_rdf/candidate_binding.py @ 1607:b21885181e35

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
more modules, types. Maybe less repeated computation on BoundLhs
author drewp@bigasterisk.com
date Mon, 06 Sep 2021 15:38:48 -0700
parents
children bcfa368e5498
comparison
equal deleted inserted replaced
1606:6cf39d43fd40 1607:b21885181e35
1 from dataclasses import dataclass
2 from typing import Dict, Iterator
3
4 from prometheus_client import Summary
5 from rdflib import BNode, Graph
6 from rdflib.term import Node, Variable
7
8 from inference_types import BindableTerm, BindingUnknown, Triple
9
10
11 @dataclass
12 class CandidateBinding:
13 binding: Dict[BindableTerm, Node]
14
15 def __repr__(self):
16 b = " ".join("%s=%s" % (k, v) for k, v in sorted(self.binding.items()))
17 return f'CandidateBinding({b})'
18
19 def apply(self, g: Graph) -> Iterator[Triple]:
20 for stmt in g:
21 try:
22 bound = (self._applyTerm(stmt[0]), self._applyTerm(stmt[1]), self._applyTerm(stmt[2]))
23 except BindingUnknown:
24 continue
25 yield bound
26
27 def _applyTerm(self, term: Node):
28 if isinstance(term, (Variable, BNode)):
29 if term in self.binding:
30 return self.binding[term]
31 else:
32 raise BindingUnknown()
33 return term
34
35 def addNewBindings(self, newBindings: 'CandidateBinding'):
36 for k, v in newBindings.binding.items():
37 if k in self.binding and self.binding[k] != v:
38 raise ValueError(
39 f'conflict- thought {k} would be {self.binding[k]} but another Evaluation said it should be {v}')
40 self.binding[k] = v