Mercurial > code > home > repos > homeauto
annotate lib/patchablegraph/patchsource.py @ 1651:20474ad4968e
WIP - functions are broken as i move most layers to work in Chunks not Triples
A Chunk is a Triple plus any rdf lists.
author | drewp@bigasterisk.com |
---|---|
date | Sat, 18 Sep 2021 23:57:20 -0700 |
parents | 32939cddf360 |
children |
rev | line source |
---|---|
592 | 1 import logging, time |
302 | 2 import traceback |
3 from rdflib import ConjunctiveGraph | |
4 from rdflib.parser import StringInputSource | |
514 | 5 from twisted.internet import reactor, defer |
302 | 6 |
351
7716b1810d6c
reasoning & collector move into docker images
drewp@bigasterisk.com
parents:
312
diff
changeset
|
7 from rdfdb.patch import Patch |
569 | 8 from twisted_sse.eventsource import EventSource |
9 | |
10 from .patchablegraph import patchFromJson | |
302 | 11 |
312
170dc9b1e789
fix input graph web display by dirtying combinedGraph better.
drewp@bigasterisk.com
parents:
306
diff
changeset
|
12 log = logging.getLogger('fetch') |
302 | 13 |
14 class PatchSource(object): | |
15 """wrap EventSource so it emits Patch objects and has an explicit stop method.""" | |
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
16 def __init__(self, url, agent): |
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
17 self.url = str(url) |
302 | 18 |
19 # add callbacks to these to learn if we failed to connect | |
20 # (approximately) or if the ccnnection was unexpectedly lost | |
21 self.connectionFailed = defer.Deferred() | |
22 self.connectionLost = defer.Deferred() | |
23 | |
24 self._listeners = set() | |
25 log.info('start read from %s', url) | |
592 | 26 self._startReadTime = time.time() |
27 self._patchesReceived = 0 # including fullgraph | |
312
170dc9b1e789
fix input graph web display by dirtying combinedGraph better.
drewp@bigasterisk.com
parents:
306
diff
changeset
|
28 # note: fullGraphReceived isn't guaranteed- the stream could |
170dc9b1e789
fix input graph web display by dirtying combinedGraph better.
drewp@bigasterisk.com
parents:
306
diff
changeset
|
29 # start with patches |
302 | 30 self._fullGraphReceived = False |
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
31 self._eventSource = EventSource(url.toPython().encode('utf8'), |
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
32 userAgent=agent) |
302 | 33 |
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
34 self._eventSource.addEventListener(b'fullGraph', self._onFullGraph) |
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
35 self._eventSource.addEventListener(b'patch', self._onPatch) |
302 | 36 self._eventSource.onerror(self._onError) |
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
37 self._eventSource.onConnectionLost = self._onDisconnect |
306 | 38 |
439
124c921ad52d
stats->state to make room for greplin stats
drewp@bigasterisk.com
parents:
429
diff
changeset
|
39 def state(self): |
306 | 40 return { |
41 'url': self.url, | |
42 'fullGraphReceived': self._fullGraphReceived, | |
592 | 43 'patchesReceived': self._patchesReceived, |
44 'time': { | |
45 'open': getattr(self, '_startReadTime', None), | |
46 'fullGraph': getattr(self, '_fullGraphTime', None), | |
47 'latestPatch': getattr(self, '_latestPatchTime', None), | |
48 }, | |
49 'closed': self._eventSource is None, | |
306 | 50 } |
302 | 51 |
52 def addPatchListener(self, func): | |
53 """ | |
54 func(patch, fullGraph=[true if the patch is the initial fullgraph]) | |
55 """ | |
56 self._listeners.add(func) | |
57 | |
58 def stop(self): | |
59 log.info('stop read from %s', self.url) | |
60 try: | |
61 self._eventSource.protocol.stopProducing() # needed? | |
62 except AttributeError: | |
63 pass | |
64 self._eventSource = None | |
65 | |
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
66 def _onDisconnect(self, reason): |
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
67 log.debug('PatchSource._onDisconnect from %s (%s)', self.url, reason) |
302 | 68 # skip this if we're doing a stop? |
69 self.connectionLost.callback(None) | |
70 | |
71 def _onError(self, msg): | |
72 log.debug('PatchSource._onError from %s %r', self.url, msg) | |
73 if not self._fullGraphReceived: | |
74 self.connectionFailed.callback(msg) | |
75 else: | |
76 self.connectionLost.callback(msg) | |
77 | |
78 def _onFullGraph(self, message): | |
79 try: | |
80 g = ConjunctiveGraph() | |
81 g.parse(StringInputSource(message), format='json-ld') | |
82 p = Patch(addGraph=g) | |
83 self._sendPatch(p, fullGraph=True) | |
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
84 except Exception: |
302 | 85 log.error(traceback.format_exc()) |
86 raise | |
87 self._fullGraphReceived = True | |
592 | 88 self._fullGraphTime = time.time() |
89 self._patchesReceived += 1 | |
302 | 90 |
91 def _onPatch(self, message): | |
92 try: | |
93 p = patchFromJson(message) | |
94 self._sendPatch(p, fullGraph=False) | |
95 except: | |
96 log.error(traceback.format_exc()) | |
97 raise | |
592 | 98 self._latestPatchTime = time.time() |
99 self._patchesReceived += 1 | |
302 | 100 |
101 def _sendPatch(self, p, fullGraph): | |
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
102 log.debug('PatchSource %s received patch %s (fullGraph=%s)', |
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
103 self.url, p.shortSummary(), fullGraph) |
302 | 104 for lis in self._listeners: |
105 lis(p, fullGraph=fullGraph) | |
106 | |
107 def __del__(self): | |
108 if self._eventSource: | |
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
109 raise ValueError("PatchSource wasn't stopped before del") |
302 | 110 |
111 class ReconnectingPatchSource(object): | |
112 """ | |
113 PatchSource api, but auto-reconnects internally and takes listener | |
114 at init time to not miss any patches. You'll get another | |
115 fullGraph=True patch if we have to reconnect. | |
116 | |
117 todo: generate connection stmts in here | |
118 """ | |
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
119 def __init__(self, url, listener, reconnectSecs=60, agent='unset'): |
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
120 # type: (str, Any, Any, str) |
302 | 121 self.url = url |
122 self._stopped = False | |
123 self._listener = listener | |
429
fcd2c026f51e
exportStats for sending scales data to influxdb
drewp@bigasterisk.com
parents:
351
diff
changeset
|
124 self.reconnectSecs = reconnectSecs |
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
125 self.agent = agent |
302 | 126 self._reconnect() |
127 | |
128 def _reconnect(self): | |
129 if self._stopped: | |
130 return | |
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
131 self._ps = PatchSource(self.url, agent=self.agent) |
302 | 132 self._ps.addPatchListener(self._onPatch) |
133 self._ps.connectionFailed.addCallback(self._onConnectionFailed) | |
134 self._ps.connectionLost.addCallback(self._onConnectionLost) | |
135 | |
136 def _onPatch(self, p, fullGraph): | |
137 self._listener(p, fullGraph=fullGraph) | |
306 | 138 |
439
124c921ad52d
stats->state to make room for greplin stats
drewp@bigasterisk.com
parents:
429
diff
changeset
|
139 def state(self): |
306 | 140 return { |
439
124c921ad52d
stats->state to make room for greplin stats
drewp@bigasterisk.com
parents:
429
diff
changeset
|
141 'reconnectedPatchSource': self._ps.state(), |
306 | 142 } |
302 | 143 |
144 def stop(self): | |
145 self._stopped = True | |
146 self._ps.stop() | |
147 | |
148 def _onConnectionFailed(self, arg): | |
429
fcd2c026f51e
exportStats for sending scales data to influxdb
drewp@bigasterisk.com
parents:
351
diff
changeset
|
149 reactor.callLater(self.reconnectSecs, self._reconnect) |
302 | 150 |
151 def _onConnectionLost(self, arg): | |
429
fcd2c026f51e
exportStats for sending scales data to influxdb
drewp@bigasterisk.com
parents:
351
diff
changeset
|
152 reactor.callLater(self.reconnectSecs, self._reconnect) |
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
153 |