homeauto: service/mqtt_to_rdf/inference.py comparison

comparison service/mqtt_to_rdf/inference.py @ 1600:89a50242cb5e

cleanup internal names, imports

author	drewp@bigasterisk.com
date	Sun, 05 Sep 2021 22:50:15 -0700
parents	abbf0eb0e640
children	30463df12d89

comparison

equal deleted inserted replaced

-:abbf0eb0e640
+:89a50242cb5e
 def verify(self, lhs: 'Lhs', workingSet: ReadOnlyWorkingSet, usedByFuncs: Graph) -> bool:
 """Can this lhs be true all at once in workingSet? Does it match with these bindings?"""
 boundLhs = list(self.apply(lhs.graph))
 boundUsedByFuncs = list(self.apply(usedByFuncs))
-self.logVerifyBanner(boundLhs, workingSet, boundUsedByFuncs)
+self._logVerifyBanner(boundLhs, workingSet, boundUsedByFuncs)
 for stmt in boundLhs:
 log.debug(f'{INDENT*4} check for {stmt}')
 if stmt in boundUsedByFuncs:
 else:
 log.debug(f'{INDENT*5} stmt not known to be true')
 return False
 return True
-def logVerifyBanner(self, boundLhs, workingSet: ReadOnlyWorkingSet, boundUsedByFuncs):
+def _logVerifyBanner(self, boundLhs, workingSet: ReadOnlyWorkingSet, boundUsedByFuncs):
 if not log.isEnabledFor(logging.DEBUG):
 return
 log.debug(f'{INDENT*4}/ verify all bindings against this boundLhs:')
 for stmt in sorted(boundLhs):
 log.debug(f'{INDENT*4}|{INDENT} {stmt}')
 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}')
-if not self.allStaticStatementsMatch(workingSet):
+if not self._allStaticStatementsMatch(workingSet):
 return
-candidateTermMatches: Dict[BindableTerm, Set[Node]] = self.allCandidateTermMatches(workingSet)
+candidateTermMatches: Dict[BindableTerm, Set[Node]] = self._allCandidateTermMatches(workingSet)
-orderedVars, orderedValueSets = organize(candidateTermMatches)
+orderedVars, orderedValueSets = _organize(candidateTermMatches)
-self.logCandidates(orderedVars, orderedValueSets)
+self._logCandidates(orderedVars, orderedValueSets)
 log.debug(f'{INDENT*3} trying all permutations:')
 for perm in itertools.product(*orderedValueSets):
 binding = CandidateBinding(dict(zip(orderedVars, perm)))
 if not binding.verify(self, workingSet, usedByFuncs):
 log.debug(f'{INDENT*4} this binding did not verify')
 continue
 yield binding
-def allStaticStatementsMatch(self, workingSet: ReadOnlyWorkingSet) -> bool:
+def _allStaticStatementsMatch(self, workingSet: ReadOnlyWorkingSet) -> bool:
 for ruleStmt in self.staticRuleStmts:
 if ruleStmt not in workingSet:
 log.debug(f'{INDENT*3} {ruleStmt} not in working set- skip rule')
 return False
 return True
-def allCandidateTermMatches(self, workingSet: ReadOnlyWorkingSet) -> Dict[BindableTerm, Set[Node]]:
+def _allCandidateTermMatches(self, workingSet: ReadOnlyWorkingSet) -> Dict[BindableTerm, Set[Node]]:
 """the total set of terms each variable could possibly match"""
 candidateTermMatches: Dict[BindableTerm, Set[Node]] = defaultdict(set)
 for lhsStmt in self.graph:
 log.debug(f'{INDENT*4} possibles for this lhs stmt: {lhsStmt}')
 for stmt in self.graph:
 if stmt not in usedByFuncs:
 g.add(stmt)
 return g
-def logCandidates(self, orderedVars, orderedValueSets):
+def _logCandidates(self, orderedVars, orderedValueSets):
 if not log.isEnabledFor(logging.DEBUG):
 return
 log.debug(f'{INDENT*3} resulting candidate terms:')
 for v, vals in zip(orderedVars, orderedValueSets):
 log.debug(f'{INDENT*4} {v!r} could be:')
 """
 @staticmethod
 def findEvals(graph: Graph) -> Iterator['Evaluation']:
 for stmt in graph.triples((None, MATH['sum'], None)):
-operands, operandsStmts = parseList(graph, stmt[0])
+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, [])
 log.info(f'{INDENT*1} {st}')
 return implied
 def _iterateAllRules(self, workingSet: Graph, implied: Graph):
 for i, r in enumerate(self.rules):
-self.logRuleApplicationHeader(workingSet, i, r)
+self._logRuleApplicationHeader(workingSet, i, r)
-applyRule(Lhs(r[0]), r[2], workingSet, implied)
+_applyRule(Lhs(r[0]), r[2], workingSet, implied)
-def logRuleApplicationHeader(self, workingSet, i, r):
+def _logRuleApplicationHeader(self, workingSet, i, r):
 if not log.isEnabledFor(logging.DEBUG):
 return
 log.debug('')
 log.debug(f'{INDENT*2} workingSet:')
 log.debug(f'{INDENT*2}-applying rule {i}')
 log.debug(f'{INDENT*3} rule def lhs: {graphDump(r[0])}')
 log.debug(f'{INDENT*3} rule def rhs: {graphDump(r[2])}')
-def applyRule(lhs: Lhs, rhs: Graph, workingSet: Graph, implied: Graph):
+def _applyRule(lhs: Lhs, rhs: Graph, workingSet: Graph, implied: Graph):
 for binding in lhs.findCandidateBindings(ReadOnlyGraphAggregate([workingSet])):
 log.debug(f'{INDENT*3} rule has a working binding:')
 for lhsBoundStmt in binding.apply(lhs.graphWithoutEvals(binding)):
 log.debug(f'{INDENT*5} adding {lhsBoundStmt=}')
 log.debug(f'{INDENT*5} adding {newStmt=}')
 workingSet.add(newStmt)
 implied.add(newStmt)
-def parseList(graph, subj) -> Tuple[List[Node], Set[Triple]]:
+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
 lines = cast(bytes, g.serialize(format='n3')).decode('utf8').splitlines()
 lines = [line for line in lines if not line.startswith('@prefix')]
 return ' '.join(lines)
-def organize(candidateTermMatches: Dict[BindableTerm, Set[Node]]) -> Tuple[List[BindableTerm], List[List[Node]]]:
+def _organize(candidateTermMatches: Dict[BindableTerm, Set[Node]]) -> Tuple[List[BindableTerm], List[List[Node]]]:
 items = list(candidateTermMatches.items())
 items.sort()
 orderedVars: List[BindableTerm] = []
 orderedValueSets: List[List[Node]] = []
 for v, vals in items:

Mercurial > code > home > repos > homeauto

comparison service/mqtt_to_rdf/inference.py @ 1600:89a50242cb5e