Mercurial > code > home > repos > homeauto

comparison service/mqtt_to_rdf/inference/lhs_evaluation.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/lhs_evaluation.py@80f4e741ca4f
children
comparison
equal deleted inserted replaced
1726:7d3797ed6681 1727:23e6154e6c11
1 import logging
2 from decimal import Decimal
3 from typing import Dict, Iterator, List, Optional, Type, Union, cast
4
5 from rdflib import Literal, Namespace, URIRef
6 from rdflib.term import Node, Variable
7
8 from inference.candidate_binding import CandidateBinding
9 from inference.inference_types import BindableTerm
10 from inference.stmt_chunk import Chunk
11
12 log = logging.getLogger('infer')
13
14 INDENT = ' '
15
16 ROOM = Namespace("http://projects.bigasterisk.com/room/")
17 LOG = Namespace('http://www.w3.org/2000/10/swap/log#')
18 MATH = Namespace('http://www.w3.org/2000/10/swap/math#')
19
20
21 def _numericNode(n: Node):
22 if not isinstance(n, Literal):
23 raise TypeError(f'expected Literal, got {n=}')
24 val = n.toPython()
25 if not isinstance(val, (int, float, Decimal)):
26 raise TypeError(f'expected number, got {val=}')
27 return val
28
29
30 class Function:
31 """any rule stmt that runs a function (not just a statement match)"""
32 pred: URIRef
33
34 def __init__(self, chunk: Chunk):
35 self.chunk = chunk
36 if chunk.predicate != self.pred:
37 raise TypeError
38
39 def getOperandNodes(self, existingBinding: CandidateBinding) -> List[Node]:
40 raise NotImplementedError
41
42 def getNumericOperands(self, existingBinding: CandidateBinding) -> List[Union[int, float, Decimal]]:
43 out = []
44 for op in self.getOperandNodes(existingBinding):
45 out.append(_numericNode(op))
46
47 return out
48
49 def bind(self, existingBinding: CandidateBinding) -> Optional[CandidateBinding]:
50 """either any new bindings this function makes (could be 0), or None if it doesn't match"""
51 raise NotImplementedError
52
53 def valueInObjectTerm(self, value: Node) -> Optional[CandidateBinding]:
54 objVar = self.chunk.primary[2]
55 if not isinstance(objVar, Variable):
56 raise TypeError(f'expected Variable, got {objVar!r}')
57 return CandidateBinding({cast(BindableTerm, objVar): value})
58
59
60 class SubjectFunction(Function):
61 """function that depends only on the subject term"""
62
63 def getOperandNodes(self, existingBinding: CandidateBinding) -> List[Node]:
64 if self.chunk.primary[0] is None:
65 raise ValueError(f'expected one operand on {self.chunk}')
66 return [existingBinding.applyTerm(self.chunk.primary[0])]
67
68
69 class SubjectObjectFunction(Function):
70 """a filter function that depends on the subject and object terms"""
71
72 def getOperandNodes(self, existingBinding: CandidateBinding) -> List[Node]:
73 if self.chunk.primary[0] is None or self.chunk.primary[2] is None:
74 raise ValueError(f'expected one operand on each side of {self.chunk}')
75 return [existingBinding.applyTerm(self.chunk.primary[0]), existingBinding.applyTerm(self.chunk.primary[2])]
76
77
78 class ListFunction(Function):
79 """function that takes an rdf list as input"""
80
81 def getOperandNodes(self, existingBinding: CandidateBinding) -> List[Node]:
82 if self.chunk.subjList is None:
83 raise ValueError(f'expected subject list on {self.chunk}')
84 return [existingBinding.applyTerm(x) for x in self.chunk.subjList]
85
86
87 _registeredFunctionTypes: List[Type['Function']] = []
88
89
90 def register(cls: Type['Function']):
91 _registeredFunctionTypes.append(cls)
92 return cls
93
94
95 import inference.inference_functions as inference_functions # calls register() on some classes
96
97 _byPred: Dict[URIRef, Type[Function]] = dict((cls.pred, cls) for cls in _registeredFunctionTypes)
98
99
100 def functionsFor(pred: URIRef) -> Iterator[Type[Function]]:
101 try:
102 yield _byPred[pred]
103 except KeyError:
104 return