homeauto: service/mqtt_to_rdf/mqtt_to

comparison service/mqtt_to_rdf/mqtt_to_rdf.py @ 1706:2085ed9cfcc4

reworking UI to reflect the new inferencing code

author	drewp@bigasterisk.com
date	Sat, 23 Oct 2021 13:22:40 -0700
parents	34eb87f68ab8
children	7d3797ed6681

comparison

equal deleted inserted replaced

-:250f4c27d56f
+:2085ed9cfcc4
 from mqtt_message import graphFromMessage
 import os
 import time
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Callable, Sequence, Set, Tuple, Union, cast
+from typing import Callable, Set, Tuple, Union, cast
 import cyclone.sse
 import cyclone.web
 import export_to_influxdb
 import prometheus_client
 from mqtt_client import MqttClient
 from patchablegraph import (CycloneGraphEventsHandler, CycloneGraphHandler, PatchableGraph)
 from prometheus_client import Counter, Gauge, Histogram, Summary
 from prometheus_client.exposition import generate_latest
 from prometheus_client.registry import REGISTRY
-from rdfdb.rdflibpatch import graphFromQuads
 from rdflib import RDF, XSD, Graph, Literal, Namespace, URIRef
 from rdflib.graph import ConjunctiveGraph
 from rdflib.term import Node
-from rx.core import Observable
+from rx.core.observable.observable import Observable
 from rx.core.typing import Mapper
 from standardservice.logsetup import log, verboseLogging
 from twisted.internet import reactor, task
 from inference import Inference
 from button_events import button_events
 pass
 class MqttStatementSource:
-def __init__(self, uri: URIRef, config: Graph, masterGraph: PatchableGraph, mqtt, internalMqtt, debugPageData,
+def __init__(self, uri: URIRef, topic: bytes, masterGraph: PatchableGraph, mqtt, internalMqtt, debugPageData,
-influxExport: InfluxExporter, inference: Inference):
+# influxExport: InfluxExporter,
+inference: Inference):
 self.uri = uri
-self.config = config
 self.masterGraph = masterGraph
 self.debugPageData = debugPageData
 self.mqtt = mqtt  # deprecated
 self.internalMqtt = internalMqtt
-self.influxExport = influxExport
+# self.influxExport = influxExport
 self.inference = inference
