Mercurial > code > home > repos > homeauto

comparison service/mqtt_to_rdf/inference/candidate_binding.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/candidate_binding.py@3cf7f313b285
children
comparison
equal deleted inserted replaced
1726:7d3797ed6681 1727:23e6154e6c11
1 import logging
2 from dataclasses import dataclass
3 from typing import Dict, Iterable, Iterator, Union
4
5 from rdflib import Graph
6 from rdflib.term import Node, Variable
7
8 from inference.inference_types import (BindableTerm, BindingUnknown, RuleUnboundBnode, Triple)
9
10 log = logging.getLogger('cbind')
11 INDENT = ' '
12
13
14 class BindingConflict(ValueError): # might be the same as `Inconsistent`
15 pass
16
17
18 @dataclass
19 class CandidateBinding:
20 binding: Dict[BindableTerm, Node]
21
22 def __post_init__(self):
23 for n in self.binding.values():
24 if isinstance(n, RuleUnboundBnode):
25 raise TypeError(repr(self))
26
27 def __repr__(self):
28 b = " ".join("%r=%r" % (var, value) for var, value in sorted(self.binding.items()))
29 return f'CandidateBinding({b})'
30
31 def key(self):
32 """note this is only good for the current value, and self.binding is mutable"""
33 return tuple(sorted(self.binding.items()))
34
35 def apply(self, g: Union[Graph, Iterable[Triple]], returnBoundStatementsOnly=True) -> Iterator[Triple]:
36 for stmt in g:
37 try:
38 bound = (
39 self.applyTerm(stmt[0], returnBoundStatementsOnly), #
40 self.applyTerm(stmt[1], returnBoundStatementsOnly), #
41 self.applyTerm(stmt[2], returnBoundStatementsOnly))
42 except BindingUnknown:
43 if log.isEnabledFor(logging.DEBUG):
44 log.debug(f'{INDENT*7} CB.apply cant bind {stmt} using {self.binding}')
45
46 continue
47 if log.isEnabledFor(logging.DEBUG):
48 log.debug(f'{INDENT*7} CB.apply took {stmt} to {bound}')
49
50 yield bound
51
52 def applyTerm(self, term: Node, failUnbound=True):
53 if isinstance(term, (Variable, RuleUnboundBnode)):
54 if term in self.binding:
55 return self.binding[term]
56 else:
57 if failUnbound:
58 raise BindingUnknown()
59 return term
60
61 def addNewBindings(self, newBindings: 'CandidateBinding'):
62 for k, v in newBindings.binding.items():
63 if k in self.binding and self.binding[k] != v:
64 raise BindingConflict(f'thought {k} would be {self.binding[k]} but another Evaluation said it should be {v}')
65 self.binding[k] = v
66
67 def copy(self):
68 return CandidateBinding(self.binding.copy())
69
70 def contains(self, term: BindableTerm):
71 return term in self.binding