|
1080
|
1 import logging, time
|
|
|
2
|
|
|
3 from rdflib import Graph, ConjunctiveGraph
|
|
|
4 from rdflib import Namespace, URIRef, Literal, RDF
|
|
1086
|
5 from rdflib.parser import StringInputSource
|
|
1080
|
6
|
|
|
7 from twisted.python.filepath import FilePath
|
|
|
8 from twisted.internet.defer import inlineCallbacks, gatherResults
|
|
|
9
|
|
|
10 from rdflibtrig import addTrig
|
|
|
11 from graphop import graphEqual
|
|
|
12
|
|
|
13 log = logging.getLogger('fetch')
|
|
|
14
|
|
|
15 ROOM = Namespace("http://projects.bigasterisk.com/room/")
|
|
|
16 DEV = Namespace("http://projects.bigasterisk.com/device/")
|
|
|
17
|
|
|
18
|
|
1086
|
19 def parseRdf(text, contentType):
|
|
|
20 g = Graph()
|
|
|
21 g.parse(StringInputSource(text), format={
|
|
|
22 'text/n3': 'n3',
|
|
|
23 }[contentType])
|
|
|
24 return g
|
|
|
25
|
|
|
26
|
|
1080
|
27 class InputGraph(object):
|
|
|
28 def __init__(self, inputDirs, onChange, sourceSubstr=None):
|
|
|
29 """
|
|
|
30 this has one Graph that's made of:
|
|
|
31 - all .n3 files from inputDirs (read at startup)
|
|
|
32 - all the remote graphs, specified in the file graphs
|
|
|
33
|
|
|
34 call updateFileData or updateRemoteData to reread those
|
|
|
35 graphs. getGraph to access the combined graph.
|
|
|
36
|
|
|
37 onChange(self) is called if the contents of the full graph
|
|
|
38 change (in an interesting way) during updateFileData or
|
|
|
39 updateRemoteData. Interesting means statements other than the
|
|
|
40 ones with the predicates on the boring list. onChange(self,
|
|
|
41 oneShot=True) means: don't store the result of this change
|
|
|
42 anywhere; it needs to be processed only once
|
|
|
43
|
|
|
44 sourceSubstr filters to only pull from sources containing the
|
|
|
45 string (for debugging).
|
|
|
46 """
|
|
|
47 self.inputDirs = inputDirs
|
|
|
48 self.onChange = onChange
|
|
|
49 self.sourceSubstr = sourceSubstr
|
|
|
50 self._fileGraph = Graph()
|
|
|
51 self._remoteGraph = None
|
|
|
52 self._combinedGraph = None
|
|
|
53 self._oneShotAdditionGraph = None
|
|
|
54 self._lastErrLog = {} # source: error
|
|
|
55
|
|
|
56 def updateFileData(self):
|
|
|
57 """
|
|
|
58 make sure we contain the correct data from the files in inputDirs
|
|
|
59 """
|
|
|
60 # this sample one is actually only needed for the output, but I don't
|
|
|
61 # think I want to have a separate graph for the output
|
|
|
62 # handling
|
|
|
63 log.debug("read file graphs")
|
|
|
64 for fp in FilePath("input").walk():
|
|
|
65 if fp.isdir():
|
|
|
66 continue
|
|
|
67 if fp.splitext()[1] != '.n3':
|
|
|
68 continue
|
|
|
69 log.debug("read %s", fp)
|
|
|
70 # todo: if this fails, leave the report in the graph
|
|
|
71 self._fileGraph.parse(fp.open(), format="n3")
|
|
|
72 self._combinedGraph = None
|
|
|
73
|
|
|
74 self.onChange(self)
|
|
|
75
|
|
|
76 @inlineCallbacks
|
|
|
77 def updateRemoteData(self):
|
|
|
78 """
|
|
|
79 read all remote graphs (which are themselves enumerated within
|
|
|
80 the file data)
|
|
|
81 """
|
|
|
82 t1 = time.time()
|
|
|
83 log.debug("read remote graphs")
|
|
|
84 g = ConjunctiveGraph()
|
|
|
85
|
|
|
86 @inlineCallbacks
|
|
|
87 def fetchOne(source):
|
|
|
88 try:
|
|
|
89 fetchTime = yield addTrig(g, source, timeout=5)
|
|
|
90 except Exception, e:
|
|
|
91 e = str(e)
|
|
|
92 if self._lastErrLog.get(source) != e:
|
|
|
93 log.error(" can't add source %s: %s", source, e)
|
|
|
94 self._lastErrLog[source] = e
|
|
|
95 g.add((URIRef(source), ROOM['graphLoadError'], Literal(e)))
|
|
|
96 g.add((URIRef(source), RDF.type, ROOM['FailedGraphLoad']))
|
|
|
97 else:
|
|
|
98 if self._lastErrLog.get(source):
|
|
|
99 log.warning(" source %s is back", source)
|
|
|
100 self._lastErrLog[source] = None
|
|
|
101 g.add((URIRef(source), ROOM['graphLoadMs'],
|
|
|
102 Literal(round(fetchTime * 1000, 1))))
|
|
|
103
|
|
|
104 fetchDone = []
|
|
|
105 filtered = 0
|
|
|
106 for source in self._fileGraph.objects(ROOM['reasoning'],
|
|
|
107 ROOM['source']):
|
|
|
108 if self.sourceSubstr and self.sourceSubstr not in source:
|
|
|
109 filtered += 1
|
|
|
110 continue
|
|
|
111 fetchDone.append(fetchOne(source))
|
|
|
112 yield gatherResults(fetchDone, consumeErrors=True)
|
|
|
113 log.debug("loaded %s (skipping %s) in %.1f ms", len(fetchDone),
|
|
|
114 filtered, 1000 * (time.time() - t1))
|
|
|
115
|
|
|
116 prevGraph = self._remoteGraph
|
|
|
117 self._remoteGraph = g
|
|
|
118 self._combinedGraph = None
|
|
|
119 if (prevGraph is None or
|
|
|
120 not graphEqual(g, prevGraph, ignorePredicates=[
|
|
|
121 ROOM['signalStrength'],
|
|
|
122 # perhaps anything with a number-datatype for its
|
|
|
123 # object should be filtered out, and you have to make
|
|
|
124 # an upstream quantization (e.g. 'temp high'/'temp
|
|
|
125 # low') if you want to do reasoning on the difference
|
|
|
126 URIRef("http://bigasterisk.com/map#lastSeenAgoSec"),
|
|
|
127 URIRef("http://bigasterisk.com/map#lastSeenAgo"),
|
|
|
128 ROOM['usingPower'],
|
|
|
129 ROOM['idleTimeMinutes'],
|
|
|
130 ROOM['idleTimeMs'],
|
|
|
131 ROOM['graphLoadMs'],
|
|
|
132 ROOM['localTimeToSecond'],
|
|
|
133 ROOM['history'],
|
|
|
134 ROOM['temperatureF'],
|
|
|
135 ROOM['connectedAgo'],
|
|
|
136 ])):
|
|
|
137 log.debug(" remote graph changed")
|
|
|
138 self.onChange(self)
|
|
|
139 else:
|
|
|
140 log.debug(" remote graph has no changes to trigger rules")
|
|
|
141
|
|
|
142 def addOneShot(self, g):
|
|
|
143 """
|
|
|
144 add this graph to the total, call onChange, and then revert
|
|
|
145 the addition of this graph
|
|
|
146 """
|
|
|
147 self._oneShotAdditionGraph = g
|
|
|
148 self._combinedGraph = None
|
|
|
149 try:
|
|
|
150 self.onChange(self, oneShot=True, oneShotGraph=g)
|
|
|
151 finally:
|
|
|
152 self._oneShotAdditionGraph = None
|
|
|
153 self._combinedGraph = None
|
|
|
154
|
|
1086
|
155 def addOneShotFromString(self, body, contentType):
|
|
|
156 g = parseRdf(body, contentType)
|
|
|
157 if not len(g):
|
|
|
158 log.warn("incoming oneshot graph had no statements: %r", body)
|
|
|
159 return 0
|
|
|
160 t1 = time.time()
|
|
|
161 self.addOneShot(g)
|
|
|
162 return time.time() - t1
|
|
|
163
|
|
1080
|
164 def getGraph(self):
|
|
|
165 """rdflib Graph with the file+remote contents of the input graph"""
|
|
|
166 # this could be much faster with the combined readonly graph
|
|
|
167 # view from rdflib
|
|
|
168 if self._combinedGraph is None:
|
|
|
169 self._combinedGraph = Graph()
|
|
|
170 if self._fileGraph:
|
|
|
171 for s in self._fileGraph:
|
|
|
172 self._combinedGraph.add(s)
|
|
|
173 if self._remoteGraph:
|
|
|
174 for s in self._remoteGraph:
|
|
|
175 self._combinedGraph.add(s)
|
|
|
176 if self._oneShotAdditionGraph:
|
|
|
177 for s in self._oneShotAdditionGraph:
|
|
|
178 self._combinedGraph.add(s)
|
|
|
179
|
|
|
180 return self._combinedGraph
|
