homeauto: service/mqtt_to_rdf/inference.py comparison

comparison service/mqtt_to_rdf/inference.py @ 1605:449746d1598f

WIP move evaluation to new file

author	drewp@bigasterisk.com
date	Mon, 06 Sep 2021 01:13:55 -0700
parents	7f8bf68534ed
children	b21885181e35

comparison

equal deleted inserted replaced

-:e78464befd24
+:449746d1598f
 from prometheus_client import Summary
 from rdflib import RDF, BNode, Graph, Literal, Namespace, URIRef
 from rdflib.graph import ConjunctiveGraph, ReadOnlyGraphAggregate
 from rdflib.term import Node, Variable
+from lhs_evaluation import EvaluationFailed, Evaluation
 log = logging.getLogger('infer')
 INDENT = '    '
 Triple = Tuple[Node, Node, Node]
 Rule = Tuple[Graph, Node, Graph]
 MATH = Namespace('http://www.w3.org/2000/10/swap/math#')
 # Graph() makes a BNode if you don't pass
 # identifier, which can be a bottleneck.
 GRAPH_ID = URIRef('dont/care')
-class EvaluationFailed(ValueError):
-"""e.g. we were given (5 math:greaterThan 6)"""
 class BindingUnknown(ValueError):
 """e.g. we were asked to make the bound version
 of (A B ?c) and we don't have a binding for ?c
 if not varsAndBnodesInStmt:
 self.staticRuleStmts.add(ruleStmt)
 self.evaluations = list(Evaluation.findEvals(self.graph))
 def findCandidateBindings(self, workingSet: ReadOnlyWorkingSet) -> Iterator[CandidateBinding]:
 """bindings that fit the LHS of a rule, using statements from workingSet and functions
 from LHS"""
 log.debug(f'{INDENT*3} nodesToBind: {self.lhsBindables}')
 self.stats['findCandidateBindingsCalls'] += 1
 orderedVars, orderedValueSets = _organize(candidateTermMatches)
 self._logCandidates(orderedVars, orderedValueSets)
 log.debug(f'{INDENT*3} trying all permutations:')
 for perm in itertools.product(*orderedValueSets):
 binding = CandidateBinding(dict(zip(orderedVars, perm)))
 log.debug('')
 log.debug(f'{INDENT*4}*trying {binding}')
 log.debug(f'{INDENT*4} {v!r} could be:')
 for val in vals:
 log.debug(f'{INDENT*5}{val!r}')
-class Evaluation:
-"""some lhs statements need to be evaluated with a special function
-(e.g. math) and then not considered for the rest of the rule-firing
-process. It's like they already 'matched' something, so they don't need
-to match a statement from the known-true working set.
-One Evaluation instance is for one function call.
-"""
-@staticmethod
-def findEvals(graph: Graph) -> Iterator['Evaluation']:
-for stmt in graph.triples((None, MATH['sum'], None)):
-operands, operandsStmts = _parseList(graph, stmt[0])
-yield Evaluation(operands, stmt, operandsStmts)
-for stmt in graph.triples((None, MATH['greaterThan'], None)):
-yield Evaluation([stmt[0], stmt[2]], stmt, [])
-for stmt in graph.triples((None, ROOM['asFarenheit'], None)):
-yield Evaluation([stmt[0]], stmt, [])
-# internal, use findEvals
-def __init__(self, operands: List[Node], mainStmt: Triple, otherStmts: Iterable[Triple]) -> None:
-self.operands = operands
-self.operandsStmts = Graph(identifier=GRAPH_ID)
-self.operandsStmts += otherStmts  # may grow
-self.operandsStmts.add(mainStmt)
-self.stmt = mainStmt
-def resultBindings(self, inputBindings) -> Tuple[Dict[BindableTerm, Node], Graph]:
-"""under the bindings so far, what would this evaluation tell us, and which stmts would be consumed from doing so?"""
-pred = self.stmt[1]
-objVar: Node = self.stmt[2]
-boundOperands = []
-for op in self.operands:
-if isinstance(op, Variable):
-try:
-op = inputBindings[op]
-except KeyError:
-return {}, self.operandsStmts
-boundOperands.append(op)
-if pred == MATH['sum']:
-obj = Literal(sum(map(numericNode, boundOperands)))
-if not isinstance(objVar, Variable):
-raise TypeError(f'expected Variable, got {objVar!r}')
-res: Dict[BindableTerm, Node] = {objVar: obj}
-elif pred == ROOM['asFarenheit']:
-if len(boundOperands) != 1:
-raise ValueError(":asFarenheit takes 1 subject operand")
-f = Literal(Decimal(numericNode(boundOperands[0])) * 9 / 5 + 32)
-if not isinstance(objVar, Variable):
-raise TypeError(f'expected Variable, got {objVar!r}')
-res: Dict[BindableTerm, Node] = {objVar: f}
-elif pred == MATH['greaterThan']:
-if not (numericNode(boundOperands[0]) > numericNode(boundOperands[1])):
-raise EvaluationFailed()
-res: Dict[BindableTerm, Node] = {}
-else:
-raise NotImplementedError(repr(pred))
-return res, self.operandsStmts
-def numericNode(n: Node):
-if not isinstance(n, Literal):
-raise TypeError(f'expected Literal, got {n=}')
-val = n.toPython()
-if not isinstance(val, (int, float, Decimal)):
-raise TypeError(f'expected number, got {val=}')
-return val
 class Inference:
 def __init__(self) -> None:
 self.rules = ConjunctiveGraph()
 """
 returns new graph of inferred statements.
 """
 log.info(f'{INDENT*0} Begin inference of graph len={graph.__len__()} with rules len={len(self.rules)}:')
 startTime = time.time()
-self.stats: Dict[str, Union[int,float]] = defaultdict(lambda: 0)
+self.stats: Dict[str, Union[int, float]] = defaultdict(lambda: 0)
 # everything that is true: the input graph, plus every rule conclusion we can make
 workingSet = Graph()
 workingSet += graph
 # just the statements that came from RHS's of rules that fired.
 log.debug(f'{INDENT*5} adding {newStmt=}')
 workingSet.add(newStmt)
 implied.add(newStmt)
-def _parseList(graph, subj) -> Tuple[List[Node], Set[Triple]]:
-""""Do like Collection(g, subj) but also return all the
-triples that are involved in the list"""
-out = []
-used = set()
-cur = subj
-while cur != RDF.nil:
-out.append(graph.value(cur, RDF.first))
-used.add((cur, RDF.first, out[-1]))
-next = graph.value(cur, RDF.rest)
-used.add((cur, RDF.rest, next))
-cur = next
-return out, used
 def graphDump(g: Union[Graph, List[Triple]]):
 if not isinstance(g, Graph):
 g2 = Graph()
 g2 += g
 g = g2

Mercurial > code > home > repos > homeauto

comparison service/mqtt_to_rdf/inference.py @ 1605:449746d1598f