-self.mqttTopic = self.topicFromConfig(self.config)
+self.mqttTopic = topic
 if self.mqttTopic == b'':
 raise EmptyTopicError(f"empty topic for {uri=}")
 log.debug(f'new mqttTopic {self.mqttTopic}')
 self.debugSub = {
 rawBytes.subscribe(on_next=self.countIncomingMessage)
 rawBytes.subscribe_(self.onMessage)
 def onMessage(self, raw: bytes):
-g = graphFromMessage(self.mqttTopic, raw)
+g = graphFromMessage(self.uri, self.mqttTopic, raw)
 logGraph(log.debug, 'message graph', g)
 appendLimit(
 self.debugSub['recentMessageGraphs'],
 {  #
 't': truncTime(),
 'n3': serializeWithNs(g, hidePrefixes=True)
 })
 implied = self.inference.infer(g)
 self.updateMasterGraph(implied)
-def topicFromConfig(self, config) -> bytes:
-topicParts = list(config.items(config.value(self.uri, ROOM['mqttTopic'])))
-return b'/'.join(t.encode('ascii') for t in topicParts)
 def subscribeMqtt(self, topic: bytes):
 # goal is to get everyone on the internal broker and eliminate this
 mqtt = self.internalMqtt if topic.startswith(b'frontdoorlock') else self.mqtt
 return mqtt.subscribe(topic)
 def unbind(self):
 self.loop.stop()
-@dataclass
+# @dataclass
-class WatchFiles:
+# class WatchFiles:
-# could be merged with rdfdb.service and GraphFile code
+#     # could be merged with rdfdb.service and GraphFile code
-globPattern: str
+#     globPattern: str
-outGraph: Graph
+#     outGraph: Graph
-def __post_init__(self):
+#     def __post_init__(self):
-self.lastUpdate = 0
+#         self.lastUpdate = 0
-task.LoopingCall(self.refresh).start(1)
+#         task.LoopingCall(self.refresh).start(1)
-log.info(f'start watching {self.globPattern}')
+#         log.info(f'start watching {self.globPattern}')
-def refresh(self):
+#     def refresh(self):
-files = glob.glob(self.globPattern)
+#         files = glob.glob(self.globPattern)
-for fn in files:
+#         for fn in files:
-if os.path.getmtime(fn) > self.lastUpdate:
+#             if os.path.getmtime(fn) > self.lastUpdate:
-break
+#                 break
-else:
+#         else:
-return
+#             return
-self.lastUpdate = time.time()
+#         self.lastUpdate = time.time()
-self.outGraph.remove((None, None, None))
+#         self.outGraph.remove((None, None, None))
-log.info('reread config')
+#         log.info('reread config')
-for fn in files:
+#         for fn in files:
-# todo: handle read errors
+#             # todo: handle read errors
-self.outGraph.parse(fn, format='n3')
+#             self.outGraph.parse(fn, format='n3')
-# and notify this change,so we can recalc the latest output
+#         # and notify this change,so we can recalc the latest output
+class RunState:
+"""this is rebuilt upon every config reload"""
+def __init__(self,
+expandedConfigPatchableCopy: PatchableGraph,  # for output and display
+masterGraph: PatchableGraph,  # current sensor outputs
+mqtt: MqttClient,
+internalMqtt: MqttClient,
+#  influxExport: InfluxExporter,
+inference: Inference):
+loadedConfig = ConjunctiveGraph()
+loadedConfig.parse('conf/rules.n3', format='n3')
+inference.setRules(loadedConfig)
+self.expandedConfig = inference.infer(loadedConfig)
+self.expandedConfig += inference.nonRuleStatements()
+ecWithQuads = ConjunctiveGraph()
+for s, p, o in self.expandedConfig:
+ecWithQuads.add((s, p, o, URIRef('/config')))
+expandedConfigPatchableCopy.setToGraph(ecWithQuads)
+self.srcs = []
+for src in sorted(self.expandedConfig.subjects(RDF.type, ROOM['MqttStatementSource'])):
+log.info(f'setup source {src=}')
+try:
+self.srcs.append(
+MqttStatementSource(src, self.topicFromConfig(self.expandedConfig, src),
+masterGraph, mqtt=mqtt, internalMqtt=internalMqtt, debugPageData=debugPageData,
+# influxExport=influxExport,
+inference=inference))
+except EmptyTopicError:
+continue
+log.info(f'set up {len(self.srcs)} sources')
+def topicFromConfig(self, config, src) -> bytes:
+topicParts = list(config.items(config.value(src, ROOM['mqttTopic'])))
+return b'/'.join(t.encode('ascii') for t in topicParts)
 if __name__ == '__main__':
 arg = docopt("""
 Usage: mqtt_to_rdf.py [options]
 logging.getLogger('infer').setLevel(logging.INFO)
 logging.getLogger('cbind').setLevel(logging.INFO)
 # log.setLevel(logging.DEBUG)
 log.info('log start')
-config = ConjunctiveGraph()
+# config = ConjunctiveGraph()
-watcher = WatchFiles('conf/rules.n3', config)
+# watcher = WatchFiles('conf/rules.n3', config)
-# for fn in Path('.').glob('conf/*.n3'):
-#     if not arg['--cs'] or str(arg['--cs']) in str(fn):
-#         log.debug(f'loading {fn}')
-#         config.parse(str(fn), format='n3')
-#     else:
-#         log.debug(f'skipping {fn}')
 masterGraph = PatchableGraph()
+inference = Inference()
 brokerHost = 'mosquitto-frontdoor.default.svc.cluster.local'
 brokerPort = 10210
+mqtt = MqttClient(clientId='mqtt_to_rdf', brokerHost='mosquitto-ext.default.svc.cluster.local', brokerPort=1883)  # deprecated
+internalMqtt = MqttClient(clientId='mqtt_to_rdf', brokerHost=brokerHost, brokerPort=brokerPort)
+# needs rework since the config can change:
+# influxExport = InfluxExporter(config, influxHost=os.environ['INFLUXDB_SERVICE_HOST'])
 debugPageData = {
 # schema in index.ts
 'server': f'{brokerHost}:{brokerPort}',
 'messagesSeen': 0,
 'subscribed': [],
+"rules": "",
+"rulesInferred": "",
 }
-inference = Inference()
+expandedConfigPatchableCopy = PatchableGraph()
-inference.setRules(config)
-expandedConfig = inference.infer(config)
+runState = RunState(expandedConfigPatchableCopy, masterGraph, mqtt, internalMqtt, inference)
-expandedConfig += inference.nonRuleStatements()
-log.info('expanded config:')
-for stmt in sorted(expandedConfig):
-log.info(f'  {stmt}')
-mqtt = MqttClient(clientId='mqtt_to_rdf', brokerHost='mosquitto-ext.default.svc.cluster.local', brokerPort=1883)  # deprecated
-internalMqtt = MqttClient(clientId='mqtt_to_rdf', brokerHost=brokerHost, brokerPort=brokerPort)
-influxExport = InfluxExporter(config, influxHost=os.environ['INFLUXDB_SERVICE_HOST'])
-# this needs to be part of the config reload. Maybe GraphFile patch output would be better?
-srcs = []
-for src in sorted(expandedConfig.subjects(RDF.type, ROOM['MqttStatementSource'])):
-log.info(f'setup source {src=}')
-try:
-srcs.append(
-MqttStatementSource(src,
-expandedConfig,
-masterGraph,
-mqtt=mqtt,
-internalMqtt=internalMqtt,
-debugPageData=debugPageData,
-influxExport=influxExport,
-inference=inference))
-except EmptyTopicError:
-continue
-log.info(f'set up {len(srcs)} sources')
-peg = PatchableGraph()
-peg.patch(Patch(addQuads=[(s,p,o,URIRef('/config')) for s,p,o in expandedConfig]))
 port = 10018
 reactor.listenTCP(port,
 cyclone.web.Application([
-(r"/()", cyclone.web.StaticFileHandler, {
+(r"/()", cyclone.web.StaticFileHandler, {"path": ".", "default_filename": "index.html"}),
-"path": ".",
+(r"/build/(bundle.js)", cyclone.web.StaticFileHandler, {"path": "build"}),
-"default_filename": "index.html"
+(r"/graph/config", CycloneGraphHandler, {'masterGraph': expandedConfigPatchableCopy}),
-}),
+(r"/graph/mqtt", CycloneGraphHandler, {'masterGraph': masterGraph}),
-(r"/build/(bundle.js)", cyclone.web.StaticFileHandler, {
+(r"/graph/mqtt/events", CycloneGraphEventsHandler, {'masterGraph': masterGraph}),
-"path": "build"
-}),
-(r"/graph/config", CycloneGraphHandler, {
-'masterGraph': peg,
-}),
-(r"/graph/mqtt", CycloneGraphHandler, {
-'masterGraph': masterGraph
-}),
-(r"/graph/mqtt/events", CycloneGraphEventsHandler, {
-'masterGraph': masterGraph
-}),
 (r'/debugPageData', DebugPageData),
 (r'/metrics', Metrics),
 ],
-mqtt=mqtt,
+#   mqtt=mqtt,
-internalMqtt=internalMqtt,
+#   internalMqtt=internalMqtt,
-expandedConfig=expandedConfig,
+#   masterGraph=masterGraph,
-masterGraph=masterGraph,
 debugPageData=debugPageData,
 debug=arg['-v']),
 interface='::')
 log.info('serving on %s', port)

Mercurial > code > home > repos > homeauto

comparison service/mqtt_to_rdf/mqtt_to_rdf.py @ 1706:2085ed9cfcc4